How to test Python 3.4 asyncio code?

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

from unittest import IsolatedAsyncioTestCaseclass 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.