How to deal with asyncio.sleep?

[fish-face]

I'm not sure whether this is a missing feature or if I'm just missing how to do it. I have some async code I wish to test which schedules something to happen by creating an asyncio task on the event loop which awaits asyncio.sleep() for the right amount of time, then does something.

In the test if I freeze time, initiate this then tick by the amount the task is sleeping for (or say, one second extra to allow for inaccuracies) the event never goes off. On the other hand if I don't use freezegun and just time.sleep() for the same amount of time, it does.

Is there anything special I should or can do to get this to work? asyncio of course uses its own monotonic timer, and my test is running synchronously so I don't know if something special needs to be done to allow the event loop to notice that time has advanced. However I don't do anything special to make this work with time.sleep().

For some more details: This is within a django unittest test, using a django-channels communicator to try and retrieve websocket data. After the time has elapsed I am using the test's own event loop (obtained using asyncio.get_event_loop()) to run the communicator's receive_json_from() method. We are successfully using freezegun in other tests.

[boxed]

Could you supply a minimal test case? I'm not very versed in asyncio stuff so I'm not sure I can reproduce just given your description.

[fish-face]

Something like the following:

def schedule_thing(event_loop):
    event_loop.create_task(do_thing(1000))

async def do_thing(delay):
    await asyncio.sleep(delay)
    # await really_do_thing()

# tests.py

class ThingTests(TestCase):
    def test_scheduling_things():
        # get appropriate event loop
        schedule_thing(event_loop)

        with freezegun.freeze_time() as frozen_datetime:
            frozen_datetime.tick(1500)
            # self.assertTrue(thing_happened)

At the moment the only way to do this is by actually waiting that amount of time and hoping - and so your tests get very slow, and unreliable as well if something affects the timing (which is sensitive because you're trying to get it as low as possible to be as fast as possible!)

[spapinistarkware]

I have a suggestion:

import asyncio
from selectors import DefaultSelector
import queue


class FFSelector(DefaultSelector):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._current_time = 0

    def select(self, timeout):
        # There are tasks to be scheduled. Continue simulating.
        self._current_time += timeout
        return DefaultSelector.select(self, 0)


class FFEventLoop(asyncio.SelectorEventLoop):  # type: ignore
    def __init__(self):
        super().__init__(selector=FFSelector())

    def time(self):
        return self._selector._current_time

This event loop simulates time. You can also add a check in select() method, if we passed a target time, and stop there.