site logo

Marico's space

site logo

Marico's space

首页
我的日志
我的游记
世界·视野
编程技术

技术合作:

look@marico.cc

+86 18660050334

 

应用 API 测试框架：真实微服务案例

前端技术 2026-06-30 14:50:08 5

大部分讲 API 测试的教程，都是测一个独立的接口。真实项目没这么干净：服务之间互相调 HTTP、共享 JWT 令牌、只有在全部跑起来才会出 bug。

这篇换个做法。我们来搭一个小型但真实的微服务系统——user-service（用户服务）、product-service（商品服务）、order-service（订单服务），然后用两套不同的测试框架写完整测试套件：Node.js 侧用 Jest + Supertest，Python 侧用 Pytest + HTTPX。文章里所有代码示例都在可运行的仓库里。

待测系统长什么样

动手写测试之前，先搞清楚我们在折腾什么。

┌──────────────┐ JWT 认证 ┌────────────────┐
│ user-service│ ◄───────────────► │ order-service │
│ :3001 │ │ :3003 │
└──────────────┘ └──────┬─────────┘ │ 库存校验 ▼ ┌────────────────┐ │product-service │ │ :3002 │ └────────────────┘
服务 职责
user-service 注册、JWT 登录、用户信息
product-service 商品目录、库存管理
order-service 创建订单；调用 product-service 校验库存并扣减

order-service 用和 user-service 相同的 JWT 密钥来验证调用方身份——这是生产环境中很常见的去中心化认证模式。

两套框架，两种策略

我们采用两种不同的测试策略，分别匹配各自框架最擅长的场景：

Jest + Supertest Pytest + HTTPX
测试层级 单元测试 / 进程内 集成测试 / E2E
服务是否运行？ 否 — 直接导入 app 是 — 子进程方式
HTTP mock？ 是（order-service 测试） 否 — 真实 HTTP
速度 快（约 1 秒） 较慢（约 5–10 秒）
能发现的问题 逻辑 bug、契约违背 跨服务连接问题

两层都不可或缺。单元测试快且精准；集成测试证明整个系统真的能跑通。

第一部分 — Jest + Supertest（Node.js）

为什么选 Supertest？

Supertest 包装你的 Express 应用，让你可以对它发起真实的 HTTP 请求但不需要启动服务器。它在内部绑定了临时端口，所以测试既快又完全和网络隔离。

cd tests/jest && npm install

测试 user-service

user-service 负责注册、登录、获取用户信息。先看注册接口的测试：

