()
| 126 | |
| 127 | |
| 128 | async def test_websocket_completion() -> None: |
| 129 | # Ensure that the connecion callable returns on completion |
| 130 | app = Quart(__name__) |
| 131 | scope: WebsocketScope = { |
| 132 | "type": "websocket", |
| 133 | "asgi": {}, |
| 134 | "http_version": "1.1", |
| 135 | "scheme": "wss", |
| 136 | "path": "/", |
| 137 | "raw_path": b"/", |
| 138 | "query_string": b"", |
| 139 | "root_path": "", |
| 140 | "headers": [(b"host", b"quart")], |
| 141 | "client": ("127.0.0.1", 80), |
| 142 | "server": None, |
| 143 | "subprotocols": [], |
| 144 | "extensions": {"websocket.http.response": {}}, |
| 145 | "state": {}, # type: ignore[typeddict-item] |
| 146 | } |
| 147 | connection = ASGIWebsocketConnection(app, scope) |
| 148 | |
| 149 | queue: asyncio.Queue = asyncio.Queue() |
| 150 | queue.put_nowait({"type": "websocket.connect"}) |
| 151 | |
| 152 | async def receive() -> ASGIReceiveEvent: |
| 153 | # This will block after returning the first and only entry |
| 154 | return await queue.get() |
| 155 | |
| 156 | async def send(message: ASGISendEvent) -> None: |
| 157 | pass |
| 158 | |
| 159 | # This test fails if a timeout error is raised here |
| 160 | await asyncio.wait_for(connection(receive, send), timeout=1) |
| 161 | |
| 162 | |
| 163 | def test_http_path_from_absolute_target() -> None: |
nothing calls this directly
no test coverage detected
searching dependent graphs…