(ws_protocol_cls: WSProtocol, http_protocol_cls: HTTPProtocol, unused_tcp_port: int)
| 514 | |
| 515 | |
| 516 | async def test_client_close(ws_protocol_cls: WSProtocol, http_protocol_cls: HTTPProtocol, unused_tcp_port: int): |
| 517 | disconnect_message: WebSocketDisconnectEvent | None = None |
| 518 | |
| 519 | async def app(scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable): |
| 520 | nonlocal disconnect_message |
| 521 | while True: |
| 522 | message = await receive() |
| 523 | if message["type"] == "websocket.connect": |
| 524 | await send({"type": "websocket.accept"}) |
| 525 | elif message["type"] == "websocket.receive": |
| 526 | pass |
| 527 | elif message["type"] == "websocket.disconnect": |
| 528 | disconnect_message = message |
| 529 | break |
| 530 | |
| 531 | async def websocket_session(url: str): |
| 532 | async with websockets.client.connect(url) as websocket: |
| 533 | await websocket.ping() |
| 534 | await websocket.send("abc") |
| 535 | await websocket.close(code=1001, reason="custom reason") |
| 536 | |
| 537 | config = Config(app=app, ws=ws_protocol_cls, http=http_protocol_cls, lifespan="off", port=unused_tcp_port) |
| 538 | async with run_server(config): |
| 539 | await websocket_session(f"ws://127.0.0.1:{unused_tcp_port}") |
| 540 | |
| 541 | assert disconnect_message == {"type": "websocket.disconnect", "code": 1001, "reason": "custom reason"} |
| 542 | |
| 543 | |
| 544 | async def test_client_connection_lost( |
nothing calls this directly
no test coverage detected