// tests/jest/user-service.test.js
const request = require("supertest");
const { app, users } = require("../../services/user-service/src/index"); // 注册用户的辅助函数
const registerUser = (overrides = {}) => request(app) .post("/auth/register") .send({ name: "Alice", email: "alice@example.com", password: "Pass1234!", ...overrides }); beforeEach(() => { users.clear(); // 在测试之间重置内存状态
}); describe("POST /auth/register", () => { it("creates a new user and returns 201 with user data (no password)", async () => { const res = await registerUser(); expect(res.status).toBe(201); expect(res.body).toMatchObject({ name: "Alice", email: "alice@example.com", role: "user" }); expect(res.body).not.toHaveProperty("password"); // 永远不暴露密码哈希 }); it("returns 409 when email is already registered", async () => { await registerUser(); const res = await registerUser(); // 重复注册 expect(res.status).toBe(409); expect(res.body.error).toMatch(/already registered/i); });
});

几个值得注意的点：

  1. beforeEach(() => users.clear()) — 服务导出它的内存 Map，这样测试可以在不重启进程的情况下重置状态。这是为可测试性专门做的设计。
  2. expect(res.body).not.toHaveProperty("password") — 这个断言很容易被忘掉，但很关键。一旦回归，密码哈希就泄漏给客户端了。
  3. toMatch(/already registered/i) — 用正则匹配而非精确字符串，让测试对措辞的小改动更宽容。

测试登录流程

describe("POST /auth/login", () => { beforeEach(async () => { await registerUser(); // 确保用户已存在 }); it("returns a JWT token on valid credentials", async () => { const res = await request(app) .post("/auth/login") .send({ email: "alice@example.com", password: "Pass1234!" }); expect(res.status).toBe(200); expect(res.body).toHaveProperty("token"); expect(typeof res.body.token).toBe("string"); }); it("returns 401 on wrong password", async () => { const res = await request(app) .post("/auth/login") .send({ email: "alice@example.com", password: "wrongpassword" }); expect(res.status).toBe(401); });
});

Mock order-service 中的跨服务 HTTP 调用

这里开始有意思了。order-service 通过 HTTP 调用 product-service 来校验库存、下单后扣减库存。在单元测试中，不想启动真正的 product-service，需要精确控制它的返回值。

这个服务专门留了一个口子，就是为了方便测试：

// services/order-service/src/index.js (简化版)
let httpClient = null; const getHttpClient = () => { if (httpClient) return httpClient; // 注入的 mock return require("node-fetch"); // 生产环境用真实库
}; module.exports = { app, orders, setHttpClient };

测试时注入一个 jest.fn() 来模拟 product-service 的响应：

// tests/jest/order-service.test.js
const { app, orders, setHttpClient } = require("../../services/order-service/src/index"); const mockFetch = (productOverrides = {}) => { const defaultProduct = { id: 1, name: "Mechanical Keyboard", price: 129.99, stock: 50, ...productOverrides, }; return jest.fn().mockImplementation((url, options = {}) => { const method = options.method || "GET"; if (method === "GET" && url.includes("/products/")) { return Promise.resolve({ ok: true, json: () => Promise.resolve(defaultProduct), }); } if (method === "PATCH" && url.includes("/stock")) { return Promise.resolve({ ok: true, json: () => Promise.resolve({ ...defaultProduct, stock: defaultProduct.stock - 2 }), }); } });
}; afterEach(() => { setHttpClient(null); // 恢复真实 fetch
});

现在可以测试"库存不足"路径了，完全不碰真实服务：

it("returns 422 when stock is insufficient", async () => { setHttpClient(mockFetch({ stock: 1 })); // 库存只有 1 const token = makeToken(); const res = await request(app) .post("/orders") .set("Authorization", `Bearer ${token}`) .send({ items: [{ productId: 1, quantity: 5 }] }); // 申请 5 个 expect(res.status).toBe(422); expect(res.body.error).toMatch(/insufficient stock/i);
});

再测正常路径，验证总价计算是否正确：

it("creates an order and returns 201 with confirmed status", async () => { setHttpClient(mockFetch()); // 价格 129.99 const token = makeToken(); const res = await request(app) .post("/orders") .set("Authorization", `Bearer ${token}`) .send({ items: [{ productId: 1, quantity: 2 }] }); expect(res.status).toBe(201); expect(res.body).toMatchObject({ status: "confirmed", userId: 1, total: 259.98, // 129.99 * 2 });
});

跑 Jest 测试套件

cd tests/jest
npm test

应该看到类似输出：

PASS user-service.test.js
PASS product-service.test.js
PASS order-service.test.js Test Suites: 3 passed, 3 total
Tests: 22 passed, 22 total
Time: 1.4 s

第二部分 — Pytest + HTTPX（Python）

为什么选 HTTPX？

HTTPX 是现代 Python HTTP 客户端，API 干净，支持异步。配合 pytest-asyncio，可以写感觉很自然的异步测试函数：

async def test_something(client): res = await client.get("/health") assert res.status_code == 200

和 Jest 测试直接导入 app 不同，Pytest 测试针对真实运行的服务。这意味着得先启动它们。

pip install -r tests/pytest/requirements.txt

conftest.py — 共享的 fixtures

conftest.py 是 Pytest fixtures 存放的地方。我们的做三件事：以子进程方式启动 Node 服务、等待它们健康检查通过、暴露认证相关的异步辅助函数。

# tests/pytest/conftest.py
import subprocess, time, os
import httpx, pytest USER_URL = os.getenv("USER_SERVICE_URL", "http://localhost:3001")
PRODUCT_URL = os.getenv("PRODUCT_SERVICE_URL", "http://localhost:3002")
ORDER_URL = os.getenv("ORDER_SERVICE_URL", "http://localhost:3003") def _wait_for_service(url: str, retries: int = 20, delay: float = 0.5) -> None: for _ in range(retries): try: r = httpx.get(f"{url}/health", timeout=2) if r.status_code == 200: return except httpx.TransportError: pass time.sleep(delay) raise RuntimeError(f"Service at {url} did not become healthy in time") @pytest.fixture(scope="session")
def services(): """ 在整个测试 session 期间启动所有三个 Node 服务，只启动一次。 """ procs = [] root = os.path.abspath(os.path.join(os.path.dirname(__file__), "../..")) for rel_dir, port in [ ("services/user-service", 3001), ("services/product-service", 3002), ("services/order-service", 3003), ]: p = subprocess.Popen( ["node", "src/index.js"], cwd=os.path.join(root, rel_dir), env={**os.environ, "PORT": str(port)}, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) procs.append(p) for url in [USER_URL, PRODUCT_URL, ORDER_URL]: _wait_for_service(url) yield {"user": USER_URL, "product": PRODUCT_URL, "order": ORDER_URL} for p in procs: p.terminate(); p.wait() @pytest.fixture
async def client(): async with httpx.AsyncClient(timeout=10) as c: yield c @pytest.fixture
async def auth_token(client, services, registered_user): res = await client.post( f"{services['user']}/auth/login", json={"email": registered_user["email"], "password": registered_user["password"]}, ) assert res.status_code == 200 return res.json()["token"]

scope="session" 让 services fixture 在整个测试运行期间只启动一次，比每个测试都重启快得多。

user-service 的集成测试

# tests/pytest/test_user_service.py
import pytest pytestmark = pytest.mark.asyncio class TestRegister: async def test_register_returns_201_with_user_data(self, client, services): res = await client.post( f"{services['user']}/auth/register", json={"name": "Bob", "email": "bob@pytest.com", "password": "PyTest1234!"}, ) assert res.status_code == 201 body = res.json() assert body["email"] == "bob@pytest.com" assert body["role"] == "user" assert "password" not in body # 永远不暴露密码哈希 async def test_register_returns_409_on_duplicate_email(self, client, services, registered_user): res = await client.post( f"{services['user']}/auth/register", json=registered_user, # 和 fixture 相同 email ) assert res.status_code == 409 assert "already registered" in res.json()["error"].lower() class TestLogin: async def test_login_returns_jwt_with_three_segments(self, client, services, registered_user): res = await client.post( f"{services['user']}/auth/login", json={"email": registered_user["email"], "password": registered_user["password"]}, ) assert res.status_code == 200 token = res.json().get("token") assert token is not None assert len(token.split(".")) == 3 # JWT 永远有三个 base64 段

核心：端到端订单流程

这个测试在一个场景里跑遍整个系统——注册、登录、浏览、下单、验证库存、验证订单历史：

# tests/pytest/test_order_flow.py
import pytest pytestmark = pytest.mark.asyncio class TestFullOrderFlow: async def test_end_to_end_order_flow(self, client, services): user_url, product_url, order_url = ( services["user"], services["product"], services["order"] ) # Step 1: 注册 reg_res = await client.post( f"{user_url}/auth/register", json={"name": "Charlie", "email": "charlie@e2e.com", "password": "E2ETest99!"}, ) assert reg_res.status_code in (201, 409) # Step 2: 登录 login_res = await client.post( f"{user_url}/auth/login", json={"email": "charlie@e2e.com", "password": "E2ETest99!"}, ) assert login_res.status_code == 200 token = login_res.json()["token"] headers = {"Authorization": f"Bearer {token}"} # Step 3: 浏览商品 products_res = await client.get(f"{product_url}/products") assert products_res.status_code == 200 products = products_res.json() item_a, item_b = products[0], products[1] stock_before = item_a["stock"] # Step 4: 下单 order_res = await client.post( f"{order_url}/orders", headers=headers, json={ "items": [ {"productId": item_a["id"], "quantity": 2}, {"productId": item_b["id"], "quantity": 1}, ] }, ) assert order_res.status_code == 201 order = order_res.json() assert order["status"] == "confirmed" assert len(order["items"]) == 2 assert order["total"] > 0 # Step 5: 验证库存已扣减 stock_res = await client.get(f"{product_url}/products/{item_a['id']}") assert stock_res.json()["stock"] == stock_before - 2 # Step 6: 验证订单隔离 — 另一个用户不能访问这个订单 reg2 = await client.post( f"{user_url}/auth/register", json={"name": "Dave", "email": "dave@e2e.com", "password": "E2ETest99!"}, ) login2 = await client.post( f"{user_url}/auth/login", json={"email": "dave@e2e.com", "password": "E2ETest99!"}, ) headers2 = {"Authorization": f"Bearer {login2.json()['token']}"} forbidden = await client.get(f"{order_url}/orders/{order['id']}", headers=headers2) assert forbidden.status_code == 403

这一个测试就能同时抓到 order-service 的库存扣减 bug、JWT 转发连接、订单隔离规则——全部一次性暴露。这就是端到端测试的威力。

跑 Pytest 测试套件

cd tests/pytest
pytest -v

输出：

tests/pytest/test_user_service.py::TestRegister::test_register_returns_201_with_user_data PASSED
tests/pytest/test_user_service.py::TestLogin::test_login_returns_jwt_with_three_segments PASSED
tests/pytest/test_order_flow.py::TestFullOrderFlow::test_end_to_end_order_flow PASSED
tests/pytest/test_order_flow.py::TestFullOrderFlow::test_order_isolation_between_users PASSED 5 passed in 6.4s

值得带走的核心模式

1. 为可测试性设计服务

导出内部状态（如内存 Map），让测试能重置它。为 HTTP 客户端添加依赖注入口子，这样就能 mock 外部调用。这不是 hack，是好的设计。

// 导出状态供测试清理
module.exports = { app, users }; // 导出 setter 让测试可以注入 mock HTTP 客户端
module.exports = { app, orders, setHttpClient };

2. 用 beforeEach 重置状态，不要依赖测试顺序

永远不要依赖测试按特定顺序运行。每个测试应该自己准备好所需条件并清理。 beforeEach(() => users.clear()) 这个模式强制了这一点。

3. 让合适的工具匹配合适的层级

  • Supertest：快，进程内，测试 Express 路由逻辑和中间件的理想选择。
  • HTTPX async：针对真实运行服务的集成测试和 E2E 测试，异步 API 很干净。
  • 别试图用一个工具包办所有事。

4. 测试契约，而不是实现细节

expect(res.body).not.toHaveProperty("password") 这样的测试是安全契约的文档。测试确切的 SQL 查询是什么就太脆弱了，测的东西也不对。

5. 搭好测试金字塔

 ┌─────────┐ │ E2E │ ← Pytest + HTTPX（少量，慢，高置信度） ┌┴─────────┴┐ │ 集成测试 │ ← Pytest + HTTPX（服务 + 真实 HTTP） ┌┴────────────┴┐ │ 单元测试 │ ← Jest + Supertest（大量，快，隔离） └──────────────┘

大部分测试应该在底层。顶层的少量 E2E 测试能抓到单元测试无法发现的服务连接问题。

代码仓库

本文所有代码都在这里：

github.com/andre-carbajal/api-testing-microservices

仓库包含：

  • 三个完整可运行的 Express 微服务
  • 22 个 Jest + Supertest 单元测试
  • Pytest 集成测试和 E2E 测试，含异步 fixtures
  • 自动管理服务生命周期的 conftest.py

Clone 下来，跑一下测试，故意搞坏点什么，看看哪个测试先抓到你。

小结

微服务架构下测 API，不只是发请求、查状态码那么简单。是关于每个关注点选对测试层级：

  • 用Jest + Supertest 快速、隔离地测试路由逻辑，完全掌控依赖。
  • 用Pytest + HTTPX 证明服务在真实运行环境中正确连接。
  • 用可测试性思维设计服务——可导出状态、可注入的 HTTP 客户端、干净的健康检查端点。

这些模式可扩展性好：不管你是三个服务还是三十个，同样的分层方法——大量快速单元测试、少量较慢集成测试、一点点 E2E 冒烟测试——让你有信心，而不会把测试套件变成噩梦。