为外部 API 调用编写测试

作者：Harry，2020-01-25

了解更多关于此图片的信息

这是 Python 测试中人们常问的一个问题

如何为调用第三方 API 的代码编写测试？

（感谢 Brian Okken 提出这个问题）。

在本文中，我想概述几个选项，从最熟悉的（模拟）到最架构宇航员式的，并尝试讨论每种方法的优缺点。希望我能说服您至少尝试一下最后的一些想法。

我将使用物流领域的示例，我们需要将货运同步到货运供应商的 API，但您可以想象任何旧的 API——支付网关、短信通知引擎、云存储提供商。或者您可以想象与网络完全无关的外部依赖项，只是任何类型的难以进行单元测试的外部 I/O 依赖项。

但为了使事情具体化，在我们的物流示例中，我们将有一个货运模型，其中包含许多订单行。我们还关心其预计到达时间 (eta) 和一个称为贸易术语的术语（您不需要理解那是什么，我只是想说明一些现实生活中的复杂性，在这个小例子中）。

@dataclass
class OrderLine:
    sku: str  # sku="stock keeping unit", it's a product id basically
    qty: int


@dataclass
class Shipment:
    reference: str
    lines: List[OrderLine]
    eta: Optional[date]
    incoterm: str

    def save(self):
        ...  # for the sake of the example, let's imagine the model
             # knows how to save itself to the DB.  like Django.

我们希望通过他们的 API 将我们的货运模型与第三方货运公司同步。我们有几个用例：创建新货运和检查更新后的 eta。

假设我们有一些控制器函数负责执行此操作。它接受将 sku 映射到数量的字典，创建我们的模型对象，保存它们，然后调用辅助函数以同步到 API。希望这种事情看起来很熟悉

def create_shipment(quantities: Dict[str, int], incoterm):
    reference = uuid.uuid4().hex[:10]
    order_lines = [OrderLine(sku=sku, qty=qty) for sku, qty in quantities.items()]
    shipment = Shipment(reference=reference, lines=order_lines, eta=None, incoterm=incoterm)
    shipment.save()
    sync_to_api(shipment)

我们如何同步到 API？一个简单的 POST 请求，带有一些数据类型转换和处理。

def sync_to_api(shipment):
    requests.post(f'{API_URL}/shipments/', json={
        'client_reference': shipment.reference,
        'arrival_date': shipment.eta.isoformat(),
        'products': [
            {'sku': ol.sku, 'quantity': ol.quantity}
            for ol in shipment.lines
        ]
    })

还不错！

我们如何测试它？在这种情况下，典型的反应是求助于模拟，并且只要事情保持简单，它就相当容易管理

def test_create_shipment_does_post_to_external_api():
    with mock.patch('controllers.requests') as mock_requests:
        shipment = create_shipment({'sku1': 10}, incoterm='EXW')
        expected_data = {
            'client_reference': shipment.reference,
            'arrival_date': None,
            'products': [{'sku': 'sku1', 'quantity': 10}],
        }
        assert mock_requests.post.call_args == mock.call(
            API_URL + '/shipments/', json=expected_data
        )

您可以想象添加更多测试，也许一个检查我们是否正确执行了 date-to-isoformat 转换，也许一个检查我们是否可以处理多行。三个测试，每个测试一个模拟，我们没问题。

问题是它永远不会那么简单，不是吗？例如，货运公司可能已经有货运记录在案，因为某些原因。如果您在已存在的情况下执行 POST，则会发生不好的事情。因此，我们首先需要使用 GET 请求检查他们是否存档了货运，然后如果它是新的，则执行 POST，或者对于现有的货运执行 PUT

def sync_to_api(shipment):
    external_shipment_id = get_shipment_id(shipment.reference)
    if external_shipment_id is None:
        requests.post(f'{API_URL}/shipments/', json={
            'client_reference': shipment.reference,
            'arrival_date': shipment.eta,
            'products': [
                {'sku': ol.sku, 'quantity': ol.quantity}
                for ol in shipment.lines
            ]
        })

    else:
        requests.put(f'{API_URL}/shipments/{external_shipment_id}', json={
            'client_reference': shipment.reference,
            'arrival_date': shipment.eta,
            'products': [
                {'sku': ol.sku, 'quantity': ol.quantity}
                for ol in shipment.lines
            ]
        })


