Clocked test case

[lromor]

Hello,
how would you achieve something similar to this:
https://asynctest.readthedocs.io/en/latest/asynctest.case.html#asynctest.ClockedTestCase

with pytest asyncio?

-l

[albertodonato]

I had the same need, I'm using something like this, which is a straight copy of ClockedTestCase logic:

@pytest.fixture
async def advance_time(event_loop):

    class Clocker:

        time: float = 0

        def __init__(self, loop: asyncio.BaseEventLoop):
            self.loop = loop
            self.loop.time = lambda: self.time

        async def advance(self, seconds: float):
            if seconds < 0:
                raise ValueError('Advance time must not be negative')
            await self._drain_loop()

            target_time = self.time + seconds
            while True:
                next_time = self._next_scheduled()
                if next_time is None or next_time > target_time:
                    break

                self.time = next_time
                await self._drain_loop()

            self.time = target_time
            await self._drain_loop()

        def _next_scheduled(self):
            try:
                return self.loop._scheduled[0]._when
            except IndexError:
                return None

        async def _drain_loop(self):
            while True:
                next_time = self._next_scheduled()
                if (not self.loop._ready
                        and (next_time is None or next_time > self.time)):
                    break
                await asyncio.sleep(0)

    yield Clocker(event_loop).advance

in tests you can use:

@pytest.mark.asyncio
async def test_foo(advance_time):
    await do_stuff()
    await advance_time(2)
    await more_stuff()

[albertodonato]

It would be nice to have something like this builtin