Function test_request_context_isolation

tests/shared/test_sse.py:321–345

Each SSE connection's handlers see only that connection's request headers.

()

@pytest.mark.anyio
async def test_request_context_isolation() -> None:
    """Each SSE connection's handlers see only that connection's request headers."""
    factory = in_process_client_factory(make_context_server_app())
    contexts: list[dict[str, Any]] = []

    # Connect three clients in turn, each with its own headers.
    for i in range(3):
        headers = {"X-Request-Id": f"request-{i}", "X-Custom-Value": f"value-{i}"}

        async with sse_client(f"{BASE_URL}/sse", httpx_client_factory=factory, headers=headers) as streams:
            async with ClientSession(*streams) as session:
                await session.initialize()

                tool_result = await session.call_tool("echo_context", {"request_id": f"request-{i}"})

                assert len(tool_result.content) == 1
                content = tool_result.content[0]
                assert isinstance(content, TextContent)
                contexts.append(json.loads(content.text))

    assert len(contexts) == 3
    for i, ctx in enumerate(contexts):
        assert ctx["request_id"] == f"request-{i}"
        assert ctx["headers"].get("x-request-id") == f"request-{i}"
        assert ctx["headers"].get("x-custom-value") == f"value-{i}"


def test_sse_message_id_coercion() -> None: