Function running_pair

tests/shared/test_dispatcher.py:61–85

Yield `(client, server, client_recorder, server_recorder)` with both `run()` loops live.

(
    factory: PairFactory,
    *,
    server_on_request: OnRequest | None = None,
    server_on_notify: OnNotify | None = None,
    client_on_request: OnRequest | None = None,
    client_on_notify: OnNotify | None = None,
    can_send_request: bool = True,
)

@asynccontextmanager
async def running_pair(
    factory: PairFactory,
    *,
    server_on_request: OnRequest | None = None,
    server_on_notify: OnNotify | None = None,
    client_on_request: OnRequest | None = None,
    client_on_notify: OnNotify | None = None,
    can_send_request: bool = True,
) -> AsyncIterator[tuple[Dispatcher[TransportContext], Dispatcher[TransportContext], Recorder, Recorder]]:
    """Yield `(client, server, client_recorder, server_recorder)` with both `run()` loops live."""
    client, server, close = factory(can_send_request=can_send_request)
    client_rec, server_rec = Recorder(), Recorder()
    c_req, c_notify = echo_handlers(client_rec)
    s_req, s_notify = echo_handlers(server_rec)
    try:
        async with anyio.create_task_group() as tg:
            await tg.start(client.run, client_on_request or c_req, client_on_notify or c_notify)
            await tg.start(server.run, server_on_request or s_req, server_on_notify or s_notify)
            try:
                yield client, server, client_rec, server_rec
            finally:
                tg.cancel_scope.cancel()
    finally:
        await resync_tracer()
        close()


@pytest.mark.anyio