shop-java/tools/test_api_batch2.py

# -*- coding: utf-8 -*-
"""客户端 api 第 2 批接口冒烟：购物车 / 结算下单 / 订单 / 订单评价 / 收货地址。

用法：起 Java 18080 后执行
    python tools/test_api_batch2.py
环境变量：YOSHOP_API_BASE（默认 http://127.0.0.1:18080）

说明：
- 全部是需登录接口，脚本会往 Redis 写一个测试 token（yoshop:api:token:apitest-batch2）；
- 会真实下单，finally 中删除测试订单/订单商品/订单地址/评价并恢复库存与默认地址；
- 数据库为联合调试库，测试商品固定取 10009（运费模板 10002，上海市 3 元运费）。
"""
import json
import os
import re
import socket
import sys

import pymysql
import requests

from _root import ROOT
from redis_probe import cmd

BASE = os.environ.get("YOSHOP_API_BASE", "http://127.0.0.1:18080")
TOKEN = "apitest-batch2"
USER_ID = 10001
ADDRESS_ID = 10001
# 测试商品：多规格、运费模板 10002（上海 +3 元、续件 +3 元）
GOODS_ID = 10009
SKU_ID = "10009"
SKU_PRICE = "80.00"
# 单规格包邮商品
FREE_GOODS_ID = 10007
FREE_SKU_ID = "0"
FREE_SKU_PRICE = "39.90"

YML = str(ROOT / "yoshop-java/src/main/resources/application.yml")
TEXT = open(YML, encoding="utf-8").read()
DB_PWD = re.findall(r"password:\s*(.+)", TEXT)[0].strip().strip("\"'")
REDIS_PWD = re.findall(r"password:\s*(.+)", TEXT)[1].strip().strip("\"'")

S = requests.Session()
S.trust_env = False

passed = 0
failed = 0
created_orders = []
points_setting_created = False
test_coupon_id = None
stock_backup = {}
address_backup = None


def check(name, cond, detail=""):
    global passed, failed
    if cond:
        passed += 1
        print(f"  PASS  {name}")
    else:
        failed += 1
        print(f"  FAIL  {name}  {detail}")


def call(method, path, params=None, body=None, token=TOKEN):
    headers = {"platform": "MP-WEIXIN"}
    if token:
        headers["Access-Token"] = token
    if method == "GET":
        resp = S.get(BASE + path, params=params or {}, headers=headers, timeout=60)
    else:
        resp = S.post(BASE + path, params=params or {}, json=body, headers=headers, timeout=60)
    try:
        return resp.json()
    except Exception:
        return {"status": resp.status_code, "raw": resp.text[:300]}


def get(path, params=None, token=TOKEN):
    return call("GET", path, params=params, token=token)


def post(path, body=None, params=None, token=TOKEN):
    return call("POST", path, params=params, body=body, token=token)


def snake_keys(obj, prefix="$"):
    found = []
    if isinstance(obj, dict):
        for k, v in obj.items():
            if isinstance(k, str) and re.fullmatch(r"[a-z0-9]+(_[a-z0-9]+)+", k):
                found.append(f"{prefix}.{k}")
            found.extend(snake_keys(v, f"{prefix}.{k}"))
    elif isinstance(obj, list):
        for i, v in enumerate(obj[:3]):
            found.extend(snake_keys(v, f"{prefix}[{i}]"))
    return found


def db(sql):
    conn = pymysql.connect(host="192.168.1.177", port=3306, user="root", password=DB_PWD,
                           database="shop", charset="utf8mb4", autocommit=True)
    try:
        cur = conn.cursor()
        cur.execute(sql)
        return cur.fetchall()
    finally:
        conn.close()


def db_exec(sql):
    conn = pymysql.connect(host="192.168.1.177", port=3306, user="root", password=DB_PWD,
                           database="shop", charset="utf8mb4", autocommit=True)
    try:
        cur = conn.cursor()
        cur.execute(sql)
        return cur.lastrowid
    finally:
        conn.close()


