(self)
| 69 | |
| 70 | @coroutine_test |
| 71 | async def test_simple(self): |
| 72 | for length in [20, 50, 100]: |
| 73 | parallel_count = [0] |
| 74 | max_parallel_count = [0] |
| 75 | results = [] |
| 76 | ait = self.get_async_iterable(length) |
| 77 | await _parallel_asyncio( |
| 78 | ait, |
| 79 | self.CONCURRENT_ITEMS, |
| 80 | self.callable_wrapped, |
| 81 | results, |
| 82 | parallel_count, |
| 83 | max_parallel_count, |
| 84 | ) |
| 85 | assert list(range(length)) == sorted(results) |
| 86 | assert max_parallel_count[0] <= self.CONCURRENT_ITEMS |
| 87 | |
| 88 | @coroutine_test |
| 89 | async def test_delays(self): |
nothing calls this directly
no test coverage detected