# Copyright 2024 The HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.


import inspect
import json
import os
import random
import shutil
import sys
import tempfile
from pathlib import Path
from unittest.mock import patch

import numpy as np
from huggingface_hub import hf_hub_download
from parameterized import parameterized

from transformers import ProcessorMixin
from transformers.processing_utils import MODALITY_TO_AUTOPROCESSOR_MAPPING
from transformers.testing_utils import (
    check_json_file_has_correct_format,
    require_librosa,
    require_torch,
    require_torchcodec,
    require_vision,
)
from transformers.utils import is_torch_available, is_vision_available
from transformers.video_utils import get_video_size


parent_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
sys.path.append(os.path.join(parent_dir, "utils"))
from fetch_hub_objects_for_ci import url_to_local_path  # noqa: E402


global_rng = random.Random()

if is_vision_available():
    from PIL import Image

if is_torch_available():
    import torch

MODALITY_INPUT_DATA = {
    "images": [
        "https://huggingface.co/datasets/hf-internal-testing/fixtures_image_utils/resolve/main/coco_sample.png",
        "https://huggingface.co/datasets/hf-internal-testing/fixtures_image_utils/resolve/main/coco_sample.png",
    ],
    "videos": [
        "https://huggingface.co/datasets/hf-internal-testing/test-videos/resolve/main/big_buck_bunny_320x240_10s.mp4",
        "https://huggingface.co/datasets/hf-internal-testing/test-videos/resolve/main/sample_demo_1_320x240.mp4",
    ],
    "audio": [
        "https://huggingface.co/datasets/hf-internal-testing/dummy-audio-samples/resolve/main/glass-breaking-151256.mp3",
        "https://huggingface.co/datasets/hf-internal-testing/dummy-audio-samples/resolve/main/f2641_0_throatclearing.wav",
    ],
}

for modality, urls in MODALITY_INPUT_DATA.items():
    MODALITY_INPUT_DATA[modality] = [url_to_local_path(url) for url in urls]


MODALITY_TEST_SPECS = {
    "text": {
        "component_key": "tokenizer",
        "call_time_kwargs": {"return_tensors": "pt"},
        "init_time_kwargs": {},
    },
    "images": {
        "component_key": "image_processor",
        "call_time_kwargs": {"return_tensors": "pt"},
        "init_time_kwargs": {"do_rescale": True, "rescale_factor": -1.0},
    },
    "videos": {
        "component_key": "video_processor",
        # `do_sample_frames=False` is required by video processors so that
        # frame sampling doesn't change the number of resulting tokens or raise errors
        "call_time_kwargs": {"do_sample_frames": False, "return_tensors": "pt"},
        "init_time_kwargs": {
            "do_rescale": True,
            "rescale_factor": -1.0,
        },
    },
    "audio": {
        # Either a raw feature_extractor or an audio_processor attribute
        # is acceptable, whichever one the processor exposes.
        "component_key": None,
        "call_time_kwargs": {"return_tensors": "pt"},
        "init_time_kwargs": {},
    },
}


def prepare_images_inputs():
    """This function prepares a list of PIL images"""
    image_inputs = [np.random.randint(255, size=(3, 30, 400), dtype=np.uint8)]
    image_inputs = [Image.fromarray(np.moveaxis(x, 0, -1)) for x in image_inputs]
    return image_inputs


def floats_list(shape, scale=1.0, rng=None, name=None):
    """Creates a random float32 tensor"""
    if rng is None:
        rng = global_rng

    values = []
    for batch_idx in range(shape[0]):
        values.append([])
        for _ in range(shape[1]):
            values[-1].append(rng.random() * scale)

    return values


