aiofastforward 0.0.8

Fast-forward time in asyncio Python by patching loop.time, loop.call_later, loop.call_at, and asyncio.sleep

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
import aiofastforward

loop = asyncio.get_event_loop()
with aiofastforward.FastForward(loop) as forward:
    # Call production function(s)
    # ...

    # Fast-forward time 1 second.
    await forward(1)

    # More production functions or assertions
    # ...

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)])