Python:检查对第三方服务的 API 应用程序请求

对于许多人来说,在测试中使用第三方服务的模拟是一件很常见的事情。检查代码对特定响应(HTTP 代码和正文)的响应。但是,您多久检查一次 Python 应用程序向此第三方服务发出的请求?毕竟,mock 将返回任何请求正文的所需响应。

示例如下:https://github.com/sharypoff/check-mocks-httpx

对于许多人来说,在测试中使用第三方服务的模拟是一件很常见的事情。检查代码对特定响应(HTTP 代码和正文)的响应。但是,您多久检查一次 Python 应用程序向此第三方服务发出的请求?毕竟,mock 将返回任何请求正文的所需响应。

我真的很热衷于编写测试。此外,我经常使用TDD。对我来说,尽可能多地用测试覆盖代码是很重要的。我经常研究别人的项目,不仅在测试中发现差距,而且在代码本身中也发现差距。这些差距将投入生产。发生这种情况是因为有时,我们忘记了一些细节。

除了检查对第三方服务的请求外,这些还包括迁移测试、事务回滚等。

现在,我将向您展示一些示例,说明如何在测试中验证对第三方服务的请求。我将使用异步 FastAPI + httpx 的示例来做到这一点。但是,aiohttprequests 甚至 Django 都有一套类似的验证方法。

主要任务是比较我们的应用程序发送到存根的请求。如果它不同,那么我们需要在测试中返回一个错误。同时,正确的请求通常可能具有不同的视觉表示形式,因此仅将请求正文与模板进行比较是行不通的。我比较字典或列表,因为它更灵活。

举个例子,让我们以以下简单的应用程序为例:

import httpx
from fastapi import FastAPI

app = FastAPI()
client = httpx.AsyncClient()


@app.post("/")
async def root():
    result1 = await client.request(
        "POST",
        "http://example.com/user",
        json={"name": "John", "age": 21},
    )
    user_json = result1.json()
    result2 = await client.request(
        "POST",
        "http://example1.com/group/23",
        json={"user_id": user_json["id"]},
    )
    group_json = result2.json()
    return {"user": user_json, "group": group_json}

此应用程序发出两个连续请求。对于第二个请求,它使用第一个请求的结果。在这两种情况下,当使用标准模拟时,我们的应用程序可以发送绝对无效的请求,并且仍将收到所需的模拟响应。

那么,让我们看看我们能做些什么。

在不检查请求的情况下进行测试

用于向第三方服务发送请求的最流行的软件包是 httpx。我通常使用 pytest_httpx 作为 mock,对于 aiohttp,我经常使用 aiorequests

检查此终结点的简单测试如下所示:

import httpx
import pytest
import pytest_asyncio
from pytest_httpx import HTTPXMock
from app.main import app

TEST_SERVER = "test_server"


@pytest_asyncio.fixture
async def non_mocked_hosts() -> list:
    return [TEST_SERVER]


@pytest.mark.asyncio
async def test_success(httpx_mock: HTTPXMock):
    httpx_mock.add_response(
        method="POST",
        url="http://example.com/user",
        json={"id": 12, "name": "John", "age": 21},
    )
    httpx_mock.add_response(
        method="POST",
        url="http://example1.com/group/23",
        json={"id": 23, "user_id": 12},
    )
    async with httpx.AsyncClient(app=app, base_url=f"http://{TEST_SERVER}") as async_client:
        response = await async_client.post("/")
    assert response.status_code == 200
    assert response.json() == {
        'user': {'id': 12, 'name': 'John', 'age': 21},
        'group': {'id': 23, 'user_id': 12},
    }

对于对应用程序端点的请求,我们还使用 httpx。AsyncClient。为了不小心嘲笑它,我们添加了non_mocked_hosts夹具。在测试中,我们使用 httpx_mock.add_response 添加两个模拟。此处使用方法和 URL 作为匹配器。如果在测试结束前没有调用带有模拟 URL 和方法的请求,我们将收到错误。

现在,让我们尝试执行一些操作来确保请求正文是外部服务期望的正文。

第一个选项内置于 httpx_mock 中。我们可以添加内容 – 这是完整请求文本的字节对象。

httpx_mock.add_response(
    method="POST",
    url="http://example.com/user",
    json={"id": 12, "name": "John", "age": 21},
    content=b'{"name": "John", "age": 21}',
)

在这种情况下,在搜索过程中,不仅会比较方法和 URL,还会比较这个内容。但是,正如我上面提到的,我们肯定需要知道请求正文。任何字符的差异都会导致错误。例如,冒号和键之间的空格、行转换和任何格式详细信息,或者只是字典中键的不同顺序。这种测试JSON的方式非常低效。但是,该包没有内置的 JSON 格式请求常规验证程序。

简单的方法

在测试结束时获取请求并将它们与模板进行比较似乎是最简单的方法。

import json

import httpx
import pytest
import pytest_asyncio
from pytest_httpx import HTTPXMock
from app.main import app

