接口测试不止校验状态码——从4个维度保障接口质量少返工50%

# jsonschema 做结构校验，pytest-xdist 用来并行跑用例pip install jsonschema requests pytest pytest-xdist pymysql pyyaml

# -*- coding: utf-8 -*-# JSON Schema 官方文档: https://json-schema.org/import jsonschema# ============ 订单创建接口 POST /api/v1/orders ============# 这份 Schema 写一次，所有下单相关用例的结构校验就都覆盖了ORDER_CREATE_SCHEMA = {    "$schema": "http://json-schema.org/draft-07/schema#",    "type": "object",    # required：这三个顶层字段少一个就直接报错    "required": ["code", "message", "data"],    "properties": {        "code": {            "type": "integer",            "enum": [0],              # 成功场景只接受 0        },        "message": {"type": "string"},  # 允许空串，但不接受 null        "data": {            "type": "object",            "required": ["order_id", "status", "total_amount", "items"],            "properties": {                # 订单号：20 位大写字母 + 数字，防 null、防空串、防截断                "order_id": {                    "type": "string",                    "pattern": "^[A-Z0-9]{20}$",                },                # 订单状态：枚举卡死，防后端偷偷加新状态没通知测试                "status": {                    "type": "string",                    "enum": ["pending", "paid", "shipped",                             "delivered", "completed", "closed"],                },                # 金额：number 类型 + 范围，防负数、防 0 元单、防字符串金额                "total_amount": {                    "type": "number",                    "minimum": 0.01,                    "maximum": 999999.99,                },                # 时间：ISO 8601，防后端某天改成毫秒时间戳                "created_at": {                    "type": "string",                    "pattern": "^\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}",                },                # 商品列表：至少 1 件，每件的 sku/数量/单价都要合法                "items": {                    "type": "array",                    "minItems": 1,                    "maxItems": 99,                    "items": {                        "type": "object",                        "required": ["sku_id", "quantity", "price"],                        "properties": {                            "sku_id": {"type": "string", "pattern": "^[A-Z0-9]{10}$"},                            "quantity": {"type": "integer", "minimum": 1},                            "price": {"type": "number", "minimum": 0},                        },                    },                },            },            # 关键一行：不允许出现 Schema 里没定义的字段            # 成本价、内部备注这类字段一旦漏出来，这里直接红            "additionalProperties": False,        },    },}def validate_order_response(response_json: dict) -> None:    """校验下单接口响应结构，失败时抛出带定位信息的断言错误。"""    try:        jsonschema.validate(instance=response_json, schema=ORDER_CREATE_SCHEMA)    except jsonschema.ValidationError as e:        # e.path 能告诉你是哪一层的哪个字段挂了，排查时特别省事        path = " → ".join(str(p) for p in e.path) or "根节点"        raise AssertionError(            f"响应结构校验失败\n  位置: {path}\n  原因: {e.message}\n  实际值: {e.instance}"        ) from e    except jsonschema.SchemaError as e:        raise RuntimeError(f"Schema 自己写错了，先改 Schema: {e}") from e

# -*- coding: utf-8 -*-import pytestimport requestsimport jsonschemafrom typing import Dict, Any, Optionaldef assert_schema(    response: requests.Response,    schema: Dict[str, Any],    status_code: int = 200,) -> dict:    """    维度一 · 结构校验统一入口    参数:        response:    requests 的响应对象        schema:      JSON Schema 字典        status_code: 期望的 HTTP 状态码，默认 200    返回:        解析后的响应 JSON（dict），方便后续继续断言业务字段    """    # Step1 先卡 HTTP 状态码，失败时把响应体前 300 字打出来    assert response.status_code == status_code, (        f"[维度一·结构] HTTP 状态码不符：期望 {status_code}，实际 {response.status_code}\n"        f"响应体: {response.text[:300]}"    )    # Step2 能不能解析成 JSON，这一步经常挂在网关返回 HTML 错误页上    try:        data = response.json()    except ValueError:        pytest.fail(f"[维度一·结构] 响应不是合法 JSON：{response.text[:300]}")    # Step3 结构校验    try:        jsonschema.validate(instance=data, schema=schema)    except jsonschema.ValidationError as e:        path = " → ".join(str(p) for p in e.path) or "根节点"        pytest.fail(            f"[维度一·结构] 校验失败\n"            f"  位置  : {path}\n"            f"  原因  : {e.message}\n"            f"  实际值: {e.instance}"        )    return datadef assert_biz_code(response: requests.Response, expected: int = 0) -> dict:    """    校验业务状态码（响应体里的 code 字段），并把 data 返回出来    团队里常见的业务码约定：        0   成功        401 未登录 / Token 过期        403 无权限        4004 库存不足        4005 优惠券不可用    """    body = response.json()    actual = body.get("code")    assert actual == expected, (        f"[维度一·结构] 业务码不符：期望 {expected}，实际 {actual}，"        f"接口提示: {body.get('message', 'N/A')}"    )    return body.get("data") or {}

resp = session.post(f"{BASE_URL}/api/v1/orders", json=payload)data = assert_schema(resp, ORDER_CREATE_SCHEMA)   # 结构 + 状态码一起过order_id = data["data"]["order_id"]                 # 拿到就能往下传

# -*- coding: utf-8 -*-import timeimport pymysqlfrom contextlib import contextmanagerfrom typing import Optional, List, DictDB_CONF = {    "host": "10.20.30.40",    "port": 3306,    "user": "qa_readonly",        # 强烈建议给测试单独开只读账号    "password": "***",             # 实际从环境变量读，别硬编码进仓库    "database": "mall_order",    "charset": "utf8mb4",    "cursorclass": pymysql.cursors.DictCursor,   # 查出来直接是 dict，好用}@contextmanagerdef db_conn():    """每次用完就关，别让连接跨用例复用，事务状态会互相污染。"""    conn = pymysql.connect(**DB_CONF)    try:        # 关键：把隔离级别降到 READ COMMITTED        # 默认的 REPEATABLE READ 会让你在同一个连接里反复查到旧快照，        # 明明接口已经写进去了，你就是查不到——我在这上面栽过一整个下午        with conn.cursor() as cur:            cur.execute("SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED")        yield conn    finally:        conn.close()def query_one(sql: str, args: tuple = ()) -> Optional[Dict]:    """查单行。一条 SQL 把需要的字段一次查全，别为每个断言查一次库。"""    with db_conn() as conn:        with conn.cursor() as cur:            cur.execute(sql, args)            return cur.fetchone()def query_all(sql: str, args: tuple = ()) -> List[Dict]:    with db_conn() as conn:        with conn.cursor() as cur:            cur.execute(sql, args)            return cur.fetchall()def wait_until(sql: str, args: tuple = (), timeout: float = 5.0,               interval: float = 0.3) -> Optional[Dict]:    """    轮询等待数据出现，专治异步写入。    比 time.sleep(3) 好在两点：      1. 数据早到就早返回，不浪费用例执行时间      2. 超时时间可控，不会因为环境慢一点就整批用例挂掉    用法:        row = wait_until("SELECT * FROM orders WHERE order_id=%s", (oid,))        assert row, "5 秒内订单没落库"    """    deadline = time.time() + timeout    while time.time() < deadline:        row = query_one(sql, args)        if row:            return row        time.sleep(interval)    return None

# -*- coding: utf-8 -*-import pytestfrom utils.db_helper import query_one, query_all, wait_untilfrom utils.assertions import assert_schemafrom schemas.order_schema import ORDER_CREATE_SCHEMAdef test_order_create_db_consistency(api_session, base_url, test_sku):    """维度二：下单成功后，三张表的数据必须全部对得上。"""    # --- 前置：记录下单前的库存，作为对照基准 ---    before = query_one(        "SELECT available_qty, locked_qty FROM inventory WHERE sku_id=%s",        (test_sku,),    )    assert before, f"测试 SKU {test_sku} 在库存表里不存在，先造数据"    # --- 调接口 ---    payload = {        "user_id": "U100000001",        "address_id": "A20001",        "items": [{"sku_id": test_sku, "quantity": 2}],    }    resp = api_session.post(f"{base_url}/api/v1/orders", json=payload)    body = assert_schema(resp, ORDER_CREATE_SCHEMA)    order_id = body["data"]["order_id"]    # --- 校验点1：订单主表 ---    # 用 wait_until 而不是 sleep，异步写入也能兜住    order = wait_until(        "SELECT order_id, user_id, status, total_amount, is_deleted "        "FROM orders WHERE order_id=%s",        (order_id,), timeout=5,    )    assert order, f"[维度二·数据] 接口返回成功，但 5 秒内 orders 表查不到 {order_id}"    assert order["status"] == "pending", (        f"[维度二·数据] 新建订单状态应为 pending，实际 {order['status']}"    )    assert order["is_deleted"] == 0    # 注意：金额一定要用 Decimal 或者转字符串比，别用 float 直接 ==    assert str(order["total_amount"]) == str(body["data"]["total_amount"]), (        f"[维度二·数据] 接口返回金额与库中金额不一致："        f"接口 {body['data']['total_amount']} vs 库 {order['total_amount']}"    )    # --- 校验点2：订单明细子表 ---    items = query_all(        "SELECT sku_id, quantity, price FROM order_items WHERE order_id=%s",        (order_id,),    )    assert len(items) == 1, f"[维度二·数据] 明细行数不对，期望 1，实际 {len(items)}"    assert items[0]["quantity"] == 2    # --- 校验点3：库存表（这一步很容易被漏掉） ---    after = query_one(        "SELECT available_qty, locked_qty FROM inventory WHERE sku_id=%s",        (test_sku,),    )    assert after["available_qty"] == before["available_qty"] - 2, (        f"[维度二·数据] 可售库存没扣：下单前 {before['available_qty']}，"        f"下单后 {after['available_qty']}，期望减 2"    )    assert after["locked_qty"] == before["locked_qty"] + 2, (        "[维度二·数据] 预占库存没加，下单锁库逻辑可能没走到"    )

def test_order_rollback_on_insufficient_stock(api_session, base_url, low_stock_sku):    """    维度二·事务回滚：库存不足时下单，订单表 / 明细表 / 库存表都不该有变化。    low_stock_sku 这个 fixture 会先把某个 SKU 的库存改成 1，    然后我们下单 5 件，必然失败。    """    before_inv = query_one(        "SELECT available_qty, locked_qty FROM inventory WHERE sku_id=%s",        (low_stock_sku,),    )    before_cnt = query_one("SELECT COUNT(*) AS c FROM orders")["c"]    resp = api_session.post(f"{base_url}/api/v1/orders", json={        "user_id": "U100000001",        "address_id": "A20001",        "items": [{"sku_id": low_stock_sku, "quantity": 5}],    })    # 业务上应该明确失败，而不是返回一个"部分成功"    assert resp.status_code == 200    assert resp.json()["code"] == 4004, "库存不足时业务码应为 4004"    # --- 回滚校验：三处都不能有残留 ---    after_cnt = query_one("SELECT COUNT(*) AS c FROM orders")["c"]    assert after_cnt == before_cnt, (        f"[维度二·回滚] 下单失败却多了 {after_cnt - before_cnt} 条订单，事务没回滚干净"    )    after_inv = query_one(        "SELECT available_qty, locked_qty FROM inventory WHERE sku_id=%s",        (low_stock_sku,),    )    assert after_inv == before_inv, (        f"[维度二·回滚] 下单失败但库存被改了：{before_inv} → {after_inv}"    )

# -*- coding: utf-8 -*-import pytestfrom decimal import Decimalfrom utils.db_helper import query_one, wait_untilfrom utils.assertions import assert_schema, assert_biz_code@pytest.mark.chaindef test_order_pay_ship_chain(api_session, base_url, test_sku):    """    维度三 · 主链路：下单 → 支付 → 发货 → 查询    校验重点不是"每步返回 200"，而是：      · order_id 一路不变      · 金额一路不变      · 每一步之后，状态和关联字段同步更新    """    # ===== 环节1：下单 =====    r1 = api_session.post(f"{base_url}/api/v1/orders", json={        "user_id": "U100000001",        "address_id": "A20001",        "items": [{"sku_id": test_sku, "quantity": 1}],    })    d1 = assert_biz_code(r1)    order_id = d1["order_id"]    amount = Decimal(str(d1["total_amount"]))    assert d1["status"] == "pending"    # ===== 环节2：支付回调（把上一步的 order_id 原样带进来） =====    r2 = api_session.post(f"{base_url}/api/v1/payments/callback", json={        "order_id": order_id,        "pay_amount": str(amount),           # 金额用字符串传，防精度丢失        "channel": "alipay",        "trade_no": f"TRADE{order_id[-8:]}",    })    d2 = assert_biz_code(r2)    # 传递校验①：支付接口返回的订单号，必须和下单时一模一样    assert d2["order_id"] == order_id, (        f"[维度三·链路] 订单号在支付环节变了：{order_id} → {d2['order_id']}"    )    # 传递校验②：金额不能有任何漂移    assert Decimal(str(d2["pay_amount"])) == amount, (        f"[维度三·链路] 支付金额漂移：下单 {amount}，支付 {d2['pay_amount']}"    )    # 状态校验：支付后订单必须变成 paid，同时支付流水要落库    order = wait_until(        "SELECT status, paid_at FROM orders WHERE order_id=%s AND status='paid'",        (order_id,), timeout=8,    )    assert order, "[维度三·链路] 支付回调返回成功，但 8 秒内订单状态没变成 paid"    assert order["paid_at"] is not None, "[维度三·链路] 状态变了但支付时间没写"    pay = query_one(        "SELECT trade_no, amount FROM payments WHERE order_id=%s", (order_id,)    )    assert pay, "[维度三·链路] 订单状态是 paid，但支付流水表里没记录（幽灵支付）"    # ===== 环节3：发货 =====    r3 = api_session.post(f"{base_url}/api/v1/shipments", json={        "order_id": order_id,        "carrier": "SF",    })    d3 = assert_biz_code(r3)    tracking_no = d3["tracking_number"]    assert tracking_no, "[维度三·链路] 发货成功但物流单号是空的"    # ===== 环节4：反查订单详情，确认全链路数据闭环 =====    r4 = api_session.get(f"{base_url}/api/v1/orders/{order_id}")    d4 = assert_biz_code(r4)    assert d4["order_id"] == order_id    assert d4["status"] == "shipped"    assert d4["tracking_number"] == tracking_no, (        "[维度三·链路] 详情接口返回的物流单号与发货接口不一致"    )    assert Decimal(str(d4["total_amount"])) == amount, (        "[维度三·链路] 走完全链路后，订单金额和下单时对不上了"    )

