Function test_websocket_iter_json

tests/test_websockets.py:197–208

(test_client_factory: TestClientFactory)

def test_websocket_iter_json(test_client_factory: TestClientFactory) -> None:
    async def app(scope: Scope, receive: Receive, send: Send) -> None:
        websocket = WebSocket(scope, receive=receive, send=send)
        await websocket.accept()
        async for data in websocket.iter_json():
            await websocket.send_json({class="st">"message": data})

    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">"message": {class="st">"hello": class="st">"world"}}


def test_websocket_concurrency_pattern(test_client_factory: TestClientFactory) -> None: