使用响应存根进行 CarrierLookup 集成离线测试

在允许 CarrierLookup 消费者使用网络之前，注入传输层并测试明确的合成响应存根。

CarrierLookup API 模拟测试应替换 HTTP 传输层，拒绝意外的网络访问，并断言您的应用程序如何处理每种已记录的结果。合成存根可以在不提交电话号码或消耗余额的情况下测试消费者逻辑。

在 HTTP 请求之前设置测试接缝

CRM 导入程序可能通过了其“成功路径”测试，但仍将一个 无法判定 的号码标记为无效。它也可能通过了模拟浏览器测试，但其后端却发起了真实的付费请求。请将您的测试接缝放置在执行出站 HTTP 交换的后端函数中。

消费者应接受一个传输依赖项，而不是在内部发现真实的客户端。注入的测试传输层接收预期的路径和请求体，对其进行断言，并返回一个合成的信封。测试中不包含 API 密钥。同时，请确保 CI 作业中不包含实时凭据。

使用独立的网络限制作为后盾。一个返回便捷数据的模拟程序并不能证明所有出站调用都被拦截。请在 CI 容器或运行器边界拒绝出站流量，并在尝试网络访问时使作业失败。语言层面的防护有助于捕获常见错误，但无法覆盖子进程或所有网络库。

在编写存根之前定义结果

我们的同步运营商服务使用 service_type=carrier。单项响应具有扁平化的 data 字段，包括 identifier、carrier、underlying_carrier、number_type、country_code、region 和 city。读取 data.extra.carrier 会测试一种虚构的结构。

运营商值描述的是原始分配情况。它们不应填充 CRM 中名为“当前网络”的字段。同样，空分配结果或 exists=false 均不能确立电话的有效性。请使用独立的应用程序结果标签，以便后续的报告查询不会意外合并这些状态。

运行小型合成存根套件

将以下代码保存为 carrier_consumer_test.py 并运行 python3 carrier_consumer_test.py。它使用了 Python 的标准库。该示例是一个最小化的消费者接缝，带有合成标识符和特意选择的值；它从不联系 CarrierLookup。在保持存根断言的同时，将演示用的 client 和分类器替换为您自己的适配器。

import socket
import unittest
from unittest.mock import patch

FIELDS = ("carrier", "underlying_carrier", "number_type",
          "country_code", "region", "city")
A, B = "+12025550101", "+12025550102"  # 仅限合成测试数据。

def classify(envelope):
    code = envelope["code"]
    if code != 0:
        return {42200: "undetermined", 42901: "busy",
                50400: "timeout"}.get(code, "error")
    data = envelope["data"]
    if data["service_type"] != "carrier":
        raise ValueError("unexpected service")
    return "completed" if data["carrier"] else "completed_empty"

def client(transport, identifiers):
    multi = len(identifiers) > 1
    path = "/api/v1/batch-check" if multi else "/api/v1/check"
    body = {"service_type": "carrier"}
    body["identifiers" if multi else "identifier"] = (
        identifiers if multi else identifiers[0])
    return transport(path, body)

def completed(identifier, carrier=""):
    return {"code": 0, "msg": "synthetic fixture", "data": {
        "service_type": "carrier", "identifier": identifier,
        **dict.fromkeys(FIELDS, ""), "carrier": carrier}}

