(self)
| 293 | self.assertEqual({reader2.result(), reader3.result()}, {1, 2}) |
| 294 | |
| 295 | async def test_put_cancel_drop(self): |
| 296 | q = asyncio.Queue(1) |
| 297 | |
| 298 | q.put_nowait(1) |
| 299 | |
| 300 | # putting a second item in the queue has to block (qsize=1) |
| 301 | writer = asyncio.create_task(q.put(2)) |
| 302 | await asyncio.sleep(0) |
| 303 | |
| 304 | value1 = q.get_nowait() |
| 305 | self.assertEqual(value1, 1) |
| 306 | |
| 307 | writer.cancel() |
| 308 | try: |
| 309 | await writer |
| 310 | except asyncio.CancelledError: |
| 311 | # try again |
| 312 | writer = asyncio.create_task(q.put(2)) |
| 313 | await writer |
| 314 | |
| 315 | value2 = q.get_nowait() |
| 316 | self.assertEqual(value2, 2) |
| 317 | self.assertEqual(q.qsize(), 0) |
| 318 | |
| 319 | def test_nonblocking_put_exception(self): |
| 320 | q = asyncio.Queue(maxsize=1, ) |
nothing calls this directly
no test coverage detected