No good way to detect when a websocket is disconnected by the server in tests

[jamesh1999]

I have the following use-case: I'm writing auth tests for a WebSocket service. When the client has insufficient permissions for what it's trying, the connection should be closed by the server with the correct status code. If the service keeps the connection open, I have no guarantee of regular traffic.

Of course, I can do the following:

with pytest.raises(WebSocketDisconnect) as e:
    websocket.receive_text()
# Do something here to check exit code / reason

However, this will block indefinitely if the server did not close the connection - not acceptable for a test suite that could run in CI without a person monitoring.

I use anyio in my codebase so I have tried move_on_after and fail_after. Unfortunately none of these can interrupt the receive method and return execution if the server did not close the connection.

#2195 #2329 #1140 would have made this possible by adding timeouts. Note: most of the justifications given for not implementing these features do not apply in my use case. There is nothing 'flakey' about the tests. It is simply impossible to check whether the connection has been closed without potentially blocking indefinitely.

#2897 would also be viable as it would enable anyio cancellation of the receive call.

Open to suggestions for alternative solutions but I think this is a pretty big limitation in the TestClient that necessitates looking at ASGI messages directly

[jamesh1999]

I should add that this is what I am doing currently. It works for now but has some serious limitations:

async def safe_check_closed(ws: WebSocketTestSession, timeout=1) -> WebSocketDisconnect | None: 
    """
    There is NO way to check if a WebSocket is closed without
    risking an infinitely-blocking receive_text() call.

    This does its best to avoid that by running the check in a separate thread.
    However, the threads may linger because they cannot be truly cancelled.
    """

    def checker():
        try:
            ws.receive_text()
        except WebSocketDisconnect as e:
            return e
        return None

    with move_on_after(timeout):
        return await to_thread.run_sync(checker, cancellable=True)
    return None

It makes it possible to use anyio's move_on_after however, the spawned threads cannot be cancelled. They will remain executing (waiting) until an error is produced internally. This trades being able to tell whether a socket is closed for a resource leak