如何使用 pytest、aiohttp 和 aioresponses 模拟使用 asyncio.gather 发出的同时请求？

Question

我有一段代码使用asyncio.gather 发出同时请求：

estimated_income, judicial_records = await asyncio.gather(
        *(client.get_estimated_income(), client.get_judicial_records()), return_exceptions=True
    )

# `client.get_estimated_income()` calls `CREDIT_BUREAU_URL`
# `client.get_judicial_records()` calls `NATIONAL_ARCHIVES_URL`

在我的测试中，我试图通过模拟请求状态来模拟一些场景：

mock_aioresponse.get(NATIONAL_ARCHIVES_URL, status=200)
mock_aioresponse.get(CREDIT_BUREAU_URL, status=400)

如果我运行一个测试，它会按预期工作，但如果我运行多个测试（而其他测试甚至不必使用 mock_aioresponse），我会两次到达那段代码并开始得到一些 @987654325第二次出现@错误（第一次工作正常）-传播到测试使其失败。

对我来说最奇怪的事情是，如果我运行多个测试，就会两次达到该功能。

如何使用aioresponses 来完成我的测试用例？

代码：

# main.py

@app.get(
    "/leads/{lead_id}/sales_pipeline",
    response_model=LeadRead,
    responses={status.HTTP_404_NOT_FOUND: {"model": NotFoundResponse}},
)
def sales_pipeline(lead_id: int, db: Session = Depends(get_db)):
    lead = db.get(Lead, lead_id)
    if not lead:
        raise HTTPException(status_code=404, detail="Lead not found")

    pipeline_history = PipelineHistory(lead_id=lead.id)
    db.add(pipeline_history)
    db.commit()
    db.refresh(pipeline_history)
    
    # dispatch an event to handlers.py
    dispatch(event_name=SALES_PIPELINE_ENTRYPOINT_EVENT_NAME, payload={"id": pipeline_history.id})

    return lead

# handlers.py

async def _check_if_lead_is_able_to_become_prospect(
    client: LeadExternalSystemsClient,
) -> Tuple[Optional[bool], Optional[str]]:
    error_messages: List[str] = []
    estimated_income, judicial_records = await asyncio.gather(
        *(client.get_estimated_income(), client.get_judicial_records()), return_exceptions=True
    )
    
    if isinstance(estimated_income, LeadExternalSystemsClient.LeadExternalSystemsException):
        error_messages.append("Credit Bureau network error")
    if isinstance(judicial_records, LeadExternalSystemsClient.LeadExternalSystemsException):
        error_messages.append("National Archives network error")
    # more code

# `LeadExternalSystemsClient` class at client.py

class LeadExternalSystemsClient:
    class LeadExternalSystemsException(Exception):
        pass

    def __init__(self, lead: Lead, timeout: int = 30):
        self.lead = lead
        self._session = ClientSession(
            timeout=ClientTimeout(total=timeout),
            connector=TCPConnector(limit=30, ssl=False),
            raise_for_status=True,
        )

    async def __aenter__(self) -> "LeadExternalSystemsClient":
        return self

    async def __aexit__(self, *_, **__) -> None:
        await self._session.close()

    async def _request(self, method: str, url: str) -> Any:
        try:
            response = self._session.request(method=method, url=url)
            return await response.json()
        except ClientError as exception:
            raise self.LeadExternalSystemsException(str(exception))

    async def get_estimated_income(self) -> Dict[str, float]:
        result = await self._request(method="GET", url=CREDIT_BUREAU_URL)
        # more code

    async def get_judicial_records(self) -> List[Dict[str, str]]:
        result = await self._request(method="GET", url=NATIONAL_ARCHIVES_URL)
        # more code

# tests

@pytest.mark.usefixtures("mock_engine_for_test")
def test_estimated_income_network_error(client, lead, mocker, mock_aioresponse):
    # GIVEN
    mocker.patch(
        "app.consumers.handlers.LeadExternalSystemsClient.personal_information_is_valid",
        return_value=True,
    )
    mock_aioresponse.get(NATIONAL_ARCHIVES_URL, status=200)
    mock_aioresponse.get(CREDIT_BUREAU_URL, status=400)

    # WHEN
    response = client.get(f"/leads/{lead.id}/sales_pipeline")
    result = client.get(f"/leads/{lead.id}").json()

    # THEN
    assert response.status_code == status.HTTP_200_OK
    assert result["is_prospect"] is False
    assert len(result["pipeline_histories"]) == 1
    assert result["pipeline_histories"][0]["started_at"] is not None
    assert result["pipeline_histories"][0]["finished_at"] is not None
    assert result["pipeline_histories"][0]["extra_infos"] == "Credit Bureau network error"
    assert result["pipeline_histories"][0]["status"] == PipelineStatus.NETWORK_ERROR.name

Answer 1

看起来解决方案是将repeat=True 传递给aioresponses().get() https://bytemeta.vip/repo/pnuckowski/aioresponses/issues/205