| 261 | |
| 262 | @pytest.mark.anyio |
| 263 | async def test_client_disconnect_on_send() -> None: |
| 264 | async def app(scope: Scope, receive: Receive, send: Send) -> None: |
| 265 | websocket = WebSocket(scope, receive=receive, send=send) |
| 266 | await websocket.accept() |
| 267 | await websocket.send_text("Hello, world!") |
| 268 | |
| 269 | async def receive() -> Message: |
| 270 | return {"type": "websocket.connect"} |
| 271 | |
| 272 | async def send(message: Message) -> None: |
| 273 | if message["type"] == "websocket.accept": |
| 274 | return |
| 275 | # Simulate the exception the server would send to the application when the client disconnects. |
| 276 | raise OSError |
| 277 | |
| 278 | with pytest.raises(WebSocketDisconnect) as ctx: |
| 279 | await app({"type": "websocket", "path": "/"}, receive, send) |
| 280 | assert ctx.value.code == status.WS_1006_ABNORMAL_CLOSURE |
| 281 | |
| 282 | |
| 283 | def test_application_close(test_client_factory: TestClientFactory) -> None: |