(self)
| 750 | assert not f.intercepted |
| 751 | |
| 752 | async def test_wait_for_resume(self): |
| 753 | f = tflow() |
| 754 | await f.wait_for_resume() |
| 755 | |
| 756 | f = tflow() |
| 757 | f.intercept() |
| 758 | f.resume() |
| 759 | await f.wait_for_resume() |
| 760 | |
| 761 | f = tflow() |
| 762 | f.intercept() |
| 763 | with pytest.raises(asyncio.TimeoutError): |
| 764 | await asyncio.wait_for(f.wait_for_resume(), 0.2) |
| 765 | f.resume() |
| 766 | await f.wait_for_resume() |
| 767 | |
| 768 | f = tflow() |
| 769 | f.intercept() |
| 770 | with pytest.raises(asyncio.TimeoutError): |
| 771 | await asyncio.wait_for(f.wait_for_resume(), 0.2) |
| 772 | f.resume() |
| 773 | f.intercept() |
| 774 | with pytest.raises(asyncio.TimeoutError): |
| 775 | await asyncio.wait_for(f.wait_for_resume(), 0.2) |
| 776 | f.resume() |
| 777 | await f.wait_for_resume() |
| 778 | |
| 779 | def test_resume_duplicated(self): |
| 780 | f = tflow() |
nothing calls this directly
no test coverage detected