Function test_cross_context

tests/test_driver.py:175–220

(app: App, driver: Driver)

    indirect=True,
)
async def test_cross_context(app: App, driver: Driver):
    assert isinstance(driver, ASGIMixin)

    ws: WebSocket | None = None
    ws_ready = anyio.Event()
    ws_should_close = anyio.Event()

    # create a background task before the ws connection established
    async def background_task():
        try:
            await ws_ready.wait()
            assert ws is not None

            await ws.send("ping")
            data = await ws.receive()
            assert data == "pong"
        finally:
            ws_should_close.set()

    async def _handle_ws(websocket: WebSocket) -> None:
        nonlocal ws
        await websocket.accept()
        ws = websocket
        ws_ready.set()

        await ws_should_close.wait()
        await websocket.close()

    ws_setup = WebSocketServerSetup(URL("/ws_test"), "ws_test", _handle_ws)
    driver.setup_websocket_server(ws_setup)

    async with anyio.create_task_group() as tg, app.test_server(driver.asgi) as ctx:
        tg.start_soon(background_task)

        client = ctx.get_client()

        async with client.websocket_connect("/ws_test") as websocket:
            try:
                data = await websocket.receive_text()
                assert data == "ping"
                await websocket.send_text("pong")
            except Exception as e:
                if not e.args or "websocket.close" not in str(e.args[0]):
                    raise

    await anyio.sleep(1)


@pytest.mark.anyio