Function test_websocket_blocking_receive

tests/test_testclient.py:233–255

(test_client_factory: TestClientFactory)

def test_websocket_blocking_receive(test_client_factory: TestClientFactory) -> None:
    def app(scope: Scope) -> ASGIInstance:
        async def respond(websocket: WebSocket) -> None:
            await websocket.send_json({class="st">"message": class="st">"test"})

        async def asgi(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(respond, websocket)
                try:
                    class="cm"># this will block as the client does not send us data
                    class="cm"># it should not prevent `respond` from executing though
                    await websocket.receive_json()
                except WebSocketDisconnect:
                    pass

        return asgi

    client = test_client_factory(app)  class="cm"># type: ignore
    with client.websocket_connect(class="st">"/") as websocket:
        data = websocket.receive_json()
        assert data == {class="st">"message": class="st">"test"}


def test_websocket_not_block_on_close(test_client_factory: TestClientFactory) -> None: