微服务测试实战教程

def test_当用户查询他人订单时_网关拒绝(gateway, user_token):
    response = gateway.get("/api/orders/OTHER-1", token=user_token)
    assert response.status_code == 403
    assert "paidAmount" not in response.text

def test_当重复提交相同幂等键时_只创建一个订单(gateway, payload):
    headers = {"Idempotency-Key": "case-order-001"}
    first = gateway.post("/api/orders", json=payload, headers=headers)
    second = gateway.post("/api/orders", json=payload, headers=headers)
    assert first.status_code == second.status_code == 200
    assert first.json()["data"]["orderId"] == second.json()["data"]["orderId"]

def test_inventory_reserve_contract(inventory_client):
    response = inventory_client.reserve({
        "orderId": "O-CONTRACT-1", "skuId": "SKU-1", "quantity": 1
    })
    assert response.status_code == 200
    body = response.json()
    assert body["businessCode"] == "SUCCESS"
    assert isinstance(body["data"]["reservationId"], str)
    assert body["data"]["status"] == "RESERVED"

# 给支付上游增加 1500ms 延迟
curl -X POST http://localhost:8474/proxies/payment/toxics \
  -H "Content-Type: application/json" \
  -d '{"name":"latency","type":"latency","attributes":{"latency":1500}}'

# 用同一幂等键发起支付；成功时 HTTP 200，仍检查响应体
curl -X POST http://localhost:8080/api/payments \
  -H "Idempotency-Key: pay-O-1001" \
  -d '{"orderId":"O-1001","amount":"80.00"}'

def test_当优惠服务持续超时时_熔断并安全降级(shop):
    shop.inject_timeout(service="coupon", count=10)
    results = [shop.preview_order("SKU-1") for _ in range(12)]

    assert any(r.headers.get("X-Circuit-State") == "OPEN" for r in results)
    assert all(r.status_code == 200 for r in results)
    assert all(r.json()["businessCode"] == "COUPON_UNAVAILABLE" for r in results)
    assert shop.count_created_orders() == 0

def wait_until(fetch, predicate, timeout=10):
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        value = fetch()
        if predicate(value):
            return value
        time.sleep(0.2)
    raise AssertionError("business state did not converge")

def test_当支付失败时_库存最终释放(order_api, inventory_api, order):
    order_api.pay(order["orderId"], force="DECLINED")
    reservation = wait_until(
        lambda: inventory_api.get(order["reservationId"]),
        lambda x: x["status"] == "RELEASED",
    )
    assert reservation["releasedQuantity"] == 1

curl -i http://localhost:8080/api/orders/O-1001 \
  -H "Authorization: Bearer $TEST_TOKEN" \
  -H "traceparent: 00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"

# 检查：网关、订单、库存、支付 span 共享同一 traceId
# 检查：错误 span 带 order.id、peer.service 与 retry.count