(self)
| 387 | await communicator.wait() |
| 388 | |
| 389 | async def test_disconnect(self): |
| 390 | application = get_asgi_application() |
| 391 | scope = self.async_request_factory._base_scope(path="/") |
| 392 | communicator = ApplicationCommunicator(application, scope) |
| 393 | await communicator.send_input({"type": "http.disconnect"}) |
| 394 | with self.assertRaises(asyncio.TimeoutError): |
| 395 | await communicator.receive_output() |
| 396 | |
| 397 | async def test_disconnect_both_return(self): |
| 398 | # Force both the disconnect listener and the task that sends the |
nothing calls this directly
no test coverage detected