Тестирование асинхронного метода с помощью pytest не показывает вызовы исправленного метода ⇐ Python

[Anonymous]

В настоящее время я пишу несколько тестов для компонента промежуточного программного обеспечения FastAPI, используя pytest. Этот компонент представляет собой промежуточное программное обеспечение классового типа и работает именно так, как должен, очень доволен этим. Однако при написании тестов я пытаюсь выполнить метод диспетчеризации внутри компонента, чтобы затем сделать утверждения на выходе (остальную часть кода установки я прикреплю ниже, вот важная часть):

Код: Выделить всё

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

Я на 100 % уверен, что вызывается макет get_item и что он возвращает ожидаемое имитируемое значение (и то и другое путем добавления операторов print повсюду в кеше). метод диспетчеризации, и поскольку второй метод утверждения будет удерживать только его, это было именно так).
Что-то не хватает для правильного обновления свойства call_count?
Что-то не хватает для правильного обновления свойства call_count?
p>

Оставшийся код установки

Код: Выделить всё

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

Я пробовал использовать метод Assert_ Call_once с тем же результатом.
Я пробовал использовать тот же метод, но для несвязанного несвязанного -async, и это работает правильно.
Я попробовал имитировать только один метод в своем приборе с тем же результатом.
Честно говоря, мне кажется, что я мне не хватает чего-то очевидного в документации, которую я не смог найти. Благодарим за любую помощь

Подробнее здесь: https://stackoverflow.com/questions/790 ... hed-method