def test_order_auto_close_and_restore_stock(api_session, base_url, test_sku, db_write):    """维度三·逆向：超时未支付自动关单，预占库存必须释放。"""    before = query_one("SELECT locked_qty FROM inventory WHERE sku_id=%s", (test_sku,))    # 1) 正常下单，此时库存被预占    d = assert_biz_code(api_session.post(f"{base_url}/api/v1/orders", json={        "user_id": "U100000001", "address_id": "A20001",        "items": [{"sku_id": test_sku, "quantity": 1}],    }))    order_id = d["order_id"]    mid = query_one("SELECT locked_qty FROM inventory WHERE sku_id=%s", (test_sku,))    assert mid["locked_qty"] == before["locked_qty"] + 1, "下单没有预占库存"    # 2) 把创建时间往前拨 31 分钟，制造"超时"条件    db_write(        "UPDATE orders SET created_at = DATE_SUB(created_at, INTERVAL 31 MINUTE) "        "WHERE order_id=%s", (order_id,)    )    # 3) 手动触发关单任务（测试环境提供的内部接口）    api_session.post(f"{base_url}/internal/jobs/close-expired-orders")    # 4) 校验：状态变 closed + 预占库存回到原点    row = wait_until(        "SELECT status FROM orders WHERE order_id=%s AND status='closed'",        (order_id,), timeout=10,    )    assert row, "[维度三·逆向] 超时订单没有被自动关闭"    after = query_one("SELECT locked_qty FROM inventory WHERE sku_id=%s", (test_sku,))    assert after["locked_qty"] == before["locked_qty"], (        f"[维度三·逆向] 关单了但预占库存没释放："        f"关单前 {mid['locked_qty']}，关单后 {after['locked_qty']}，期望回到 {before['locked_qty']}"    )

