(self)
| 368 | await communicator.wait() |
| 369 | |
| 370 | async def test_get_query_string(self): |
| 371 | application = get_asgi_application() |
| 372 | for query_string in (b"name=Andrew", "name=Andrew"): |
| 373 | with self.subTest(query_string=query_string): |
| 374 | scope = self.async_request_factory._base_scope( |
| 375 | path="/", |
| 376 | query_string=query_string, |
| 377 | ) |
| 378 | communicator = ApplicationCommunicator(application, scope) |
| 379 | await communicator.send_input({"type": "http.request"}) |
| 380 | response_start = await communicator.receive_output() |
| 381 | self.assertEqual(response_start["type"], "http.response.start") |
| 382 | self.assertEqual(response_start["status"], 200) |
| 383 | response_body = await communicator.receive_output() |
| 384 | self.assertEqual(response_body["type"], "http.response.body") |
| 385 | self.assertEqual(response_body["body"], b"Hello Andrew!") |
| 386 | # Allow response.close() to finish |
| 387 | await communicator.wait() |
| 388 | |
| 389 | async def test_disconnect(self): |
| 390 | application = get_asgi_application() |
nothing calls this directly
no test coverage detected