Testing Asynchronous Python Code with unittest

Intermediate

Asynchronous functions return coroutine objects, so a useful test must run the coroutine and await its result. Python’s built-in unittest.IsolatedAsyncioTestCase gives each test an async-aware runner, letting you write async def test_... methods and use ordinary unittest assertions without managing a new event loop yourself.

Use IsolatedAsyncioTestCase for coroutine tests

A normal unittest.TestCase is designed for synchronous test methods. For code that must be awaited, subclass unittest.IsolatedAsyncioTestCase and define each test method with async def. Inside it, await the coroutine under test and then assert on the result. Do not call asyncio.run() inside the test: the test case already manages the async execution context.

await is essential: calling get_status() alone only creates a coroutine object. Awaiting it lets the test observe the completed result and any exception it raises. The same assertion methods you already know from TestCase work in these async test methods.

import unittest

async def fetch_value():
    return 7

class TestValue(unittest.IsolatedAsyncioTestCase):
    ?:
        self.assertEqual(await fetch_value(), 7)

if __name__ == "__main__":
    unittest.main(argv=["quick-check"], exit=False)

Check asynchronous failures and outcomes

Use the familiar TestCase assertions to check the behavior after an await. To test an expected exception, place the await inside self.assertRaises(...); the exception must occur while the context is active. This keeps the test about the operation’s contract instead of about event-loop mechanics.

import unittest

async def parse_count(text):
    raise ValueError("bad count")

class TestCount(unittest.IsolatedAsyncioTestCase):
    async def test_bad_count(self):
        with self.assertRaises(ValueError):
            ?

if __name__ == "__main__":
    unittest.main(argv=["quick-check"], exit=False)

Prepare and clean up asynchronous fixtures

Async tests can use both fixture styles. The synchronous setUp() runs before asyncSetUp(). After a test, asyncTearDown() runs before synchronous tearDown(). An async setup failure means the test and its teardown methods do not run as if setup had succeeded, so keep each step responsible for resources it actually acquired.

Prefer addAsyncCleanup() when you acquire a resource that must be closed even if the test fails. It registers an async cleanup coroutine and avoids making cleanup depend on the test reaching a particular line. enterAsyncContext() can enter an asynchronous context manager and automatically register its exit method as a cleanup.

import unittest

class TestResource(unittest.IsolatedAsyncioTestCase):
    async def close_resource(self):
        pass

    async def asyncSetUp(self):
        ?
        self.resource_ready = True

    async def test_resource_ready(self):
        self.assertTrue(self.resource_ready)

if __name__ == "__main__":
    unittest.main(argv=["quick-check"], exit=False)

Check that async collaborators are awaited

When an asynchronous dependency is replaced for a unit test, use unittest.mock.AsyncMock rather than a plain synchronous mock for the awaited call. An async mock lets the test control the awaited result and distinguish a call from an actual await with assert_awaited_once_with(). The full mock API and patching rules remain covered in the separate unittest.mock lesson.

from unittest.mock import AsyncMock

reader = AsyncMock(return_value="ok")
async def main():
    await reader(7)
    ?

import asyncio
asyncio.run(main())

Run async tests and keep them deterministic

Run these tests with the same built-in test runner as synchronous cases, for example python -m unittest -v test_async_service.py. IsolatedAsyncioTestCase manages a separate event-loop lifecycle for a test case. Tests should await the operation they exercise, close resources, avoid depending on test order, and avoid real network or service calls unless the goal is specifically an integration test.

import unittest

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

class TestAdd(unittest.IsolatedAsyncioTestCase):
    async def test_add(self):
        ?
        self.assertEqual(result, 42)

if __name__ == "__main__":
    unittest.main(argv=["quick-check"], exit=False)