TEST_SERVER = "test_server"


@pytest_asyncio.fixture
async def non_mocked_hosts() -> list:
    return [TEST_SERVER]


@pytest.mark.asyncio
async def test_with_all_requests_success(httpx_mock: HTTPXMock):
    httpx_mock.add_response(
        method="POST",
        url="http://example.com/user",
        json={"id": 12, "name": "John", "age": 21},
    )
    httpx_mock.add_response(
        method="POST",
        url="http://example1.com/group/23",
        json={"id": 23, "user_id": 12},
    )
    async with httpx.AsyncClient(app=app, base_url=f"http://{TEST_SERVER}") as async_client:
        response = await async_client.post("/")
    assert response.status_code == 200
    assert response.json() == {
        'user': {'id': 12, 'name': 'John', 'age': 21},
        'group': {'id': 23, 'user_id': 12},
    }
    expected_requests = [
        {"url": "http://example.com/user", "json": {"name": "John", "age": 21}},
        {"url": "http://example1.com/group/23", "json": {"user_id": 12}},
    ]
    for expecter_request, real_request in zip(expected_requests, httpx_mock.get_requests()):
        assert expecter_request["url"] == real_request.url
        assert expecter_request["json"] == json.loads(real_request.content)

这种方法非常庞大,需要重复代码。

更复杂的方法

我决定稍微修复一下包。我本可以通过重写必要的方法创建一个子类,但我创建的层略有不同。为此,我添加了 APIMock 类:

from typing import Any
import httpx
import json
from pytest_httpx import HTTPXMock


class APIMock:
    def __init__(self, httpx_mock: HTTPXMock):
        self.httpx_mock = httpx_mock

    async def request_mock(
        self,
        method: str,
        url: str,
        status_code: int = 200,
        request_for_check: Any = None,
        **kwargs,
    ):
        async def custom_response(request: httpx.Request) -> httpx.Response:
            response = httpx.Response(
                status_code,
                **kwargs,
            )
            if request_for_check:
                assert request_for_check == json.loads(request.content), \
                    f"{request_for_check} != {json.loads(request.content)}"
            return response
        self.httpx_mock.add_callback(custom_response, method=method, url=url)

此类中只有一个公共方法。它通过从 httpx_mock 调用 add_callback 来添加模拟。

它的结构来自原始add_response httpx_mock方法,但更短一些。如果需要更多参数,可以随时添加它们。

要使用它,只需通过固定装置导入它并调用add_request即可。

下面是使用 APIMock 类和请求验证的测试示例。

import httpx
import pytest
import pytest_asyncio
from pytest_httpx import HTTPXMock
from app.main import app
from app.api_mock import APIMock

TEST_SERVER = "test_server"


@pytest_asyncio.fixture
async def non_mocked_hosts() -> list:
    return [TEST_SERVER]


@pytest_asyncio.fixture
async def requests_mock(httpx_mock: HTTPXMock):
    return APIMock(httpx_mock)


@pytest.mark.asyncio
async def test_with_requests_success(requests_mock):
    await requests_mock.request_mock(
        method="POST",
        url="http://example.com/user",
        json={"id": 12, "name": "John", "age": 21},
        request_for_check={"name": "John", "age": 21},
    )
    await requests_mock.request_mock(
        method="POST",
        url="http://example1.com/group/23",
        json={"id": 23, "user_id": 12},
        request_for_check={"user_id": 12},
    )
    async with httpx.AsyncClient(app=app, base_url=f"http://{TEST_SERVER}") as async_client:
        response = await async_client.post("/")
    assert response.status_code == 200
    assert response.json() == {
        'user': {'id': 12, 'name': 'John', 'age': 21},
        'group': {'id': 23, 'user_id': 12},
    }

它看起来非常简单方便。该结构允许您在创建模拟时指定预期的请求,就像使用内容时一样。我发现这种方法是最方便的一种方法。此外,其他类可以从这个类继承,它们会将每个特定端点模拟到第三方服务。这非常方便。

对 aioresponses 包和响应也可以执行相同的操作。它们还支持回调函数。

从理论上讲,您必须为这样的测试编写一个测试,因为它非常合乎逻辑。但这不是问题,因为我们有一个单独的类。

结论

这两种方法都有优点和缺点。

第一种方法

简单方法的优点:

  1. 无需更改或添加任何内容。一切都是开箱即用的。
  2. 检查请求的顺序。

简单方法的缺点:

  1. 您必须在每次测试中重复验证码。
  2. 您必须维护每个测试的所有请求的列表。
  3. 如果是并行请求,则可能会违反订单。

第二种方法

第二种方法的优点:

  1. 最低码数和可重复性。
  2. 您可以方便简洁地实现每个端点的模拟。

第二种方法的缺点:

  1. 请求订单未验证。

我在测试中使用这两种方法。在每种特定情况下,什么最适合您取决于您。

示例如下:https://github.com/sharypoff/check-mocks-httpx

发表评论

邮箱地址不会被公开。 必填项已用*标注