def get_shipment_id(our_reference) -> Optional[str]:
    their_shipments = requests.get(f"{API_URL}/shipments/").json()['items']
    return next(
        (s['id'] for s in their_shipments if s['client_reference'] == our_reference),
        None
    )

和往常一样，复杂性悄然而至

  • 因为事情永远不容易，所以第三方有不同的参考编号，因此我们需要 get_shipment_id() 函数来为我们找到正确的参考编号

  • 如果它是新的货运，我们需要使用 POST，如果它是现有的货运，则需要使用 PUT。

您已经可以想象我们需要编写相当多的测试来涵盖所有这些选项。这只是一个示例

def test_does_PUT_if_shipment_already_exists():
    with mock.patch('controllers.uuid') as mock_uuid, mock.patch('controllers.requests') as mock_requests:
        mock_uuid.uuid4.return_value.hex = 'our-id'
        mock_requests.get.return_value.json.return_value = {
            'items': [{'id': 'their-id', 'client_reference': 'our-id'}]
        }

        shipment = create_shipment({'sku1': 10}, incoterm='EXW')
        assert mock_requests.post.called is False
        expected_data = {
            'client_reference': 'our-id',
            'arrival_date': None,
            'products': [{'sku': 'sku1', 'quantity': 10}],
        }
        assert mock_requests.put.call_args == mock.call(
            API_URL + '/shipments/their-id/', json=expected_data
        )

……我们的测试变得越来越不愉快。同样，细节并不太重要，希望这种测试的丑陋是熟悉的。

而这仅仅是开始，我们展示了一个只关心写入的 API 集成，但是读取呢？假设我们现在想不时轮询我们的第三方 API，以获取我们货运的更新 eta。根据 eta，我们有一些关于通知人们延迟的业务逻辑……

# another example controller,
# showing business logic getting intermingled with API calls

def get_updated_eta(shipment):
    external_shipment_id = get_shipment_id(shipment.reference)
    if external_shipment_id is None:
        logging.warning('tried to get updated eta for shipment %s not yet sent to partners', shipment.reference)
        return

    [journey] = requests.get(f"{API_URL}/shipments/{external_shipment_id}/journeys").json()['items']
    latest_eta = journey['eta']
    if latest_eta == shipment.eta:
        return
    logging.info('setting new shipment eta for %s: %s (was %s)', shipment.reference, latest_eta, shipment.eta)
    if shipment.eta is not None and latest_eta > shipment.eta:
        notify_delay(shipment_ref=shipment.reference, delay=latest_eta - shipment.eta)
    if shipment.eta is None and shipment.incoterm == 'FOB' and len(shipment.lines) > 10:
        notify_new_large_shipment(shipment_ref=shipment.reference, eta=latest_eta)

    shipment.eta = latest_eta
    shipment.save()

我还没有编写所有测试的样子，但您可以想象它们

  1. 如果货运不存在，则记录警告的测试。需要模拟 requests.get 或 get_shipment_id()
  2. 如果 eta 没有更改，我们什么也不做的测试。在 requests.get 上需要两个不同的模拟
  3. 货运 API 没有行程的错误情况的测试
  4. 货运有多个行程的边缘情况的测试
  5. 检查如果 eta 晚于当前 eta，我们是否执行通知的测试。
  6. 以及相反情况的测试，如果 eta 更早则不通知
  7. 大型货运通知的测试
  8. 以及仅在必要时才执行该通知的测试
  9. 以及更新本地 eta 并保存它的一般测试。
  10. ……我相信我们可以想象更多。

而这些测试中的每一个都需要设置三到四个模拟。我们正在进入 Ed Jung 所谓的 Mock 地狱。

除了我们的测试难以阅读和编写之外，它们也很脆弱。如果我们更改导入方式，从 import requests 到 from requests import get（并不是说您会这样做，但您明白了），那么我们所有的模拟都会中断。如果您想要一个更合理的示例，也许我们决定停止使用 requests.get()，因为我们想出于某种原因使用 requests.Session()。

