(loop, conn)
| 1194 | |
| 1195 | |
| 1196 | async def test_data_stream_exc(loop, conn) -> None: |
| 1197 | fut = loop.create_future() |
| 1198 | |
| 1199 | async def gen(): |
| 1200 | yield b"binary data" |
| 1201 | await fut |
| 1202 | |
| 1203 | req = ClientRequest("POST", URL("http://python.org/"), data=gen(), loop=loop) |
| 1204 | assert req.chunked |
| 1205 | assert req.headers["TRANSFER-ENCODING"] == "chunked" |
| 1206 | |
| 1207 | async def throw_exc(): |
| 1208 | await asyncio.sleep(0.01) |
| 1209 | fut.set_exception(ValueError) |
| 1210 | |
| 1211 | loop.create_task(throw_exc()) |
| 1212 | |
| 1213 | async with await req.send(conn): |
| 1214 | assert req._writer is not None |
| 1215 | await req._writer |
| 1216 | # assert conn.close.called |
| 1217 | assert conn.protocol is not None |
| 1218 | assert conn.protocol.set_exception.called |
| 1219 | await req.close() |
| 1220 | |
| 1221 | |
| 1222 | async def test_data_stream_exc_chain(loop, conn) -> None: |
nothing calls this directly
no test coverage detected