Buckets:
| """Unit tests for the shared vLLM backend helpers (vllm_utils.py). | |
| Run standalone (Python standard library only, no backend venv needed): | |
| cd backend/python/common && python3 -m unittest vllm_utils_test | |
| ``vllm_utils`` imports vLLM lazily (inside functions), so the module is | |
| importable without the vLLM wheel. ``AsyncEngineArgs`` is stood in for by a | |
| local dataclass that mirrors the field shapes that matter: plain scalars, | |
| ``Optional[...]`` scalars (vLLM's tri-state flags) and dict-valued fields. | |
| """ | |
| import contextlib | |
| import dataclasses | |
| import io | |
| import unittest | |
| from typing import Dict, Literal, Optional, Union | |
| from vllm_utils import apply_options_to_engine_args, normalize_option_key | |
| class FakeEngineArgs: | |
| model: str = "" | |
| quantization: Optional[str] = None | |
| kv_cache_dtype: str = "auto" | |
| enable_prefix_caching: Optional[bool] = None | |
| enforce_eager: bool = False | |
| max_model_len: Optional[int] = None | |
| tensor_parallel_size: int = 1 | |
| gpu_memory_utilization: float = 0.9 | |
| limit_mm_per_prompt: Optional[Dict[str, int]] = None | |
| # vLLM types dtype as a Literal of strings, several of which contain the | |
| # word "float" - a naive match reads that as a float field. | |
| dtype: Literal["auto", "float16", "bfloat16"] = "auto" | |
| seed: Union[int, str, None] = None | |
| # vLLM's arg_utils is annotated under PEP 563, where dataclasses.fields() hands | |
| # back annotations as plain strings rather than types. | |
| StringAnnotatedEngineArgs = dataclasses.make_dataclass( | |
| "StringAnnotatedEngineArgs", | |
| [ | |
| ("max_model_len", "int | None", dataclasses.field(default=None)), | |
| ("enable_prefix_caching", "Optional[bool]", dataclasses.field(default=None)), | |
| ("kv_cache_dtype", "str", dataclasses.field(default="auto")), | |
| ], | |
| ) | |
| def _apply(options, **overrides): | |
| """Apply options to a fresh FakeEngineArgs, returning (result, stderr).""" | |
| err = io.StringIO() | |
| with contextlib.redirect_stderr(err): | |
| out = apply_options_to_engine_args(FakeEngineArgs(**overrides), options) | |
| return out, err.getvalue() | |
| class TestApplyOptionsToEngineArgs(unittest.TestCase): | |
| def test_string_option_is_applied(self): | |
| out, _ = _apply(["--quantization:gptq_marlin"]) | |
| self.assertEqual(out.quantization, "gptq_marlin") | |
| def test_valueless_flag_enables_boolean_field(self): | |
| out, _ = _apply(["--enable-prefix-caching"]) | |
| self.assertIs(out.enable_prefix_caching, True) | |
| def test_boolean_field_accepts_explicit_false(self): | |
| out, _ = _apply(["--enable-prefix-caching:false"], enable_prefix_caching=True) | |
| self.assertIs(out.enable_prefix_caching, False) | |
| def test_integer_field_is_coerced(self): | |
| out, _ = _apply(["--max-model-len:4096", "--tensor-parallel-size:2"]) | |
| self.assertEqual(out.max_model_len, 4096) | |
| self.assertEqual(out.tensor_parallel_size, 2) | |
| def test_float_field_is_coerced(self): | |
| out, _ = _apply(["--gpu-memory-utilization:0.85"]) | |
| self.assertAlmostEqual(out.gpu_memory_utilization, 0.85) | |
| def test_equals_separator_is_accepted(self): | |
| out, _ = _apply(["--kv-cache-dtype=fp8_e5m2"]) | |
| self.assertEqual(out.kv_cache_dtype, "fp8_e5m2") | |
| def test_dict_field_is_parsed_as_json(self): | |
| out, _ = _apply(['--limit-mm-per-prompt:{"image": 4}']) | |
| self.assertEqual(out.limit_mm_per_prompt, {"image": 4}) | |
| def test_non_flag_options_are_left_alone(self): | |
| out, err = _apply(["tool_parser:hermes", "reasoning_parser:qwen3", "vad_only"]) | |
| self.assertEqual(out, FakeEngineArgs()) | |
| self.assertEqual(err, "") | |
| def test_unknown_flag_warns_and_is_skipped(self): | |
| out, err = _apply(["--not-a-real-flag:1", "--quantization:awq"]) | |
| self.assertEqual(out.quantization, "awq") | |
| self.assertIn("not_a_real_flag", err) | |
| self.assertIn("unknown", err.lower()) | |
| def test_parser_flags_are_not_reported_as_unknown(self): | |
| # LocalAI consumes these itself; whether the engine dataclass also has | |
| # the field depends on the vLLM version, and warning about them would | |
| # send users chasing a non-problem. | |
| out, err = _apply(["--tool-parser:hermes", "--reasoning-parser:qwen3"]) | |
| self.assertEqual(out, FakeEngineArgs()) | |
| self.assertEqual(err, "") | |
| def test_unknown_flag_hints_at_the_closest_field(self): | |
| _, err = _apply(["--max-model-length:4096"]) | |
| self.assertIn("max_model_len", err) | |
| def test_uncoercible_value_warns_and_is_skipped(self): | |
| out, err = _apply(["--max-model-len:lots"]) | |
| self.assertIsNone(out.max_model_len) | |
| self.assertIn("max_model_len", err) | |
| def test_valueless_flag_on_non_boolean_field_warns_and_is_skipped(self): | |
| out, err = _apply(["--quantization"]) | |
| self.assertIsNone(out.quantization) | |
| self.assertIn("quantization", err) | |
| def test_empty_options_returns_the_same_engine_args(self): | |
| original = FakeEngineArgs(model="m") | |
| self.assertIs(apply_options_to_engine_args(original, []), original) | |
| class TestFieldTypeInference(unittest.TestCase): | |
| def test_literal_typed_field_keeps_its_string_value(self): | |
| out, err = _apply(["--dtype:bfloat16"]) | |
| self.assertEqual(out.dtype, "bfloat16") | |
| self.assertEqual(err.count("skipping"), 0) | |
| def test_ambiguous_union_falls_back_to_literal_inference(self): | |
| out, _ = _apply(["--seed:42"]) | |
| self.assertEqual(out.seed, 42) | |
| def test_string_annotations_are_understood(self): | |
| err = io.StringIO() | |
| with contextlib.redirect_stderr(err): | |
| out = apply_options_to_engine_args( | |
| StringAnnotatedEngineArgs(), | |
| ["--max-model-len:4096", "--enable-prefix-caching", "--kv-cache-dtype:fp8_e5m2"], | |
| ) | |
| self.assertEqual(out.max_model_len, 4096) | |
| self.assertIs(out.enable_prefix_caching, True) | |
| self.assertEqual(out.kv_cache_dtype, "fp8_e5m2") | |
| class TestNormalizeOptionKey(unittest.TestCase): | |
| def test_cli_flag_becomes_a_field_name(self): | |
| self.assertEqual(normalize_option_key("--reasoning-parser"), "reasoning_parser") | |
| def test_plain_key_is_untouched(self): | |
| self.assertEqual(normalize_option_key("tool_parser"), "tool_parser") | |
| def test_surrounding_whitespace_is_stripped(self): | |
| self.assertEqual(normalize_option_key(" --tool-parser "), "tool_parser") | |
| if __name__ == "__main__": | |
| unittest.main() | |
Xet Storage Details
- Size:
- 6.49 kB
- Xet hash:
- 97961e64f8feda11d39befd5a33fd9fea1ad97441dc6f97a1b5d33d404da18fd
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.