def redis_set(key, value):
    sock = socket.create_connection(("192.168.1.177", 6379), timeout=5)
    try:
        cmd(sock, "AUTH", REDIS_PWD)
        cmd(sock, "SET", key, str(value))
        cmd(sock, "EXPIRE", key, "3600")
    finally:
        sock.close()


def field(obj, name, default=None):
    return obj.get(name, default) if isinstance(obj, dict) else default


def make_order(order_status, pay_status, delivery_status, receipt_status, is_comment=0, with_goods=True):
    """造一条测试订单，返回 order_id（订单商品取测试商品）"""
    order_no = "TEST" + str(os.getpid()) + str(len(created_orders))
    order_id = db_exec(
        "INSERT INTO yoshop_order (order_no, order_type, total_price, order_price, coupon_id, coupon_money, "
        "points_money, points_num, pay_price, update_price, buyer_remark, pay_method, pay_status, trade_id, "
        "delivery_type, express_price, delivery_status, receipt_status, order_status, points_bonus, "
        "is_settled, is_comment, order_source, order_source_id, order_source_data, platform, user_id, "
        "is_delete, store_id, create_time, update_time) VALUES "
        f"('{order_no}', 10, 80.00, 80.00, 0, 0.00, 0.00, 0, 83.00, 0.00, '', '', {pay_status}, 0, "
        f"10, 3.00, {delivery_status}, {receipt_status}, {order_status}, 0, 0, {is_comment}, 10, 0, '[]', "
        f"'H5', {USER_ID}, 0, 10001, UNIX_TIMESTAMP(), UNIX_TIMESTAMP())")
    created_orders.append(order_id)
    if with_goods:
        db_exec(
            "INSERT INTO yoshop_order_goods (goods_id, goods_type, goods_name, image_id, deduct_stock_type, "
            "spec_type, goods_sku_id, goods_props, content, goods_no, goods_price, line_price, goods_weight, "
            "is_user_grade, grade_ratio, grade_goods_price, grade_total_money, coupon_money, points_money, "
            "points_num, points_bonus, total_num, total_price, total_pay_price, delivery_status, delivery_num, "
            "is_comment, order_id, user_id, goods_source_id, store_id, create_time) VALUES "
            f"({GOODS_ID}, 10, '正宗荔浦芋头', 0, 10, 20, '{SKU_ID}', '', '', '', 80.00, 100.00, 4.5, "
            f"0, 0, 0.00, 0.00, 0.00, 0.00, 0, 0, 1, 80.00, 80.00, 10, 0, 0, {order_id}, {USER_ID}, 0, "
            "10001, UNIX_TIMESTAMP())")
    return order_id


def cleanup():
    try:
        ids = ",".join(str(i) for i in created_orders) or "0"
        db_exec(f"DELETE FROM yoshop_comment_image WHERE comment_id IN "
                f"(SELECT comment_id FROM yoshop_comment WHERE order_id IN ({ids}))")
        db_exec(f"DELETE FROM yoshop_comment WHERE order_id IN ({ids})")
        db_exec(f"DELETE FROM yoshop_order_goods WHERE order_id IN ({ids})")
        db_exec(f"DELETE FROM yoshop_order_address WHERE order_id IN ({ids})")
        db_exec(f"DELETE FROM yoshop_order WHERE order_id IN ({ids})")
        if stock_backup:
            db_exec(f"UPDATE yoshop_goods SET stock_total={stock_backup['stock_total']} WHERE goods_id={GOODS_ID}")
            db_exec(f"UPDATE yoshop_goods_sku SET stock_num={stock_backup['stock_num']} "
                    f"WHERE goods_id={GOODS_ID} AND goods_sku_id='{SKU_ID}'")
        if address_backup is not None:
            db_exec(f"UPDATE yoshop_user SET address_id={address_backup} WHERE user_id={USER_ID}")
        db_exec(f"UPDATE yoshop_cart SET is_delete=1 WHERE user_id={USER_ID} AND goods_id={GOODS_ID}")
        if points_setting_created:
            db_exec("DELETE FROM yoshop_store_setting WHERE store_id=10001 AND `key`='points'")
        if test_coupon_id:
            db_exec("DELETE FROM yoshop_user_coupon WHERE name='接口自测券'")
        db_exec(f"UPDATE yoshop_user SET points=0 WHERE user_id={USER_ID}")
        db_exec(f"DELETE FROM yoshop_user_points_log WHERE user_id={USER_ID} AND `describe` LIKE '%接口自测%'")
        print("  CLEAN 测试数据已清理")
    except Exception as e:  # noqa: BLE001
        print("  CLEAN 失败:", e)


