loop_factory in IsolatedAsyncioTestCase gives Python tests direct control over how an event loop is created. Instead of relying only on global event-loop policy, a test class can provide its own factory and keep asynchronous tests isolated, repeatable, and easier to debug. This is useful for networking libraries, HTTP clients, queues, background workers, services, and any project whose behavior depends on the asyncio loop.
Why loop_factory matters
Every asynchronous test needs an event loop. When loop creation depends on global state, one test module can influence another. Plugins, frameworks, operating-system defaults, and local configuration may also change the selected implementation. The result can be intermittent failures, unfinished tasks, and behavior that differs between development and continuous integration.
An explicit factory makes loop creation part of the test contract. Each test receives a clean environment, and the class documents which loop it expects. This complements the isolation already provided by unittest.IsolatedAsyncioTestCase.
Basic structure
import asyncio
import unittest
class ServiceTests(unittest.IsolatedAsyncioTestCase):
loop_factory = asyncio.new_event_loop
async def test_response(self):
await asyncio.sleep(0)
self.assertTrue(True)
The factory must be callable and return a new loop. Do not return one shared global instance. Reusing the same loop can carry callbacks, tasks, exception handlers, and state from one test into another.
Custom factories
A custom factory is valuable when you need debug mode, instrumentation, implementation-specific checks, or consistent behavior across machines.
def make_debug_loop():
loop = asyncio.new_event_loop()
loop.set_debug(True)
return loop
class ConcurrencyTests(unittest.IsolatedAsyncioTestCase):
loop_factory = staticmethod(make_debug_loop)
Debug mode can reveal slow callbacks, forgotten awaits, and suspicious scheduling behavior. It is not a replacement for assertions, but it gives the test suite useful diagnostic information.
Asynchronous setup and cleanup
IsolatedAsyncioTestCase provides asyncSetUp and asyncTearDown. Use them to create and close resources associated with the current loop, such as servers, clients, tasks, workers, queues, and connections.
class ClientTests(unittest.IsolatedAsyncioTestCase):
loop_factory = asyncio.new_event_loop
async def asyncSetUp(self):
self.queue = asyncio.Queue()
self.worker = asyncio.create_task(self.consume())
async def asyncTearDown(self):
self.worker.cancel()
with self.assertRaises(asyncio.CancelledError):
await self.worker
async def consume(self):
while True:
await self.queue.get()
Explicit cleanup prevents “task was destroyed but it is pending” warnings. It also verifies that production code exposes a predictable shutdown path. See the Academify guides to Python asyncio and unit testing with unittest for related foundations.
Testing timeouts
Timeout tests should be short and deterministic. Prefer events, futures, and queues over long real sleeps.
async def wait_forever():
await asyncio.Event().wait()
class TimeoutTests(unittest.IsolatedAsyncioTestCase):
loop_factory = asyncio.new_event_loop
async def test_timeout(self):
with self.assertRaises(asyncio.TimeoutError):
await asyncio.wait_for(wait_forever(), timeout=0.01)
Do not make thresholds unrealistically tight. Slow CI machines can introduce scheduling delays. Test the logical timeout contract rather than attempting to benchmark performance.
Testing cancellation
Cancellation is part of many asynchronous APIs. A useful test confirms that a task receives CancelledError, runs cleanup logic, and leaves no resources open.
class CancellationTests(unittest.IsolatedAsyncioTestCase):
async def test_cancel_task(self):
task = asyncio.create_task(asyncio.sleep(60))
task.cancel()
with self.assertRaises(asyncio.CancelledError):
await task
Keep references to every task you create. Closing the loop may cancel leftovers, but relying on automatic cleanup can hide design defects in application code.
Using AsyncMock
unittest.mock.AsyncMock works naturally with isolated asynchronous tests. It can verify awaited calls, arguments, return values, and exceptions without reaching external services.
from unittest.mock import AsyncMock
class RepositoryTests(unittest.IsolatedAsyncioTestCase):
async def test_save_item(self):
save = AsyncMock(return_value=42)
result = await save({"name": "Ada"})
self.assertEqual(result, 42)
save.assert_awaited_once_with({"name": "Ada"})
Additional practical references include Python mocks and pytest in Python. Even projects that use pytest benefit from understanding unittest’s built-in lifecycle.
Version compatibility
Confirm the minimum Python version supported by your package before depending on loop_factory. Multi-version libraries may need a small compatibility base class.
import sys
if sys.version_info >= (3, 13):
class AsyncCase(unittest.IsolatedAsyncioTestCase):
loop_factory = asyncio.new_event_loop
else:
class AsyncCase(unittest.IsolatedAsyncioTestCase):
pass
Keep the compatibility branch documented and tested in CI. The official unittest documentation is the authoritative source for the version you target. The asyncio event-loop documentation explains loop creation and lifecycle details.
Avoid global state
One major benefit of a class-level factory is reducing the need for asyncio.set_event_loop_policy. Global policy changes affect tests that run later and may conflict with plugins. When global mutation is unavoidable, save the previous policy and restore it reliably.
Do not move loop-bound objects between loops. Tasks, Futures, Locks, Events, and Queues are created in an asynchronous context and should remain there. Sharing them across isolated tests can produce runtime errors or subtle races.
Checking pending tasks
A defensive helper can detect and cancel tasks left behind by a test.
async def cancel_pending():
current = asyncio.current_task()
pending = [
task for task in asyncio.all_tasks()
if task is not current and not task.done()
]
for task in pending:
task.cancel()
await asyncio.gather(*pending, return_exceptions=True)
This is a safety net, not a substitute for ownership. Each component should expose a shutdown method, and each test should close what it creates.
Suite organization
Create a small base class for the factory and shared helpers. Separate tests for transport, persistence, queues, and business rules. Give methods behavior-oriented names and limit each test to one clear guarantee.
Avoid sharing clients, connections, and running workers across test methods. Per-test resources may cost a little more time, but they reduce coupling. To improve performance, replace slow dependencies with test doubles instead of weakening event-loop isolation.
Complete example
import asyncio
import unittest
async def process(queue, output):
while True:
item = await queue.get()
try:
if item is None:
return
output.append(item * 2)
finally:
queue.task_done()
class PipelineTests(unittest.IsolatedAsyncioTestCase):
loop_factory = asyncio.new_event_loop
async def asyncSetUp(self):
self.queue = asyncio.Queue()
self.output = []
self.worker = asyncio.create_task(
process(self.queue, self.output)
)
async def asyncTearDown(self):
if not self.worker.done():
self.worker.cancel()
await asyncio.gather(self.worker, return_exceptions=True)
async def test_processes_items(self):
await self.queue.put(2)
await self.queue.put(5)
await self.queue.join()
self.assertEqual(self.output, [4, 10])
async def test_shutdown(self):
await self.queue.put(None)
await self.worker
self.assertTrue(self.worker.done())
The example combines isolated loop creation, setup, teardown, queue processing, task ownership, and graceful shutdown. It does not depend on external event-loop policy.
Common mistakes
Common mistakes include returning the same loop from the factory, creating tasks outside the test lifecycle, using long sleeps, swallowing cancellation, and leaving asynchronous generators open. Another mistake is changing global policy merely to satisfy one test class.
Use explicit ownership: the code that creates a task should know how it ends. Use short deterministic synchronization primitives, collect background exceptions, and ensure teardown remains safe even when the test fails halfway through setup.
Conclusion
loop_factory makes IsolatedAsyncioTestCase more configurable and transparent. It reduces global state, supports debug and custom loop creation, and helps test suites behave consistently across machines. Its strongest results come from combining it with explicit cleanup, deterministic timeouts, cancellation tests, AsyncMock, and pending-task checks. Those practices produce asynchronous tests that are easier to trust and maintain.
