微服务接口测试实战：从契约测试到全链路验证

// 1. 安装Pact// npm install -D @pact-foundation/pact// 2. 编写消费者测试const { Pact } = require('@pact-foundation/pact');describe('Order Service Consumer', () => {  const provider = new Pact({    consumer: 'WebFrontend',    provider: 'OrderService',    port: 8080,    log: './logs/pact.log',    dir: './pacts',  });  beforeAll(() => provider.setup());  it('should create an order', async () => {    // 定义期望的接口交互    await provider.addInteraction({      state: 'order does not exist',      uponReceiving: 'a request to create an order',      withRequest: {        method: 'POST',        path: '/orders',        body: {          userId: 123,          productId: 456,          quantity: 2        }      },      willRespondWith: {        status: 201,        body: {          orderId: 'abc123',          status: 'CREATED'        }      }    });    // 执行消费者代码    const response = await createOrder({      userId: 123,      productId: 456,      quantity: 2    });    expect(response.orderId).toEqual('abc123');  });  afterAll(() => provider.finalize());});

# 1. 安装pact-python# pip install pact-python# 2. 编写提供者验证import pytestfrom pact import Verifier@pytest.fixturedef verifier():    return Verifier(        provider_name='OrderService',        provider_base_url='http://localhost:5000'    )def test_provider_fulfills_consumer_contract(verifier):    # 从Pact Broker拉取契约文件并验证    success, logs = verifier.verify_pacts(        'http://pact-broker/pacts/provider/OrderService/consumer/WebFrontend/latest'    )    assert success == 0, f"Pact verification failed:\n{logs}"

# tests/test_order_service.pyimport pytestimport requestsfrom config import BASE_URL, AUTH_TOKEN# Fixture：创建测试数据@pytest.fixturedef create_test_user():    """创建测试用户"""    response = requests.post(f"{BASE_URL}/users", json={        "username": "test_user_001",        "email": "test@example.com"    }, headers={"Authorization": f"Bearer {AUTH_TOKEN}"})    user_id = response.json()["id"]    yield user_id    # Teardown：清理测试数据    requests.delete(f"{BASE_URL}/users/{user_id}",                     headers={"Authorization": f"Bearer {AUTH_TOKEN}"})# Fixture：获取API客户端@pytest.fixturedef api_client():    session = requests.Session()    session.headers.update({        "Authorization": f"Bearer {AUTH_TOKEN}",        "Content-Type": "application/json"    })    return sessionclass TestOrderService:    """订单服务接口测试"""    def test_create_order_success(self, api_client, create_test_user):        """测试创建订单 - 成功场景"""        user_id = create_test_user
        # 调用创建订单接口        response = api_client.post(f"{BASE_URL}/orders", json={            "userId": user_id,            "productId": "PROD-001",            "quantity": 2,            "shippingAddress": {                "province": "广东",                "city": "深圳"            }        })        # 断言：状态码        assert response.status_code == 201, f"Expected 201, got {response.status_code}"
        # 断言：响应体结构        data = response.json()        assert "orderId" in data, "Response should contain orderId"        assert "status" in data, "Response should contain status"        assert data["status"] == "CREATED", "Order status should be CREATED"
        # 断言：数据结构类型        assert isinstance(data["orderId"], str), "orderId should be string"        assert len(data["orderId"]) > 0, "orderId should not be empty"    @pytest.mark.parametrize("invalid_data,expected_error", [        ({"userId": ""}, "userId is required"),        ({"productId": "INVALID"}, "product not found"),        ({"quantity": 0}, "quantity must be positive"),    ])    def test_create_order_validation(self, api_client, invalid_data, expected_error):        """参数化测试：创建订单 - 参数校验"""        response = api_client.post(f"{BASE_URL}/orders", json=invalid_data)
        assert response.status_code == 400        assert expected_error in response.json()["message"]    def test_get_order_by_id(self, api_client, create_test_user):        """测试查询订单详情"""        # 先创建订单        create_resp = api_client.post(f"{BASE_URL}/orders", json={            "userId": create_test_user,            "productId": "PROD-001",            "quantity": 1        })        order_id = create_resp.json()["orderId"]        # 查询订单        get_resp = api_client.get(f"{BASE_URL}/orders/{order_id}")
        assert get_resp.status_code == 200        assert get_resp.json()["orderId"] == order_id        assert "createdAt" in get_resp.json()    @pytest.mark.slow    def test_order_timeout_scenario(self, api_client):        """测试订单服务超时处理"""        # 模拟超时场景（需要Mock服务配合）        with pytest.raises(requests.exceptions.Timeout):            api_client.post(                f"{BASE_URL}/orders",                json={"userId": "123", "productId": "PROD-001"},                timeout=0.001  # 极短超时            )

