Method get_async_iterable_with_delays

tests/test_utils_defer.py:188–198

(length: int)

    @staticmethod
    async def get_async_iterable_with_delays(length: int) -> AsyncGenerator[int, None]:
        class="cm"># simulate a callback with delays between some of the results
        from twisted.internet import reactor

        for i in range(length):
            if random.random() < 0.1:
                dfd: Deferred[None] = Deferred()
                delay = random.random() / 20
                reactor.callLater(delay, dfd.callback, None)
                await maybe_deferred_to_future(dfd)
            yield i

    @inline_callbacks_test
    def test_simple(self):