How to Mock asyncio.open_connection in Python
Mock asyncio.open_connection with AsyncMock to test async code without a real network connection.
Python code
25 linesimport asyncio
from unittest.mock import AsyncMock, patch
async def fetch_data(reader: asyncio.StreamReader) -> str:
data = await reader.readline()
return data.decode().strip()
async def main() -> None:
# Mock asyncio.open_connection to simulate a server response
mock_reader = AsyncMock()
mock_reader.readline.return_value = b"hello from server\n"
mock_writer = AsyncMock()
with patch("asyncio.open_connection", AsyncMock(return_value=(mock_reader, mock_writer))):
reader, writer = await asyncio.open_connection("localhost", 8888)
result = await fetch_data(reader)
print(f"Received: {result}")
writer.close()
await writer.wait_closed()
if __name__ == "__main__":
asyncio.run(main())Output
Received: hello from serverHow it works
This example uses AsyncMock to simulate the network call made by asyncio.open_connection. By patching asyncio.open_connection, we replace the real socket connection with a mock that returns a mock reader and writer. The mock_reader.readline is configured to return a pre-defined bytes object, which simulates a server response. This pattern lets you test async I/O code deterministically without any real network access.
Common mistakes
- Forgetting to use `AsyncMock` for the `readline` call; regular `Mock` returns a `MagicMock` instead of a coroutine.
- Not patching the correct path; if your code imports `open_connection` directly, patch that reference instead of `asyncio`.
- Forgetting to await `writer.wait_closed()` in the mock writer, which can cause warnings or hangs.
- Assuming `mock_reader.readline` is a coroutine; it must be awaited as in real code.
Variations
- Use `patch` with `side_effect` to test multiple connection attempts or error handling.
- Use `AsyncMock` with `await` semantics directly instead of `return_value` for more complex behavior.
Real-world use cases
- Unit testing a TCP client that reads streaming data without starting a real server on CI.
- Testing retry logic in a service that reconnects to a database or cache over a socket.
- Verifying that your async function handles a malformed response regardless of the actual network.
Sponsored
More from Concurrency & performance
- Benchmark list.append vs deque.append in Python medium
- Build a Python Performance Profiler That Generates Readable Reports medium
- Graceful Shutdown Executor Context Manager in Python medium
- How to Build a Producer-Consumer Pattern with asyncio.Queue in Python medium
- How to Cancel an asyncio Task with Graceful Cleanup in Python medium
- How to Convert Data in Parallel with ThreadPoolExecutor in Python easy
Keep learning
Related tutorials and quizzes for this topic.