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