Fast-forward time in asyncio Python by patching loop.time, loop.call_later, loop.call_at, and asyncio.sleep
Project description
aiofastforward
Fast-forward time in asyncio Python by patching loop.call_later, loop.call_at, loop.time, and asyncio.sleep. This allows you to test asynchronous code synchronously.
Inspired by AngularJS $timeout.$flush.
Installation
pip install aiofastforward
Usage
Patching is done through a context manager, similar to unittest.patch.
import asyncio
from aiofastforward import FastForward
loop = asyncio.get_event_loop()
with FastForward(loop) as forward:
# Call production function(s), that call asyncio.sleep, loop.call_later,
# loop.call_at, or loop.time
# ...
# Fast-forward time 1 second.
await forward(1)
# More production functions or assertions
# ...
Differences between aiofastforward.FastForward and asynctest.ClockedTestCase
There is overlap in functionality: both support fast-forwarding time in terms of loop.call_later and loop.call_at. However, there are properties that FastForward has that ClockedTestCase does not:
- FastForward is not coupled to any particular test framework. The only requirement is that the test code must be in an async function. If you wish, you can use FastForward in an asynctest.TestCase test.
- FastForward supports fast-forwarding asyncio.sleep.
- FastForward allows fast-forwarding time in any event loop, not just the one the test code runs in.
ClockedTestCase does have an advantage over FastForward, which may be important for some uses:
- ClockedTestCase supports Python 3.4 onwards, while FastForward supports Python 3.5.0 onwards.
Examples
asyncio.sleep
# Production code
async def sleeper(callback):
await asyncio.sleep(2)
callback(0)
# Test code
from unittest.mock import Mock, call
loop = asyncio.get_event_loop()
callback = Mock()
with aiofastforward.FastForward(loop) as forward:
asyncio.ensure_future(sleeper())
await forward(1)
self.assertEqual(callback.mock_calls, [])
await forward(1)
self.assertEqual(callback.mock_calls, [call(0)])
loop.call_later
# Production code
async def schedule_callback(loop, callback):
loop.call_later(1, callback, 0)
loop.call_later(2, callback, 1)
# Test code
from unittest.mock import Mock, call
loop = asyncio.get_event_loop()
with aiofastforward.FastForward(loop) as forward:
callback = Mock()
await schedule_callback(loop, callback)
await forward(1)
self.assertEqual(callback.mock_calls, [call(0)])
await forward(1)
self.assertEqual(callback.mock_calls, [call(0), call(1)])
loop.call_at
# Production code
async def schedule_callback(loop, callback):
now = loop.time()
loop.call_at(now + 1, callback, 0)
loop.call_at(now + 2, callback, 1)
# Test code
from unittest.mock import Mock, call
loop = asyncio.get_event_loop()
with aiofastforward.FastForward(loop) as forward:
callback = Mock()
await schedule_callback(loop, callback)
await forward(1)
self.assertEqual(callback.mock_calls, [call(0)])
await forward(1)
self.assertEqual(callback.mock_calls, [call(0), call(1)])
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for aiofastforward-0.0.16-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | f02e9f59361f089f4571f5ca623c2781094ffff4d77279bcce456fc0f81a6366 |
|
MD5 | 9cf377a30c3c1377be181833ec4de9b5 |
|
BLAKE2b-256 | c0b7d07f8eecf7ad97d96d0c8f7f56aaaa7de489070ad382e8459f2296a9f409 |