@require_torch
@require_vision
class ProcessorTesterMixin:
    processor_class = None
    # Optional: set this to a real Hub repo containing a complete set of processor files
    # (tokenizer, image processor, etc.) so all components can be loaded via from_pretrained.
    model_id = None
    # Optional: set this to a Hub repo containing a complete set of processor files where some
    # components represent a tiny version (e.g. a tokenizer with a trimmed vocab) for
    # memory-sensitive tests. Must be a real Hub repo with all components loadable via from_pretrained.
    tiny_model_id = None
    text_input_name = "input_ids"
    images_input_name = "pixel_values"
    videos_input_name = "pixel_values_videos"
    audio_input_name_values = "input_values"  # raw/normalized audio
    audio_input_name = "input_features"  # computed features, e.g. Mel spectrogram, STFT

    # Max-length values used in image-text kwargs tests. Override in subclasses if needed.
    images_text_kwargs_max_length = 117
    images_text_kwargs_override_max_length = 112
    images_unstructured_max_length = 76

    # Max-length values used in audio-text kwargs tests. Override in subclasses if needed.
    audio_text_kwargs_max_length = 300
    audio_processor_tester_max_length = 117
    audio_unstructured_max_length = 76

    # Max-length values used in video-text kwargs tests. Override in subclasses if needed.
    videos_text_kwargs_max_length = 167
    videos_text_kwargs_override_max_length = 162
    videos_unstructured_max_length = 176

    # Max-length value used in chat template tests. Override in subclasses if needed.
    chat_template_max_length = 100  # max_length in test_apply_chat_template_*

    # Role used in chat template tests. Override in subclasses whose template expects another role.
    chat_template_user_role = "user"

    @classmethod
    def setUpClass(cls):
        """
        Automatically set up the processor test by creating and saving all required components.
        Individual test classes only need to set processor_class and optionally:
        - model_id: to load components from a specific pretrained model
        - prepare_processor_dict(): to provide custom kwargs for processor initialization
        """
        if cls.processor_class is None:
            raise ValueError(
                f"{cls.__name__} must define 'processor_class' attribute. Example: processor_class = MyProcessor"
            )

        cls.tmpdirname = tempfile.mkdtemp()
        cls.full_tmpdirname = None

        if cls.tiny_model_id is not None:
            # tiny_model_id is set: tmpdirname holds the lightweight processor (used by all tests),
            # full_tmpdirname holds the full processor (used only by tests that call get_processor(use_tiny_ckpt=False)).
            tiny_processor = cls._setup_from_pretrained(cls.tiny_model_id)
            cls._setup_test_attributes(tiny_processor)
            tiny_processor.save_pretrained(cls.tmpdirname)

            # If model_id is specified, load the full processor into full_tmpdirname.
            # If model_id is None, no full processor is needed: full_tmpdirname stays None,
            # and get_processor(use_tiny_ckpt=False) will fall back to tmpdirname (tiny).
            if cls.model_id is not None:
                cls.full_tmpdirname = tempfile.mkdtemp()
                full_processor = cls._setup_from_pretrained(cls.model_id)
                # TODO: make this more robust. We intentionally do NOT call _setup_test_attributes(full_processor)
                # here because it would overwrite the class attributes already set from tiny_processor (e.g.
                # image_token, video_token, audio_token). We assume these special tokens are identical between
                # the tiny and full processor — but this is not guaranteed: if the tiny tokenizer is built
                # differently (e.g. missing special tokens or using different token strings), cls.image_token
                # etc. will silently reflect the wrong values for tests that use the full processor.
                full_processor.save_pretrained(cls.full_tmpdirname)
        else:
            # No tiny_model_id: tmpdirname holds the only processor.
            # If model_id is specified, load components from that model
            if cls.model_id is not None:
                processor = cls._setup_from_pretrained(cls.model_id)
            else:
                # Otherwise, create generic components
                processor = cls._setup_from_components()
            # setup test attributes
            cls._setup_test_attributes(processor)
            processor.save_pretrained(cls.tmpdirname)

    @classmethod
    def _setup_test_attributes(cls, processor):
        # can be overriden in the child class to define more class attributes
        for token_attr in ("image_token", "video_token", "audio_token"):
            token = getattr(processor, token_attr, None)
            if token is not None:
                setattr(cls, token_attr, token)

    @classmethod
    def _setup_from_pretrained(cls, model_id, **kwargs):
        """Load all components from model_id to build the processor.

        If any component is provided via a _setup_<attribute>() hook, all remaining components
        are loaded individually from model_id so that processor_class.__init__ receives a complete
        set of components (all must be passed together when any one is customized).
        """
        # check if there are any custom components to setup
        custom_components = {}
        for attribute in cls.processor_class.get_attributes():
            if hasattr(cls, f"_setup_{attribute}"):
                custom_method = getattr(cls, f"_setup_{attribute}")
                custom_components[attribute] = custom_method()

        # if there is one custom component, we need to add all the other ones (with from_pretrained)
        if custom_components:
            for attribute in cls.processor_class.get_attributes():
                if attribute not in custom_components:
                    component_class = cls._get_component_class_from_processor(attribute)
                    custom_components[attribute] = component_class.from_pretrained(model_id)

        kwargs.update(cls.prepare_processor_dict())
        processor = cls.processor_class.from_pretrained(model_id, **custom_components, **kwargs)
        return processor

    @classmethod
    def _setup_from_components(cls):
        """Create all required components for the processor and save the complete processor."""
        # Get all required attributes for this processor
        attributes = cls.processor_class.get_attributes()

        # Create each component (but don't save them individually)
        components = {}
        for attribute in attributes:
            components[attribute] = cls._setup_component(attribute)

        processor_kwargs = cls.prepare_processor_dict()
        processor = cls.processor_class(**components, **processor_kwargs)
        return processor

    @classmethod
    def _setup_component(cls, attribute):
        """
        Create and return a component.

        This method first checks for a custom setup method (_setup_{attribute}).
        If not found, it tries to get the component class from the processor's Auto mappings
        and instantiate it without arguments.
        If that fails, it raises an error telling the user to override the setup method.

        Individual test classes should override _setup_{attribute}() for custom component setup.
        Custom methods should return the created component.

        Returns:
            The created component instance.
        """
        # Check if there's a custom setup method for this specific attribute
        custom_method = getattr(cls, f"_setup_{attribute}", None)
        if custom_method is not None:
            return custom_method()

        # Get the component class from processor's Auto mappings
        component_class = cls._get_component_class_from_processor(attribute)

        # Get the base class name for the component to provide helpful error messages
        component_type = attribute.replace("_", " ")

        # Try to instantiate the component without arguments
        try:
            component = component_class()
        except Exception as e:
            raise TypeError(
                f"Failed to instantiate {component_type} ({component_class}) without arguments.\n"
                f"Error: {e}\n\n"
                f"To fix this, override the setup method in your test class:\n\n"
                f"    @classmethod\n"
                f"    def _setup_{attribute}(cls):\n"
                f"        # Create your custom {component_type}\n"
                f"        from transformers import {component_class}\n"
                f"        component = {component_class}(...)\n"
                f"        return component\n"
            ) from e

        return component

    @classmethod
    def _get_component_class_from_processor(cls, attribute, use_fast: bool = True):
        """
        Get the component class for a given attribute from the processor's Auto mappings.

        This extracts the model type from the test file name and uses that to look up
        the config class, which is then used to find the appropriate component class.
        """
        import inspect
        import re

        from transformers.models.auto.configuration_auto import (
            CONFIG_MAPPING_NAMES,
            SPECIAL_MODEL_TYPE_TO_MODULE_NAME,
        )

        # Get the component class from the appropriate Auto mapping
        if attribute in MODALITY_TO_AUTOPROCESSOR_MAPPING:
            mapping_name = attribute
        elif "tokenizer" in attribute:
            mapping_name = "tokenizer"
        else:
            raise ValueError(
                f"Unknown attribute type: '{attribute}'. "
                f"Please override _setup_{attribute}() in your test class to provide custom setup."
            )

        # Extract model_type from the test file name
        # Test files are named like test_processing_align.py or test_processor_align.py
        test_file = inspect.getfile(cls)
        match = re.search(r"test_process(?:ing|or)_(\w+)\.py$", test_file)
        if not match:
            raise ValueError(
                f"Could not extract model type from test file name: {test_file}. "
                f"Please override _setup_{attribute}() in your test class."
            )

        model_type = match.group(1)

        if model_type not in CONFIG_MAPPING_NAMES:
            # check if the model type is a special model type
            for special_model_type, special_module_name in SPECIAL_MODEL_TYPE_TO_MODULE_NAME.items():
                if model_type != special_module_name or special_model_type not in CONFIG_MAPPING_NAMES:
                    continue

                component_class = cls.resolve_model_type_to_attribute(special_model_type, mapping_name)
                if component_class is not None:
                    break
        else:
            component_class = cls.resolve_model_type_to_attribute(model_type, mapping_name)

        if component_class is None:
            raise ValueError(
                f"Could not find {mapping_name} class for model {match.group(1)}. "
                f"Please override _setup_{attribute}() in your test class."
            )

        # Handle tuple case (some mappings return tuples of classes)
        if isinstance(component_class, tuple):
            if use_fast:
                component_class = component_class[-1] if component_class[-1] is not None else component_class[0]
            else:
                component_class = component_class[0] if component_class[0] is not None else component_class[1]
        elif isinstance(component_class, dict):
            if not use_fast:
                component_class = component_class["pil"]
            else:
                component_class = (
                    component_class["torchvision"] if "torchvision" in component_class else component_class["pil"]
                )
        return component_class

    @staticmethod
    def resolve_model_type_to_attribute(model_type, mapping_name):
        from transformers.models.auto.configuration_auto import CONFIG_MAPPING

        config_class = CONFIG_MAPPING[model_type]

        # Get the appropriate Auto mapping for this component type
        if mapping_name == "tokenizer":
            from transformers.models.auto.tokenization_auto import TOKENIZER_MAPPING
            from transformers.utils import is_tokenizers_available

            component_class = TOKENIZER_MAPPING.get(config_class, None)
            if component_class is None and is_tokenizers_available():
                from transformers.tokenization_utils_tokenizers import TokenizersBackend

                component_class = TokenizersBackend
        elif mapping_name == "image_processor":
            from transformers.models.auto.image_processing_auto import IMAGE_PROCESSOR_MAPPING

            component_class = IMAGE_PROCESSOR_MAPPING.get(config_class, None)
        elif mapping_name == "feature_extractor" or mapping_name == "audio_processor":
            from transformers.models.auto.feature_extraction_auto import FEATURE_EXTRACTOR_MAPPING

            component_class = FEATURE_EXTRACTOR_MAPPING.get(config_class, None)
        elif mapping_name == "video_processor":
            from transformers.models.auto.video_processing_auto import VIDEO_PROCESSOR_MAPPING

            component_class = VIDEO_PROCESSOR_MAPPING.get(config_class, None)
        else:
            raise ValueError(f"Unknown mapping for attribute: {mapping_name}")
        return component_class

    @classmethod
    def tearDownClass(cls):
        """Clean up the temporary directory."""
        if hasattr(cls, "tmpdirname"):
            shutil.rmtree(cls.tmpdirname, ignore_errors=True)
        if hasattr(cls, "full_tmpdirname") and cls.full_tmpdirname is not None:
            shutil.rmtree(cls.full_tmpdirname, ignore_errors=True)

    @staticmethod
    def prepare_processor_dict():
        """Override this method to provide custom kwargs for processor initialization."""
        return {}

    def get_component(self, attribute, use_tiny_ckpt=True, **kwargs):
        # use_tiny_ckpt only has effect when tiny_model_id is set. In that case, tmpdirname holds the
        # lightweight processor and full_tmpdirname holds the full one. If tiny_model_id is not set,
        # tmpdirname already contains the full processor loaded from cls.model_id, and calling this
        # function with use_tiny_ckpt=True still returns a full processor.
        dirpath = self.tmpdirname if (use_tiny_ckpt or self.full_tmpdirname is None) else self.full_tmpdirname
        if attribute not in MODALITY_TO_AUTOPROCESSOR_MAPPING and "tokenizer" in attribute:
            auto_processor_class = MODALITY_TO_AUTOPROCESSOR_MAPPING["tokenizer"]
            component = auto_processor_class.from_pretrained(dirpath, subfolder=attribute, **kwargs)  # noqa
        else:
            auto_processor_class = MODALITY_TO_AUTOPROCESSOR_MAPPING[attribute]
            component = auto_processor_class.from_pretrained(dirpath, **kwargs)  # noqa
        if "tokenizer" in attribute and not component.pad_token:
            component.pad_token = "[TEST_PAD]"
            if component.pad_token_id is None:
                component.pad_token_id = 0

        return component

    def prepare_components(self, **kwargs):
        components = {}
        for attribute in self.processor_class.get_attributes():
            component = self.get_component(attribute)
            components[attribute] = component

        return components

    def get_processor(self, use_tiny_ckpt=True):
        # use_tiny_ckpt only has effect when tiny_model_id is set. In that case, tmpdirname holds the
        # lightweight processor and full_tmpdirname holds the full one. If tiny_model_id is not set,
        # tmpdirname already contains the full processor loaded from cls.model_id, and calling this
        # function with use_tiny_ckpt=True still returns a full processor.
        if not use_tiny_ckpt and self.full_tmpdirname is not None:
            return self.processor_class.from_pretrained(self.full_tmpdirname)
        return self.processor_class.from_pretrained(self.tmpdirname)

    def prepare_text_inputs(self, batch_size: int | None = None, modalities: str | list | None = None):
        if isinstance(modalities, str):
            modalities = [modalities]

        special_token_to_add = ""
        if modalities is not None:
            for modality in modalities:
                # we have non-uniform naming conventions for image/videos
                if modality == "images":
                    modality = "image"
                elif modality == "videos":
                    modality = "video"
                special_token_to_add += getattr(self, f"{modality}_token", "")

        if batch_size is None:
            return f"lower newer {special_token_to_add}"

        if batch_size < 1:
            raise ValueError("batch_size must be greater than 0")

        if batch_size == 1:
            return [f"lower newer {special_token_to_add}"]
        return [f"lower newer {special_token_to_add}", f" {special_token_to_add} upper older longer string"] + [
            f"lower newer {special_token_to_add}"
        ] * (batch_size - 2)

    @require_vision
    def prepare_images_inputs(self, batch_size: int | None = None, nested: bool = False):
        """This function prepares a list of PIL images for testing"""
        if batch_size is None:
            return prepare_images_inputs()[0]
        if batch_size < 1:
            raise ValueError("batch_size must be greater than 0")
        if nested:
            return [prepare_images_inputs()] * batch_size
        return prepare_images_inputs() * batch_size

    @require_vision
    def prepare_videos_inputs(self, batch_size: int | None = None):
        """This function prepares a list of numpy videos."""
        video_input = [np.random.randint(255, size=(3, 30, 400), dtype=np.uint8)] * 8
        video_input = np.array(video_input)
        if batch_size is None:
            return video_input
        return [video_input] * batch_size

    def prepare_audio_inputs(self, batch_size: int | None = None):
        """This function prepares a list of numpy audio."""
        raw_speech = floats_list((1, 1000))
        raw_speech = [np.asarray(audio) for audio in raw_speech]
        if batch_size is None:
            return raw_speech
        return raw_speech * batch_size

    # Video sampling inputs and expected sampled frame length. Override for models with custom sampling
    # All tests use the same pre-loaded tiny video thus the expectations should be deterministic
    @property
    def video_sampling_expectations(self):
        return [
            {"num_frames": 3, "fps": None, "expected_dim": 1, "output_length": 3},
            {"num_frames": None, "fps": 10, "expected_dim": 1, "output_length": 3},
            {"do_sample_frames": False, "fps": 10, "expected_dim": 1, "output_length": 11},
            {"do_sample_frames": False, "expected_dim": 1, "output_length": 11},
            {"expected_dim": 1, "output_length": 11},
        ]

    def test_processor_to_json_string(self):
        processor = self.get_processor()
        obj = json.loads(processor.to_json_string())
        for key, value in self.prepare_processor_dict().items():
            # Chat template is saved as a separate file
            if key not in "chat_template":
                # json converts dict keys to str, but some processors force convert back to int when init
                if (
                    isinstance(obj[key], dict)
                    and isinstance(list(obj[key].keys())[0], str)
                    and isinstance(list(value.keys())[0], int)
                ):
                    obj[key] = {int(k): v for k, v in obj[key].items()}
                self.assertEqual(obj[key], value)
                self.assertEqual(getattr(processor, key, None), value)

    def test_processor_from_and_save_pretrained(self):
        processor_first = self.get_processor()

        with tempfile.TemporaryDirectory() as tmpdirname:
            saved_files = processor_first.save_pretrained(tmpdirname)
            if len(saved_files) > 0:
                check_json_file_has_correct_format(saved_files[0])
                processor_second = self.processor_class.from_pretrained(tmpdirname)

                self.assertEqual(processor_second.to_dict(), processor_first.to_dict())

                for attribute in processor_first.get_attributes():
                    attribute_first = getattr(processor_first, attribute)
                    attribute_second = getattr(processor_second, attribute)

                    # tokenizer repr contains model-path from where we loaded
                    if "tokenizer" not in attribute:
                        # We don't store/load `_processor_class` for subprocessors.
                        # The `_processor_class` is saved once per config, at general level
                        self.assertFalse(hasattr(attribute_second, "_processor_class"))
                        self.assertFalse(hasattr(attribute_first, "_processor_class"))

                        self.assertFalse(hasattr(attribute_second, "processor_class"))
                        self.assertFalse(hasattr(attribute_first, "processor_class"))

                        self.assertEqual(repr(attribute_first), repr(attribute_second))

    def test_processor_from_and_save_pretrained_as_nested_dict(self):
        processor_first = self.get_processor()

        with tempfile.TemporaryDirectory() as tmpdirname:
            saved_files = processor_first.save_pretrained(tmpdirname)
            check_json_file_has_correct_format(saved_files[0])

            # Load it back and check if loaded correctly
            processor_second = self.processor_class.from_pretrained(tmpdirname)
            self.assertEqual(processor_second.to_dict(), processor_first.to_dict())

            # Try to load each attribute separately from saved directory
            for attribute in processor_first.get_attributes():
                if attribute not in MODALITY_TO_AUTOPROCESSOR_MAPPING and "tokenizer" in attribute:
                    auto_processor_class = MODALITY_TO_AUTOPROCESSOR_MAPPING["tokenizer"]
                    attribute_reloaded = auto_processor_class.from_pretrained(tmpdirname, subfolder=attribute)
                else:
                    auto_processor_class = MODALITY_TO_AUTOPROCESSOR_MAPPING[attribute]
                    attribute_reloaded = auto_processor_class.from_pretrained(tmpdirname)
                attribute_first = getattr(processor_first, attribute)

                # tokenizer repr contains model-path from where we loaded
                if "tokenizer" not in attribute:
                    self.assertEqual(repr(attribute_first), repr(attribute_reloaded))

    def test_save_load_pretrained_additional_features(self):
        """
        Tests that additional kwargs passed to from_pretrained are correctly applied to components.
        """
        attributes = self.processor_class.get_attributes()

        if not any(
            attr in ["tokenizer", "image_processor", "feature_extractor", "audio_processor", "video_processor"]
            for attr in attributes
        ):
            self.skipTest("Processor has no tokenizer or image_processor to test additional features")
        additional_kwargs = {}

        has_tokenizer = "tokenizer" in attributes
        if has_tokenizer:
            additional_kwargs["cls_token"] = "(CLS)"
            additional_kwargs["sep_token"] = "(SEP)"

        has_image_processor = "image_processor" in attributes
        if has_image_processor:
            additional_kwargs["do_normalize"] = False
        has_video_processor = "video_processor" in attributes
        if has_video_processor:
            additional_kwargs["do_normalize"] = False

        processor_second = self.processor_class.from_pretrained(self.tmpdirname, **additional_kwargs)
        if has_tokenizer:
            self.assertEqual(processor_second.tokenizer.cls_token, "(CLS)")
            self.assertEqual(processor_second.tokenizer.sep_token, "(SEP)")
        if has_image_processor:
            self.assertEqual(processor_second.image_processor.do_normalize, False)
        if has_video_processor:
            self.assertEqual(processor_second.video_processor.do_normalize, False)

    def test_processor_from_pretrained_vs_from_components(self):
        """
        Tests that loading a processor fully with from_pretrained produces the same result as
        loading each component individually with from_pretrained and building the processor from them.
        """
        # Load processor fully with from_pretrained
        processor_full = self.get_processor()

        # Load each component individually with from_pretrained
        components = {}
        for attribute in self.processor_class.get_attributes():
            components[attribute] = self.get_component(attribute)

        # Build processor from components + prepare_processor_dict() kwargs
        processor_kwargs = self.prepare_processor_dict()
        processor_from_components = self.processor_class(**components, **processor_kwargs)

        self.assertEqual(processor_from_components.to_dict(), processor_full.to_dict())

    def test_model_input_names(self):
        processor = self.get_processor()

        text = self.prepare_text_inputs(modalities=["image", "video", "audio"])
        image_input = self.prepare_images_inputs()
        video_inputs = self.prepare_videos_inputs()
        audio_inputs = self.prepare_audio_inputs()
        inputs_dict = {"text": text, "images": image_input, "videos": video_inputs, "audio": audio_inputs}

        call_signature = inspect.signature(processor.__call__)
        input_args = [param.name for param in call_signature.parameters.values()]
        inputs_dict = {k: v for k, v in inputs_dict.items() if k in input_args}
        # Shouldn't sample when input is a decoded video without metadata (fpx/duration/etc.)
        if "videos" in inputs_dict:
            inputs_dict["do_sample_frames"] = False

        inputs = processor(**inputs_dict, return_tensors="pt")

        self.assertSetEqual(set(inputs.keys()), set(processor.model_input_names))

    @parameterized.expand(
        [
            ("text",),
            ("images",),
            ("videos",),
            ("audio",),
        ]
    )
    def test_subprocessor_defaults(self, modality):
        """
        Tests that sub-processor is called correctly when passing each modality input to the processor.
        This test verifies that processor(single_modality_data) produces the same output as subprocessor(single_modality_data).
        """
        # Skip if processor doesn't have image_processor
        parameterized_config = MODALITY_TEST_SPECS[modality]
        attributes = self.processor_class.get_attributes()
        component_key = self.get_subprocessor_name(modality, attributes)

        if component_key not in attributes:
            self.skipTest(f"{component_key} attribute not present in {self.processor_class}")

        subprocessor = self.get_component(component_key)

        # Get all other required components for processor
        components = {}
        for attribute in self.processor_class.get_attributes():
            components[attribute] = self.get_component(attribute)

        processor = self.processor_class(**components, **self.prepare_processor_dict())
        modality_input = self._prepare_modality_input(modality)

        # merge processor defaults when calling a subprocessor
        kwargs = parameterized_config["call_time_kwargs"]
        merged_kwargs = processor._merge_kwargs(
            processor.valid_processor_kwargs,
            tokenizer_init_kwargs=processor.tokenizer.init_kwargs if hasattr(processor, "tokenizer") else {},
            **kwargs,
        )
        kwargs = merged_kwargs[f"{modality}_kwargs"]

        input_subproc = subprocessor(modality_input, **kwargs)
        try:
            input_processor = processor(**{modality: modality_input, **kwargs})
        except Exception:
            input_processor = {}

        # Verify outputs match
        for key in input_subproc:
            if input_processor and key in processor.model_input_names:
                torch.testing.assert_close(input_subproc[key], input_processor[key])

    def test_tokenizer_decode_defaults(self):
        """
        Tests that processor.batch_decode() correctly forwards to tokenizer.batch_decode().
        """
        # Skip if processor doesn't have tokenizer
        if "tokenizer" not in self.processor_class.get_attributes():
            self.skipTest(f"tokenizer attribute not present in {self.processor_class}")

        # Get all required components for processor
        components = {}
        for attribute in self.processor_class.get_attributes():
            components[attribute] = self.get_component(attribute)

        processor = self.processor_class(**components)
        tokenizer = components["tokenizer"]

        predicted_ids = [[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

        # Test batch_decode
        decoded_processor = processor.batch_decode(predicted_ids)
        decoded_tok = tokenizer.batch_decode(predicted_ids)

        self.assertListEqual(decoded_tok, decoded_processor)

    def test_processor_with_multiple_inputs(self):
        """
        Tests that processor correctly handles multiple modality inputs together.
        Verifies that the output contains expected keys and raises error when no input is provided.
        """
        # Skip if processor doesn't have multiple attributes (not multimodal)
        attributes = self.processor_class.get_attributes()
        if len(attributes) <= 1:
            self.skipTest(f"Processor only has {len(attributes)} attribute(s), test requires multimodal processor")

        processor = self.get_processor()

        # Prepare inputs dynamically based on processor attributes
        processor_inputs = {}
        expected_output_keys = []
        processing_kwargs = {}

        for modality, metadata in MODALITY_TEST_SPECS.items():
            attribute = self.get_subprocessor_name(modality, attributes)
            if attribute in attributes:
                processing_kwargs.update(metadata["call_time_kwargs"])
                prepare_method = getattr(self, f"prepare_{modality}_inputs")
                if modality == "text":
                    processor_inputs[modality] = prepare_method(modalities=["image", "video", "audio"])
                else:
                    processor_inputs[modality] = prepare_method()
                # Track expected output keys
                expected_output_keys.append(getattr(self, f"{modality}_input_name"))

        # Test combined processing
        inputs = processor(**processor_inputs, **processing_kwargs)

        # Verify output contains all expected keys
        for key in expected_output_keys:
            if key == self.audio_input_name:
                self.assertTrue(
                    self.audio_input_name_values in inputs or self.audio_input_name in inputs,
                    f"Expected either '{self.audio_input_name_values}' or '{self.audio_input_name}' in inputs",
                )
            else:
                self.assertIn(key, inputs)

        # Test that it raises error when no input is passed
        with self.assertRaises((TypeError, ValueError)):
            processor()

    def test_processor_text_has_no_visual(self):
        """
        Tests that multimodal models can process batch of inputs where samples can
        be with images/videos or without. See https://github.com/huggingface/transformers/issues/40263
        """
        processor = self.get_processor()
        call_signature = inspect.signature(processor.__call__)
        input_args = [param.name for param in call_signature.parameters.values() if param.annotation != param.empty]

        if not (
            "text" in input_args
            and "images" in input_args
            and hasattr(processor, "image_processor")
            and hasattr(processor, "tokenizer")
        ):
            self.skipTest(f"{self.processor_class} doesn't support images with text.")

        # Prepare inputs and filter by input signature. Make sure to use a high batch size, we'll set some
        # samples to text-only later
        image_inputs = self.prepare_images_inputs(batch_size=3)
        image_inputs_nested = [[image] if not isinstance(image, list) else image for image in image_inputs]
        inputs_dict_nested = {"images": image_inputs_nested}
        modalities = ["image"]

        processing_kwargs = {"return_tensors": "pt", "padding": True}
        # Shouldn't sample when input is a decoded video without metadata (fpx/duration/etc.)
        if "videos" in input_args and hasattr(processor, "video_processor"):
            video_inputs = self.prepare_videos_inputs(batch_size=3)
            inputs_dict_nested["videos"] = [[video] for video in video_inputs]
            modalities.append("video")
            processing_kwargs["do_sample_frames"] = False

        # First call processor with all inputs and use nested input type, which is the format supported by all multimodal processors
        text = self.prepare_text_inputs(batch_size=3, modalities=modalities)
        inputs_dict_nested["text"] = text
        inputs = processor(**inputs_dict_nested, **processing_kwargs)
        self.assertTrue(self.text_input_name in inputs)

        # Now call with one of the samples with no associated vision input. Let's set the first input to be a plain text
        # with no placeholder tokens and no images/videos. The final format would be `images = [[], [image2], [image3]]`
        plain_text = "lower newer"
        inputs_dict_no_vision = {key: [[]] + value[1:] for key, value in inputs_dict_nested.items() if key != "text"}
        inputs_dict_no_vision["text"] = [plain_text] + text[1:]
        inputs_nested = processor(**inputs_dict_no_vision, **processing_kwargs)

        # Check that text samples are same and are expanded with placeholder tokens correctly. First sample
        # has no vision input associated, so we skip it and check it has no vision
        self.assertListEqual(
            inputs[self.text_input_name][1:].tolist(), inputs_nested[self.text_input_name][1:].tolist()
        )

        # Now test if we can apply chat templates with no vision inputs in one of the samples
        # NOTE: we don't skip the test as we want the above to be checked even if process has to chat template
        if processor.chat_template is not None:
            messages = [
                [
                    {
                        "role": "user",
                        "content": [
                            {"type": "text", "text": "What is the capital of France?"},
                        ],
                    },
                ],
                [
                    {
                        "role": "user",
                        "content": [
                            {"type": "text", "text": "What is the capital of France?"},
                            {
                                "type": "image",
                                "url": url_to_local_path(
                                    "https://huggingface.co/datasets/hf-internal-testing/fixtures_image_utils/resolve/main/coco_sample.png"
                                ),
                            },
                        ],
                    },
                ],
            ]

            inputs_chat_template = processor.apply_chat_template(
                messages,
                add_generation_prompt=False,
                tokenize=True,
                return_dict=True,
                return_tensors="pt",
                processor_kwargs={"padding": True},
            )
            self.assertTrue(self.text_input_name in inputs_chat_template)

    # These kwargs-related tests ensure that processors are correctly instantiated.
    # they need to be applied only if an image_processor exists.

    def maybe_skip_typed_test_for_modality(self, modality: str, attributes: list):
        # we have only a few processors that are old and dont follow TypedDict format
        if any(
            name in self.processor_class.__name__.lower()
            for name in ["instructblipvideo", "mgpstr", "sam3", "layoutlmv", "layoutxlm"]
        ):
            self.skipTest(f"{self.processor_class} doesn't have typed kwargs.")

        if "tokenizer" not in attributes:
            self.skipTest(f"tokenizer attribute not present in {self.processor_class}")

        component_key = self.get_subprocessor_name(modality, attributes)
        if component_key not in attributes:
            self.skipTest(f"{component_key} attribute not present in {self.processor_class}")

    def get_subprocessor_name(self, modality: str, attributes: list):
        if modality == "audio":
            if "audio_processor" in attributes:
                component_key = "audio_processor"
            else:
                component_key = "feature_extractor"
        else:
            component_key = MODALITY_TEST_SPECS[modality]["component_key"]
        return component_key

    def _prepare_modality_input(self, modality: str, batch_size: int | None = None):
        # to be overriden by special models if needed
        prepare_fn = getattr(self, f"prepare_{modality}_inputs")
        return prepare_fn(batch_size=batch_size) if batch_size is not None else prepare_fn()

    def _check_modality_outputs(self, inputs: dict, modality: str):
        # to be overriden by special models if needed
        input_key = getattr(self, f"{modality}_input_name")
        # FIXME eric, eustache - audio has no single kwarg that works same way in all models
        if modality in ["images", "videos"]:
            self.assertLessEqual(inputs[input_key][0][0].mean(), 0)

    def _test_modality_processor_defaults_preserved_by_modality_kwargs(self, modality):
        attributes = self.processor_class.get_attributes()
        self.maybe_skip_typed_test_for_modality(modality, attributes)

        component_key = self.get_subprocessor_name(modality, attributes)
        processor_components = self.prepare_components()

        init_time_kwargs = MODALITY_TEST_SPECS[modality]["init_time_kwargs"]
        processor_components[component_key] = self.get_component(component_key, **init_time_kwargs)

        max_length = getattr(self, f"{modality}_text_kwargs_max_length")
        processor_components["tokenizer"] = self.get_component(
            "tokenizer", max_length=max_length, padding="max_length"
        )

        processor_kwargs = self.prepare_processor_dict()
        processor = self.processor_class(**processor_components, **processor_kwargs)

        input_str = self.prepare_text_inputs(modalities=modality)
        modal_input = self._prepare_modality_input(modality)

        call_kwargs = MODALITY_TEST_SPECS[modality]["call_time_kwargs"]
        inputs = processor(text=input_str, **{modality: modal_input, **call_kwargs})

        self._check_modality_outputs(inputs, modality)
        self.assertEqual(len(inputs[self.text_input_name][0]), max_length)

    def _test_kwargs_overrides_default_modality_processor_kwargs(self, modality):
        attributes = self.processor_class.get_attributes()
        component_key = self.get_subprocessor_name(modality, attributes)
        self.maybe_skip_typed_test_for_modality(modality, attributes)

        processor_components = self.prepare_components()
        processor_components[component_key] = self.get_component(component_key)
        processor_components["tokenizer"] = self.get_component("tokenizer", padding=False)
        processor_kwargs = self.prepare_processor_dict()
        processor = self.processor_class(**processor_components, **processor_kwargs)

        init_time_kwargs = MODALITY_TEST_SPECS[modality]["init_time_kwargs"]
        max_length = getattr(self, f"{modality}_text_kwargs_max_length")
        input_str = self.prepare_text_inputs(modalities=modality)
        modal_input = self._prepare_modality_input(modality)

        call_kwargs = MODALITY_TEST_SPECS[modality]["call_time_kwargs"]
        inputs = processor(
            text=input_str,
            max_length=max_length,
            padding="max_length",
            **{modality: modal_input, **call_kwargs, **init_time_kwargs},
        )

        self._check_modality_outputs(inputs, modality)
        self.assertEqual(len(inputs[self.text_input_name][0]), max_length)

    def _test_unstructured_kwargs(self, modality):
        attributes = self.processor_class.get_attributes()
        processor = self.get_processor()
        self.maybe_skip_typed_test_for_modality(modality, attributes)

        input_str = self.prepare_text_inputs(modalities=modality)
        modal_input = self._prepare_modality_input(modality)
        max_length = getattr(self, f"{modality}_unstructured_max_length")

        init_time_kwargs = MODALITY_TEST_SPECS[modality]["init_time_kwargs"]

        call_kwargs = MODALITY_TEST_SPECS[modality]["call_time_kwargs"]
        inputs = processor(
            text=input_str,
            max_length=max_length,
            padding="max_length",
            **{modality: modal_input, **call_kwargs, **init_time_kwargs},
        )

        self._check_modality_outputs(inputs, modality)
        self.assertEqual(inputs[self.text_input_name].shape[-1], max_length)

    def _test_unstructured_kwargs_batched(self, modality):
        attributes = self.processor_class.get_attributes()
        processor = self.get_processor()
        self.maybe_skip_typed_test_for_modality(modality, attributes)

        input_str = self.prepare_text_inputs(batch_size=2, modalities=modality)
        modal_input = self._prepare_modality_input(modality, batch_size=2)
        max_length = getattr(self, f"{modality}_unstructured_max_length")
        init_time_kwargs = MODALITY_TEST_SPECS[modality]["init_time_kwargs"]

        call_kwargs = MODALITY_TEST_SPECS[modality]["call_time_kwargs"]
        inputs = processor(
            text=input_str,
            max_length=max_length,
            padding="longest",
            **{modality: modal_input, **call_kwargs, **init_time_kwargs},
        )

        self._check_modality_outputs(inputs, modality)
        self.assertTrue(
            len(inputs[self.text_input_name][0]) == len(inputs[self.text_input_name][1])
            and len(inputs[self.text_input_name][1]) < max_length
        )

    def _test_doubly_passed_kwargs(self, modality):
        attributes = self.processor_class.get_attributes()
        processor = self.get_processor()
        self.maybe_skip_typed_test_for_modality(modality, attributes)

        input_str = [self.prepare_text_inputs(modalities=modality)]
        modal_input = self._prepare_modality_input(modality)
        modality_kwargs_key = f"{modality}_kwargs"
        init_time_kwargs = MODALITY_TEST_SPECS[modality]["init_time_kwargs"]
        with self.assertRaises(ValueError):
            processor(
                text=input_str, **{modality: modal_input, modality_kwargs_key: init_time_kwargs, **init_time_kwargs}
            )

    def _test_structured_kwargs_nested_from_dict(self, modality):
        attributes = self.processor_class.get_attributes()
        processor = self.get_processor()
        self.maybe_skip_typed_test_for_modality(modality, attributes)

        input_str = self.prepare_text_inputs(modalities=modality)
        modal_input = self._prepare_modality_input(modality)
        max_length = getattr(self, f"{modality}_unstructured_max_length")
        modality_kwargs_key = f"{modality}_kwargs"
        modality_kwargs = MODALITY_TEST_SPECS[modality]["init_time_kwargs"]
        call_kwargs = MODALITY_TEST_SPECS[modality]["call_time_kwargs"]

        all_kwargs = {
            "common_kwargs": {"return_tensors": "pt"},
            modality_kwargs_key: modality_kwargs,
            "text_kwargs": {"padding": "max_length", "max_length": max_length},
            **call_kwargs,
        }

        inputs = processor(text=input_str, **{modality: modal_input}, **all_kwargs)

        self._check_modality_outputs(inputs, modality)
        self.assertEqual(inputs[self.text_input_name].shape[-1], max_length)

    def _test_overlapping_text_modality_kwargs_handling(self, modality):
        attributes = self.processor_class.get_attributes()
        processor = self.get_processor()
        self.maybe_skip_typed_test_for_modality(modality, attributes)

        input_str = self.prepare_text_inputs(modalities=modality)
        modal_input = self._prepare_modality_input(modality)

        with self.assertRaises(ValueError):
            call_kwargs = MODALITY_TEST_SPECS[modality]["call_time_kwargs"]
            processor(
                text=input_str,
                text_kwargs={"padding": "do_not_pad"},
                padding="max_length",
                **{modality: modal_input, **call_kwargs},
            )

    # ------------------------------------------------------------------
    # Public entrypoints - one per historical test name, each a thin
    # delegate into the shared, modality-parameterized helper above.
    # ------------------------------------------------------------------

    @parameterized.expand(["images", "videos", "audio"])
    def test_subprocessor_defaults_preserved_by_kwargs(self, modality):
        self._test_modality_processor_defaults_preserved_by_modality_kwargs(modality)

    @parameterized.expand(["images", "videos", "audio"])
    def test_kwargs_overrides_default_subprocessor_kwargs(self, modality):
        self._test_kwargs_overrides_default_modality_processor_kwargs(modality)

    @parameterized.expand(["images", "videos", "audio"])
    def test_unstructured_kwargs(self, modality):
        self._test_unstructured_kwargs(modality)

    @parameterized.expand(["images", "videos", "audio"])
    def test_unstructured_kwargs_batched(self, modality):
        self._test_unstructured_kwargs_batched(modality)

    # skip audio - no single shared arg that behaves same way - can't test
    @parameterized.expand(["images", "videos"])
    def test_doubly_passed_kwargs(self, modality):
        self._test_doubly_passed_kwargs(modality)

    @parameterized.expand(["images", "videos", "audio"])
    def test_structured_kwargs_nested_from_dict(self, modality):
        self._test_structured_kwargs_nested_from_dict(modality)

    @parameterized.expand(["images", "videos", "audio"])
    def test_overlapping_text_image_kwargs_handling(self, modality):
        self._test_overlapping_text_modality_kwargs_handling(modality)

    def test_flat_kwarg_applied_when_modality_dict_lacks_it(self):
        # Regression for #46192: a flat top-level kwarg (e.g. return_tensors) was silently
        # dropped when its modality dict (e.g. text_kwargs={...}) was also passed without
        # that key. Companion to test_doubly_passed_kwargs which covers the conflict case.
        processor = self.get_processor()
        text = self.prepare_text_inputs(modalities=["image", "video", "audio"])
        inputs_dict = {
            "text": text,
            "images": self.prepare_images_inputs(),
            "videos": self.prepare_videos_inputs(),
            "audio": self.prepare_audio_inputs(),
        }
        call_signature = inspect.signature(processor.__call__)
        input_args = [param.name for param in call_signature.parameters.values()]
        inputs_dict = {k: v for k, v in inputs_dict.items() if k in input_args}

        # Sampling frames from a numpy video array can raise on some video processors.
        extra_kwargs = {"do_sample_frames": False} if "videos" in inputs_dict else {}

        inputs = processor(**inputs_dict, **extra_kwargs, text_kwargs={}, return_tensors="pt")
        for k, v in inputs.items():
            self.assertIsInstance(v, torch.Tensor, msg=f"{k} should be a torch.Tensor")

    def test_args_overlap_kwargs(self):
        if "image_processor" not in self.processor_class.get_attributes():
            self.skipTest(f"image_processor attribute not present in {self.processor_class}")
        processor_first = self.get_processor()
        image_processor = processor_first.image_processor
        image_processor.is_override = True

        with tempfile.TemporaryDirectory() as tmpdirname:
            processor_first.save_pretrained(tmpdirname)
            processor_second = self.processor_class.from_pretrained(tmpdirname, image_processor=image_processor)
            self.assertTrue(processor_second.image_processor.is_override)

    def test_chat_template_save_loading(self):
        processor = self.processor_class.from_pretrained(self.tmpdirname)
        signature = inspect.signature(processor.__init__)
        if "chat_template" not in {*signature.parameters.keys()}:
            self.skipTest("Processor doesn't accept chat templates at input")

        processor.chat_template = "test template"
        with tempfile.TemporaryDirectory() as tmpdirname:
            processor.save_pretrained(tmpdirname)
            with open(Path(tmpdirname, "chat_template.json"), "w", encoding="utf-8") as fp:
                json.dump({"chat_template": processor.chat_template}, fp)
            os.remove(Path(tmpdirname, "chat_template.jinja"))

            reloaded_processor = self.processor_class.from_pretrained(tmpdirname)
            self.assertEqual(processor.chat_template, reloaded_processor.chat_template)

        with tempfile.TemporaryDirectory() as tmpdirname:
            processor.save_pretrained(tmpdirname)
            self.assertTrue(Path(tmpdirname, "chat_template.jinja").is_file())
            self.assertFalse(Path(tmpdirname, "chat_template.json").is_file())
            self.assertFalse(Path(tmpdirname, "additional_chat_templates").is_dir())
            reloaded_processor = self.processor_class.from_pretrained(tmpdirname)
            self.assertEqual(processor.chat_template, reloaded_processor.chat_template)
            # When we save as single files, tokenizers and processors share a chat template, which means
            # the reloaded tokenizer should get the chat template as well
            self.assertEqual(reloaded_processor.chat_template, reloaded_processor.tokenizer.chat_template)

        with tempfile.TemporaryDirectory() as tmpdirname:
            processor.chat_template = {"default": "a", "secondary": "b"}
            processor.save_pretrained(tmpdirname)
            self.assertTrue(Path(tmpdirname, "chat_template.jinja").is_file())
            self.assertFalse(Path(tmpdirname, "chat_template.json").is_file())
            self.assertTrue(Path(tmpdirname, "additional_chat_templates").is_dir())
            reloaded_processor = self.processor_class.from_pretrained(tmpdirname)
            self.assertEqual(processor.chat_template, reloaded_processor.chat_template)
            # When we save as single files, tokenizers and processors share a chat template, which means
            # the reloaded tokenizer should get the chat template as well
            self.assertEqual(reloaded_processor.chat_template, reloaded_processor.tokenizer.chat_template)

    def test_chat_template_saving_rejects_path_traversal(self):
        # A malicious chat_template dict key must not be usable to escape the save directory and write
        # attacker-controlled content to an arbitrary path (path traversal, CWE-22). The dict key is used
        # verbatim as a `<name>.jinja` filename, so `save_pretrained` must reject names that are not plain
        # filenames instead of silently writing outside the target directory.
        processor = self.processor_class.from_pretrained(self.tmpdirname)
        signature = inspect.signature(processor.__init__)
        if "chat_template" not in {*signature.parameters.keys()}:
            self.skipTest("Processor doesn't accept chat templates at input")

        processor.chat_template = {"default": "a", "../../PWNED": "attacker content"}
        with tempfile.TemporaryDirectory() as tmpdirname:
            save_dir = os.path.join(tmpdirname, "save")
            # Where the "../../PWNED" key would land if traversal succeeded:
            # save/additional_chat_templates/../../PWNED.jinja -> tmpdirname/PWNED.jinja
            canary = Path(tmpdirname, "PWNED.jinja")
            with self.assertRaises(ValueError):
                processor.save_pretrained(save_dir)
            self.assertFalse(canary.exists())

    def _test_apply_chat_template(
        self,
        modality: str,
        batch_size: int,
        return_tensors: str,
        input_name: str,
        processor_name: str,
        input_data: list[str],
    ):
        processor = self.get_processor()
        if processor.chat_template is None:
            self.skipTest("Processor has no chat template")

        if processor_name not in self.processor_class.get_attributes():
            self.skipTest(f"{processor_name} attribute not present in {self.processor_class}")

        # some models have only Fast image processor
        if getattr(processor, processor_name).__class__.__name__.endswith("Fast"):
            return_tensors = "pt"

        batch_messages = [
            [
                {"role": "system", "content": [{"type": "text", "text": "You are a helpful assistant."}]},
                {"role": "user", "content": [{"type": "text", "text": "Describe this."}]},
            ]
        ] * batch_size

        # Test that jinja can be applied
        formatted_prompt = processor.apply_chat_template(batch_messages, add_generation_prompt=True, tokenize=False)
        self.assertEqual(len(formatted_prompt), batch_size)

        # Test that tokenizing with template and directly with `self.tokenizer` gives same output
        formatted_prompt_tokenized = processor.apply_chat_template(
            batch_messages, add_generation_prompt=True, tokenize=True, return_tensors=return_tensors
        )
        add_special_tokens = True
        if processor.tokenizer.bos_token is not None and formatted_prompt[0].startswith(processor.tokenizer.bos_token):
            add_special_tokens = False
        tok_output = processor.tokenizer(
            formatted_prompt, return_tensors=return_tensors, add_special_tokens=add_special_tokens
        )
        expected_output = tok_output.input_ids
        self.assertListEqual(expected_output.tolist(), formatted_prompt_tokenized.tolist())

        # Test that kwargs passed to processor's `__call__` are actually used
        tokenized_prompt_100 = processor.apply_chat_template(
            batch_messages,
            add_generation_prompt=True,
            tokenize=True,
            return_tensors=return_tensors,
            processor_kwargs={
                "padding": "max_length",
                "truncation": True,
                "max_length": self.chat_template_max_length,
            },
        )
        self.assertEqual(len(tokenized_prompt_100[0]), self.chat_template_max_length)

        # Test that `return_dict=True` returns text related inputs in the dict
        out_dict_text = processor.apply_chat_template(
            batch_messages,
            add_generation_prompt=True,
            tokenize=True,
            return_dict=True,
            return_tensors=return_tensors,
        )
        self.assertTrue(all(key in out_dict_text for key in ["input_ids", "attention_mask"]))
        self.assertEqual(len(out_dict_text["input_ids"]), batch_size)
        self.assertEqual(len(out_dict_text["attention_mask"]), batch_size)

        # Test that with modality URLs and `return_dict=True`, we get modality inputs in the dict
        for idx, url in enumerate(input_data[:batch_size]):
            batch_messages[idx][1]["content"] = [batch_messages[idx][1]["content"][0], {"type": modality, "url": url}]

        out_dict = processor.apply_chat_template(
            batch_messages,
            add_generation_prompt=True,
            tokenize=True,
            return_dict=True,
            return_tensors=return_tensors,
            # No more than 2 frames and explicitly disable other ways to sample
            processor_kwargs={"num_frames": 2, "fps": None},
        )
        input_name = getattr(self, input_name)
        self.assertTrue(input_name in out_dict)
        self.assertEqual(len(out_dict["input_ids"]), batch_size)
        self.assertEqual(len(out_dict["attention_mask"]), batch_size)

        # Qwen-style pixels don't scale with bs same way as other models
        # calculate expected video token count based on video_grid_thw
        if (grid_thw := out_dict.get(f"{modality}_grid_thw")) is not None:
            mm_len = sum(thw[0] * thw[1] * thw[2] for thw in grid_thw)
        elif (num_frames_per_video := out_dict.get("num_frames_per_video")) is not None:
            # Models that concatenate the frames of every video along a single axis
            mm_len = sum(num_frames_per_video)
        else:
            mm_len = batch_size

        self.assertEqual(len(out_dict[input_name]), mm_len)

        return_tensor_to_type = {"pt": torch.Tensor, "np": np.ndarray, None: list}
        for k in out_dict:
            self.assertIsInstance(out_dict[k], return_tensor_to_type[return_tensors])

        # Test continue from final message
        assistant_message = {
            "role": "assistant",
            "content": [{"type": "text", "text": "It is the sound of"}],
        }
        for idx, url in enumerate(input_data[:batch_size]):
            batch_messages[idx] = batch_messages[idx] + [assistant_message]
        continue_prompt = processor.apply_chat_template(batch_messages, continue_final_message=True, tokenize=False)
        for prompt in continue_prompt:
            self.assertTrue(prompt.endswith("It is the sound of"))  # no `eos` token at the end

    @require_librosa
    @parameterized.expand([(1, "np"), (1, "pt"), (2, "np"), (2, "pt")])
    def test_apply_chat_template_audio(self, batch_size: int, return_tensors: str):
        if "feature_extractor" in self.processor_class.get_attributes():
            self._test_apply_chat_template(
                "audio",
                batch_size,
                return_tensors,
                "audio_input_name",
                "feature_extractor",
                MODALITY_INPUT_DATA["audio"],
            )
        else:
            self._test_apply_chat_template(
                "audio",
                batch_size,
                return_tensors,
                "audio_input_name",
                "audio_processor",
                MODALITY_INPUT_DATA["audio"],
            )

    @require_librosa
    def test_chat_template_audio_sampling_rate(self):
        """Audio decoded by `apply_chat_template` reaches the processor with the `sampling_rate` it was decoded at,
        so audio processors don't warn about a missing one. Users may pass it flat or nested under `audio_kwargs`,
        and a nested one must be forwarded as-is instead of being duplicated by a flat one."""
        processor = self.get_processor()
        if processor.chat_template is None:
            self.skipTest("Processor has no chat template")

        audio_processor = getattr(processor, "feature_extractor", getattr(processor, "audio_processor", None))
        if audio_processor is None:
            self.skipTest(f"{self.processor_class} has no audio processor")

        sampling_rate = audio_processor.sampling_rate
        messages = [
            {
                "role": self.chat_template_user_role,
                "content": [
                    {"type": "audio", "url": MODALITY_INPUT_DATA["audio"][0]},
                    {"type": "text", "text": "What is happening in this audio?"},
                ],
            },
        ]

        for processor_kwargs in [
            None,
            {"sampling_rate": sampling_rate},
            {"audio_kwargs": {"sampling_rate": sampling_rate}},
            {"sampling_rate": None},
            {"audio_kwargs": {"sampling_rate": None}},
        ]:
            with patch.object(
                type(processor), "__call__", autospec=True, side_effect=type(processor).__call__
            ) as mocked_call:
                out_dict = processor.apply_chat_template(
                    messages,
                    add_generation_prompt=True,
                    tokenize=True,
                    return_dict=True,
                    return_tensors="pt",
                    processor_kwargs=processor_kwargs,
                )

            call_kwargs = mocked_call.call_args.kwargs
            passed_sampling_rate = call_kwargs.get(
                "sampling_rate", call_kwargs.get("audio_kwargs", {}).get("sampling_rate")
            )
            self.assertEqual(passed_sampling_rate, sampling_rate)
            self.assertIn(self.audio_input_name, out_dict)

    @require_torchcodec
    @parameterized.expand([(1, "pt")])
    def test_apply_chat_template_decoded_video(self, batch_size: int, return_tensors: str):
        dummy_preloaded_video = np.array(self.prepare_videos_inputs())
        input_data = [dummy_preloaded_video]
        self._test_apply_chat_template(
            "video", batch_size, return_tensors, "videos_input_name", "video_processor", input_data
        )

    @require_torchcodec
    @parameterized.expand([(1, "pt"), (2, "pt")])  # video processor supports only torchvision
    def test_apply_chat_template_video(self, batch_size: int, return_tensors: str):
        self._test_apply_chat_template(
            "video", batch_size, return_tensors, "videos_input_name", "video_processor", MODALITY_INPUT_DATA["videos"]
        )

    @parameterized.expand([(1, "pt"), (2, "pt")])  # fast image processors supports only torchvision
    def test_apply_chat_template_image(self, batch_size: int, return_tensors: str):
        self._test_apply_chat_template(
            "image", batch_size, return_tensors, "images_input_name", "image_processor", MODALITY_INPUT_DATA["images"]
        )

    def test_apply_chat_template_video_frame_sampling(self):
        processor = self.get_processor()

        if processor.chat_template is None:
            self.skipTest("Processor has no chat template")

        if "video_processor" not in self.processor_class.get_attributes():
            self.skipTest("Processor doesn't accept videos at input")

        messages = [
            [
                {
                    "role": "user",
                    "content": [
                        {
                            "type": "video",
                            "url": url_to_local_path(
                                "https://huggingface.co/datasets/hf-internal-testing/test-videos/resolve/main/tiny_video_320x240.mp4"
                            ),
                        },
                        {"type": "text", "text": "What is shown in this video?"},
                    ],
                },
            ]
        ]

        for processor_kwargs in self.video_sampling_expectations:
            exp_output_length = processor_kwargs.pop("output_length")
            expected_dim = processor_kwargs.pop("expected_dim")

            out_dict_with_video = processor.apply_chat_template(
                messages,
                add_generation_prompt=True,
                tokenize=True,
                return_dict=True,
                return_tensors="pt",
                processor_kwargs=processor_kwargs,
            )
            self.assertTrue(self.videos_input_name in out_dict_with_video)
            self.assertEqual(out_dict_with_video[self.videos_input_name].shape[expected_dim], exp_output_length)

        messages[0][0]["content"][0] = {
            "type": "video",
            "url": [
                url_to_local_path(
                    "https://huggingface.co/datasets/hf-internal-testing/fixtures_image_utils/resolve/main/australia.jpg"
                )
            ]
            * 2,
        }

        # Load video as a list of frames (i.e. images).
        # NOTE: each frame should have same size because we assume they come from one video
        out_dict_with_video = processor.apply_chat_template(
            messages,
            add_generation_prompt=True,
            tokenize=True,
            return_dict=True,
            return_tensors="pt",
            processor_kwargs={"do_sample_frames": False},
        )
        self.assertTrue(self.videos_input_name in out_dict_with_video)

        # When the inputs are frame URLs/paths we expect that those are already
        # sampled and will raise an error is asked to sample again.
        with self.assertRaisesRegex(
            ValueError, "Sampling frames from a list of images is not supported! Set `do_sample_frames=False`"
        ):
            out_dict_with_video = processor.apply_chat_template(
                messages,
                add_generation_prompt=True,
                tokenize=True,
                return_dict=True,
                return_tensors="pt",
                processor_kwargs={"do_sample_frames": True},
            )

    @require_librosa
    @require_torchcodec
    def test_chat_template_audio_from_video(self):
        processor = self.get_processor()
        if processor.chat_template is None:
            self.skipTest("Processor has no chat template")

        signature = inspect.signature(processor.__call__)
        if "videos" not in {*signature.parameters.keys()} or (
            signature.parameters.get("videos") is not None
            and signature.parameters["videos"].annotation == inspect._empty
        ):
            self.skipTest(f"{self.processor_class} does not support video inputs")

        if (
            "feature_extractor" not in self.processor_class.get_attributes()
            or "audio_processor" not in self.processor_class.get_attributes()
        ):
            self.skipTest(f"feature_extractor attribute not present in {self.processor_class}")

        video_file_path = hf_hub_download(
            repo_id="hf-internal-testing/test-videos", filename="sample_demo_1_320x240.mp4", repo_type="dataset"
        )
        messages = [
            {
                "role": "user",
                "content": [
                    {"type": "video", "path": video_file_path},
                    {"type": "text", "text": "Which of these animals is making the sound?"},
                ],
            },
            {
                "role": "assistant",
                "content": [{"type": "text", "text": "It is a cow."}],
            },
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": "Tell me all about this animal."},
                ],
            },
        ]

        formatted_prompt = processor.apply_chat_template([messages], add_generation_prompt=True, tokenize=False)
        self.assertEqual(len(formatted_prompt), 1)  # batch size=1

        out_dict = processor.apply_chat_template(
            messages,
            add_generation_prompt=True,
            tokenize=True,
            return_dict=True,
            return_tensors="pt",
            load_audio_from_video=True,
        )
        self.assertTrue(self.audio_input_name in out_dict)
        self.assertTrue(self.videos_input_name in out_dict)

        # should always have input_ids and attention_mask
        self.assertEqual(len(out_dict["input_ids"]), 1)  # batch-size=1
        self.assertEqual(len(out_dict["attention_mask"]), 1)  # batch-size=1
        self.assertEqual(len(out_dict[self.audio_input_name]), 1)  # 1 audio in the conversation
        self.assertEqual(len(out_dict[self.videos_input_name]), 1)  # 1 video in the conversation

    def test_chat_template_jinja_kwargs(self):
        """Tests that users can pass any kwargs and they will be used in jinja templates."""
        processor = self.get_processor()
        if processor.chat_template is None:
            self.skipTest("Processor has no chat template")

        messages = [
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": "Which of these animals is making the sound?"},
                ],
            },
            {
                "role": "assistant",
                "content": [{"type": "text", "text": "It is a cow."}],
            },
        ]

        dummy_template = (
            "{% for message in messages %}"
            "{% if add_system_prompt %}"
            "{{'You are a helpful assistant.'}}"
            "{% endif %}"
            "{% if (message['role'] != 'assistant') %}"
            "{{'<|special_start|>' + message['role'] + '\n' + message['content'][0]['text'] + '<|special_end|>' + '\n'}}"
            "{% elif (message['role'] == 'assistant')%}"
            "{{'<|special_start|>' + message['role'] + '\n'}}"
            "{{message['content'][0]['text'] + '<|special_end|>' + '\n'}}"
            "{% endif %}"
            "{% endfor %}"
        )

        formatted_prompt = processor.apply_chat_template(
            messages, add_system_prompt=True, tokenize=False, chat_template=dummy_template
        )
        expected_prompt = "You are a helpful assistant.<|special_start|>user\nWhich of these animals is making the sound?<|special_end|>\nYou are a helpful assistant.<|special_start|>assistant\nIt is a cow.<|special_end|>\n"
        self.assertEqual(formatted_prompt, expected_prompt)

    def test_apply_chat_template_assistant_mask(self):
        processor = self.get_processor()

        if processor.chat_template is None:
            self.skipTest("Processor has no chat template")

        # The second conversation is shorter, so it gets padded when the two are batched together
        messages = [
            [
                {"role": "user", "content": [{"type": "text", "text": "What is the capital of France?"}]},
                {"role": "assistant", "content": [{"type": "text", "text": "The capital of France is Paris."}]},
                {"role": "user", "content": [{"type": "text", "text": "What about Italy?"}]},
                {"role": "assistant", "content": [{"type": "text", "text": "The capital of Italy is Rome."}]},
            ],
            [
                {"role": "user", "content": [{"type": "text", "text": "What is the capital of Spain?"}]},
                {"role": "assistant", "content": [{"type": "text", "text": "The capital of Spain is Madrid."}]},
            ],
        ]
        dummy_template = (
            "{% for message in messages %}"
            "{% if (message['role'] != 'assistant') %}"
            "{{'<|special_start|>' + message['role'] + '\n'}}"
            "{% for content in message['content'] %}"
            "{{ image_token if content['type'] == 'image' else content['text'] }}"
            "{% endfor %}"
            "{{'<|special_end|>' + '\n'}}"
            "{% elif (message['role'] == 'assistant')%}"
            "{{'<|special_start|>' + message['role'] + '\n'}}"
            "{% generation %}"
            "{{message['content'][0]['text'] + '<|special_end|>'}}"
            "{% endgeneration %}"
            "{{'\n'}}"
            "{% endif %}"
            "{% endfor %}"
        )

        # The tokenizer's own implementation on the text-only conversations is the reference for the assistant ids. Note
        # that the generation span ends on a non-whitespace char above: `char_to_token` has no token for stripped whitespace
        reference = processor.tokenizer.apply_chat_template(
            messages, tokenize=True, return_dict=True, return_assistant_tokens_mask=True, chat_template=dummy_template
        )
        expected_ids = [
            [token_id for token_id, is_assistant in zip(input_ids, assistant_mask) if is_assistant]
            for input_ids, assistant_mask in zip(reference["input_ids"], reference["assistant_masks"])
        ]

        # Regression test for #44521: expanding each placeholder into N image tokens must not shift the assistant spans.
        # Use several images in one turn and images in several turns, since every expansion shifts the spans after it
        image_token = getattr(self, "image_token", None)
        if image_token and self.does_processor_return_mm_offsets(processor.__class__, "replace_image_token"):
            for turn, num_images in ((0, 2), (2, 1)):
                for _ in range(num_images):
                    messages[0][turn]["content"].insert(0, {"type": "image", "image": self.prepare_images_inputs()})

        # A single conversation first, then a batch mixing it with the text-only one. Padding tokens have `(0, 0)`
        # offsets, so the spans must be mapped correctly whichever side the padding is on
        for batch, padding_side in ((messages[:1], None), (messages, "right"), (messages, "left")):
            inputs = processor.apply_chat_template(
                batch,
                add_generation_prompt=False,
                tokenize=True,
                return_dict=True,
                return_tensors="pt",
                return_assistant_tokens_mask=True,
                chat_template=dummy_template,
                image_token=image_token,
                padding=True,
                padding_side=padding_side,
            )
            self.assertIn("assistant_masks", inputs)
            self.assertEqual(len(inputs["assistant_masks"]), len(inputs["input_ids"]))

            masks = inputs["assistant_masks"].bool()
            for input_ids, mask, expected in zip(inputs["input_ids"], masks, expected_ids):
                self.assertEqual(input_ids[mask].tolist(), expected)

    def test_apply_chat_template_tool_calls_no_content(self):
        processor = self.get_processor()

        if processor.chat_template is None:
            self.skipTest("Processor has no chat template")

        if "tool" not in processor.chat_template:  # good heuristic to check if template supports tools
            self.skipTest("Chat template does not support tools")

        messages = [
            {
                "role": "user",
                "content": [{"type": "text", "text": "What is the weather?"}],
            },
            {
                "role": "assistant",
                "tool_calls": [{"type": "function", "function": {"name": "get_weather", "arguments": "{}"}}],
            },
        ]

        # Regression test for #45290: tokenize=True used to raise KeyError when "content" was missing
        result = processor.apply_chat_template(messages, tokenize=True)
        self.assertIsInstance(result, list)

    def test_get_num_multimodal_tokens_matches_processor_call(self):
        "Tests that the helper used internally in vLLM works correctly"

        processor = self.get_processor()

        if not hasattr(processor, "_get_num_multimodal_tokens"):
            self.skipTest("Processor doesn't support `_get_num_multimodal_tokens` yet")

        if processor.tokenizer.pad_token_id is None:
            processor.tokenizer.pad_token_id = processor.tokenizer.eos_token_id

        image_sizes = [(100, 100), (300, 100), (500, 30), (213, 167)]
        image_inputs = []
        for h, w in image_sizes:
            image_inputs.append(np.random.randint(255, size=(h, w, 3), dtype=np.uint8))

        image_token = getattr(self, "image_token", "")
        text = [f"This is an image {image_token}"] * len(image_inputs)
        inputs = processor(
            text=text, images=image_inputs, padding=True, return_mm_token_type_ids=True, return_tensors="pt"
        )

        if "mm_token_type_ids" not in inputs:
            self.skipTest("Processor doesn't support `mm_token_type_ids`")

        num_image_tokens_from_call = inputs.mm_token_type_ids.sum(-1).tolist()
        num_image_tokens_from_helper = processor._get_num_multimodal_tokens(image_sizes=image_sizes)
        self.assertListEqual(num_image_tokens_from_call, num_image_tokens_from_helper["num_image_tokens"])

        # Test with two images per single text
        text = [f"These are two images {image_token}{image_token}"] * len(image_inputs)
        inputs = processor(
            text=text,
            images=image_inputs * 2,
            padding=True,
            return_mm_token_type_ids=True,
            return_tensors="pt",
        )

        num_image_tokens_from_call = inputs.mm_token_type_ids.sum(-1).tolist()
        num_image_tokens_from_helper = processor._get_num_multimodal_tokens(image_sizes=image_sizes * 2)
        self.assertEqual(sum(num_image_tokens_from_call), sum(num_image_tokens_from_helper["num_image_tokens"]))

    def test_get_num_multimodal_tokens_matches_processor_call_video(self):
        "Tests that the helper used internally in vLLM works correctly"

        processor = self.get_processor()

        if not hasattr(processor, "_get_num_multimodal_tokens"):
            self.skipTest("Processor doesn't support `_get_num_multimodal_tokens` yet")

        if processor.tokenizer.pad_token_id is None:
            processor.tokenizer.pad_token_id = processor.tokenizer.eos_token_id

        if getattr(processor, "video_processor", None) is None:
            self.skipTest("Processor has no video processor")

        if "video_sizes" not in inspect.signature(processor._get_num_multimodal_tokens).parameters:
            self.skipTest("Processor doesn't count video tokens yet")

        video_inputs = self.prepare_videos_inputs(batch_size=2)
        # An odd frame count, so that the temporal padding of the counters is covered
        video_inputs = [video_inputs[0][:7], video_inputs[1][:7, :, :, :200]]
        video_sizes = [(len(video), *get_video_size(video)) for video in video_inputs]

        try:
            num_video_tokens_from_helper = processor._get_num_multimodal_tokens(video_sizes=video_sizes)
        except AttributeError:
            self.skipTest("Video processor doesn't support `get_num_of_video_patches` yet")
        if num_video_tokens_from_helper["num_video_tokens"] is None:
            self.skipTest("Processor doesn't count video tokens yet")

        video_token = getattr(self, "video_token", "")
        text = [f"This is a video {video_token}"] * len(video_inputs)
        inputs = processor(
            text=text,
            videos=video_inputs,
            padding=True,
            do_sample_frames=False,
            return_mm_token_type_ids=True,
            return_tensors="pt",
        )

        if "mm_token_type_ids" not in inputs:
            self.skipTest("Processor doesn't support `mm_token_type_ids`")

        num_video_tokens_from_call = (inputs.mm_token_type_ids == 2).sum(-1).tolist()
        self.assertListEqual(num_video_tokens_from_call, num_video_tokens_from_helper["num_video_tokens"])

        # Test with two videos per single text
        text = [f"These are two videos {video_token}{video_token}"] * len(video_inputs)
        inputs = processor(
            text=text,
            videos=video_inputs * 2,
            padding=True,
            do_sample_frames=False,
            return_mm_token_type_ids=True,
            return_tensors="pt",
        )

        num_video_tokens_from_call = (inputs.mm_token_type_ids == 2).sum(-1).tolist()
        num_video_tokens_from_helper = processor._get_num_multimodal_tokens(video_sizes=video_sizes * 2)
        self.assertEqual(sum(num_video_tokens_from_call), sum(num_video_tokens_from_helper["num_video_tokens"]))

    @staticmethod
    def does_processor_return_mm_offsets(processor_class, method_name: str):
        """
        Updated processor need to override a `replace_xxx_method` which is the only
        reliable way to tell if processor is old format or not.
        """
        child_method = getattr(processor_class, method_name)
        base_method = getattr(ProcessorMixin, method_name)
        return child_method is not base_method

    def test_replacement_offsets(self):
        """
        Tests that the returned replacement offsets show correct text and spans for multimodal tokens
        It is used mainly in vllm currently, and will be used for adding offsets in assistant mask
        """

        # Tiny model IDs have small vocab and squash all tokens to `UNK` Test becomes pointless
        # if we are comparing unk to unk token, the differences are lost
        processor = self.get_processor(use_tiny_ckpt=False)

        if not (
            self.does_processor_return_mm_offsets(processor.__class__, "replace_image_token")
            or self.does_processor_return_mm_offsets(processor.__class__, "replace_video_token")
        ):
            self.skipTest("Processor doesn't support `_get_num_multimodal_tokens` yet")

        # Prepare inputs dynamically based on processor attributes
        processor_inputs = {}
        modalities = []
        attributes = self.processor_class.get_attributes()

        processor_inputs = {}
        processing_kwargs = {}
        for modality, metadata in MODALITY_TEST_SPECS.items():
            attribute = self.get_subprocessor_name(modality, attributes)
            if attribute in attributes:
                modalities.append(modality.rstrip("s"))
                processing_kwargs.update(metadata["call_time_kwargs"])
                prepare_method = getattr(self, f"prepare_{modality}_inputs")
                if modality == "text":
                    processor_inputs[modality] = prepare_method(modalities=["image", "video", "audio"])
                else:
                    processor_inputs[modality] = prepare_method()

        # Test combined processing
        model_inputs = processor(
            **processor_inputs,
            add_special_tokens=False,
            padding=False,
            truncation=False,
            max_length=None,
            return_text_replacement_offsets=True,
            **processing_kwargs,
        )
        detokenized_text = processor.tokenizer.batch_decode(
            model_inputs["input_ids"], skip_special_tokens=False, clean_up_tokenization_spaces=True
        )
        self.assertIn("text_replacement_offsets", model_inputs)

        for i, sample_offsets in enumerate(model_inputs["text_replacement_offsets"]):
            for curr_dict in sample_offsets:
                self.assertIn(curr_dict["type"], modalities)
                start, end = curr_dict["new_span"]
                self.assertEqual(detokenized_text[i][start:end], curr_dict["replacement"])