How to test Python 3.4 asyncio code?

Question

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?

Answer 1

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) 

Answer 2

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.