Function connected_runner

tests/server/test_runner.py:93–148

Yield `(client, runner)` running over an in-memory JSON-RPC dispatcher pair. Starts the client (echo handlers) and the server-side dispatcher loop (kernel `on_request`/`on_notify` + `aclose_shielded` teardown - the `serve_connection` shape) in a task group, wraps the body in `anyio.

(
    server: SrvT,
    *,
    initialized: bool = True,
    init_options: InitializationOptions | None = None,
    dispatch_middleware: list[DispatchMiddleware] | None = None,
    connection: Connection | None = None,
)

@asynccontextmanager
async def connected_runner(
    server: SrvT,
    *,
    initialized: bool = True,
    init_options: InitializationOptions | None = None,
    dispatch_middleware: list[DispatchMiddleware] | None = None,
    connection: Connection | None = None,
) -> AsyncIterator[tuple[JSONRPCDispatcher[TransportContext], ServerRunner[dict[str, Any]]]]:
    """Yield `(client, runner)` running over an in-memory JSON-RPC dispatcher pair.

    Starts the client (echo handlers) and the server-side dispatcher loop
    (kernel `on_request`/`on_notify` + `aclose_shielded` teardown - the
    `serve_connection` shape) in a task group, wraps the body in
    `anyio.fail_after(5)`, and cancels on exit. When `initialized` is true the
    helper performs the real `initialize` request before yielding, so tests
    start past the init-gate via the public path.

    `connection` defaults to `Connection.for_loop(server_dispatcher)`. Pass a
    factory-built connection (e.g. `Connection.from_envelope(...)`) to exercise
    the born-ready path; the kernel reads it as a fact and is mode-agnostic.
    """
    client, server_d, close = jsonrpc_pair()
    assert isinstance(client, JSONRPCDispatcher) and isinstance(server_d, JSONRPCDispatcher)
    if connection is None:
        connection = Connection.for_loop(server_d)
    runner = ServerRunner(
        server=server,
        connection=connection,
        lifespan_state={},
        init_options=init_options,
        dispatch_middleware=dispatch_middleware or [],
    )
    c_req, c_notify = echo_handlers(Recorder())
    body_exc: BaseException | None = None

    async def _drive(*, task_status: anyio.abc.TaskStatus[None]) -> None:
        try:
            await server_d.run(runner.on_request, runner.on_notify, task_status=task_status)
        finally:
            await aclose_shielded(connection)

    async with anyio.create_task_group() as tg:
        await tg.start(client.run, c_req, c_notify)
        await tg.start(_drive)
        try:
            with anyio.fail_after(5):
                if initialized:
                    await client.send_raw_request("initialize", _initialize_params())
                yield client, runner
        except BaseException as e:
            # Capture and re-raise outside the task group so test failures
            # surface as the original exception, not an ExceptionGroup wrapper.
            body_exc = e
        close()
    if body_exc is not None:
        raise body_exc