138

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?

Improve this question
2
  • 5
    you could use loop.run_until_complete() instead of yield from. See also asyncio.test_utils. Commented Apr 12, 2014 at 20:35
  • For python 3.5+ async def and await syntax, see: stackoverflow.com/questions/41263988/… Commented Dec 27, 2016 at 21:03

11 Answers 11

Reset to default
238

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)
Improve this answer
Sign up to request clarification or add additional context in comments.

3 Comments

Too bad this answer is shown after no less than 5 workarounds only as of today.
@Marvin Killing could accept this answer and that would change maybe...
Very great answer. This saves my life! Async tests should not be a permanent stop and frustrated issue for py developers.
63

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.

Improve this answer

6 Comments

Is there a way to modify this solution so the decorator uses a specific loop, not the threads default loop?
Yes, function annotations can take arguments in Python, so you can pass an event loop there. Note that writing annotations that take arguments is a liitle confusing at first: stackoverflow.com/a/5929165/823869
@JackO'Connor I think you mean function decorators not function annotations as function annotations has a specific meaning in Python: docs.python.org/3/tutorial/…
I ran into issues with asyncio.get_event_loop() and used asyncio.new_event_loop()
Warning that asyncio.coroutine is deprecated and will be removed in py3.10: docs.python.org/3/library/…
|
55

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.

Improve this answer

6 Comments

Is there any reason why you do asyncio.set_event_loop(None) and later pass self.loop explicitly to asyncio.open_connection() instead of just doing asyncio.set_event_loop(self.loop) right from the start?
Well, it's just my habit. When I working on asyncio and or aio-based libraries I use asyncio.set_event_loop(None) to directly specify the fact that the library should not relay on global loop existence and safely work by explicit loop passing. It's code style for asyncio tests itself, I use it in my libraries too.
This example should also mock asyncio.open_connection shouldn't it? Running it produces ConnectionRefusedError: [Errno 61] Connect call failed ('127.0.0.1', 8888)
@terrycojones mock is not always required. In example I use local address, so I may setup test server on the address before test run or in setUp method. Concrete implementation depends on your needs.
Adds more boilterplate in place but definitively this is the way of make tests unitary and isolated
|
32

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())
Improve this answer

4 Comments

for anyone using nosetests, you might want to rename the decorator or nose thinks it's actually a test as well, with a mysterious message about async_test missing a required positional argument. I renamed to asynctest and added an additional decorator @nose.tools.istest to make the testcase autodiscoverable
Wrap async_test with nose.tools.nottest decorator if using nosetests.
this is beautiful. Incorporating in my project. Thank you!
In my Odoo installation, even though I'm using Python 3.10, the top answer to use IsolatedAsyncioTestCase is not working, but this is; thanks!
19

pytest-asyncio looks promising:

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

3 Comments

There is an issue with the pytest approach when using unittest.TestCase, that make very limited for me. jacobbridges.github.io/post/unit-testing-with-asyncio
Looks like an issue was filed with them here. No solution yet. github.com/pytest-dev/pytest-asyncio/issues/15
Also mocking classes via mock.patch stops working. github.com/pytest-dev/pytest-asyncio/issues/42
13

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)
Improve this answer

Comments

12

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.

Improve this answer

2 Comments

This is a great solution. Added in a slight change here: stackoverflow.com/a/60986764/328059
Please add a description to your code. Code only is not an answer.
2

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())
Improve this answer

Comments

2

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.

Improve this answer

Comments

2

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
Improve this answer

Comments

2

I found python test file have a similar 'async_test' function like Marvin Killing's answer. Because "@coroutine" decorator is deprecated since Python 3.8. When I use python3.8 or above. I got a "DeprecationWarning".

If you use Python 3.5+. This answer maybe a good option. Hope helps.

import asyncio
import functools


def async_test(func):
    """Decorator to turn an async function into a test case."""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        coro = func(*args, **kwargs)
        asyncio.run(coro)
    return wrapper

Test example:

import unittest


async def add_func(a, b):
    return a + b


class TestSomeCase(unittest.TestCase):
    @async_test
    async def test_add_func(self):
        self.assertEqual(await add_func(1, 2), 3)
Improve this answer

Comments

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.