(self)
| 715 | await f(lock, asyncio.Condition(lock)) |
| 716 | |
| 717 | async def test_ambiguous_loops(self): |
| 718 | loop = asyncio.new_event_loop() |
| 719 | self.addCleanup(loop.close) |
| 720 | |
| 721 | async def wrong_loop_in_lock(): |
| 722 | with self.assertRaises(TypeError): |
| 723 | asyncio.Lock(loop=loop) # actively disallowed since 3.10 |
| 724 | lock = asyncio.Lock() |
| 725 | lock._loop = loop # use private API for testing |
| 726 | async with lock: |
| 727 | # acquired immediately via the fast-path |
| 728 | # without interaction with any event loop. |
| 729 | cond = asyncio.Condition(lock) |
| 730 | # cond.acquire() will trigger waiting on the lock |
| 731 | # and it will discover the event loop mismatch. |
| 732 | with self.assertRaisesRegex( |
| 733 | RuntimeError, |
| 734 | "is bound to a different event loop", |
| 735 | ): |
| 736 | await cond.acquire() |
| 737 | |
| 738 | async def wrong_loop_in_cond(): |
| 739 | # Same analogy here with the condition's loop. |
| 740 | lock = asyncio.Lock() |
| 741 | async with lock: |
| 742 | with self.assertRaises(TypeError): |
| 743 | asyncio.Condition(lock, loop=loop) |
| 744 | cond = asyncio.Condition(lock) |
| 745 | cond._loop = loop |
| 746 | with self.assertRaisesRegex( |
| 747 | RuntimeError, |
| 748 | "is bound to a different event loop", |
| 749 | ): |
| 750 | await cond.wait() |
| 751 | |
| 752 | await wrong_loop_in_lock() |
| 753 | await wrong_loop_in_cond() |
| 754 | |
| 755 | async def test_timeout_in_block(self): |
| 756 | condition = asyncio.Condition() |
nothing calls this directly
no test coverage detected