关键是 mock.patch 将您束缚于特定的实现细节

我们甚至还没有谈到其他类型的测试。为了让您确信事情真的有效，您可能需要一两个集成测试，甚至可能需要一个 E2E 测试。

这是对模拟方法优缺点的简要回顾。每次我们引入新选项时，我们都会有其中一个。

模拟和修补：权衡

优点
  • 客户端代码无需更改
  • 低工作量
  • 对于（大多数？许多？）开发人员来说，它很熟悉
缺点
  • 紧密耦合
  • 脆弱。requests.get -> requests.Session().get 将会破坏它。
  • 需要记住在每个可能最终调用该 API 的测试中 @mock.patch
  • 容易将业务逻辑和 I/O 问题混在一起
  • 可能还需要集成和 E2E 测试。

建议：构建适配器（外部 API 的包装器）

我们真的希望将我们的业务逻辑与我们的 API 集成分离。围绕 API 构建一个抽象、一个包装器，它只是为我们在代码中调用公开了友好的、可读的方法。

我们在 端口和适配器 意义上称之为“适配器”，但您不必完全采用六边形架构来使用此模式。

class RealCargoAPI:
    API_URL = 'https://example.org'

    def sync(self, shipment: Shipment) -> None:
        external_shipment_id = self._get_shipment_id(shipment.reference)
        if external_shipment_id is None:
            requests.post(f'{self.API_URL}/shipments/', json={
              ...

        else:
            requests.put(f'{self.API_URL}/shipments/{external_shipment_id}/', json={
              ...


    def _get_shipment_id(self, our_reference) -> Optional[str]:
        try:
            their_shipments = requests.get(f"{self.API_URL}/shipments/").json()['items']
            return next(
              ...
        except requests.exceptions.RequestException:
            ...

现在我们的测试看起来如何？

def test_create_shipment_syncs_to_api():
    with mock.patch('controllers.RealCargoAPI') as mock_RealCargoAPI:
        mock_cargo_api = mock_RealCargoAPI.return_value
        shipment = create_shipment({'sku1': 10}, incoterm='EXW')
        assert mock_cargo_api.sync.call_args == mock.call(shipment)

更易于管理！

但是

  • 我们仍然有 mock.patch 脆弱性，这意味着如果我们改变我们对事物导入方式的想法，我们需要更改我们的模拟

  • 我们仍然需要测试 API 适配器本身

def test_sync_does_post_for_new_shipment():
    api = RealCargoAPI()
    line = OrderLine('sku1', 10)
    shipment = Shipment(reference='ref', lines=[line], eta=None, incoterm='foo')
    with mock.patch('cargo_api.requests') as mock_requests:
        api.sync(shipment)

        expected_data = {
            'client_reference': shipment.reference,
            'arrival_date': None,
            'products': [{'sku': 'sku1', 'quantity': 10}],
        }
        assert mock_requests.post.call_args == mock.call(
            API_URL + '/shipments/', json=expected_data
        )

建议：使用（仅？）集成测试来测试您的适配器

现在我们可以将我们的适配器与我们的主要应用程序代码分开测试，我们可以考虑测试它的最佳方法是什么。由于它只是外部系统的一个薄包装器，因此最好的测试类型是集成测试

def test_can_create_new_shipment():
    api = RealCargoAPI('https://sandbox.example.com/')
    line = OrderLine('sku1', 10)
    ref = random_reference()
    shipment = Shipment(reference=ref, lines=[line], eta=None, incoterm='foo')

    api.sync(shipment)

    shipments = requests.get(api.api_url + '/shipments/').json()['items']
    new_shipment = next(s for s in shipments if s['client_reference'] == ref)
    assert new_shipment['arrival_date'] is None
    assert new_shipment['products'] == [{'sku': 'sku1', 'quantity': 10}]


def test_can_update_a_shipment():
    api = RealCargoAPI('https://sandbox.example.com/')
    line = OrderLine('sku1', 10)
    ref = random_reference()
    shipment = Shipment(reference=ref, lines=[line], eta=None, incoterm='foo')

    api.sync(shipment)

    shipment.lines[0].qty = 20

    api.sync(shipment)

    shipments = requests.get(api.api_url + '/shipments/').json()['items']
    new_shipment = next(s for s in shipments if s['client_reference'] == ref)
    assert new_shipment['products'] == [{'sku': 'sku1', 'quantity': 20}]

这依赖于您的第三方 API 是否有一个体面的沙箱，您可以针对该沙箱进行测试。您需要考虑

  • 您如何清理？每天在开发和 CI 中运行数十次测试将开始用测试数据填充沙箱。

  • 沙箱测试起来是否缓慢且令人讨厌？开发人员是否会对等待集成测试在其机器上或在 CI 中完成感到恼火？

  • 沙箱是否完全不稳定？您现在是否在您的构建中引入了随机失败的测试？

围绕 API 的适配器，带有集成测试，权衡

优点
  • 遵守“不要模拟你不拥有的东西”规则。
  • 我们呈现一个简单的 API，更容易模拟
  • 我们停止摆弄诸如 requests.get.return_value.json.return_value 之类的模拟
  • 如果我们曾经更改我们的第三方，我们的适配器的 API 很有可能不会更改。因此我们的核心应用程序代码（及其测试）不需要更改。
缺点
  • 我们在我们的应用程序代码中添加了一个额外的层，对于简单的情况，这可能是不必要的复杂性
  • 集成测试强烈依赖于您的第三方提供良好的测试沙箱
  • 集成测试可能很慢且不稳定

选项：vcr.py

此时我想快速提及 vcr.py。

VCR 是一个非常简洁的解决方案。它允许您针对真实端点运行测试，然后它捕获传出和传入的请求，并将它们序列化到磁盘。下次您运行测试时，它会拦截您的 HTTP 请求，将它们与保存的请求进行比较，并重放过去的响应。

最终结果是，您有一种运行集成测试的方法，该方法具有真实的模拟响应，但实际上无需与外部第三方对话。

在任何时候，您也可以触发针对真实 API 的测试运行，它将更新您保存的响应文件。这为您提供了一种检查事物是否定期发生更改并在发生更改时更新记录响应的方法。

正如我所说，这是一个非常简洁的解决方案，我已成功使用它，但它确实有一些缺点

  • 首先，工作流程可能非常令人困惑。当您仍在改进您的集成时，您的代码将会更改，并且罐头响应也会更改，并且很难跟踪磁盘上的内容、什么是假的以及什么不是假的。一个人通常可以理解它，但对于团队的其他成员来说，这是一个陡峭的学习曲线。如果代码只是不经常更改，那可能会特别痛苦，因为时间足够长，每个人都会忘记。

  • 其次，当您的请求中具有随机数据（例如唯一 ID）时，vcr.py 很难配置。默认情况下，它会查找与它记录的请求完全相同的请求。您可以配置“匹配器”以在识别请求时有选择地忽略某些字段，但这仅解决了一半的问题。

  • 如果您发出 POST 并跟进对同一 ID 的 GET，您或许可以配置匹配器以忽略请求中的 ID，但响应仍将包含旧 ID。这将破坏您自己方面基于这些 ID 执行任何逻辑的任何逻辑。

vcr.py 权衡

优点
  • 通过重放罐头响应，为您提供了一种将测试与外部依赖项隔离的方法
  • 可以随时针对真实 API 重新运行
  • 无需更改应用程序代码
缺点
  • 团队成员可能难以理解
  • 处理随机生成的数据很困难
  • 模拟基于状态的工作流程具有挑战性

选项：为您自己的集成测试构建自己的伪造品

我们现在进入了危险领域，我们即将提出的解决方案不一定是所有情况下的好主意。就像您在互联网上的随机博客上找到的任何解决方案一样，但仍然如此。

那么您何时会考虑这样做呢？

  • 如果集成不是您应用程序的核心，即它是附带功能
  • 如果您编写的大部分代码以及您想要的反馈不是关于集成问题，而是关于您应用程序中的其他内容
  • 如果您真的无法弄清楚如何以另一种方式解决集成测试的问题（重试？也许无论如何它们都是一个好主意？）

那么您可以考虑构建您自己的外部 API 的伪造版本。然后，您可以在 docker 容器中启动它，与您的测试代码一起运行它，并与它而不是真实 API 对话。

伪造第三方通常很简单。围绕 CRUD 数据模型的 REST API 可能只是将 json 对象弹出和弹出内存字典，例如

from flask import Flask, request

app = Flask('fake-cargo-api')

SHIPMENTS = {}  # type: Dict[str, Dict]

@app.route('/shipments/', methods=["GET"])
def list_shipments():
    print('returning', SHIPMENTS)
    return {'items': list(SHIPMENTS.values())}


@app.route('/shipments/', methods=["POST"])
def create_shipment():
    new_id = uuid.uuid4().hex
    refs = {s['client_reference'] for s in SHIPMENTS.values()}
    if request.json['client_reference'] in refs:
        return 'already exists', 400
    SHIPMENTS[new_id] = {'id': new_id, **request.json}
    print('saved', SHIPMENTS)
    return 'ok', 201


@app.route('/shipments/<shipment_id>/', methods=["PUT"])
def update_shipment(shipment_id):
    existing = SHIPMENTS[shipment_id]
    SHIPMENTS[shipment_id] = {**existing, **request.json}
    print('updated', SHIPMENTS)
    return 'ok', 200

这并不意味着您永远不会针对第三方 API 进行测试，但您现在已经给自己选择不这样做。

  • 也许您在 CI 中针对真实 API 进行测试，但在开发中不进行测试

  • 也许您有一种方法可以将某些 PR 标记为需要“真实” API 集成测试

  • 也许您在 CI 中有一些逻辑会查看给定 PR 中已更改的代码，尝试发现任何与第三方 API 相关的内容，然后仅针对真实 API 运行

选项：合同测试

我不确定“合同测试”是否是一个真正的术语，但其思想是测试第三方 API 的行为是否符合合同。它是否执行您需要它执行的操作。

它们与集成测试不同，因为您可能没有测试您的适配器本身，并且它们倾向于一次针对单个端点。诸如

  • 检查给定端点的数据格式和数据类型。您需要的所有字段都在那里吗？

  • 如果第三方 API 有您需要解决的错误，您可以在测试中重现该错误，以便您知道他们何时修复它

这些测试往往比集成测试更轻量级，因为它们通常是只读的，因此它们受与清理相关的问题的影响较小。您可能会认为它们除了集成测试之外也很有用，或者如果无法进行适当的集成测试，它们可能是一个有用的备用选项。以类似的方式，您可能需要有选择地针对您的第三方运行合同测试的方法。

您还可以针对您的伪造 API 运行您的合同测试。

当您针对您自己的伪造 API 以及真实 API 运行您的合同测试时，您正在确认您的伪造品的质量。有些人称之为已验证的伪造品（另请参阅“停止模拟并开始测试”。）

选项：DI

我们仍然存在使用 mock.patch 将我们束缚于导入适配器的特定方式的问题。我们还需要记住在任何可能使用第三方适配器的测试中设置该模拟。

使依赖项显式化并使用 DI 可以解决这些问题

同样，我们在这里处于危险领域。Python 人们对 DI 持怀疑态度，并且这两个问题都不是那么大的问题。但是 DI 确实为我们带来了一些好处，所以请以开放的心态继续阅读。

首先，您可能想为您的依赖项显式定义一个接口。您可以使用 abc.ABC，或者如果您反对继承，可以使用新式的 typing.Protocol

class CargoAPI(Protocol):

    def get_latest_eta(self, reference: str) -> date:
        ...

    def sync(self, shipment: Shipment) -> None:
        ...

现在我们可以在需要的地方添加我们的显式依赖项，用某个函数的新显式参数替换硬编码的 import。可能甚至带有类型提示

def create_shipment(
    quantities: Dict[str, int],
    incoterm: str,
    cargo_api: CargoAPI
) -> Shipment:
    ...
    # rest of controller code essentially unchanged.

这对我们的测试有什么影响？好吧，我们不需要调用 with mock.patch()，我们可以创建一个独立的模拟，并将其传入

def test_create_shipment_syncs_to_api():
    mock_api = mock.Mock()
    shipment = create_shipment({'sku1': 10}, incoterm='EXW', cargo_api=mock_api)
    assert mock_api.sync.call_args == mock.call(shipment)

DI 权衡

优点
  • 无需记住执行 mock.patch()，函数参数始终需要依赖项
缺点
  • 我们为我们的函数添加了一个“不必要”的额外参数

Yeray Díaz 的演讲 将导入作为反模式 中令人难忘地倡导了这种将 import 更改为显式依赖项的做法

到目前为止，您可能会认为优点不足以证明缺点的合理性？好吧，如果我们再进一步并真正致力于 DI，您可能会同意。

选项：为您自己的单元测试构建自己的伪造品

就像我们可以为集成测试构建我们自己的伪造品一样，我们也可以为单元测试构建我们自己的伪造品。是的，它比 mock_api = mock.Mock() 行数更多，但不多

class FakeCargoAPI:
    def __init__(self):
        self._shipments = {}

    def get_latest_eta(self, reference) -> date:
        return self._shipments[reference].eta

    def sync(self, shipment: Shipment):
        self._shipments[shipment.reference] = shipment

    def __contains__(self, shipment):
        return shipment in self._shipments.values()

这次伪造品是在内存中和进程中的，但同样，它只是某种容器（在本例中为字典）的薄包装器。

get_latest_eta() 和 sync() 是我们需要定义的两个方法，以使其模拟真实 API（并符合 Protocol）。

当您做对这件事时，或者如果您需要更改它，mypy 会告诉您

__contains__ 只是一个小的语法糖，它允许我们在我们的测试中使用 assert in，这看起来不错。这是一个 Bob 的东西。

def test_create_shipment_syncs_to_api():
    api = FakeCargoAPI()
    shipment = create_shipment({'sku1': 10}, incoterm='EXW', cargo_api=api)
    assert shipment in api

为什么要为此烦恼？

用于单元测试的手工伪造品，权衡

优点
  • 测试可以更具可读性，不再有 mock.call_args == call(foo,bar) 之类的东西
  • 👉我们的伪造品对我们的适配器的 API 施加了设计压力👈
缺点
  • 测试中更多的代码
  • 需要使伪造品与真实事物保持同步

在我们看来，设计压力是关键论点。因为手工制作伪造品是更多的工作，它迫使我们思考我们的适配器的 API，并且它激励我们保持简单。

如果您回想起我们最初构建包装器的决定，在我们的玩具示例中，很容易决定适配器应该是什么样子，我们只需要一个名为 sync() 的公共方法。在现实生活中，有时很难弄清楚什么属于适配器，什么保留在业务逻辑中。通过强迫我们自己构建伪造品，我们真正看到了我们正在抽象出的事物的形状。

为了获得奖励积分，您甚至可以在用于单元测试的伪造类和用于集成测试的伪造类之间共享代码。

回顾

  • 一旦您与外部 API 的集成变得超出琐碎的范围，模拟和修补就开始变得非常痛苦

  • 考虑在您的 API 周围抽象出一个包装器

  • 使用集成测试来测试您的适配器，并使用单元测试来测试您的业务逻辑（并检查您是否正确调用了您的适配器）

  • 考虑为您自己的单元测试编写您自己的伪造品。它们将帮助您找到一个好的抽象。

  • 如果您想要一种让开发人员或 CI 在不依赖外部 API 的情况下运行测试的方法，请考虑也编写第三方 API 的功能齐全的伪造品（实际的 Web 服务器）。

  • 为了获得奖励积分，这两个伪造品可以共享代码。

  • 有选择地针对伪造 API 和真实 API 运行集成测试可以验证两者是否随着时间的推移继续工作。

  • 您还可以考虑为此目的添加更有针对性的“合同测试”。

如果您想使用此博客文章中的代码进行实验，您可以在此处查看

先前技术