Function test_websocket_concurrency_pattern

tests/test_websockets.py:211–238

(test_client_factory: TestClientFactory)

def test_websocket_concurrency_pattern(test_client_factory: TestClientFactory) -> None:
    stream_send: ObjectSendStream[MutableMapping[str, Any]]
    stream_receive: ObjectReceiveStream[MutableMapping[str, Any]]
    stream_send, stream_receive = anyio.create_memory_object_stream()

    async def reader(websocket: WebSocket) -> None:
        async with stream_send:
            async for data in websocket.iter_json():
                await stream_send.send(data)

    async def writer(websocket: WebSocket) -> None:
        async with stream_receive:
            async for message in stream_receive:
                await websocket.send_json(message)

    async def app(scope: Scope, receive: Receive, send: Send) -> None:
        websocket = WebSocket(scope, receive=receive, send=send)
        await websocket.accept()
        async with anyio.create_task_group() as task_group:
            task_group.start_soon(reader, websocket)
            await writer(websocket)
        await websocket.close()

    client = test_client_factory(app)
    with client.websocket_connect(class="st">"/") as websocket:
        websocket.send_json({class="st">"hello": class="st">"world"})
        data = websocket.receive_json()
        assert data == {class="st">"hello": class="st">"world"}


def test_client_close(test_client_factory: TestClientFactory) -> None: