shop-java/tools/test_api_batch4.py

# -*- coding: utf-8 -*-
"""客户端 api 第 4 批接口冒烟：图形/短信验证码、登录、收银台、支付回调。

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

说明：
- 图形验证码的答案从 Redis 读取（模拟用户看图输入），验证码接口本身走真实链路；
- 短信通道未接入，验证码只落 Redis，脚本同样从 Redis 取值完成登录；
- 会真实下单/支付/回调，finally 中删除测试订单、交易记录、充值单、余额流水与测试用户，
  并恢复用户余额 / 总支付金额 / 商品销量。
"""
import hashlib
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-batch4"
USER_ID = 10001
# 登录用测试手机号（脚本会创建并在 finally 删除）
TEST_MOBILE = "13900001111"
# 支付链路测试商品（下单减库存、单规格由 make_order 固定为 goods_sku_id）
GOODS_ID = 10009
SKU_ID = "10009"

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("\"'")
# 是否校验短信验证码：跟随后端 yoshop.sms.verify-code（未配置时为 false）
_m = re.search(r"verify-code:\s*(\w+)", TEXT)
SMS_VERIFY = bool(_m) and _m.group(1).lower() in ("true", "1", "yes")

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

passed = 0
failed = 0
created_orders = []
created_trades = []
created_recharge_orders = []
created_payments = []
backup = {}


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 raw_post(path, text=None, data=None, headers=None):
    resp = S.post(BASE + path, data=text if text is not None else data,
                  headers=headers or {}, timeout=60)
    return resp.text


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_cmd(*args):
    sock = socket.create_connection(("192.168.1.177", 6379), timeout=5)
    try:
        cmd(sock, "AUTH", REDIS_PWD)
        return cmd(sock, *args)
    finally:
        sock.close()


def redis_get(key):
    return redis_cmd("GET", key)


def redis_del(key):
    return redis_cmd("DEL", key)


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


def new_captcha():
    """取一张新图形验证码，返回 (接口数据, 明文验证码)"""
    data = get("/api/captcha/image", token=None).get("data") or {}
    raw = redis_get("yoshop:api:captcha:" + str(data.get("key")))
    code = json.loads(raw)["code"] if raw else None
    return data, code


def send_sms(mobile):
    """完整走一遍「图形验证码 → 发送短信」，返回 (响应, 短信验证码, 所用图形验证码)"""
    data, code = new_captcha()
    resp = post("/api/captcha/sendSmsCaptcha",
                {"mobile": mobile, "captchaCode": code, "captchaKey": data.get("key")}, token=None)
    raw = redis_get("yoshop:api:sms:" + mobile)
    sms = json.loads(raw)["code"] if raw else None
    return resp, sms, data


def make_order(order_status=10, pay_status=10, order_no=None):
    order_no = order_no or ("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, 10, 10, {order_status}, 0, 0, 0, 10, 0, '[]', "
        f"'H5', {USER_ID}, 0, 10001, UNIX_TIMESTAMP(), UNIX_TIMESTAMP())")
    created_orders.append(order_id)
    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, order_no


def make_trade(order_id, order_no, method="wechat", order_type=10, user_id=USER_ID):
    out_trade_no = "NT" + str(os.getpid()) + str(len(created_trades)) + str(int(order_id))
    trade_id = db_exec(
        "INSERT INTO yoshop_payment_trade (out_trade_no, client, pay_method, order_type, order_id, order_no, "
        "user_id, trade_no, prepay_id, trade_state, store_id, create_time, update_time) VALUES "
        f"('{out_trade_no}', 'MP-WEIXIN', '{method}', {order_type}, {order_id}, '{order_no}', {user_id}, "
        f"'', '', 10, 10001, UNIX_TIMESTAMP(), UNIX_TIMESTAMP())")
    created_trades.append(trade_id)
    return trade_id, out_trade_no


def cleanup():
    try:
        ids = ",".join(str(i) for i in created_orders) or "0"
        tids = ",".join(str(i) for i in created_trades) or "0"
        rids = ",".join(str(i) for i in created_recharge_orders) or "0"
        db_exec(f"DELETE FROM yoshop_payment_trade WHERE trade_id IN ({tids})")
        db_exec(f"DELETE FROM yoshop_order_goods WHERE order_id IN ({ids})")
        db_exec(f"DELETE FROM yoshop_order WHERE order_id IN ({ids})")
        db_exec(f"DELETE FROM yoshop_recharge_order WHERE order_id IN ({rids})")
        pids = ",".join(str(i) for i in created_payments) or "0"
        db_exec(f"DELETE FROM yoshop_payment WHERE payment_id IN ({pids})")
        db_exec(f"DELETE FROM yoshop_user_balance_log WHERE user_id={USER_ID} AND "
                f"(`describe` LIKE '用户消费：TEST%' OR `describe` LIKE '用户充值：TEST%')")
        if backup:
            db_exec(f"UPDATE yoshop_user SET balance={backup['balance']}, pay_money={backup['pay_money']}, "
                    f"points={backup['points']} WHERE user_id={USER_ID}")
            db_exec(f"UPDATE yoshop_goods SET sales_actual={backup['sales_actual']} WHERE goods_id={GOODS_ID}")
        db_exec(f"DELETE FROM yoshop_user_oauth WHERE user_id IN "
                f"(SELECT user_id FROM yoshop_user WHERE mobile='{TEST_MOBILE}')")
        db_exec(f"DELETE FROM yoshop_user WHERE mobile='{TEST_MOBILE}'")
        for key in ["yoshop:api:sms:" + TEST_MOBILE, "yoshop:api:sms:send:" + TEST_MOBILE,
                    "yoshop:api:token:" + TOKEN]:
            redis_del(key)
        print("  CLEAN 测试数据已清理")
    except Exception as e:  # noqa: BLE001
        print("  CLEAN 失败:", e)