# -*- coding: utf-8 -*-import uuidimport pytestimport requestsfrom collections import Counterfrom concurrent.futures import ThreadPoolExecutor, as_completedfrom utils.db_helper import query_one, query_all@pytest.mark.concurrencydef test_no_oversell_under_concurrency(base_url, token, db_write, test_sku):    """    维度四 · 超卖验证    场景：库存精确设为 5，20 个线程同时各下 1 件。    期望：成功 5 单，失败 15 单，库存归零，一件都不能多卖。    """    # --- 关键前置：把库存重置为已知值，别指望环境是干净的 ---    STOCK = 5    CONCURRENCY = 20    db_write(        "UPDATE inventory SET available_qty=%s, locked_qty=0 WHERE sku_id=%s",        (STOCK, test_sku),    )    url = f"{base_url}/api/v1/orders"    headers = {"Authorization": f"Bearer {token}"}    def place_order(idx: int):        """单个下单请求。每个线程用独立 session，避免连接池竞争。"""        payload = {            # 每个线程用不同用户，排除"同用户限购"逻辑的干扰            "user_id": f"U1000000{idx:02d}",            "address_id": "A20001",            "items": [{"sku_id": test_sku, "quantity": 1}],        }        try:            r = requests.post(url, json=payload, headers=headers, timeout=10)            body = r.json()            return {"idx": idx, "http": r.status_code,                    "code": body.get("code"),                    "order_id": (body.get("data") or {}).get("order_id")}        except Exception as e:            return {"idx": idx, "http": -1, "code": None, "error": str(e)}    # --- 并发发射 ---    results = []    with ThreadPoolExecutor(max_workers=CONCURRENCY) as pool:        futures = [pool.submit(place_order, i) for i in range(CONCURRENCY)]        for f in as_completed(futures):            results.append(f.result())    # --- 校验点1：成功数量必须等于库存数量 ---    success = [r for r in results if r["code"] == 0]    assert len(success) == STOCK, (        f"[维度四·并发] 库存 {STOCK} 件，却成功了 {len(success)} 单。\n"        f"业务码分布: {Counter(r['code'] for r in results)}"    )    # --- 校验点2：库存不能变成负数（超卖的直接证据） ---    inv = query_one(        "SELECT available_qty, locked_qty FROM inventory WHERE sku_id=%s",        (test_sku,),    )    assert inv["available_qty"] >= 0, (        f"[维度四·并发] 库存被扣成负数：{inv['available_qty']}，确认超卖"    )    assert inv["available_qty"] == 0, (        f"[维度四·并发] 库存应恰好扣完，实际剩 {inv['available_qty']}，"        f"可能有请求扣了库存却没建单"    )    # --- 校验点3：订单数量与成功数一致，不能有多出来的脏单 ---    order_ids = [r["order_id"] for r in success]    assert len(set(order_ids)) == STOCK, "[维度四·并发] 出现了重复的订单号"@pytest.mark.concurrencydef test_idempotency_with_same_key(base_url, token, test_sku):    """    维度四 · 幂等验证    场景：拿同一个幂等键，并发打 10 次下单请求。    期望：只产生 1 笔订单，10 次响应内容完全一致。    """    idem_key = f"IDEM-{uuid.uuid4().hex}"    url = f"{base_url}/api/v1/orders"    payload = {        "user_id": "U100000099",        "address_id": "A20001",        "items": [{"sku_id": test_sku, "quantity": 1}],    }    headers = {        "Authorization": f"Bearer {token}",        "Idempotency-Key": idem_key,      # 关键：10 次请求共用同一个键    }    def fire(_):        r = requests.post(url, json=payload, headers=headers, timeout=10)        return r.json()    with ThreadPoolExecutor(max_workers=10) as pool:        bodies = list(pool.map(fire, range(10)))    # 校验点1：所有响应的订单号必须相同（返回同一笔单）    ids = {b.get("data", {}).get("order_id") for b in bodies}    assert len(ids) == 1, f"[维度四·幂等] 同一幂等键产生了多个订单号：{ids}"    # 校验点2：库里确实只有一条记录    rows = query_all(        "SELECT order_id FROM orders WHERE idempotency_key=%s", (idem_key,)    )    assert len(rows) == 1, (        f"[维度四·幂等] 幂等键 {idem_key} 在库里对应了 {len(rows)} 条订单"    )    # 校验点3：真正的幂等，是每次响应体都完全一致    # 很多接口做到了"不重复创建"，但重复请求返回的是错误码而不是首次结果，    # 这会让客户端的重试逻辑很难写    first = bodies[0]    for i, b in enumerate(bodies[1:], start=2):        assert b == first, (            f"[维度四·幂等] 第 {i} 次响应与首次不一致\n"            f"首次: {first}\n第{i}次: {b}"        )

// ========== 请求 ==========Headers:  Authorization: Bearer <token>  Idempotency-Key: <客户端生成的唯一键，可选但强烈建议>  Content-Type: application/jsonBody:{  "user_id":    "U100000001",        // 必填，用户ID  "address_id": "A20001",            // 必填，收货地址ID  "coupon_id":  "CP20240801001",     // 选填，本次迭代新增  "items": [    {"sku_id": "SKU0000001", "quantity": 2}  ]}// ========== 成功响应 =========={  "code": 0,  "message": "",  "data": {    "order_id":      "OD20240801XK92MT3B01",   // 20位    "status":        "pending",    "original_amount": 199.80,               // 优惠前金额    "discount_amount": 20.00,                // 优惠券抵扣    "total_amount":    179.80,               // 实付金额    "created_at":   "2024-08-01T14:23:05+08:00",    "items": [      {"sku_id": "SKU0000001", "quantity": 2, "price": 99.90}    ]  }}// ========== 业务错误码 ==========// 4004 库存不足　　4005 优惠券不可用　　4006 地址无效// 4009 幂等键冲突（已有处理中的同键请求）

Body:{  "order_id":  "OD20240801XK92MT3B01",  "trade_no":  "TRADE202408011423001",   // 支付渠道流水号  "pay_amount": "179.80",                 // 字符串传，防精度丢失  "channel":   "alipay"}Response:{  "code": 0,  "data": {    "order_id": "OD20240801XK92MT3B01",    "status":   "paid",    "paid_at":  "2024-08-01T14:25:11+08:00"  }}// 注意：支付渠道会重复推送回调（这是行业惯例，不是 bug）// 所以这个接口天然要求幂等 —— 这是维度四的重点靶子

Response:{  "code": 0,  "data": {    "order_id":      "OD20240801XK92MT3B01",    "status":        "paid",    "total_amount":  179.80,    "coupon_id":     "CP20240801001",    "paid_at":       "2024-08-01T14:25:11+08:00",    "items": [ ... ]  }}

# -*- coding: utf-8 -*-"""电商订单接口 · 四维验证完整用例集"""import uuidimport pytestimport requestsfrom decimal import Decimalfrom concurrent.futures import ThreadPoolExecutorfrom utils.assertions import assert_schema, assert_biz_codefrom utils.db_helper import query_one, query_all, wait_untilfrom schemas.order_schema import ORDER_CREATE_SCHEMA# ============================================================# 用例1：下单主流程 —— 一次覆盖维度一 + 维度二（8 个校验点）# ============================================================def test_create_order_with_coupon(api_session, base_url, ready_sku, ready_coupon):    user_id = "U100000001"    inv_before = query_one(        "SELECT available_qty, locked_qty FROM inventory WHERE sku_id=%s",        (ready_sku,),    )    resp = api_session.post(f"{base_url}/api/v1/orders", json={        "user_id": user_id,        "address_id": "A20001",        "coupon_id": ready_coupon,        "items": [{"sku_id": ready_sku, "quantity": 2}],    })    # ---------- 维度一：结构 ----------    body = assert_schema(resp, ORDER_CREATE_SCHEMA)    d = body["data"]    order_id = d["order_id"]    # 金额勾稽：优惠前 - 优惠 = 实付。这一条抓到过 Bug#1    assert (Decimal(str(d["original_amount"])) - Decimal(str(d["discount_amount"]))            == Decimal(str(d["total_amount"]))), (        f"[维度一·结构] 金额勾稽不上："        f"{d['original_amount']} - {d['discount_amount']} != {d['total_amount']}"    )    # ---------- 维度二：数据 ----------    order = wait_until(        "SELECT order_id, status, original_amount, discount_amount, "        "       total_amount, coupon_id, is_deleted "        "FROM orders WHERE order_id=%s", (order_id,), timeout=5,    )    assert order, f"[维度二·数据] 接口成功但 orders 表无记录：{order_id}"    assert order["status"] == "pending"    assert Decimal(str(order["total_amount"])) == Decimal(str(d["total_amount"])), (        "[维度二·数据] 库中实付金额与接口返回不一致"    )    assert order["coupon_id"] == ready_coupon, "[维度二·数据] 订单没有记录所用优惠券"    # 明细行金额求和，必须等于优惠前金额（这一条抓到过 Bug#2）    items = query_all(        "SELECT sku_id, quantity, price FROM order_items WHERE order_id=%s",        (order_id,),    )    assert len(items) == 1    items_sum = sum(Decimal(str(i["price"])) * i["quantity"] for i in items)    assert items_sum == Decimal(str(order["original_amount"])), (        f"[维度二·数据] 明细金额求和 {items_sum} != 主表优惠前金额 {order['original_amount']}"    )    # 库存：可售减少、预占增加    inv_after = query_one(        "SELECT available_qty, locked_qty FROM inventory WHERE sku_id=%s",        (ready_sku,),    )    assert inv_after["available_qty"] == inv_before["available_qty"] - 2    assert inv_after["locked_qty"] == inv_before["locked_qty"] + 2    # 优惠券：状态置为已使用，并回填订单号（这一条抓到过 Bug#3）    cp = query_one(        "SELECT status, used_order_id FROM user_coupons "        "WHERE coupon_id=%s AND user_id=%s", (ready_coupon, user_id),    )    assert cp["status"] == "used", (        f"[维度二·数据] 优惠券状态应为 used，实际 {cp['status']}——券可能被重复使用"    )    assert cp["used_order_id"] == order_id, "[维度二·数据] 券没有绑定到本次订单"# ============================================================# 用例2：支付回调幂等 —— 维度三 + 维度四# ============================================================def test_payment_callback_idempotent(api_session, base_url, pending_order):    """支付渠道会重复推回调，这是行业常态，必须幂等。"""    order_id, amount = pending_order    trade_no = f"TRADE{uuid.uuid4().hex[:14].upper()}"    payload = {        "order_id": order_id, "trade_no": trade_no,        "pay_amount": str(amount), "channel": "alipay",    }    # 连续推 5 次，模拟渠道重复回调    bodies = [api_session.post(f"{base_url}/api/v1/payments/callback",                              json=payload).json() for _ in range(5)]    # 维度四：5 次响应必须完全一致    for i, b in enumerate(bodies[1:], start=2):        assert b == bodies[0], f"[维度四·幂等] 第 {i} 次回调响应与首次不同"    # 维度二：支付流水只能有 1 条（这一条抓到过 Bug#4）    pays = query_all("SELECT id, amount FROM payments WHERE order_id=%s", (order_id,))    assert len(pays) == 1, (        f"[维度四·幂等] 5 次回调产生了 {len(pays)} 条支付流水，财务对账要炸"    )    # 维度三：状态正确流转，且伴随字段齐全    order = wait_until(        "SELECT status, paid_at FROM orders WHERE order_id=%s AND status='paid'",        (order_id,), timeout=8,    )    assert order and order["paid_at"], "[维度三·链路] 支付后状态或支付时间缺失"    # 维度三：详情接口反查，金额一路不变    d = assert_biz_code(api_session.get(f"{base_url}/api/v1/orders/{order_id}"))    assert Decimal(str(d["total_amount"])) == amount# ============================================================# 用例3：同一张优惠券并发下单 —— 维度四# ============================================================def test_coupon_no_double_use(base_url, token, ready_sku, ready_coupon):    """一张券并发下 5 单，只允许 1 单用券成功。"""    url = f"{base_url}/api/v1/orders"    headers = {"Authorization": f"Bearer {token}"}    def fire(i):        r = requests.post(url, headers=headers, timeout=10, json={            "user_id": "U100000001", "address_id": "A20001",            "coupon_id": ready_coupon,            "items": [{"sku_id": ready_sku, "quantity": 1}],        })        return r.json()    with ThreadPoolExecutor(max_workers=5) as pool:        bodies = list(pool.map(fire, range(5)))    ok = [b for b in bodies if b.get("code") == 0]    assert len(ok) == 1, (        f"[维度四·并发] 一张券被 {len(ok)} 笔订单同时用掉了，直接资损"    )    # 库里也确认一遍：这张券只被一个订单绑定    rows = query_all(        "SELECT order_id FROM orders WHERE coupon_id=%s AND is_deleted=0",        (ready_coupon,),    )    assert len(rows) == 1, f"[维度四·并发] 券 {ready_coupon} 关联了 {len(rows)} 笔订单"

# -*- coding: utf-8 -*-import osimport timeimport uuidimport pytestimport requestsimport pymysqlfrom utils.db_helper import DB_CONF, query_one# ---------- 基础配置 ----------@pytest.fixture(scope="session")def base_url():    # 环境地址从环境变量读，方便在 CI 里切 dev/test/staging    return os.getenv("API_BASE_URL", "http://api-test.internal")# ---------- Token 自动续期（解决"跑着跑着全挂"） ----------class TokenManager:    """    Token 不硬编码，且在快过期时自动续。    很多团队的用例集跑到一半突然全红，十有八九是 Token 到期了。    """    def __init__(self, base_url):        self.base_url = base_url        self._token = None        self._expire_at = 0    def get(self) -> str:        # 剩余有效期不足 5 分钟就提前换新的        if self._token and time.time() < self._expire_at - 300:            return self._token        r = requests.post(f"{self.base_url}/api/v1/auth/login", json={            "username": os.environ["QA_USER"],            "password": os.environ["QA_PASS"],        }, timeout=10)        r.raise_for_status()        data = r.json()["data"]        self._token = data["access_token"]        self._expire_at = time.time() + data.get("expires_in", 7200)        return self._token@pytest.fixture(scope="session")def token_mgr(base_url):    return TokenManager(base_url)@pytest.fixturedef token(token_mgr):    return token_mgr.get()@pytest.fixturedef api_session(token):    """带认证头的 session，连接复用，比每次新建 requests 快很多。"""    s = requests.Session()    s.headers.update({        "Authorization": f"Bearer {token}",        "Content-Type": "application/json",        # 打标记，方便后端在日志里区分自动化流量        "X-Request-Source": "qa-automation",    })    s.request = _with_timeout(s.request, default=15)    yield s    s.close()def _with_timeout(func, default=15):    """给所有请求兜一个默认超时，防止某个接口卡死拖垮整批用例。"""    def wrapper(method, url, **kw):        kw.setdefault("timeout", default)        return func(method, url, **kw)    return wrapper# ---------- 数据库写操作（造数据 / 清理用） ----------@pytest.fixturedef db_write():    """    提供一个可执行写 SQL 的函数。    注意：这个 fixture 用的是有写权限的账号，和查询用的只读账号分开。    造数据、改状态、重置库存走这里，普通断言查询走 query_one。    """    conn = pymysql.connect(**{**DB_CONF, "user": os.environ["DB_RW_USER"],                              "password": os.environ["DB_RW_PASS"]})    def _exec(sql: str, args: tuple = ()):        with conn.cursor() as cur:            cur.execute(sql, args)        conn.commit()    yield _exec    conn.close()# ---------- 测试数据 fixture：天然隔离，不靠清理逻辑 ----------@pytest.fixturedef ready_sku(db_write):    """    每个用例造一个独立 SKU，避免并行执行时互相抢库存。    这里用 yield 而不是 return，是为了保证 teardown 一定会执行——    即使用例中途断言失败，yield 之后的清理代码照样跑。    """    sku_id = f"SKU{uuid.uuid4().hex[:7].upper()}"    db_write(        "INSERT INTO inventory (sku_id, available_qty, locked_qty, price) "        "VALUES (%s, 100, 0, 99.90)", (sku_id,)    )    yield sku_id    # 清理用软删除，不做物理 DELETE —— WHERE 写错一次就回不来了    try:        db_write("UPDATE inventory SET is_deleted=1 WHERE sku_id=%s", (sku_id,))    except Exception as e:        # 清理失败不能影响用例本身的结论，打日志就好        print(f"[teardown] 清理 SKU {sku_id} 失败: {e}")@pytest.fixturedef ready_coupon(db_write):    """造一张满199减20的未使用优惠券。"""    cid = f"CP{uuid.uuid4().hex[:11].upper()}"    db_write(        "INSERT INTO user_coupons (coupon_id, user_id, threshold, discount, status) "        "VALUES (%s, 'U100000001', 199.00, 20.00, 'unused')", (cid,)    )    yield cid    try:        db_write("UPDATE user_coupons SET is_deleted=1 WHERE coupon_id=%s", (cid,))    except Exception as e:        print(f"[teardown] 清理券 {cid} 失败: {e}")# ---------- 自定义 marker，方便按维度筛选执行 ----------def pytest_configure(config):    config.addinivalue_line("markers", "schema: 维度一 结构校验")    config.addinivalue_line("markers", "db: 维度二 数据一致性")    config.addinivalue_line("markers", "chain: 维度三 业务链路")    config.addinivalue_line("markers", "concurrency: 维度四 并发幂等")

# 每次提交：只跑快的（结构 + 数据），2 分钟出结果pytest -m "schema or db" -n 4 --tb=short# 每日构建：加上链路，10 分钟pytest -m "schema or db or chain" -n 4# 发版前 / 每周：全量，并发用例单独串行跑，避免互相干扰pytest -m "not concurrency" -n 4pytest -m concurrency          # 并发用例不要加 -n，会自己打自己

# 下单接口参数化数据# 字段说明：#   name     用例名，会显示在 pytest 报告里#   payload  请求体#   expect   期望的业务码和关键字段- name: "正常下单_单件商品_无优惠券"  payload:    quantity: 1    coupon: null  expect:    code: 0    status: "pending"    discount: "0.00"- name: "正常下单_满减券_刚好达到门槛"  payload:    quantity: 2          # 2 × 99.90 = 199.80 ≥ 199 门槛    coupon: "THRESHOLD_199_MINUS_20"  expect:    code: 0    discount: "20.00"- name: "异常_满减券_差一分钱不到门槛"  payload:    quantity: 1          # 99.90 < 199，券用不了    coupon: "THRESHOLD_199_MINUS_20"  expect:    code: 4005    message_contains: "未达到使用门槛"- name: "边界_数量上限99件"  payload:    quantity: 99    coupon: null  expect:    code: 0- name: "异常_数量超上限100件"  payload:    quantity: 100    coupon: null  expect:    code: 4002    message_contains: "单次购买数量"- name: "异常_数量为0"  payload:    quantity: 0    coupon: null  expect:    code: 4002

import yamlimport pytestfrom pathlib import PathDATA = yaml.safe_load(    (Path(__file__).parent / "test_data/order_cases.yaml").read_text("utf-8"))@pytest.mark.parametrize("case", DATA, ids=[c["name"] for c in DATA])def test_order_scenarios(api_session, base_url, ready_sku, case):    """一份 YAML 驱动 6 个场景，加场景不用改代码。"""    payload = {        "user_id": "U100000001", "address_id": "A20001",        "items": [{"sku_id": ready_sku, "quantity": case["payload"]["quantity"]}],    }    if case["payload"].get("coupon"):        payload["coupon_id"] = resolve_coupon(case["payload"]["coupon"])    resp = api_session.post(f"{base_url}/api/v1/orders", json=payload)    body = resp.json()    assert body["code"] == case["expect"]["code"], (        f"场景【{case['name']}】业务码不符："        f"期望 {case['expect']['code']}，实际 {body['code']}，"        f"提示：{body.get('message')}"    )    if "message_contains" in case["expect"]:        assert case["expect"]["message_contains"] in body.get("message", "")

# ❌ 错误：断言失败后，清理代码永远执行不到def test_bad(api_session):    sku = create_sku()    resp = api_session.post(url, json=payload)    assert resp.json()["code"] == 0      # ← 这里挂了    cleanup(sku)                            # ← 永远跑不到# ✅ 正确：清理放 fixture teardown，且做容错@pytest.fixturedef sku(db_write):    s = create_sku()    yield s    try:        # 用软删除而不是物理 DELETE，WHERE 写错还有救        db_write("UPDATE inventory SET is_deleted=1 WHERE sku_id=%s", (s,))    except Exception as e:        print(f"[teardown] 清理失败但不影响用例结论: {e}")