[Python] Testing async method with pytest does not show calls to patched method

I'm currently writing some tests for a FastAPI middleware component using pytest. This component is a class-type middleware and works exactly as it should, very happy with it. However, when writing tests I'm trying to execute the dispatch method within the component to then make assertions on the output (I'll attach the rest of the setup code below, here is the important part):

import asyncio

pytest_plugins = ("pytest_asyncio",)

@pytest.fixture(scope="function", name="cache_service_table_get_mock")
def fixture_cache_service_table_get_mock(mocker: MockerFixture):
mock_table = mocker.patch("api.utils.cache_middleware.dynamo.table")

mock_table.get_item.return_value = {"Items": []}
mock_table.put_item.return_value = None

yield mock_table

@pytest.mark.asyncio
async def test_cache_no_hit(cache_service_table_get_mock: MagicMock):
cache = CacheMiddleware(app)
request = create_request({"query": "some random query", "product_code": "dummy"})
result = await cache.dispatch(request=request, call_next=mock_call_next)

assert result.status_code == 200
assert result.body == json.dumps({"answer": "dummy response"}, indent=2).encode(
"utf-8"
)
# This generates an error indicating that call_count is 0
assert cache_service_table_get_mock.get_item.call_count == 1


I am 100% sure that the get_item mock is getting called and that it returns the expected mocked value (both by adding print statements everywhere in the cache.dispatch method and because the second assert would only hold it this was the case).

Is there something missing for the call_count property to update correctly?

Remaining setup code


import json
from unittest.mock import MagicMock

import pytest
from fastapi import Request, Response
from pytest_mock import MockerFixture
from starlette.testclient import TestClient

from api import app
from api.utils.cache_middleware import CacheMiddleware

client = TestClient(app)

# Class used to mock the body_iterator prop used in the cache middleware
class AsyncIterator:
def __init__(self, item):
self.length = 1
self.item = item

def __aiter__(self):
return self

async def __anext__(self):
if self.length == 0:
raise StopAsyncIteration
self.length -= 1
return self.item

pass


async def mock_call_next(request: Request) -> Response:
content = json.dumps({"answer": "dummy response"}, indent=2).encode("utf-8")
response = Response(
content=content,
status_code=200,
headers=request.headers,
media_type="text",
)
body_iterator = AsyncIterator(content)
response.__dict__["body_iterator"] = body_iterator

return response


def create_request(body):
request = Request(
scope={
"method": "POST",
"url": "test.com/customapi",
"body": body,
"type": "http",
"path": "/customapi",
"headers": {},
}
)

async def receive():
return {
"type": "http.request",
"body": json.dumps(body, indent=2).encode("utf-8"),
}

request._receive = receive
return request


I have tried using the assert_called_once method with the same result.

I have tried using the same method but for an unrelated non-async function and this works correctly.

I have tried only mocking one method in my fixture with the same result.

It honestly feels like I'm missing something obvious from the documentation that I've been unable to find. Appreciate any help

Continue reading...