def main():
    print("== 客户端 api 第 4 批（验证码 / 登录 / 收银台 / 支付回调） ==")
    redis_cmd("SET", "yoshop:api:token:" + TOKEN, str(USER_ID))
    redis_cmd("EXPIRE", "yoshop:api:token:" + TOKEN, "3600")
    row = db(f"SELECT balance, pay_money, points FROM yoshop_user WHERE user_id={USER_ID}")[0]
    backup.update(balance=row[0], pay_money=row[1], points=row[2],
                  sales_actual=db(f"SELECT sales_actual FROM yoshop_goods WHERE goods_id={GOODS_ID}")[0][0])
    for key in ["yoshop:api:sms:" + TEST_MOBILE, "yoshop:api:sms:send:" + TEST_MOBILE]:
        redis_del(key)

    # ---------------------------------------------------------------- 图形验证码
    print("\n[1] 图形验证码")
    r = get("/api/captcha/image", token=None)
    data = field(r, "data") or {}
    check("captcha/image 成功", r.get("status") == 200, r)
    check("出参仅 base64/key/md5", sorted(data.keys()) == ["base64", "key", "md5"], list(data.keys()))
    base64_text = str(data.get("base64"))
    check("base64 为 png data URI", base64_text.startswith("data:image/png;base64,") and len(base64_text) > 1000,
          base64_text[:40])
    check("key 为 bcrypt 摘要", str(data.get("key", "")).startswith("$2"), str(data.get("key"))[:10])

    cache = redis_get("yoshop:api:captcha:" + str(data.get("key")))
    cached = json.loads(cache) if cache else {}
    check("验证码写入 Redis", cached.get("code") and cached.get("times") == 5, cache)
    check("md5 为验证码摘要",
          data.get("md5") == hashlib.md5(str(cached.get("code")).encode()).hexdigest(), data.get("md5"))
    ttl = redis_cmd("TTL", "yoshop:api:captcha:" + str(data.get("key")))
    check("验证码 TTL 为 1800 秒内", isinstance(ttl, int) and 0 < ttl <= 1800, ttl)

    # ---------------------------------------------------------------- 发送短信验证码
    print("\n[2] 发送短信验证码")
    r = post("/api/captcha/sendSmsCaptcha", {"captchaCode": "abc", "captchaKey": "x"}, token=None)
    check("缺手机号报错", r.get("status") == 500 and "手机号不能为空" in (r.get("message") or ""), r)
    r = post("/api/captcha/sendSmsCaptcha", {"mobile": TEST_MOBILE}, token=None)
    check("缺图形验证码报错", r.get("status") == 500 and "图形验证码code不能为空" in (r.get("message") or ""), r)
    r = post("/api/captcha/sendSmsCaptcha",
             {"mobile": TEST_MOBILE, "captchaCode": "wrong", "captchaKey": data.get("key")}, token=None)
    check("图形验证码错误报错", r.get("status") == 500 and "图形验证码不正确" in (r.get("message") or ""), r)

    r = post("/api/captcha/sendSmsCaptcha",
             {"mobile": TEST_MOBILE, "captchaCode": "x"}, token=None)
    check("图形验证码 key 为空报错", r.get("status") == 500 and "图形验证码key不能为空" in (r.get("message") or ""), r)

    r, sms_code, used_captcha = send_sms(TEST_MOBILE)
    check("短信发送成功", r.get("status") == 200 and "发送成功" in (r.get("message") or ""), r)
    check("发送成功 data 为空数组", r.get("data") == [], r.get("data"))
    check("短信验证码为 6 位数字", bool(sms_code) and re.fullmatch(r"\d{6}", sms_code), sms_code)
    r = post("/api/captcha/sendSmsCaptcha",
             {"mobile": TEST_MOBILE, "captchaCode": "x", "captchaKey": used_captcha.get("key")}, token=None)
    check("图形验证码用后即焚", r.get("status") == 500 and "图形验证码不存在" in (r.get("message") or ""), r)

    # 每日发送次数限制：首次已用 1 次，再发 9 次后应为 0 次
    last = None
    for _ in range(9):
        last, _, _ = send_sms(TEST_MOBILE)
    check("第 10 次发送仍成功", last.get("status") == 200, last)
    last, _, _ = send_sms(TEST_MOBILE)
    check("超出每日发送次数被拒", last.get("status") == 500 and "已超出今日最大发送次数限制" in (last.get("message") or ""), last)
    left = redis_get("yoshop:api:sms:send:" + TEST_MOBILE)
    check("发送次数计为 0", str(left) == "0", left)
    # 重置发送次数窗口，避免影响后续登录用例
    redis_del("yoshop:api:sms:send:" + TEST_MOBILE)

    # ---------------------------------------------------------------- 登录
    print("\n[3] 手机号 + 短信验证码登录")
    r = post("/api/passport/login", {"smsCode": "123456"}, token=None)
    check("缺手机号报错", r.get("status") == 500 and "手机号不能为空" in (r.get("message") or ""), r)
    r = post("/api/passport/login", {"mobile": "12345", "smsCode": "123456"}, token=None)
    check("手机号格式校验", r.get("status") == 500 and "手机号格式不正确" in (r.get("message") or ""), r)

    if SMS_VERIFY:
        r = post("/api/passport/login", {"mobile": TEST_MOBILE}, token=None)
        check("缺短信验证码报错", r.get("status") == 500 and "短信验证码不能为空" in (r.get("message") or ""), r)
        redis_del("yoshop:api:sms:" + TEST_MOBILE)
        r = post("/api/passport/login", {"mobile": TEST_MOBILE, "smsCode": "123456"}, token=None)
        check("验证码不存在报错", r.get("status") == 500 and "短信验证码不存在" in (r.get("message") or ""), r)

        _, sms_code, _ = send_sms(TEST_MOBILE)
        r = post("/api/passport/login", {"mobile": TEST_MOBILE, "smsCode": "000000"}, token=None)
        check("验证码错误报错", r.get("status") == 500 and "短信验证码不正确" in (r.get("message") or ""), r)
        check("验证码错误后次数减一", json.loads(redis_get("yoshop:api:sms:" + TEST_MOBILE)).get("times") == 4,
              redis_get("yoshop:api:sms:" + TEST_MOBILE))

        r = post("/api/passport/login", {"mobile": TEST_MOBILE, "smsCode": sms_code}, token=None)
        check("登录后验证码用后即焚", redis_get("yoshop:api:sms:" + TEST_MOBILE) is None)
    else:
        print("  （短信通道未接入，yoshop.sms.verify-code=false：验证码随便填）")
        r = post("/api/passport/login", {"mobile": TEST_MOBILE}, token=None)

    login_data = field(r, "data") or {}
    check("登录成功", r.get("status") == 200 and r.get("message") == "登录成功", r)
    if not SMS_VERIFY:
        check("空短信验证码也能登录（联调开关）", r.get("status") == 200, r)
    check("返回 userId 与 token", isinstance(login_data.get("userId"), int)
          and re.fullmatch(r"[0-9a-f]{32}", str(login_data.get("token"))) is not None, login_data)
    user_id = login_data.get("userId")
    check("新用户写入库", bool(user_id) and db(f"SELECT COUNT(*) FROM yoshop_user WHERE user_id={user_id}")[0][0] == 1)
    nick = db(f"SELECT nick_name, platform FROM yoshop_user WHERE user_id={user_id}")[0] if user_id else ("", "")
    check("默认昵称为打码手机号", nick[0] == "139****1111", nick)
    check("注册来源取 platform 头", nick[1] == "MP-WEIXIN", nick)

    login_token = str(login_data.get("token"))
    r = get("/api/user/info", token=login_token)
    info = field(field(r, "data"), "userInfo") or {}
    check("登录态可用", r.get("status") == 200 and info.get("userId") == user_id, r)
    check("用户信息手机号打码", info.get("mobile") == "139****1111", info.get("mobile"))

    if SMS_VERIFY:
        _, sms_code, _ = send_sms(TEST_MOBILE)
        r = post("/api/passport/login", {"mobile": TEST_MOBILE, "smsCode": sms_code},
                 token=None)
    else:
        r = post("/api/passport/login", {"mobile": TEST_MOBILE, "smsCode": "000000"},
                 token=None)
    check("重复登录复用同一账号", field(r, "data", {}).get("userId") == user_id, r)

    # ---------------------------------------------------------------- 微信小程序登录
    print("\n[4] 微信小程序登录")
    r = post("/api/passport/loginMpWx", {"partyData": {"oauth": "MP-WEIXIN", "userInfo": {}}}, token=None)
    check("缺 code 报错", r.get("status") == 500 and "code 不能为空" in (r.get("message") or ""), r)
    r = post("/api/passport/loginMpWx",
             {"partyData": {"oauth": "MP-WEIXIN", "code": "fake-code", "userInfo": {}}}, token=None)
    check("未配置小程序参数时报错",
          r.get("status") == 500 and "AppID和AppSecret" in (r.get("message") or ""), r)
    r = post("/api/passport/loginMpWxMobile",
             {"code": "fake-code", "encryptedData": "x", "iv": "y"}, token=None)
    check("授权手机号登录同样受小程序配置限制",
          r.get("status") == 500 and "AppID和AppSecret" in (r.get("message") or ""), r)
    r = post("/api/passport/isPersonalMpweixin", {"code": "fake-code"}, token=None)
    check("isPersonalMpweixin 关闭时直接返回 false",
          r.get("status") == 200 and field(r, "data", {}).get("isPersonalMpweixin") is False, r)
    check("微信登录接口零 snake key", not snake_keys(r), snake_keys(r))

    # ---------------------------------------------------------------- 收银台
    print("\n[5] 收银台")
    r = get("/api/cashier/orderInfo", {"orderId": 1, "client": "MP-WEIXIN"}, token=None)
    check("未登录访问收银台 401", r.get("status") == 401, r)

    order_id, order_no = make_order()
    cancelled_id, _ = make_order(order_status=20)
    r = get("/api/cashier/orderInfo", {"orderId": 999999, "client": "MP-WEIXIN"})
    check("订单不存在报错", r.get("status") == 500 and "订单不存在" in (r.get("message") or ""), r)
    r = get("/api/cashier/orderInfo", {"orderId": cancelled_id, "client": "MP-WEIXIN"})
    check("已取消订单不可支付", r.get("status") == 500 and "不允许支付" in (r.get("message") or ""), r)

    r = get("/api/cashier/orderInfo", {"orderId": order_id, "client": "MP-WEIXIN"})
    info = field(r, "data") or {}
    order_part = info.get("order") or {}
    check("收银台返回订单/用户/支付方式", r.get("status") == 200
          and sorted(info.keys()) == ["order", "paymentMethods", "personal"], list(info.keys()))
    check("订单金额两位小数字符串", order_part.get("payPrice") == "83.00", order_part)
    check("返回超时信息", order_part.get("showExpiration") is True
          and re.fullmatch(r"\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}", str(order_part.get("expirationTime"))) is not None,
          order_part)
    check("订单字段齐全", {"orderId", "orderNo", "payStatus", "orderStatus", "createTime"} <= set(order_part.keys()),
          list(order_part.keys()))
    methods = [m.get("method") for m in (info.get("paymentMethods") or [])]
    check("支付方式含余额支付", "balance" in methods, methods)
    check("收银台零 snake key", not snake_keys(info), snake_keys(info))

    r = post("/api/cashier/orderPay", {"orderId": order_id, "client": "MP-WEIXIN"})
    check("缺支付方式报错", r.get("status") == 500 and "请选择支付方式" in (r.get("message") or ""), r)
    r = post("/api/cashier/orderPay", {"orderId": order_id, "method": "unknown", "client": "MP-WEIXIN"})
    check("支付方式不可用报错", r.get("status") == 500 and "支付方式不可用" in (r.get("message") or ""), r)
    db_exec(f"UPDATE yoshop_user SET balance=0 WHERE user_id={USER_ID}")
    r = post("/api/cashier/orderPay", {"orderId": order_id, "method": "balance", "client": "MP-WEIXIN"})
    check("余额不足报错", r.get("status") == 500 and "余额不足" in (r.get("message") or ""), r)
    wechat_payment_id = db_exec(
        "INSERT INTO yoshop_payment (client, method, is_must_template, template_id, is_enable, is_default, "
        "others, store_id, create_time, update_time) VALUES ('MP-WEIXIN', 'wechat', 0, 0, 1, 0, '{}', 10001, "
        "UNIX_TIMESTAMP(), UNIX_TIMESTAMP())")
    created_payments.append(wechat_payment_id)
    r = post("/api/cashier/orderPay", {"orderId": order_id, "method": "wechat", "client": "MP-WEIXIN"})
    check("微信支付未接入有明确文案", r.get("status") == 500 and "尚未接入" in (r.get("message") or ""), r)

    pay_money_before = float(db(f"SELECT pay_money FROM yoshop_user WHERE user_id={USER_ID}")[0][0])
    db_exec(f"UPDATE yoshop_user SET balance=500 WHERE user_id={USER_ID}")
    r = post("/api/cashier/orderPay", {"orderId": order_id, "method": "balance", "client": "MP-WEIXIN"})
    check("余额支付成功", r.get("status") == 200 and r.get("message") == "恭喜您，订单已付款成功", r)
    check("余额支付不返回第三方参数", field(r, "data", {}).get("payment") == [], r.get("data"))
    row = db(f"SELECT pay_status, pay_method, trade_id, pay_time FROM yoshop_order WHERE order_id={order_id}")[0]
    check("订单置为已付款", row[0] == 20 and row[1] == "balance" and row[3] > 0, row)
    balance = float(db(f"SELECT balance FROM yoshop_user WHERE user_id={USER_ID}")[0][0])
    check("余额扣减订单金额", abs(balance - 417.0) < 0.001, balance)
    pay_money = float(db(f"SELECT pay_money FROM yoshop_user WHERE user_id={USER_ID}")[0][0])
    check("累加用户总支付金额", abs(pay_money - pay_money_before - 83.0) < 0.001, pay_money)
    logs = db(f"SELECT scene, money, `describe` FROM yoshop_user_balance_log WHERE user_id={USER_ID} "
              f"AND `describe`='用户消费：{order_no}'")
    check("写余额消费明细", len(logs) == 1 and logs[0][0] == 20 and float(logs[0][1]) == -83.0, logs)
    sales = db(f"SELECT sales_actual FROM yoshop_goods WHERE goods_id={GOODS_ID}")[0][0]
    check("累计商品销量", sales == backup["sales_actual"] + 1, (sales, backup["sales_actual"]))

    r = post("/api/cashier/orderPay", {"orderId": order_id, "method": "balance", "client": "MP-WEIXIN"})
    check("已支付订单不可重复支付", r.get("status") == 500 and "已支付" in (r.get("message") or ""), r)

    r = get("/api/cashier/tradeQuery", {"method": "wechat", "client": "MP-WEIXIN"})
    check("交易查询缺单号报错", r.get("status") == 500 and "商户订单号不能为空" in (r.get("message") or ""), r)
    r = get("/api/cashier/tradeQuery", {"outTradeNo": "X1", "method": "wechat", "client": "MP-WEIXIN"})
    check("交易查询失败仍返回成功响应",
          r.get("status") == 200 and field(r, "data", {}).get("isPay") is False, r)
    check("交易查询命中未接入文案", "尚未接入" in (r.get("message") or ""), r.get("message"))
    r = get("/api/cashier/tradeQuery", {"outTradeNo": "X1", "method": "balance", "client": "MP-WEIXIN"})
    check("非第三方交易查询返回未支付",
          r.get("status") == 200 and field(r, "data", {}).get("isPay") is False
          and "未支付" in (r.get("message") or ""), r)

    # ---------------------------------------------------------------- 支付回调
    print("\n[6] 支付回调")
    text = raw_post("/api/notify/wechatV2", text="<xml><out_trade_no>NOT-EXIST</out_trade_no></xml>",
                    headers={"Content-Type": "text/xml"})
    check("微信 V2 未知单号返回 FAIL", text == "FAIL", text)
    text = raw_post("/api/notify/wechatV2", text="not-xml", headers={"Content-Type": "text/xml"})
    check("微信 V2 报文异常返回 FAIL", text == "FAIL", text)
    text = raw_post("/api/notify/wechatV3", text="{}", headers={"Content-Type": "application/json"})
    check("微信 V3 未接入返回失败 JSON", '"code": "FAIL"' in text, text)

    order2, order_no2 = make_order()
    _, out2 = make_trade(order2, order_no2, method="wechat")
    xml = (f"<xml><out_trade_no><![CDATA[{out2}]]></out_trade_no>"
           f"<transaction_id><![CDATA[WX{out2}]]></transaction_id>"
           f"<return_code><![CDATA[SUCCESS]]></return_code></xml>")
    text = raw_post("/api/notify/wechatV2", text=xml, headers={"Content-Type": "text/xml"})
    check("微信 V2 回调返回 SUCCESS", "SUCCESS" in text, text)
    row = db(f"SELECT pay_status, pay_method FROM yoshop_order WHERE order_id={order2}")[0]
    check("微信 V2 回调置订单已付款", row[0] == 20 and row[1] == "wechat", row)
    trade = db(f"SELECT trade_state, trade_no FROM yoshop_payment_trade WHERE out_trade_no='{out2}'")[0]
    check("交易记录置为已支付", trade[0] == 20 and trade[1] == "WX" + out2, trade)
    text = raw_post("/api/notify/wechatV2", text=xml, headers={"Content-Type": "text/xml"})
    check("重复回调幂等返回 SUCCESS", "SUCCESS" in text, text)

    order3, order_no3 = make_order()
    _, out3 = make_trade(order3, order_no3, method="alipay")
    text = raw_post("/api/notify/alipay", data={"out_trade_no": out3, "trade_no": "ALI" + out3})
    check("支付宝回调返回 success", text == "success", text)
    row = db(f"SELECT pay_status, pay_method FROM yoshop_order WHERE order_id={order3}")[0]
    check("支付宝回调置订单已付款", row[0] == 20 and row[1] == "alipay", row)
    text = raw_post("/api/notify/alipay", data={"out_trade_no": "NOT-EXIST"})
    check("支付宝未知单号返回 fail", text == "fail", text)

    # 充值订单回调
    recharge_no = "TESTR" + str(os.getpid())
    recharge_id = db_exec(
        "INSERT INTO yoshop_recharge_order (order_no, user_id, recharge_type, plan_id, pay_price, gift_money, "
        "actual_money, pay_method, pay_status, pay_time, trade_id, platform, store_id, create_time, update_time) "
        f"VALUES ('{recharge_no}', {USER_ID}, 10, 0, 100.00, 10.00, 110.00, '', 10, 0, 0, "
        f"'MP-WEIXIN', 10001, UNIX_TIMESTAMP(), UNIX_TIMESTAMP())")
    created_recharge_orders.append(recharge_id)
    _, out4 = make_trade(recharge_id, recharge_no, method="alipay", order_type=100)
    balance_before = float(db(f"SELECT balance FROM yoshop_user WHERE user_id={USER_ID}")[0][0])
    text = raw_post("/api/notify/alipay", data={"out_trade_no": out4, "trade_no": "ALI" + out4})
    check("充值订单回调返回 success", text == "success", text)
    row = db(f"SELECT pay_status FROM yoshop_recharge_order WHERE order_id={recharge_id}")[0]
    check("充值订单置为已支付", row[0] == 20, row)
    balance = float(db(f"SELECT balance FROM yoshop_user WHERE user_id={USER_ID}")[0][0])
    check("充值金额到账", abs(balance - balance_before - 110.0) < 0.001, (balance, balance_before))
    logs = db(f"SELECT COUNT(*) FROM yoshop_user_balance_log WHERE user_id={USER_ID} "
              f"AND `describe`='用户充值：{recharge_no}'")
    check("写充值明细", logs[0][0] == 1, logs)

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


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