Repository: vllm-project/vllm
File: tests/entrypoints/openai/test_serving_responses.py (Python)

Code before:

    def test_extract_allowed_tools_filters_non_mcp(self):
        """Test that non-MCP tools are ignored during extraction."""
        tools = [
            Mcp(
                type="mcp",
                server_label="server1",
                allowed_tools=["tool1"],
            ),
            LocalShell(type="local_shell"),  # Non-MCP tool should be ignored
            Mcp(
                type="mcp",
                server_label="server2",
                allowed_tools=["tool2"],
            ),
        ]
        result = _extract_allowed_tools_from_mcp_requests(tools)
        # Non-MCP tools should be ignored
        assert result == {
            "server1": ["tool1"],
            "server2": ["tool2"],
        }


@pytest.mark.asyncio
async def test_simple_streaming_emits_reasoning_part_events(monkeypatch):
    """Verify _process_simple_streaming_events emits reasoning_part.added
    and reasoning_part.done for reasoning content, not content_part.added.

    Before the fix, the simple streaming path always emitted
    ResponseContentPartAddedEvent even for reasoning items and never
    emitted ResponseReasoningPartDoneEvent.
    """

    class FakeTokenizer:
        def __init__(self):
            self._vocab = {}

        def get_vocab(self):
            return self._vocab

    class FakeReasoningParser:
        """Returns reasoning for the first call, content for the second."""

        def __init__(self, tokenizer):
            self._call_count = 0

        def extract_reasoning_streaming(self, **kwargs):
            self._call_count += 1
            if self._call_count == 1:
                return DeltaMessage(reasoning="thinking")

Review comment (suggestion):
can you make this an E2E test in test_simple.py instead? similar to how we use pairs_of_event_types in test_streaming_types inside test_harmony.py. this way harmony / simpleContext have consistent testing logic

Code after:
                server_label="server2",
                allowed_tools=["tool2"],
            ),
        ]
        result = _extract_allowed_tools_from_mcp_requests(tools)
        # Non-MCP tools should be ignored
        assert result == {
            "server1": ["tool1"],
            "server2": ["tool2"],
        }


class TestHarmonyPreambleStreaming:
    """Tests for preamble (commentary with no recipient) streaming events."""

    @staticmethod
    def _make_ctx(*, channel, recipient, delta="hello"):
        """Build a lightweight mock StreamingHarmonyContext."""
        ctx = MagicMock()
        ctx.last_content_delta = delta
        ctx.parser.current_channel = channel
        ctx.parser.current_recipient = recipient
        return ctx

    @staticmethod
    def _make_previous_item(*, channel, recipient, text="preamble text"):
        """Build a lightweight mock previous_item (openai_harmony Message)."""
        content_part = MagicMock()
        content_part.text = text
        item = MagicMock()
        item.channel = channel
        item.recipient = recipient
        item.content = [content_part]
        return item

    def test_preamble_delta_emits_text_events(self) -> None:
        """commentary + recipient=None should emit output_text.delta events."""
        from vllm.entrypoints.openai.responses.streaming_events import (
            emit_content_delta_events,
        )

        ctx = self._make_ctx(channel="commentary", recipient=None)
        state = StreamingState()

        events = emit_content_delta_events(ctx, state)

        type_names = [e.type for e in events]
        assert "response.output_text.delta" in type_names
        assert "response.output_item.added" in type_names

    def test_preamble_delta_second_token_no_added(self) -> None: