How to test Python 3.4 asyncio code?

What's the best way to write unit tests for code using the Python 3.4 asyncio library? Assume I want to test a TCP client (SocketConnection):

import asyncio
import unittest

class TestSocketConnection(unittest.TestCase):
    def setUp(self):
        self.mock_server = MockServer("localhost", 1337)
        self.socket_connection = SocketConnection("localhost", 1337)

    @asyncio.coroutine
    def test_sends_handshake_after_connect(self):
        yield from self.socket_connection.connect()
        self.assertTrue(self.mock_server.received_handshake())

When running this test case with the default test runner, the test will always succeed as the method executes only up until the first yield from instruction, after which it returns before executing any assertions. This causes tests to always succeed.

Is there a prebuilt test runner that is able to handle asynchronous code like this?

Since Python 3.8 unittest comes with the IsolatedAsyncioTestCase function, designed for this purpose.

from unittest import IsolatedAsyncioTestCase

class Test(IsolatedAsyncioTestCase):

    async def test_functionality(self):
        result = await functionality()
        self.assertEqual(expected, result)

I temporarily solved the problem using a decorator inspired by Tornado's gen_test:

def async_test(f):
    def wrapper(*args, **kwargs):
        coro = asyncio.coroutine(f)
        future = coro(*args, **kwargs)
        loop = asyncio.get_event_loop()
        loop.run_until_complete(future)
    return wrapper

Like J.F. Sebastian suggested, this decorator will block until the test method coroutine has finished. This allows me to write test cases like this:

class TestSocketConnection(unittest.TestCase):
    def setUp(self):
        self.mock_server = MockServer("localhost", 1337)
        self.socket_connection = SocketConnection("localhost", 1337)

    @async_test
    def test_sends_handshake_after_connect(self):
        yield from self.socket_connection.connect()
        self.assertTrue(self.mock_server.received_handshake())

This solution probably misses some edge cases.

I think a facility like this should added to Python's standard library to make asyncio and unittest interaction more convenient out of the box.

async_test, suggested by Marvin Killing, definitely can help -- as well as direct calling loop.run_until_complete()

But I also strongly recommend to recreate new event loop for every test and directly pass loop to API calls (at least asyncio itself accepts loop keyword-only parameter for every call that need it).

Like

class Test(unittest.TestCase):
    def setUp(self):
        self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(None)

    def test_xxx(self):
        @asyncio.coroutine
        def go():
            reader, writer = yield from asyncio.open_connection(
                '127.0.0.1', 8888, loop=self.loop)
            yield from asyncio.sleep(0.01, loop=self.loop)
        self.loop.run_until_complete(go())

that isolates tests in test case and prevents strange errors like longstanding coroutine that has been created in test_a but finished only on test_b execution time.

Really like the async_test wrapper mentioned in https://stackoverflow.com/a/23036785/350195, here is an updated version for Python 3.5+

def async_test(coro):
    def wrapper(*args, **kwargs):
        loop = asyncio.new_event_loop()
        try:
            return loop.run_until_complete(coro(*args, **kwargs))
        finally:
            loop.close()
    return wrapper



class TestSocketConnection(unittest.TestCase):
    def setUp(self):
        self.mock_server = MockServer("localhost", 1337)
        self.socket_connection = SocketConnection("localhost", 1337)

    @async_test
    async def test_sends_handshake_after_connect(self):
        await self.socket_connection.connect()
        self.assertTrue(self.mock_server.received_handshake())

pytest-asyncio looks promising:

@pytest.mark.asyncio
async def test_some_asyncio_code():
    res = await library.do_something()
    assert b'expected result' == res

Use this class instead of unittest.TestCase base class:

import asyncio
import unittest


class AioTestCase(unittest.TestCase):

    # noinspection PyPep8Naming
    def __init__(self, methodName='runTest', loop=None):
        self.loop = loop or asyncio.get_event_loop()
        self._function_cache = {}
        super(AioTestCase, self).__init__(methodName=methodName)

    def coroutine_function_decorator(self, func):
        def wrapper(*args, **kw):
            return self.loop.run_until_complete(func(*args, **kw))
        return wrapper

    def __getattribute__(self, item):
        attr = object.__getattribute__(self, item)
        if asyncio.iscoroutinefunction(attr):
            if item not in self._function_cache:
                self._function_cache[item] = self.coroutine_function_decorator(attr)
            return self._function_cache[item]
        return attr


class TestMyCase(AioTestCase):

    async def test_dispatch(self):
        self.assertEqual(1, 1)

EDIT 1:

Please note the @Nitay answer about nested tests.

You can also use aiounittest that takes similar approach as @Andrew Svetlov, @Marvin Killing answers and wrap it in easy to use AsyncTestCase class:

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)

As you can see the async case is handled by AsyncTestCase. It supports also synchronous test. There is a possibility to provide custom event loop, just override AsyncTestCase.get_event_loop.

If you prefer (for some reason) the other TestCase class (eg unittest.TestCase), you might use async_test decorator:

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)

I usually define my async tests as coroutines and use a decorator for "syncing" them:

import asyncio
import unittest

def sync(coro):
    def wrapper(*args, **kwargs):
        loop = asyncio.get_event_loop()
        loop.run_until_complete(coro(*args, **kwargs))
    return wrapper

class TestSocketConnection(unittest.TestCase):
    def setUp(self):
        self.mock_server = MockServer("localhost", 1337)
        self.socket_connection = SocketConnection("localhost", 1337)

    @sync
    async def test_sends_handshake_after_connect(self):
        await self.socket_connection.connect()
        self.assertTrue(self.mock_server.received_handshake())

pylover answer is correct and is something that should be added to unittest IMO.

I would add in a slight change to support nested async tests:

class TestCaseBase(unittest.TestCase):
    # noinspection PyPep8Naming
    def __init__(self, methodName='runTest', loop=None):
        self.loop = loop or asyncio.get_event_loop()
        self._function_cache = {}
        super(BasicRequests, self).__init__(methodName=methodName)

    def coroutine_function_decorator(self, func):
        def wrapper(*args, **kw):
            # Is the io loop is already running? (i.e. nested async tests)
            if self.loop.is_running():
                t = func(*args, **kw)
            else:
                # Nope, we are the first
                t = self.loop.run_until_complete(func(*args, **kw))
            return t

        return wrapper

    def __getattribute__(self, item):
        attr = object.__getattribute__(self, item)
        if asyncio.iscoroutinefunction(attr):
            if item not in self._function_cache:
                self._function_cache[item] = self.coroutine_function_decorator(attr)
            return self._function_cache[item]
        return attr

In addition to pylover's answer, if you intend to use some other asynchronous method from the test class itself, the following implementation will work better -

import asyncio
import unittest

class AioTestCase(unittest.TestCase):

    # noinspection PyPep8Naming
    def __init__(self, methodName='runTest', loop=None):
        self.loop = loop or asyncio.get_event_loop()
        self._function_cache = {}
        super(AioTestCase, self).__init__(methodName=methodName)

    def coroutine_function_decorator(self, func):
        def wrapper(*args, **kw):
            return self.loop.run_until_complete(func(*args, **kw))
        return wrapper

    def __getattribute__(self, item):
        attr = object.__getattribute__(self, item)
        if asyncio.iscoroutinefunction(attr) and item.startswith('test_'):
            if item not in self._function_cache:
                self._function_cache[item] = 
                    self.coroutine_function_decorator(attr)
            return self._function_cache[item]
        return attr


class TestMyCase(AioTestCase):

    async def multiplier(self, n):
        await asyncio.sleep(1)  # just to show the difference
        return n*2

    async def test_dispatch(self):
        m = await self.multiplier(2)
        self.assertEqual(m, 4)

the only change was - and item.startswith('test_') in the __getattribute__ method.