()
| 46 | |
| 47 | |
| 48 | async def test_http_completion() -> None: |
| 49 | # Ensure that the connecion callable returns on completion |
| 50 | app = Quart(__name__) |
| 51 | scope: HTTPScope = { |
| 52 | "type": "http", |
| 53 | "asgi": {}, |
| 54 | "http_version": "1.1", |
| 55 | "method": "GET", |
| 56 | "scheme": "https", |
| 57 | "path": "/", |
| 58 | "raw_path": b"/", |
| 59 | "query_string": b"", |
| 60 | "root_path": "", |
| 61 | "headers": [(b"host", b"quart")], |
| 62 | "client": ("127.0.0.1", 80), |
| 63 | "server": None, |
| 64 | "extensions": {}, |
| 65 | "state": {}, # type: ignore[typeddict-item] |
| 66 | } |
| 67 | connection = ASGIHTTPConnection(app, scope) |
| 68 | |
| 69 | queue: asyncio.Queue = asyncio.Queue() |
| 70 | queue.put_nowait({"type": "http.request", "body": b"", "more_body": False}) |
| 71 | |
| 72 | async def receive() -> ASGIReceiveEvent: |
| 73 | # This will block after returning the first and only entry |
| 74 | return await queue.get() |
| 75 | |
| 76 | async def send(message: ASGISendEvent) -> None: |
| 77 | pass |
| 78 | |
| 79 | # This test fails if a timeout error is raised here |
| 80 | await asyncio.wait_for(connection(receive, send), timeout=1) |
| 81 | |
| 82 | |
| 83 | @pytest.mark.parametrize( |
nothing calls this directly
no test coverage detected
searching dependent graphs…