How to Mock an Asynchronous Iterator with pytest-mock in Python
Unit testing asynchronous Python code can sometimes lead to cryptic errors, especially when dealing with asynchronous generators and third-party libraries like discord.py, aiohttp, or httpx. A common stumbling block occurs when attempting to mock a method intended to be used in an async for loop, resulting in the dreaded:
TypeError: 'async for' requires an object with __aiter__ method, got coroutine
In this guide, we will break down why this error occurs when using pytest-mock and how to correctly mock an asynchronous iterator.
Understanding the Error
Consider a method like guild.fetch_members() in discord.py. Under normal circumstances, this method returns an asynchronous iterator (an object implementing __aiter__ and __anext__), allowing you to loop over items like so:
async for member in guild.fetch_members():
# Process member
pass
The error TypeError: 'async for' requires an object with __aiter__ method, got coroutine usually happens due to one of two reasons:
- Mocking the instance container as an
AsyncMock: When an entire object (likeguild) is instantiated as anAsyncMock, accessing methods on it automatically generates coroutines unless explicitly overridden. - Patching with the wrong mock type: When you patch a method with
mocker.patch()without configuring the proper return value behavior, Python expects a coroutine (usingawait) rather than a synchronous call returning an async iterator.
The Solution: Use MagicMock with an Async Generator Return Value
To correctly mock an asynchronous iterator, your mock method needs to be a standard synchronous callable (or regular MagicMock) whose return_value is an async generator or an object implementing __aiter__.
Here is how you can write the test using pytest and pytest-mock:
import pytest
from pytest_mock import MockerFixture
# Helper function to generate an asynchronous stream
async def async_generator(items):
for item in items:
yield item
class TestScraper:
@pytest.fixture
def scraper(self):
return Scraper()
@pytest.mark.asyncio
async def test_scrape(self, mocker: MockerFixture, scraper):
# 1. Create a MagicMock for the parent object (NOT AsyncMock)
mock_guild = mocker.MagicMock()
# 2. Create mock items to be yielded
mock_member = mocker.MagicMock(name="MockMember")
# 3. Assign the async generator to the return_value of the method
mock_guild.fetch_members.return_value = async_generator([mock_member])
# 4. Run the code under test
await scraper.activate(mock_guild)
# 5. Assertions
mock_guild.fetch_members.assert_called_once()
Alternative: Mocking __aiter__ Directly
If you prefer not to write a custom generator helper function, Python 3.8+ introduced native support for __aiter__ on MagicMock and AsyncMock. You can set the return value of __aiter__ directly to an iterable:
@pytest.mark.asyncio
async def test_scrape_with_magic_mock(mocker: MockerFixture, scraper):
mock_guild = mocker.MagicMock()
mock_member = mocker.MagicMock()
# Create a mock that implements __aiter__
mock_iterator = mocker.MagicMock()
mock_iterator.__aiter__.return_value = [mock_member]
# fetch_members() returns the mock iterator
mock_guild.fetch_members.return_value = mock_iterator
await scraper.activate(mock_guild)
Key Takeaways
- Use
AsyncMockfor coroutines: UseAsyncMockwhen the function is defined withasync defand called withawait function(). - Use
MagicMockfor async iterables: When a synchronous method call returns an object to be consumed byasync for, the method itself should be aMagicMockreturning an async generator or a mock with__aiter__defined. - Configure mocks on instances directly: Rather than patching class definitions globally using
mocker.patch("module.Class.method"), prefer creating mock instances (e.g.,mock_guild = mocker.MagicMock()) and injecting them into your test target for cleaner, isolated tests.