| 1 | # SPDX-License-Identifier: Apache-2.0 |
| 2 | # SPDX-FileCopyrightText: Copyright contributors to the vLLM project |
| 3 | |
| 4 | import importlib |
| 5 | |
| 6 | from transformers.processing_utils import ProcessingKwargs |
| 7 | from typing_extensions import Unpack |
| 8 | |
| 9 | from vllm.transformers_utils.processor import ( |
| 10 | get_processor_kwargs_keys, |
| 11 | get_processor_kwargs_type, |
| 12 | ) |
| 13 | |
| 14 | |
| 15 | class _FakeProcessorKwargs(ProcessingKwargs, total=False): # type: ignore |
| 16 | pass |
| 17 | |
| 18 | |
| 19 | def _assert_has_all_expected(keys: set[str]) -> None: |
| 20 | # text |
| 21 | for k in ("text_pair", "text_target", "text_pair_target"): |
| 22 | assert k in keys |
| 23 | # image |
| 24 | for k in ("do_convert_rgb", "do_resize"): |
| 25 | assert k in keys |
| 26 | # audio |
| 27 | for k in ( |
| 28 | "fps", |
| 29 | "do_sample_frames", |
| 30 | "input_data_format", |
| 31 | "default_to_square", |
| 32 | ): |
| 33 | assert k in keys |
| 34 | # audio |
| 35 | for k in ("padding", "return_attention_mask"): |
| 36 | assert k in keys |
| 37 | |
| 38 | |
| 39 | # Path 1: __call__ method has kwargs: Unpack[*ProcessorKwargs] |
| 40 | class _ProcWithUnpack: |
| 41 | def __call__(self, *args, **kwargs: Unpack[_FakeProcessorKwargs]): # type: ignore |
| 42 | return None |
| 43 | |
| 44 | |
| 45 | def test_get_processor_kwargs_from_processor_unpack_path_returns_full_union(): |
| 46 | proc = _ProcWithUnpack() |
| 47 | keys = get_processor_kwargs_keys(get_processor_kwargs_type(proc)) |
| 48 | _assert_has_all_expected(keys) |
| 49 | |
| 50 | |
| 51 | # ---- Path 2: No Unpack, fallback to scanning *ProcessorKwargs in module ---- |
| 52 | |
| 53 | |
| 54 | class _ProcWithoutUnpack: |
| 55 | def __call__(self, *args, **kwargs): |
| 56 | return None |
| 57 | |
| 58 | |
| 59 | def test_get_processor_kwargs_from_processor_module_scan_returns_full_union(): |
| 60 | # ensure the module scanned by fallback is this test module |
| 61 | module_name = _ProcWithoutUnpack.__module__ |
| 62 | mod = importlib.import_module(module_name) |
| 63 | assert hasattr(mod, "_FakeProcessorKwargs") |
| 64 | |
| 65 | proc = _ProcWithoutUnpack() |
| 66 | keys = get_processor_kwargs_keys(get_processor_kwargs_type(proc)) |
| 67 | _assert_has_all_expected(keys) |