(
ws_protocol_cls: WSProtocol,
http_protocol_cls: HTTPProtocol,
unused_tcp_port: int,
code: int | None,
reason: str | None,
)
| 478 | @pytest.mark.parametrize("code", [None, 1000, 1001]) |
| 479 | @pytest.mark.parametrize("reason", [None, "test", False], ids=["none_as_reason", "normal_reason", "without_reason"]) |
| 480 | async def test_app_close( |
| 481 | ws_protocol_cls: WSProtocol, |
| 482 | http_protocol_cls: HTTPProtocol, |
| 483 | unused_tcp_port: int, |
| 484 | code: int | None, |
| 485 | reason: str | None, |
| 486 | ): |
| 487 | async def app(scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable): |
| 488 | while True: |
| 489 | message = await receive() |
| 490 | if message["type"] == "websocket.connect": |
| 491 | await send({"type": "websocket.accept"}) |
| 492 | elif message["type"] == "websocket.receive": |
| 493 | reply: WebSocketCloseEvent = {"type": "websocket.close"} |
| 494 | |
| 495 | if code is not None: |
| 496 | reply["code"] = code |
| 497 | |
| 498 | if reason is not False: |
| 499 | reply["reason"] = reason |
| 500 | |
| 501 | await send(reply) |
| 502 | elif message["type"] == "websocket.disconnect": |
| 503 | break |
| 504 | |
| 505 | config = Config(app=app, ws=ws_protocol_cls, http=http_protocol_cls, lifespan="off", port=unused_tcp_port) |
| 506 | async with run_server(config): |
| 507 | async with websockets.client.connect(f"ws://127.0.0.1:{unused_tcp_port}") as websocket: |
| 508 | await websocket.ping() |
| 509 | await websocket.send("abc") |
| 510 | with pytest.raises(websockets.exceptions.ConnectionClosed): |
| 511 | await websocket.recv() |
| 512 | assert websocket.close_code == (code or 1000) |
| 513 | assert websocket.close_reason == (reason or "") |
| 514 | |
| 515 | |
| 516 | async def test_client_close(ws_protocol_cls: WSProtocol, http_protocol_cls: HTTPProtocol, unused_tcp_port: int): |
nothing calls this directly
no test coverage detected