(loop, conn)
| 1333 | |
| 1334 | |
| 1335 | async def test_oserror_on_write_bytes(loop, conn) -> None: |
| 1336 | req = ClientRequest("POST", URL("http://python.org/"), loop=loop) |
| 1337 | req.body = b"test data" # type: ignore[assignment] # https://github.com/python/mypy/issues/12892 |
| 1338 | |
| 1339 | writer = WriterMock() |
| 1340 | writer.write.side_effect = OSError |
| 1341 | |
| 1342 | await req.write_bytes(writer, conn, None) |
| 1343 | |
| 1344 | assert conn.protocol.set_exception.called |
| 1345 | exc = conn.protocol.set_exception.call_args[0][0] |
| 1346 | assert isinstance(exc, aiohttp.ClientOSError) |
| 1347 | |
| 1348 | await req.close() |
| 1349 | |
| 1350 | |
| 1351 | @pytest.mark.skipif(sys.version_info < (3, 11), reason="Needs Task.cancelling()") |
nothing calls this directly
no test coverage detected