(loop: asyncio.AbstractEventLoop, conn: mock.Mock)
| 1365 | |
| 1366 | |
| 1367 | async def test_terminate(loop: asyncio.AbstractEventLoop, conn: mock.Mock) -> None: |
| 1368 | req = ClientRequest("get", URL("http://python.org"), loop=loop) |
| 1369 | |
| 1370 | async def _mock_write_bytes(*args, **kwargs): |
| 1371 | # Ensure the task is scheduled |
| 1372 | await asyncio.sleep(0) |
| 1373 | |
| 1374 | with mock.patch.object(req, "write_bytes", _mock_write_bytes): |
| 1375 | resp = await req.send(conn) |
| 1376 | |
| 1377 | assert req._writer is not None |
| 1378 | assert resp._writer is not None |
| 1379 | await resp._writer |
| 1380 | writer = WriterMock() |
| 1381 | writer.done = mock.Mock(return_value=False) |
| 1382 | writer.cancel = mock.Mock() |
| 1383 | req._writer = writer |
| 1384 | resp._writer = writer |
| 1385 | |
| 1386 | assert req._writer is not None |
| 1387 | assert resp._writer is not None |
| 1388 | req.terminate() |
| 1389 | writer.cancel.assert_called_with() |
| 1390 | writer.done.assert_called_with() |
| 1391 | resp.close() |
| 1392 | |
| 1393 | await req.close() |
| 1394 | |
| 1395 | |
| 1396 | def test_terminate_with_closed_loop(loop, conn) -> None: |
nothing calls this directly
no test coverage detected