aiounittest 1.0.1

AsyncTestCase

import asyncio
import aiounittest


async def add(x, y):
    await asyncio.sleep(0.1)
    return x + y

class MyTest(aiounittest.AsyncTestCase):

    async def test_async_add(self):
        ret = await add(5, 6)
        self.assertEqual(ret, 11)

    # or 3.4 way
    @asyncio.coroutine
    def test_sleep(self):
        ret = yield from add(5, 6)
        self.assertEqual(ret, 11)

    # some regular test code
    def test_something(self):
        self.assertTrue(true)

By default AsyncTestCase use the main event loop and cleans it ups on every test. If you want to change that behaiour override the AsyncTestCase.get_event_loop method.

futurized

This helper wraps object in the asyncio’s Future. It can be used to mock coroutines. Decorate any kind of object with it and pass to unittest.mock.Mock’s (MagicMock as well) return_value or side_effect. If the given object is an Exception (or its sublcass), futurized will treat it accordingly and exception will be raised upon await.

Some dummy module:

# dummy_math.py

from asyncio import sleep

async def add(x, y):
    await sleep(666)
    return x + y

And the test with mocked asyncio.sleep:

from aiounittest import futurized, AsyncTestCase
from unittest.mock import Mock, patch

import dummy_math

class MyTest(AsyncTestCase):

    def tearDown(self):
        super().tearDown()
        patch.stopall()

    async def test_add(self):
        mock_sleep = Mock(return_value=futurized('whatever'))
        patch('dummy_math.sleep', mock_sleep).start()
        ret = await dummy_math.add(5, 6)
        self.assertEqual(ret, 11)

    async def test_fail(self):
        mock_sleep = Mock(return_value=futurized(Exception('whatever')))
        patch('dummy_math.sleep', mock_sleep).start()
        with self.assertRaises(Exception) as e:
            await dummy_math.add(5, 6)

async_test

It’s a decorator that let you to use the standard unittest.TestCase and run asynchronous tests, just decorate them.

import asyncio
import unittest
from aiounittest import async_test


async def add(x, y):
    await asyncio.sleep(0.1)
    return x + y

class MyTest(unittest.TestCase):

    @async_test
    async def test_async_add(self):
        ret = await add(5, 6)
        self.assertEqual(ret, 11)

You can also pass the custom event loop with loop argument. Note if the loop is provided, decorator won’t close it.

import asyncio
from aiounittest import async_test


my_loop = asycio.get_event_loop()

@async_test(loop=my_loop)
async def my_func():
    await asyncio.sleep(1)
    return 'OK'

# now call the function in the sync manner
my_func()
assert loop.is_closed() == False

Above examples also shows a handy way to call async code in sync way. If the name is a problem, the same functionality provide aiounittest.helpers.run_sync. For example:

import asyncio
from aiounittest.helpers import run_sync

# using decorator syntax

@run_sync
async def my_func1():
    await asyncio.sleep(1)
    return 'OK1'

ret = my_func1()
print(ret)

# using as a wrapper

async def my_func2():
    await asyncio.sleep(1)
    return 'OK2'

sync_my_func2 = run_sync(my_func2)
ret = sync_my_func2()
print(ret)