def main():
    global address_backup, points_setting_created, test_coupon_id
    print("== 客户端 api 第 2 批（购物车 / 结算下单 / 订单 / 评价 / 地址） ==")
    redis_set("yoshop:api:token:" + TOKEN, USER_ID)
    rows = db(f"SELECT address_id FROM yoshop_user WHERE user_id={USER_ID}")
    address_backup = rows[0][0] if rows else 0
    stock = db(f"SELECT stock_total FROM yoshop_goods WHERE goods_id={GOODS_ID}")[0][0]
    sku_stock = db(f"SELECT stock_num FROM yoshop_goods_sku WHERE goods_id={GOODS_ID} "
                   f"AND goods_sku_id='{SKU_ID}'")[0][0]
    stock_backup.update(stock_total=stock, stock_num=sku_stock)
    db_exec(f"UPDATE yoshop_cart SET is_delete=1 WHERE user_id={USER_ID}")

    # ---------------------------------------------------------------- 未登录
    r = get("/api/cart/list", token=None)
    check("未登录购物车返回 401", r.get("status") == 401, r.get("message"))

    # ---------------------------------------------------------------- 收货地址
    print("-- 收货地址")
    r = get("/api/address/list")
    check("address/list status=200", r.get("status") == 200, r.get("message"))
    addr_list = field(r.get("data"), "list") or []
    check("address/list 是数组且含默认地址", any(a.get("addressId") == ADDRESS_ID for a in addr_list), addr_list)
    check("地址节点无 snake_case", not snake_keys(addr_list), snake_keys(addr_list)[:3])

    r = post("/api/address/add", {
        "name": "接口自测", "phone": "13800000000", "detail": "自测详细地址",
        "region": [{"value": 110000}, {"value": 110100}, {"value": 110101}],
    })
    check("address/add 添加成功", r.get("status") == 200 and r.get("message") == "添加成功", r)
    new_id = db(f"SELECT MAX(address_id) FROM yoshop_user_address WHERE user_id={USER_ID}")[0][0]
    row = db(f"SELECT name, province_id, city_id, region_id FROM yoshop_user_address WHERE address_id={new_id}")[0]
    check("address/add 省市区解析正确", tuple(row[1:]) == (110000, 110100, 110101), row)

    r = get("/api/address/default_id")
    check("address/default_id 原默认地址不变", field(r.get("data"), "defaultId") == ADDRESS_ID,
          r.get("data"))

    r = get("/api/address/detail", {"addressId": new_id})
    check("address/detail 返回新地址", field(field(r.get("data"), "detail"), "name") == "接口自测", r.get("data"))

    r = post("/api/address/edit", {
        "addressId": new_id, "name": "接口自测2", "phone": "13900000000", "detail": "改后地址",
        "region": [{"value": 310000}, {"value": 310100}, {"value": 310104}],
    })
    check("address/edit 更新成功", r.get("status") == 200 and r.get("message") == "更新成功", r)
    row = db(f"SELECT name, province_id, city_id FROM yoshop_user_address WHERE address_id={new_id}")[0]
    check("address/edit 已落库", row == ("接口自测2", 310000, 310100), row)

    r = post("/api/address/set_default", {"addressId": new_id})
    check("address/set_default 设置成功", r.get("status") == 200, r)
    check("address/set_default 落库", db(f"SELECT address_id FROM yoshop_user WHERE user_id={USER_ID}")[0][0] == new_id)

    r = post("/api/address/add", {"name": "缺省市区"})
    check("address/add 缺省市区报错", r.get("status") == 500, r)

    r = post("/api/address/remove", {"addressId": new_id})
    check("address/remove 删除成功", r.get("status") == 200 and r.get("message") == "删除成功", r)
    check("address/remove 清空默认地址", db(f"SELECT address_id FROM yoshop_user WHERE user_id={USER_ID}")[0][0] == 0)
    db_exec(f"UPDATE yoshop_user SET address_id={ADDRESS_ID} WHERE user_id={USER_ID}")

    r = get("/api/address/detail", {"addressId": new_id})
    check("address/detail 已删除不可见", r.get("status") == 500, r)

    # ---------------------------------------------------------------- 购物车
    print("-- 购物车")
    r = get("/api/cart/list")
    check("cart/list status=200", r.get("status") == 200, r.get("message"))
    check("cart/list 初始 cartTotal=0", field(r.get("data"), "cartTotal") == 0, r.get("data"))

    r = post("/api/cart/add", {"goodsId": GOODS_ID, "goodsSkuId": SKU_ID, "goodsNum": 1})
    check("cart/add 加入购物车", r.get("status") == 200 and field(r.get("data"), "cartTotal") == 1, r)

    r = post("/api/cart/add", {"goodsId": GOODS_ID, "goodsSkuId": SKU_ID, "goodsNum": 1})
    check("cart/add 同规格累加数量", field(r.get("data"), "cartTotal") == 2, r.get("data"))

    r = get("/api/cart/list")
    cart_list = field(r.get("data"), "list") or []
    check("cart/list 一条记录", len(cart_list) == 1, len(cart_list))
    if cart_list:
        item = cart_list[0]
        goods = field(item, "goods") or {}
        check("cart/list 含 goods.skuInfo", field(goods, "skuInfo") is not None, list(goods))
        check("cart/list skuInfo.goodsPrice 正确",
              str(field(field(goods, "skuInfo"), "goodsPrice")) == SKU_PRICE, field(goods, "skuInfo"))
        check("cart/list 记录隐藏 storeId/createTime",
              not ({"storeId", "createTime", "isDelete"} & set(item)), list(item))
    check("cart/list 无 snake_case 键", not snake_keys(cart_list), snake_keys(cart_list)[:3])

    r = post("/api/cart/update", {"goodsId": GOODS_ID, "goodsSkuId": SKU_ID, "goodsNum": 3})
    check("cart/update 更新数量", field(r.get("data"), "cartTotal") == 3, r.get("data"))
    check("cart/update 落库", db(f"SELECT goods_num FROM yoshop_cart WHERE user_id={USER_ID} "
                                 f"AND is_delete=0 AND goods_id={GOODS_ID}")[0][0] == 3)

    r = get("/api/cart/total")
    check("cart/total 返回 3", field(r.get("data"), "cartTotal") == 3, r.get("data"))

    r = post("/api/cart/add", {"goodsId": 999999, "goodsSkuId": "0", "goodsNum": 1})
    check("cart/add 商品不存在报错", r.get("status") == 500, r)

    r = post("/api/cart/add", {"goodsId": GOODS_ID, "goodsSkuId": SKU_ID, "goodsNum": 99999})
    check("cart/add 超库存报错", r.get("status") == 500 and "库存" in (r.get("message") or ""), r)

    # ---------------------------------------------------------------- 结算台
    print("-- 结算台")
    r = get("/api/checkout/order", {"mode": "buyNow", "goodsId": FREE_GOODS_ID,
                                    "goodsSkuId": FREE_SKU_ID, "goodsNum": 2})
    order = field(r.get("data"), "order") or {}
    check("checkout 包邮商品订单总额", order.get("orderTotalPrice") == "79.80", order.get("orderTotalPrice"))
    check("checkout 包邮运费 0.00", order.get("expressPrice") == "0.00", order.get("expressPrice"))
    check("checkout 含 personal/setting", {"personal", "setting"} <= set(r.get("data") or {}), list(r.get("data") or {}))
    check("checkout personal 四项",
          set(field(r.get("data"), "personal") or {}) == {"userId", "balance", "points", "addressId"},
          field(r.get("data"), "personal"))
    check("checkout setting 三项",
          set(field(r.get("data"), "setting") or {}) == {"deliveryType", "pointsName", "pointsDescribe"},
          field(r.get("data"), "setting"))
    check("checkout 有地址且 isIntraRegion", order.get("existAddress") is True and order.get("isIntraRegion") is True,
          order.get("existAddress"))

    r = get("/api/checkout/order", {"mode": "buyNow", "goodsId": GOODS_ID,
                                    "goodsSkuId": SKU_ID, "goodsNum": 1})
    order = field(r.get("data"), "order") or {}
    check("checkout 首件运费 3.00", order.get("expressPrice") == "3.00", order.get("expressPrice"))
    check("checkout 订单实付 83.00", order.get("orderPayPrice") == "83.00", order.get("orderPayPrice"))
    check("checkout 商品行字段完整", {"goodsPrice", "totalNum", "totalPrice", "totalPayPrice", "skuInfo"}
          <= set((order.get("goodsList") or [{}])[0]), list((order.get("goodsList") or [{}])[0]))

    r = get("/api/checkout/order", {"mode": "buyNow", "goodsId": GOODS_ID,
                                    "goodsSkuId": SKU_ID, "goodsNum": 2})
    order = field(r.get("data"), "order") or {}
    check("checkout 续件运费 6.00", order.get("expressPrice") == "6.00", order.get("expressPrice"))
    check("checkout 续件实付 166.00", order.get("orderPayPrice") == "166.00", order.get("orderPayPrice"))
    check("checkout 无 snake_case 键", not snake_keys(r.get("data")), snake_keys(r.get("data"))[:3])

    r = get("/api/checkout/order", {"mode": "cart", "cartIds": ",".join(
        str(x[0]) for x in db(f"SELECT id FROM yoshop_cart WHERE user_id={USER_ID} AND is_delete=0"))})
    order = field(r.get("data"), "order") or {}
    check("checkout 购物车模式商品数 3", order.get("orderTotalNum") == 3, order.get("orderTotalNum"))
    check("checkout 购物车模式总额 240.00", order.get("orderTotalPrice") == "240.00", order.get("orderTotalPrice"))

    r = get("/api/checkout/order", {"mode": "buyNow", "goodsId": GOODS_ID,
                                    "goodsSkuId": SKU_ID, "goodsNum": 99999})
    order = field(r.get("data"), "order") or {}
    check("checkout 超库存 hasError", order.get("hasError") is True and "库存" in (order.get("errorMsg") or ""),
          order.get("errorMsg"))

    r = get("/api/checkout/order", {"mode": "unknown"})
    check("checkout 非法模式报错", r.get("status") == 500, r.get("message"))

    # ---------------------------------------------------------------- 下单
    print("-- 下单")
    before = db(f"SELECT stock_total FROM yoshop_goods WHERE goods_id={GOODS_ID}")[0][0]
    r = post("/api/checkout/submit", {"mode": "buyNow", "goodsId": GOODS_ID,
                                      "goodsSkuId": SKU_ID, "goodsNum": 1, "remark": "接口自测"})
    order_id = field(r.get("data"), "orderId")
    check("checkout/submit 下单成功", r.get("status") == 200 and order_id, r)
    if not order_id:
        print(f"\n结果：PASS {passed} / FAIL {failed}（下单失败，后续订单断言跳过）")
        return 1
    created_orders.append(order_id)
    row = db(f"SELECT order_no, user_id, order_status, pay_status, delivery_type, express_price, "
             f"pay_price, buyer_remark, is_delete FROM yoshop_order WHERE order_id={order_id}")[0]
    check("订单已落库且待支付", row[2] == 10 and row[3] == 10, row)
    check("订单运费 3.00 / 实付 83.00", str(row[5]) == "3.00" and str(row[6]) == "83.00", row)
    check("订单备注已保存", row[7] == "接口自测", row[7])
    goods_row = db(f"SELECT goods_id, goods_sku_id, goods_price, total_num, total_price, total_pay_price "
                   f"FROM yoshop_order_goods WHERE order_id={order_id}")[0]
    check("订单商品已落库", goods_row[0] == GOODS_ID and str(goods_row[2]) == SKU_PRICE and goods_row[3] == 1, goods_row)
    check("订单收货地址已落库", db(f"SELECT COUNT(*) FROM yoshop_order_address WHERE order_id={order_id}")[0][0] == 1)
    after = db(f"SELECT stock_total FROM yoshop_goods WHERE goods_id={GOODS_ID}")[0][0]
    check("下单扣减商品库存", after == before - 1, (before, after))

    # ---------------------------------------------------------------- 订单接口
    print("-- 订单")
    r = get("/api/order/detail", {"orderId": order_id})
    detail = field(r.get("data"), "order") or {}
    check("order/detail status=200", r.get("status") == 200, r.get("message"))
    check("order/detail 含 goods/address/stateText", {"goods", "address", "stateText"} <= set(detail), list(detail))
    check("order/detail 隐藏 merchantRemark/tradeId/isDelete",
          not ({"merchantRemark", "tradeId", "isDelete", "storeId"} & set(detail)), list(detail))
    check("order/detail 商品隐藏 image/content",
          not ({"image", "content", "storeId"} & set((detail.get("goods") or [{}])[0])),
          list((detail.get("goods") or [{}])[0]))
    check("order/detail setting.pointsName 键存在", "pointsName" in (field(r.get("data"), "setting") or {}))
    check("order/detail 无 snake_case 键", not snake_keys(r.get("data")), snake_keys(r.get("data"))[:3])

    r = get("/api/order/list", {"dataType": "payment"})
    page = field(r.get("data"), "list") or {}
    ids = [o.get("orderId") for o in (page.get("data") or [])]
    check("order/list 待付款含新订单", order_id in ids, ids)
    check("order/list 分页字段 perPage=15", page.get("perPage") == 15, page.get("perPage"))

    r = get("/api/order/todo_counts")
    counts = field(r.get("data"), "counts") or {}
    check("order/todo_counts 四个键", set(counts) == {"payment", "delivery", "received", "refund"}, counts)
    check("order/todo_counts 待付款 >=1", counts.get("payment", 0) >= 1, counts)

    r = get("/api/order/express", {"orderId": order_id})
    check("order/express 未发货返回空数组", r.get("status") == 200 and r.get("data") == {"express": []}, r.get("data"))

    r = post("/api/order/cancel", {"orderId": order_id})
    check("order/cancel 未支付订单取消成功",
          r.get("status") == 200 and r.get("message") == "订单已取消成功", r)
    check("order/cancel 订单状态 20",
          db(f"SELECT order_status FROM yoshop_order WHERE order_id={order_id}")[0][0] == 20)
    check("order/cancel 回滚库存",
          db(f"SELECT stock_total FROM yoshop_goods WHERE goods_id={GOODS_ID}")[0][0] == before)

    # 收货：造已发货订单
    receipt_order = make_order(10, 20, 20, 10)
    r = post("/api/order/receipt", {"orderId": receipt_order})
    check("order/receipt 确认收货成功", r.get("status") == 200 and r.get("message") == "确认收货成功", r)
    row = db(f"SELECT order_status, receipt_status FROM yoshop_order WHERE order_id={receipt_order}")[0]
    check("order/receipt 状态已更新", row == (30, 20), row)

    r = post("/api/order/receipt", {"orderId": receipt_order})
    check("order/receipt 重复收货报错", r.get("status") == 500 and "不合法" in (r.get("message") or ""), r)

    r = get("/api/order/detail", {"orderId": 99999999})
    check("order/detail 不存在报错", r.get("status") == 500 and "订单不存在" in (r.get("message") or ""), r)

    # ---------------------------------------------------------------- 订单评价
    print("-- 订单评价")
    comment_order = make_order(30, 20, 20, 20)
    r = get("/api/order.comment/list", {"orderId": comment_order})
    goods_list = field(r.get("data"), "goodsList") or []
    check("order.comment/list 待评价商品 1 条", len(goods_list) == 1, len(goods_list))
    order_goods_id = field(goods_list[0], "orderGoodsId") if goods_list else None
    check("order.comment/list 商品含 goodsImage", "goodsImage" in (goods_list[0] if goods_list else {}))

    r = post("/api/order.comment/submit", [
        {"orderGoodsId": order_goods_id, "goodsId": GOODS_ID, "score": 10,
         "content": "接口自测评价", "uploaded": []},
    ], params={"orderId": comment_order})
    check("order.comment/submit 评价成功", r.get("status") == 200 and r.get("message") == "评价发表成功", r)
    check("评价已落库", db(f"SELECT content FROM yoshop_comment WHERE order_id={comment_order}")[0][0] == "接口自测评价")
    check("订单商品标记已评价",
          db(f"SELECT is_comment FROM yoshop_order_goods WHERE order_id={comment_order}")[0][0] == 1)
    check("订单标记已评价", db(f"SELECT is_comment FROM yoshop_order WHERE order_id={comment_order}")[0][0] == 1)

    r = get("/api/order.comment/list", {"orderId": comment_order})
    check("已评价后列表报错", r.get("status") == 500 and "已完成评价" in (r.get("message") or ""), r)

    # 未完成订单不可评价
    pending_order = make_order(10, 10, 10, 10)
    r = get("/api/order.comment/list", {"orderId": pending_order})
    check("未完成订单不可评价", r.get("status") == 500 and "未完成" in (r.get("message") or ""), r)

    # ---------------------------------------------------------------- 优惠券 + 积分抵扣
    print("-- 优惠券与积分抵扣（临时造数据）")
    points_setting = json.dumps({
        "pointsName": "积分", "describe": "接口自测",
        "isShoppingDiscount": 1, "isShoppingGift": 1, "giftRatio": 100,
        "discount": {"fullOrderPrice": 0, "maxMoneyRatio": 100, "discountRatio": 0.01},
    }, ensure_ascii=False)
    db_exec("INSERT INTO yoshop_store_setting (`key`, `describe`, `values`, store_id, create_time, update_time) "
            f"VALUES ('points', '积分设置', '{points_setting}', 10001, 0, UNIX_TIMESTAMP())")
    points_setting_created = True
    db_exec(f"UPDATE yoshop_user SET points=1000 WHERE user_id={USER_ID}")
    coupon_row = db_exec(
        "INSERT INTO yoshop_user_coupon (coupon_id, name, coupon_type, reduce_price, discount, min_price, "
        "expire_type, expire_day, start_time, end_time, apply_range, apply_range_config, `describe`, "
        "is_expire, is_use, user_id, store_id, create_time, update_time) VALUES "
        "(0, '接口自测券', 10, 5.00, 0, 0.00, 10, 30, UNIX_TIMESTAMP()-3600, UNIX_TIMESTAMP()+86400, "
        f"10, '{{\"goodsIds\":[]}}', '', 0, 0, {USER_ID}, 10001, UNIX_TIMESTAMP(), UNIX_TIMESTAMP())")
    test_coupon_id = db(f"SELECT user_coupon_id FROM yoshop_user_coupon WHERE name='接口自测券'")[0][0]

    r = get("/api/checkout/order", {"mode": "buyNow", "goodsId": GOODS_ID, "goodsSkuId": SKU_ID,
                                    "goodsNum": 1, "couponId": test_coupon_id, "isUsePoints": 1})
    order = field(r.get("data"), "order") or {}
    check("结算台返回可用优惠券", len(order.get("couponList") or []) >= 1, order.get("couponList"))
    check("优惠券抵扣 5.00", order.get("couponMoney") == "5.00", order.get("couponMoney"))
    check("积分可抵扣 isAllowPoints", order.get("isAllowPoints") is True, order.get("isAllowPoints"))
    check("积分抵扣数量 1000", order.get("pointsNum") == 1000, order.get("pointsNum"))
    check("积分抵扣金额 10.00", order.get("pointsMoney") == "10.00", order.get("pointsMoney"))
    check("订单金额 65.00 / 实付 68.00",
          order.get("orderPrice") == "65.00" and order.get("orderPayPrice") == "68.00",
          (order.get("orderPrice"), order.get("orderPayPrice")))
    check("赠送积分 65", order.get("pointsBonus") == 65, order.get("pointsBonus"))
    line = (order.get("goodsList") or [{}])[0]
    check("商品行优惠券/积分/实付",
          line.get("couponMoney") == "5.00" and line.get("pointsMoney") == "10.00"
          and line.get("totalPayPrice") == "65.00",
          (line.get("couponMoney"), line.get("pointsMoney"), line.get("totalPayPrice")))

    r = post("/api/checkout/submit", {"mode": "buyNow", "goodsId": GOODS_ID, "goodsSkuId": SKU_ID,
                                      "goodsNum": 1, "couponId": test_coupon_id, "isUsePoints": 1})
    coupon_order = field(r.get("data"), "orderId")
    check("带优惠券+积分下单成功", r.get("status") == 200 and coupon_order, r)
    if coupon_order:
        created_orders.append(coupon_order)
        row = db(f"SELECT coupon_id, coupon_money, points_money, points_num, pay_price, points_bonus "
                 f"FROM yoshop_order WHERE order_id={coupon_order}")[0]
        check("订单优惠券已落库", row[0] == test_coupon_id and str(row[1]) == "5.00", row)
        check("订单积分抵扣已落库", str(row[2]) == "10.00" and row[3] == 1000, row)
        check("订单实付 68.00 / 赠送积分 65", str(row[4]) == "68.00" and row[5] == 65, row)
        check("优惠券已核销",
              db(f"SELECT is_use FROM yoshop_user_coupon WHERE user_coupon_id={test_coupon_id}")[0][0] == 1)
        check("用户积分已扣减 1000→0", db(f"SELECT points FROM yoshop_user WHERE user_id={USER_ID}")[0][0] == 0)
        check("积分流水已记录",
              db(f"SELECT COUNT(*) FROM yoshop_user_points_log WHERE user_id={USER_ID} AND value=-1000")[0][0] >= 1)

        r = post("/api/order/cancel", {"orderId": coupon_order})
        check("取消订单成功", r.get("status") == 200, r)
        check("取消回退积分", db(f"SELECT points FROM yoshop_user WHERE user_id={USER_ID}")[0][0] == 1000)
        check("取消回退优惠券",
              db(f"SELECT is_use FROM yoshop_user_coupon WHERE user_coupon_id={test_coupon_id}")[0][0] == 0)

    print(f"\n结果：PASS {passed} / FAIL {failed}")
    return 1 if failed else 0


if __name__ == "__main__":
    code = 1
    try:
        code = main()
    finally:
        cleanup()
    sys.exit(code)