Max Isbey
1b74b06753
Cut comment and docstring volume roughly in half across src, tests, examples, and docs_src: removed comments that restate the adjacent code, leftover development narration, section banners, and self-evident Args/Returns blocks, and compressed the remaining docstrings to a Google-style summary line plus only the detail that earns its place. Kept (and tightened) the load-bearing content: Raises sections, deprecation and version-availability notes, spec/RFC/issue references, why-comments for non-obvious decisions, and all coverage pragmas. The generated mcp_types.v* wire modules are untouched.
60 lines
2.0 KiB
Python
60 lines
2.0 KiB
Python
"""Shared fixtures for `Dispatcher` contract tests.
|
|
|
|
The `pair_factory` fixture parametrizes contract tests over every implementation:
|
|
`DirectDispatcher` (in-memory) and `JSONRPCDispatcher` (over crossed anyio memory streams).
|
|
"""
|
|
|
|
from collections.abc import Callable
|
|
|
|
import anyio
|
|
import pytest
|
|
|
|
from mcp.shared.direct_dispatcher import create_direct_dispatcher_pair
|
|
from mcp.shared.dispatcher import Dispatcher
|
|
from mcp.shared.jsonrpc_dispatcher import JSONRPCDispatcher
|
|
from mcp.shared.message import SessionMessage
|
|
from mcp.shared.transport_context import TransportContext
|
|
|
|
DispatcherTriple = tuple[Dispatcher[TransportContext], Dispatcher[TransportContext], Callable[[], None]]
|
|
PairFactory = Callable[..., DispatcherTriple]
|
|
|
|
|
|
def direct_pair(*, can_send_request: bool = True) -> DispatcherTriple:
|
|
client, server = create_direct_dispatcher_pair(can_send_request=can_send_request)
|
|
|
|
def close() -> None:
|
|
client.close()
|
|
server.close()
|
|
|
|
return client, server, close
|
|
|
|
|
|
def jsonrpc_pair(*, can_send_request: bool = True) -> DispatcherTriple:
|
|
c2s_send, c2s_recv = anyio.create_memory_object_stream[SessionMessage | Exception](32)
|
|
s2c_send, s2c_recv = anyio.create_memory_object_stream[SessionMessage | Exception](32)
|
|
|
|
def builder(_meta: object) -> TransportContext:
|
|
return TransportContext(kind="jsonrpc", can_send_request=can_send_request)
|
|
|
|
client: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher(s2c_recv, c2s_send, transport_builder=builder)
|
|
server: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher(c2s_recv, s2c_send, transport_builder=builder)
|
|
|
|
def close() -> None:
|
|
for s in (c2s_send, c2s_recv, s2c_send, s2c_recv):
|
|
s.close()
|
|
|
|
return client, server, close
|
|
|
|
|
|
@pytest.fixture(
|
|
params=[
|
|
pytest.param(direct_pair, id="direct"),
|
|
pytest.param(jsonrpc_pair, id="jsonrpc"),
|
|
]
|
|
)
|
|
def pair_factory(request: pytest.FixtureRequest) -> PairFactory:
|
|
return request.param
|
|
|
|
|
|
__all__ = ["PairFactory", "direct_pair", "jsonrpc_pair"]
|