# 使用JSON文件驱动测试数据# data/test_cases.json[  {    "name": "正常下单",    "request": {"userId": "123", "productId": "PROD-001", "quantity": 2},    "expected_status": 201,    "expected_message": "Order created successfully"  },  {    "name": "库存不足",    "request": {"userId": "123", "productId": "PROD-001", "quantity": 9999},    "expected_status": 409,    "expected_message": "Insufficient stock"  }]# tests/test_data_driven.pyimport jsonimport pytest@pytest.mark.parametrize("test_case",     json.load(open("data/test_cases.json")))def test_create_order_data_driven(api_client, test_case):    """数据驱动：从JSON文件读取测试用例"""    response = api_client.post(        f"{BASE_URL}/orders",         json=test_case["request"]    )
    assert response.status_code == test_case["expected_status"]    assert test_case["expected_message"] in response.text

# tests/test_order_e2e.pyimport pytestimport requestsimport timefrom utils.api_client import APIClientfrom utils.data_factory import create_test_product, cleanup_test_dataclass TestOrderE2EFlow:    """订单全链路E2E测试"""    def setup_class(self):        self.api = APIClient()        self.test_user = self.api.create_user({"username": "e2e_test_user"})        self.test_product = create_test_product({"name": "测试商品", "price": 100, "stock": 50})    def teardown_class(self):        cleanup_test_data(self.test_user, self.test_product)    def test_complete_order_flow(self):        """测试完整下单流程"""        api = self.api        # 1. 用户登录        login_resp = api.post("/auth/login", json={            "username": self.test_user["username"],            "password": self.test_user["password"]        })        assert login_resp.status_code == 200        token = login_resp.json()["token"]        api.set_token(token)        # 2. 添加商品到购物车        cart_resp = api.post("/cart/add", json={            "productId": self.test_product["id"],            "quantity": 2        })        assert cart_resp.status_code == 200        cart_id = cart_resp.json()["cartId"]        # 3. 提交订单        order_resp = api.post("/orders/create", json={            "cartId": cart_id,            "shippingAddress": {                "province": "广东",                "city": "深圳",                "detail": "南山区科技园"            }        })        assert order_resp.status_code == 201        order_id = order_resp.json()["orderId"]        assert order_resp.json()["status"] == "CREATED"        # 4. 获取支付信息        payment_resp = api.post("/payments/create", json={            "orderId": order_id,            "paymentMethod": "ALIPAY"        })        assert payment_resp.status_code == 201        payment_id = payment_resp.json()["paymentId"]        pay_url = payment_resp.json()["payUrl"]        # 5. 模拟支付回调（Mock支付网关）        callback_resp = api.post("/payments/callback", json={            "paymentId": payment_id,            "status": "SUCCESS",            "transactionId": "MOCK_TXN_001"        })        assert callback_resp.status_code == 200        # 6. 验证订单状态更新为已支付        time.sleep(1)  # 等待异步处理        order_detail_resp = api.get(f"/orders/{order_id}")        assert order_detail_resp.status_code == 200        assert order_detail_resp.json()["status"] == "PAID"        # 7. 验证库存扣减        product_resp = api.get(f"/products/{self.test_product['id']}")        assert product_resp.json()["stock"] == 48  # 50 - 2        print(f"订单全链路测试通过！Order ID: {order_id}")    @pytest.mark.chaos    def test_order_flow_with_payment_timeout(self):        """混沌测试：支付服务超时场景"""        # 使用Chaos Mesh注入网络延迟        # 预期：订单状态应为"PAYMENT_PENDING"，并有重试机制        # ...（省略具体实现）        pass