class ConsumerTests(unittest.TestCase):
    def setUp(self):
        self.guard = patch.object(
            socket.socket, "connect",
            side_effect=AssertionError("network forbidden"))
        self.guard.start()
        self.addCleanup(self.guard.stop)

    def test_single_flat_data_and_outcomes(self):
        for value, expected in [("Example Allocation", "completed"),
                                ("", "completed_empty")]:
            def fake(path, body):
                self.assertEqual(path, "/api/v1/check")
                self.assertEqual(body, {"service_type": "carrier",
                                        "identifier": A})
                return completed(A, value)
            result = client(fake, [A])
            self.assertNotIn("extra", result["data"])
            self.assertNotIn("registered", result["data"])
            self.assertEqual(classify(result), expected)

    def test_error_branches(self):
        for code, expected in [(42200, "undetermined"),
                               (42901, "busy"), (50400, "timeout")]:
            with self.subTest(code=code):
                result = client(lambda path, body: {
                    "code": code, "msg": "synthetic error",
                    "data": None}, [A])
                self.assertEqual(classify(result), expected)

    def test_multi_order_and_no_result(self):
        def fake(path, body):
            self.assertEqual(path, "/api/v1/batch-check")
            self.assertEqual(body, {"service_type": "carrier",
                                    "identifiers": [B, A]})
            return {"code": 0, "msg": "synthetic fixture", "data": {
                "service_type": "carrier", "total": 2,
                "succeeded": 1, "failed": 1, "results": [
                    {"identifier": B, "exists": True,
                     **dict.fromkeys(FIELDS, "")},
                    {"identifier": A, "exists": False}]}}
        data = client(fake, [B, A])["data"]
        rows = data["results"]
        self.assertEqual(data["service_type"], "carrier")
        self.assertEqual([r["identifier"] for r in rows], [B, A])
        self.assertEqual(len(rows), data["total"])
        outcomes = ["completed" if r["exists"] else "no_result"
                    for r in rows]
        self.assertEqual(outcomes, ["completed", "no_result"])
        self.assertNotIn("carrier", rows[1])

    def test_guard_rejects_an_escape(self):
        with socket.socket() as sock:
            with self.assertRaisesRegex(AssertionError, "network forbidden"):
                sock.connect(("127.0.0.1", 9))

if __name__ == "__main__":
    unittest.main()

该示例从其传输层返回已解码的信封。它测试的是请求构建和结果处理，而非 HTTP 解析。请在您的真实传输边界处保留针对 HTTP 状态、标头和格式错误 JSON 的存根。特别是，42901 响应包含 Retry-After；响应标头存根应验证您的传输层是否将其保留给调用者。请勿将新的重试间隔硬编码到结果分类器中。

套接字防护旨在测试其自身的失败路径。它会阻止此 Python 进程的套接字连接方法；对于其他路径，独立的 CI 出站限制仍然是必要的。意外的付费调用应导致作业失败，而不是成为一个未被注意到的额外存根。

使多项结果断言对排序错误敏感

多项存根按 B 在 A 之前的顺序提交。这一点很重要：如果一个测试提交按字母顺序排序的标识符并接收到相同的排序结果，可能会漏掉一个会对响应进行排序的适配器。请将每个返回的标识符与其输入位置进行匹配，并断言总长度。

第一个多项条目产生一个带有空分配值的已完成结果。第二个条目具有 exists=false 且没有运营商字段。在映射到应用程序记录时，请保留这种差异。如果您的生产适配器丢弃了失败的行，存根测试应该失败，因为否则后续行可能会偏移到错误的联系人上。

在实际的 CRM 映射函数中添加一个断言：无法判定或失败的查询必须保持之前的分配观察结果不变，或者根据您的策略将其明确标记为陈旧。它绝不能将记录覆盖为“无效号码”。同样的测试应拒绝从分配的地区或城市推导当前位置。

同步多项超时会导致整个请求失败，而不会产生部分结果。请将整个请求的超时分支与包含不成功条目的成功信封分开测试。对于消费者而言，这些是不同的输入。

保持离线信心与实时验证的分离

这些测试确立了您的代码如何响应存根所代表的契约。它们并不证明当前的提供商可用性、国家/地区接受度、账户权限或计费情况。虚构的存根也无法证明实时提供商会为特定号码返回特定的分配字段。

请将存根来源保留在测试旁边：CarrierLookup API reference、审查日期以及您的适配器所消耗的字段。当已记录的契约发生变化时，请更新存根，然后检查旧的应用程序数据是否需要兼容的读取路径。

当集成验证需要时，请运行单独授权的实时冒烟测试，使用已批准的输入及其自身的余额预算。不要让普通的 CI 依赖于该调用。首先，将测试套件中的一个实时客户端导入替换为注入的传输层，并确认尝试的网络逃逸会立即失败。

了解更多

选择适合您工作流程下一步的产品信息。

• 查看产品文档 →
• 联系团队 →