Skip to content

add advance_time fixture and test (closes #83, #95, #96 #110) #113

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 9 commits into
base: main
Choose a base branch
from
Prev Previous commit
Next Next commit
add advance_time fixture and test (closes #83, #95, #96 #110)
  • Loading branch information
derekbrokeit committed Feb 18, 2019
commit ddc08ba529dde33aab68c5ed81edd2a345576dc7
43 changes: 43 additions & 0 deletions pytest_asyncio/plugin.py
Original file line number Diff line number Diff line change
@@ -201,6 +201,44 @@ def pytest_runtest_setup(item):
}


class EventLoopClockAdvancer:
"""
A helper object that when called will advance the event loop's time. If the
call is awaited, the caller task will wait an iteration for the update to
wake up any awaiting handlers.
"""
__slots__ = ("offset", "loop", "_base_time",)

def __init__(self, loop):
self.offset = 0.0
self._base_time = loop.time
self.loop = loop

# incorporate offset timing into the event loop
self.loop.time = self.time

def time(self):
"""
Return the time according to the event loop's clock. The time is
adjusted by an offset.
"""
return self._base_time() + self.offset

def __call__(self, seconds):
"""
Advance time by a given offset in seconds. Returns an awaitable
that will complete after all tasks scheduled for after advancement
of time are proceeding.
"""
if seconds > 0:
# advance the clock by the given offset
self.offset += abs(seconds)

# Once the clock is adjusted, new tasks may have just been
# scheduled for running in the next pass through the event loop
return self.loop.create_task(asyncio.sleep(0))


@pytest.yield_fixture
def event_loop(request):
"""Create an instance of the default event loop for each test case."""
@@ -237,3 +275,8 @@ def factory():

return port
return factory


@pytest.fixture
def advance_time(event_loop):
return EventLoopClockAdvancer(event_loop)
21 changes: 21 additions & 0 deletions tests/test_simple_35.py
Original file line number Diff line number Diff line change
@@ -86,3 +86,24 @@ async def test_asyncio_marker_method(self, event_loop):
def test_async_close_loop(event_loop):
event_loop.close()
return 'ok'


@pytest.mark.asyncio
async def test_advance_time_fixture(event_loop, advance_time):
"""
Test the `advance_time` fixture
"""
# A task is created that will sleep some number of seconds
SLEEP_TIME = 10

# create the task
task = event_loop.create_task(asyncio.sleep(SLEEP_TIME))
assert not task.done()

# start the task
await advance_time(0)
assert not task.done()

# process the timeout
await advance_time(SLEEP_TIME)
assert task.done()