(self)
| 909 | self.assertIs(protocol._loop, self.loop) |
| 910 | |
| 911 | def test_multiple_drain(self): |
| 912 | # See https://github.com/python/cpython/issues/74116 |
| 913 | drained = 0 |
| 914 | |
| 915 | async def drainer(stream): |
| 916 | nonlocal drained |
| 917 | await stream._drain_helper() |
| 918 | drained += 1 |
| 919 | |
| 920 | async def main(): |
| 921 | loop = asyncio.get_running_loop() |
| 922 | stream = asyncio.streams.FlowControlMixin(loop) |
| 923 | stream.pause_writing() |
| 924 | loop.call_later(0.1, stream.resume_writing) |
| 925 | await asyncio.gather(*[drainer(stream) for _ in range(10)]) |
| 926 | self.assertEqual(drained, 10) |
| 927 | |
| 928 | self.loop.run_until_complete(main()) |
| 929 | |
| 930 | def test_drain_raises(self): |
| 931 | # See http://bugs.python.org/issue25441 |
nothing calls this directly
no test coverage detected