# SPDX-License-Identifier: Apache-2.0 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project import json from dataclasses import dataclass, field from typing import Any from unittest.mock import AsyncMock, MagicMock import pytest from vllm.config.multimodal import MultiModalConfig from vllm.entrypoints.generate.base.protocol import RequestResponseMetadata from vllm.entrypoints.openai.completion.protocol import CompletionRequest from vllm.entrypoints.openai.completion.serving import OpenAIServingCompletion from vllm.entrypoints.openai.models.protocol import BaseModelPath from vllm.entrypoints.openai.models.serving import OpenAIServingModels from vllm.entrypoints.scale_out.render.serving import ServingRender from vllm.entrypoints.serve.engine.protocol import ErrorResponse from vllm.exceptions import GenerationError, VLLMValidationError from vllm.logprobs import Logprob from vllm.outputs import CompletionOutput, RequestOutput, WeightVersionSpan from vllm.renderers.deepseek_v4 import DeepseekV4Renderer from vllm.renderers.hf import HfRenderer from vllm.renderers.online_renderer import OnlineRenderer from vllm.tokenizers.registry import cached_tokenizer_from_config from vllm.v1.engine.async_llm import AsyncLLM from vllm.v1.metrics.stats import RequestSpecDecodeMetrics, RequestStateStats MODEL_NAME = "openai-community/gpt2" MODEL_NAME_SHORT = "gpt2" DEEPSEEK_V4_FIM_BEGIN = "<｜fim▁begin｜>" DEEPSEEK_V4_FIM_HOLE = "<｜fim▁hole｜>" DEEPSEEK_V4_FIM_END = "<｜fim▁end｜>" _PER_REQUEST_STATS = RequestStateStats( queued_ts=1.0, scheduled_ts=1.5, first_token_ts=2.0, last_token_ts=3.0, num_generation_tokens=2, ) BASE_MODEL_PATHS = [ BaseModelPath(name=MODEL_NAME, model_path=MODEL_NAME), BaseModelPath(name=MODEL_NAME_SHORT, model_path=MODEL_NAME_SHORT), ] @dataclass class MockHFConfig: model_type: str = "any" @dataclass class MockModelConfig: task = "generate" runner_type = "generate" model = MODEL_NAME tokenizer = MODEL_NAME trust_remote_code = False tokenizer_mode = "auto" max_model_len = 100 revision = None code_revision = None tokenizer_revision = None multimodal_config = MultiModalConfig() hf_config = MockHFConfig() logits_processors: list[str] | None = None diff_sampling_param: dict | None = None allowed_local_media_path: str = "" allowed_media_domains: list[str] | None = None encoder_config = None generation_config: str = "auto" media_io_kwargs: dict[str, dict[str, Any]] = field(default_factory=dict) skip_tokenizer_init: bool = False is_encoder_decoder: bool = False is_multimodal_model: bool = False supports_multimodal_inputs: bool = False renderer_num_workers: int = 1 def get_diff_sampling_param(self): return self.diff_sampling_param or {} @dataclass class MockParallelConfig: _api_process_rank: int = 0 @dataclass class MockVllmConfig: model_config: MockModelConfig parallel_config: MockParallelConfig def _build_serving_completion(engine: AsyncLLM) -> OpenAIServingCompletion: models = OpenAIServingModels( engine_client=engine, base_model_paths=BASE_MODEL_PATHS, ) online_renderer = OnlineRenderer( model_config=engine.model_config, renderer=engine.renderer, request_logger=None, chat_template=None, chat_template_content_format="auto", ) return OpenAIServingCompletion( engine, models, online_renderer=online_renderer, request_logger=None, ) def _build_minimal_metrics_serving_completion( enable_per_request_metrics: bool, ) -> OpenAIServingCompletion: serving = OpenAIServingCompletion.__new__(OpenAIServingCompletion) serving.enable_prompt_tokens_details = False serving.system_fingerprint = None serving.enable_per_request_metrics = enable_per_request_metrics return serving def _make_metrics_request_output( metrics: RequestStateStats | None = _PER_REQUEST_STATS, ) -> RequestOutput: return RequestOutput( request_id="test-id", prompt="Test prompt", prompt_token_ids=[1, 2, 3], prompt_logprobs=None, outputs=[ CompletionOutput( index=0, text="Hello", token_ids=[100, 101], cumulative_logprob=None, logprobs=None, finish_reason="stop", ) ], finished=True, metrics=metrics, ) def _build_renderer(model_config: MockModelConfig): return HfRenderer( MockVllmConfig(model_config, parallel_config=MockParallelConfig()), cached_tokenizer_from_config(model_config), ) def test_completion_per_request_metrics_follow_server_flag(): request = CompletionRequest(model=MODEL_NAME, prompt="Test prompt", max_tokens=10) request_output = _make_metrics_request_output() disabled_serving = _build_minimal_metrics_serving_completion( enable_per_request_metrics=False ) disabled_response = disabled_serving.request_output_to_completion_response( [request_output], request, "cmpl-test-id", 0, MODEL_NAME, None, RequestResponseMetadata(request_id="cmpl-test-id"), ) assert disabled_response.metrics is None enabled_serving = _build_minimal_metrics_serving_completion( enable_per_request_metrics=True ) enabled_response = enabled_serving.request_output_to_completion_response( [request_output], request, "cmpl-test-id", 0, MODEL_NAME, None, RequestResponseMetadata(request_id="cmpl-test-id"), ) assert enabled_response.metrics is not None assert enabled_response.metrics.time_to_first_token_ms == pytest.approx(500.0) def test_completion_per_request_metrics_suppressed_for_multiple_prompts(): serving = _build_minimal_metrics_serving_completion(enable_per_request_metrics=True) response = serving.request_output_to_completion_response( [_make_metrics_request_output(), _make_metrics_request_output()], CompletionRequest( model=MODEL_NAME, prompt=["Test prompt", "Another prompt"], max_tokens=10, ), "cmpl-test-id", 0, MODEL_NAME, None, RequestResponseMetadata(request_id="cmpl-test-id"), ) assert response.metrics is None def _spec_decode_metrics() -> RequestSpecDecodeMetrics: # Two verify steps: accept 3 drafts, then 1 -> histogram [0, 1, 0, 1]. m = RequestSpecDecodeMetrics.new(num_spec_tokens=3) m.observe(num_draft_tokens=3, num_accepted=3) m.observe(num_draft_tokens=3, num_accepted=1) return m def _make_spec_decode_request_output( num_seqs: int = 1, with_metrics: bool = True ) -> RequestOutput: outputs = [ CompletionOutput( index=i, text="Hello", token_ids=[100, 101], cumulative_logprob=None, logprobs=None, finish_reason="stop", spec_decode_metrics=_spec_decode_metrics() if with_metrics else None, ) for i in range(num_seqs) ] return RequestOutput( request_id="test-id", prompt="Test prompt", prompt_token_ids=[1, 2, 3], prompt_logprobs=None, outputs=outputs, finished=True, metrics=None, ) def _completion_response(serving, request, request_output): return serving.request_output_to_completion_response( [request_output], request, "cmpl-test-id", 0, MODEL_NAME, None, RequestResponseMetadata(request_id="cmpl-test-id"), ) def test_completion_spec_decode_metrics_present_for_single_sequence(): # Timing off, but the sequence carries acceptance metrics -> the metrics # object is created just to hold metrics.speculative_decoding. serving = _build_minimal_metrics_serving_completion( enable_per_request_metrics=False ) response = _completion_response( serving, CompletionRequest(model=MODEL_NAME, prompt="Test prompt", max_tokens=10), _make_spec_decode_request_output(num_seqs=1), ) assert response.metrics is not None assert response.metrics.time_to_first_token_ms is None # timing not requested spec = response.metrics.speculative_decoding assert spec is not None assert spec.acceptance_histogram == [0, 1, 0, 1] # dense, index j assert spec.num_spec_steps == 2 assert spec.num_spec_tokens == 3 assert spec.mean_acceptance_length == pytest.approx(3.0) # 1 + (3 + 1) / 2 def test_completion_spec_decode_metrics_suppressed_for_n_gt_1(): # Per-request metrics can't be attributed to one of the n sequences. serving = _build_minimal_metrics_serving_completion( enable_per_request_metrics=False ) response = _completion_response( serving, CompletionRequest(model=MODEL_NAME, prompt="Test prompt", n=2, max_tokens=10), _make_spec_decode_request_output(num_seqs=2), ) assert response.metrics is None def test_completion_spec_decode_metrics_absent_when_not_collected(): # Flag off -> the sequence carries no acceptance metrics -> no metrics object. serving = _build_minimal_metrics_serving_completion( enable_per_request_metrics=False ) response = _completion_response( serving, CompletionRequest(model=MODEL_NAME, prompt="Test prompt", max_tokens=10), _make_spec_decode_request_output(num_seqs=1, with_metrics=False), ) assert response.metrics is None def test_completion_metrics_carries_both_timing_and_spec_decode(): serving = _build_minimal_metrics_serving_completion(enable_per_request_metrics=True) request_output = _make_spec_decode_request_output(num_seqs=1) request_output.metrics = _PER_REQUEST_STATS # timing source response = _completion_response( serving, CompletionRequest(model=MODEL_NAME, prompt="Test prompt", max_tokens=10), request_output, ) assert response.metrics is not None assert response.metrics.time_to_first_token_ms == pytest.approx(500.0) assert response.metrics.speculative_decoding is not None assert response.metrics.speculative_decoding.num_spec_steps == 2 @pytest.mark.asyncio @pytest.mark.parametrize("stream", [False, True]) @pytest.mark.parametrize("opt_in", [False, True]) async def test_completion_weight_versions(stream: bool, opt_in: bool): """Each choice gets its spans on its final output, even with no tokens.""" expected = [{"version": "v1", "start": 0, "end": 2}] spans = [WeightVersionSpan(**span) for span in expected] def output(index, token_ids, finish_reason=None, weight_versions=None): return CompletionOutput( index=index, text="x" * len(token_ids), token_ids=token_ids, cumulative_logprob=None, logprobs=None, finish_reason=finish_reason, weight_versions=weight_versions, ) # Choice 1 is aborted before its first token. final = [output(1, [], "abort", [])] if stream: steps = [[output(0, [100])], [output(0, [101], "length", spans), *final]] else: steps = [[output(0, [100, 101], "length", spans), *final]] async def mock_generate(*args, **kwargs): for i, outputs in enumerate(steps): yield RequestOutput( request_id="test-id", prompt="Test prompt", prompt_token_ids=[1, 2, 3], prompt_logprobs=None, outputs=outputs, finished=i == len(steps) - 1, ) mock_engine = MagicMock(spec=AsyncLLM) mock_engine.errored = False mock_engine.model_config = MockModelConfig() mock_engine.input_processor = MagicMock() mock_engine.renderer = _build_renderer(mock_engine.model_config) mock_engine.generate = MagicMock(side_effect=mock_generate) request = CompletionRequest( model=MODEL_NAME, prompt="Test prompt", max_tokens=10, n=2, stream=stream, return_weight_versions=opt_in, ) response = await _build_serving_completion(mock_engine).create_completion(request) if stream: finals = {} async for line in response: payload = line.removeprefix("data: ").strip() if payload == "[DONE]": continue for choice in json.loads(payload)["choices"]: if choice.get("finish_reason"): finals[choice["index"]] = choice else: assert choice.get("weight_versions") is None got = [finals[i].get("weight_versions") for i in (0, 1)] else: got = [choice["weight_versions"] for choice in response.model_dump()["choices"]] assert got == ([expected, []] if opt_in else [None, None]) def _build_online_renderer_for_completion( model_config: MockModelConfig, renderer: Any | None = None, ) -> OnlineRenderer: if renderer is None: renderer = MagicMock() renderer.render_completion_suffix.return_value = None return OnlineRenderer( model_config=model_config, renderer=renderer, request_logger=None, chat_template=None, chat_template_content_format="auto", ) def _deepseek_v4_suffix_renderer( model_config: MockModelConfig, ) -> DeepseekV4Renderer: return DeepseekV4Renderer( MockVllmConfig(model_config, parallel_config=MockParallelConfig()), None, ) @pytest.mark.asyncio @pytest.mark.skip_global_cleanup async def test_completion_suffix_uses_renderer_capability(): model_config = MockModelConfig() online_renderer = _build_online_renderer_for_completion( model_config, renderer=_deepseek_v4_suffix_renderer(model_config), ) online_renderer.preprocess_completion = AsyncMock(return_value=[{"ok": True}]) request = CompletionRequest( model=MODEL_NAME, prompt="def fib(n):\n return ", suffix="\n\nprint(fib(10))", max_tokens=64, ) result = await online_renderer.render_completion(request) assert result == [{"ok": True}] online_renderer.preprocess_completion.assert_awaited_once() assert online_renderer.preprocess_completion.call_args.kwargs["prompt_input"] == ( f"{DEEPSEEK_V4_FIM_BEGIN}def fib(n):\n return " f"{DEEPSEEK_V4_FIM_HOLE}\n\nprint(fib(10))" f"{DEEPSEEK_V4_FIM_END}" ) @pytest.mark.asyncio @pytest.mark.skip_global_cleanup async def test_completion_suffix_supports_text_prompt_list(): model_config = MockModelConfig() online_renderer = _build_online_renderer_for_completion( model_config, renderer=_deepseek_v4_suffix_renderer(model_config), ) online_renderer.preprocess_completion = AsyncMock(return_value=[{"ok": True}]) request = CompletionRequest( model=MODEL_NAME, prompt=["prefix A", "prefix B"], suffix=" suffix", ) result = await online_renderer.render_completion(request) assert result == [{"ok": True}] assert online_renderer.preprocess_completion.call_args.kwargs["prompt_input"] == [ ( f"{DEEPSEEK_V4_FIM_BEGIN}prefix A" f"{DEEPSEEK_V4_FIM_HOLE} suffix{DEEPSEEK_V4_FIM_END}" ), ( f"{DEEPSEEK_V4_FIM_BEGIN}prefix B" f"{DEEPSEEK_V4_FIM_HOLE} suffix{DEEPSEEK_V4_FIM_END}" ), ] @pytest.mark.asyncio @pytest.mark.skip_global_cleanup async def test_completion_suffix_rejects_renderer_without_fim_support(): model_config = MockModelConfig() online_renderer = _build_online_renderer_for_completion(model_config) online_renderer.preprocess_completion = AsyncMock() request = CompletionRequest( model=MODEL_NAME, prompt="prefix", suffix="suffix", ) result = await online_renderer.render_completion(request) assert isinstance(result, ErrorResponse) assert "FIM completion rendering" in result.error.message online_renderer.preprocess_completion.assert_not_awaited() @pytest.mark.asyncio @pytest.mark.skip_global_cleanup async def test_completion_suffix_rejects_echo(): model_config = MockModelConfig() online_renderer = _build_online_renderer_for_completion( model_config, renderer=_deepseek_v4_suffix_renderer(model_config), ) online_renderer.preprocess_completion = AsyncMock() request = CompletionRequest( model=MODEL_NAME, prompt="prefix", suffix="suffix", echo=True, ) result = await online_renderer.render_completion(request) assert isinstance(result, ErrorResponse) assert "Echo is unsupported with suffix" in result.error.message online_renderer.preprocess_completion.assert_not_awaited() @pytest.mark.asyncio @pytest.mark.skip_global_cleanup async def test_completion_suffix_rejects_prompt_embeds(): model_config = MockModelConfig() online_renderer = _build_online_renderer_for_completion( model_config, renderer=_deepseek_v4_suffix_renderer(model_config), ) online_renderer.preprocess_completion = AsyncMock() request = CompletionRequest( model=MODEL_NAME, prompt="prefix", suffix="suffix", prompt_embeds=b"embeds", ) result = await online_renderer.render_completion(request) assert isinstance(result, ErrorResponse) assert "suffix is not supported with prompt_embeds" in result.error.message online_renderer.preprocess_completion.assert_not_awaited() @pytest.mark.asyncio @pytest.mark.skip_global_cleanup @pytest.mark.parametrize("prompt", ([1, 2, 3], [[1, 2, 3]])) async def test_completion_suffix_rejects_token_prompt(prompt): model_config = MockModelConfig() online_renderer = _build_online_renderer_for_completion( model_config, renderer=_deepseek_v4_suffix_renderer(model_config), ) online_renderer.preprocess_completion = AsyncMock() request = CompletionRequest( model=MODEL_NAME, prompt=prompt, suffix="suffix", ) result = await online_renderer.render_completion(request) assert isinstance(result, ErrorResponse) assert "requires text prompt input" in result.error.message online_renderer.preprocess_completion.assert_not_awaited() @pytest.mark.asyncio async def test_completion_error_non_stream(): """Test finish_reason='error' returns 500 InternalServerError (non-streaming).""" mock_engine = MagicMock(spec=AsyncLLM) mock_engine.errored = False mock_engine.model_config = MockModelConfig() mock_engine.input_processor = MagicMock() mock_engine.renderer = _build_renderer(mock_engine.model_config) serving_completion = _build_serving_completion(mock_engine) completion_output = CompletionOutput( index=0, text="", token_ids=[], cumulative_logprob=None, logprobs=None, finish_reason="error", ) request_output = RequestOutput( request_id="test-id", prompt="Test prompt", prompt_token_ids=[1, 2, 3], prompt_logprobs=None, outputs=[completion_output], finished=True, metrics=None, lora_request=None, encoder_prompt=None, encoder_prompt_token_ids=None, ) async def mock_generate(*args, **kwargs): yield request_output mock_engine.generate = MagicMock(side_effect=mock_generate) request = CompletionRequest( model=MODEL_NAME, prompt="Test prompt", max_tokens=10, stream=False, ) with pytest.raises(GenerationError): await serving_completion.create_completion(request) @pytest.mark.asyncio async def test_openai_completion_keeps_mm_cache_for_engine_execution(): mock_engine = MagicMock(spec=AsyncLLM) mock_engine.errored = False mock_engine.model_config = MockModelConfig() mock_engine.input_processor = MagicMock() mock_engine.renderer = _build_renderer(mock_engine.model_config) serving_completion = _build_serving_completion(mock_engine) serving_completion.online_renderer.preprocess_completion = AsyncMock( return_value=[{"prompt_token_ids": [1, 2, 3]}] ) request = CompletionRequest( model=MODEL_NAME, prompt="Test prompt", ) result = await serving_completion.render_completion_request(request) assert isinstance(result, list) assert ( serving_completion.online_renderer.preprocess_completion.call_args.kwargs[ "skip_mm_cache" ] is False ) def _build_serving_render(engine: AsyncLLM) -> ServingRender: models = OpenAIServingModels( engine_client=engine, base_model_paths=BASE_MODEL_PATHS, ) online_renderer = OnlineRenderer( model_config=engine.model_config, renderer=engine.renderer, request_logger=None, chat_template=None, chat_template_content_format="auto", ) serving_render = ServingRender(models, online_renderer) async def _fake_preprocess_chat(*args, **kwargs): # return conversation, engine_inputs return ( [{"role": "user", "content": "Test"}], [{"prompt_token_ids": [1, 2, 3]}], ) serving_render.online_renderer.preprocess_chat = AsyncMock( side_effect=_fake_preprocess_chat ) return serving_render @pytest.mark.asyncio async def test_renderer_only_completion_request_skips_mm_cache(): mock_engine = MagicMock(spec=AsyncLLM) mock_engine.errored = False mock_engine.model_config = MockModelConfig() mock_engine.input_processor = MagicMock() mock_engine.renderer = _build_renderer(mock_engine.model_config) serving_render = _build_serving_render(mock_engine) serving_render.online_renderer.preprocess_completion = AsyncMock( return_value=[{"prompt_token_ids": [1, 2, 3]}] ) request = CompletionRequest( model=MODEL_NAME, prompt="Test prompt", ) result = await serving_render.render_completion_request(request) assert isinstance(result, list) assert ( serving_render.online_renderer.preprocess_completion.call_args.kwargs[ "skip_mm_cache" ] is True ) @pytest.mark.asyncio @pytest.mark.parametrize("include_usage", [False, True]) async def test_completion_error_stream(include_usage: bool): """Test finish_reason='error' returns 500 InternalServerError (streaming).""" mock_engine = MagicMock(spec=AsyncLLM) mock_engine.errored = False mock_engine.model_config = MockModelConfig() mock_engine.input_processor = MagicMock() mock_engine.renderer = _build_renderer(mock_engine.model_config) serving_completion = _build_serving_completion(mock_engine) completion_output_1 = CompletionOutput( index=0, text="Hello", token_ids=[100], cumulative_logprob=None, logprobs=None, finish_reason=None, ) request_output_1 = RequestOutput( request_id="test-id", prompt="Test prompt", prompt_token_ids=[1, 2, 3], prompt_logprobs=None, outputs=[completion_output_1], finished=False, metrics=None, lora_request=None, encoder_prompt=None, encoder_prompt_token_ids=None, ) completion_output_2 = CompletionOutput( index=0, text="Hello", token_ids=[100], cumulative_logprob=None, logprobs=None, finish_reason="error", ) request_output_2 = RequestOutput( request_id="test-id", prompt="Test prompt", prompt_token_ids=[1, 2, 3], prompt_logprobs=None, outputs=[completion_output_2], finished=True, metrics=None, lora_request=None, encoder_prompt=None, encoder_prompt_token_ids=None, ) async def mock_generate(*args, **kwargs): yield request_output_1 yield request_output_2 mock_engine.generate = MagicMock(side_effect=mock_generate) request = CompletionRequest( model=MODEL_NAME, prompt="Test prompt", max_tokens=10, stream=True, stream_options={"include_usage": include_usage}, ) response = await serving_completion.create_completion(request) chunks = [] async for chunk in response: chunks.append(chunk) assert len(chunks) >= 2 assert any("Internal server error" in chunk for chunk in chunks), ( f"Expected error message in chunks: {chunks}" ) assert chunks[-1] == "data: [DONE]\n\n" @pytest.mark.asyncio @pytest.mark.parametrize("include_usage", [False, True]) async def test_completion_error_stream_first_chunk(include_usage: bool): """An error before the first token must reach the streaming client.""" mock_engine = MagicMock(spec=AsyncLLM) mock_engine.errored = False mock_engine.model_config = MockModelConfig() mock_engine.input_processor = MagicMock() mock_engine.renderer = _build_renderer(mock_engine.model_config) serving_completion = _build_serving_completion(mock_engine) completion_output = CompletionOutput( index=0, text="", token_ids=[], cumulative_logprob=None, logprobs=None, finish_reason="error", ) request_output = RequestOutput( request_id="test-id", prompt="Test prompt", prompt_token_ids=[1, 2, 3], prompt_logprobs=None, outputs=[completion_output], finished=True, metrics=None, lora_request=None, encoder_prompt=None, encoder_prompt_token_ids=None, ) async def mock_generate(*args, **kwargs): yield request_output mock_engine.generate = MagicMock(side_effect=mock_generate) request = CompletionRequest( model=MODEL_NAME, prompt="Test prompt", max_tokens=10, stream=True, stream_options={"include_usage": include_usage}, ) response = await serving_completion.create_completion(request) chunks = [] async for chunk in response: chunks.append(chunk) assert len(chunks) >= 2 assert any("Internal server error" in chunk for chunk in chunks), ( f"Expected error message in chunks: {chunks}" ) assert chunks[-1] == "data: [DONE]\n\n" def test_json_schema_response_format_missing_schema(): """When response_format type is 'json_schema' but the json_schema field is not provided, request construction should raise a validation error so the API returns 400 instead of 500.""" with pytest.raises(Exception, match="json_schema.*must be provided"): CompletionRequest( model=MODEL_NAME, prompt="Test prompt", max_tokens=10, response_format={"type": "json_schema"}, ) @pytest.mark.parametrize("format_value", [None, {}]) def test_structural_tag_response_format_invalid(format_value): """Malformed structural tags should be rejected during request validation.""" with pytest.raises( VLLMValidationError, match="Invalid response_format structural_tag", ): CompletionRequest( model=MODEL_NAME, prompt="Test prompt", max_tokens=10, response_format={"type": "structural_tag", "format": format_value}, ) @pytest.mark.parametrize("structural_tag", ["not json", ""]) def test_structured_outputs_structural_tag_invalid(structural_tag): """Malformed direct structured_outputs structural tags should be rejected.""" with pytest.raises( VLLMValidationError, match="Invalid structured_outputs structural_tag", ): CompletionRequest( model=MODEL_NAME, prompt="Test prompt", max_tokens=10, structured_outputs={"structural_tag": structural_tag}, ) def test_negative_prompt_token_ids_nested(): """Negative token IDs in prompt (nested list) should raise validation error.""" with pytest.raises(Exception, match="greater than or equal to 0"): CompletionRequest( model=MODEL_NAME, prompt=[[-1]], max_tokens=10, ) def test_negative_prompt_token_ids_flat(): """Negative token IDs in prompt (flat list) should raise validation error.""" with pytest.raises(Exception, match="greater than or equal to 0"): CompletionRequest( model=MODEL_NAME, prompt=[-1], max_tokens=10, ) def test_logprobs_minus_one_allowed(): """logprobs=-1 means "return all logprobs". The sampling layer and the chat top_logprobs / prompt_logprobs validators all accept -1, so the completion logprobs validator must accept it too instead of rejecting it as negative.""" request = CompletionRequest( model=MODEL_NAME, prompt="Test prompt", max_tokens=10, logprobs=-1, ) assert request.logprobs == -1 def _minimal_serving_completion() -> OpenAIServingCompletion: """A serving object with only what `_create_completion_logprobs` reads.""" serving = OpenAIServingCompletion.__new__(OpenAIServingCompletion) serving.return_tokens_as_token_ids = False return serving @pytest.mark.parametrize( ("num_output_top_logprobs", "expected"), [(-1, 3), (1, 2), (2, 3)], ) def test_completion_logprobs_honors_requested_count( num_output_top_logprobs: int, expected: int ): """`logprobs=-1` asks for every logprob, so none may be filtered out. The validator accepts -1 and the sampler expands it to the vocabulary, so dropping the candidates here would return an empty `top_logprobs` for every position while still charging for the full computation. """ tokenizer = MagicMock() tokenizer.decode = lambda token_ids: f"tok{token_ids[0]}" step_logprobs = { 10: Logprob(logprob=-0.1, decoded_token="a"), 11: Logprob(logprob=-0.2, decoded_token="b"), 12: Logprob(logprob=-0.3, decoded_token="c"), } logprobs = _minimal_serving_completion()._create_completion_logprobs( token_ids=[10], top_logprobs=[step_logprobs], num_output_top_logprobs=num_output_top_logprobs, tokenizer=tokenizer, ) assert logprobs.top_logprobs is not None assert len(logprobs.top_logprobs[0]) == expected def test_logprobs_below_minus_one_rejected(): """Values more negative than -1 stay invalid.""" with pytest.raises(Exception, match="must be a positive value or -1"): CompletionRequest( model=MODEL_NAME, prompt="Test prompt", max_tokens=10, logprobs=-2, ) @pytest.mark.parametrize("max_tokens", [5, 0]) def test_beam_search_echo_logprobs_rejected(max_tokens): """Beam search does not compute prompt logprobs, so echo + logprobs is rejected.""" with pytest.raises(VLLMValidationError, match="not supported with beam search"): CompletionRequest( model=MODEL_NAME, prompt="Test prompt", max_tokens=max_tokens, use_beam_search=True, echo=True, logprobs=1, ) class TestCompletionPromptListLimit: """Regression tests for CVE: unbounded prompt list fan-out.""" def test_scalar_prompt_allowed(self): request = CompletionRequest( model=MODEL_NAME, prompt="hello", max_tokens=1, ) assert request.prompt == "hello" def test_single_token_list_allowed(self): request = CompletionRequest( model=MODEL_NAME, prompt=[1, 2, 3], max_tokens=1, ) assert request.prompt == [1, 2, 3] def test_bounded_text_prompt_list_allowed(self, monkeypatch): monkeypatch.setenv("VLLM_MAX_COMPLETION_PROMPTS", "10") from vllm import envs if hasattr(envs.__getattr__, "cache_clear"): envs.__getattr__.cache_clear() request = CompletionRequest( model=MODEL_NAME, prompt=["a", "b", "c"], max_tokens=1, ) assert request.prompt == ["a", "b", "c"] def test_bounded_token_id_prompt_list_allowed(self, monkeypatch): monkeypatch.setenv("VLLM_MAX_COMPLETION_PROMPTS", "10") from vllm import envs if hasattr(envs.__getattr__, "cache_clear"): envs.__getattr__.cache_clear() request = CompletionRequest( model=MODEL_NAME, prompt=[[1], [2], [3]], max_tokens=1, ) assert request.prompt == [[1], [2], [3]] def test_oversized_text_prompt_list_rejected(self, monkeypatch): monkeypatch.setenv("VLLM_MAX_COMPLETION_PROMPTS", "5") from vllm import envs if hasattr(envs.__getattr__, "cache_clear"): envs.__getattr__.cache_clear() with pytest.raises( Exception, match="prompt list length 10 exceeds the maximum" ): CompletionRequest( model=MODEL_NAME, prompt=["x"] * 10, max_tokens=1, ) def test_oversized_token_id_prompt_list_rejected(self, monkeypatch): monkeypatch.setenv("VLLM_MAX_COMPLETION_PROMPTS", "5") from vllm import envs if hasattr(envs.__getattr__, "cache_clear"): envs.__getattr__.cache_clear() with pytest.raises( Exception, match="prompt list length 10 exceeds the maximum" ): CompletionRequest( model=MODEL_NAME, prompt=[[1]] * 10, max_tokens=1, ) def test_exact_limit_allowed(self, monkeypatch): monkeypatch.setenv("VLLM_MAX_COMPLETION_PROMPTS", "5") from vllm import envs if hasattr(envs.__getattr__, "cache_clear"): envs.__getattr__.cache_clear() request = CompletionRequest( model=MODEL_NAME, prompt=["x"] * 5, max_tokens=1, ) assert len(request.prompt) == 5 def test_one_over_limit_rejected(self, monkeypatch): monkeypatch.setenv("VLLM_MAX_COMPLETION_PROMPTS", "5") from vllm import envs if hasattr(envs.__getattr__, "cache_clear"): envs.__getattr__.cache_clear() with pytest.raises(Exception, match="prompt list length 6 exceeds the maximum"): CompletionRequest( model=MODEL_NAME, prompt=["x"] * 6, max_tokens=1, ) def test_oversized_prompt_embeds_list_rejected(self, monkeypatch): monkeypatch.setenv("VLLM_MAX_COMPLETION_PROMPTS", "5") from vllm import envs if hasattr(envs.__getattr__, "cache_clear"): envs.__getattr__.cache_clear() with pytest.raises(Exception, match="prompt_embeds list length 10 exceeds"): CompletionRequest( model=MODEL_NAME, prompt_embeds=[b"\x00"] * 10, max_tokens=1, ) def test_bounded_prompt_embeds_list_allowed(self, monkeypatch): monkeypatch.setenv("VLLM_MAX_COMPLETION_PROMPTS", "5") from vllm import envs if hasattr(envs.__getattr__, "cache_clear"): envs.__getattr__.cache_clear() request = CompletionRequest( model=MODEL_NAME, prompt_embeds=[b"\x00"] * 5, max_tokens=1, ) assert len(request.prompt_embeds) == 5 @pytest.mark.parametrize("field_name", ["prompt_logprobs", "logprobs"]) def test_non_numeric_logprobs_rejected(field_name): """A non-numeric logprobs value must be a clean 400 validation error, not a TypeError from the mode='before' comparison (which surfaces as HTTP 500).""" with pytest.raises(VLLMValidationError, match=f"`{field_name}` must be an integer"): CompletionRequest( model=MODEL_NAME, prompt="Test prompt", max_tokens=10, **{field_name: "2"}, )