shop-java/tools/test_api_batch3.py

# -*- coding: utf-8 -*-
"""客户端 api 第 3 批接口冒烟：用户中心 / 我的优惠券 / 余额积分明细 / 售后 / 充值 / 上传。

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

说明：
- 全部是需登录接口，脚本会往 Redis 写测试 token（yoshop:api:token:apitest-batch3）与短信验证码；
- 会临时造订单、售后单、充值订单、支付方式、优惠券领取记录、上传文件、临时用户，
  finally 中全部清理并把用户资料（手机号 / 昵称 / 头像）恢复原值。
"""
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-batch3"
USER_ID = 10001
GOODS_ID = 10009
SKU_ID = "10009"
# 领取测试用券（满 100 减 10、领取后 7 天有效）
COUPON_ID = 10001
PLAN_ID = 10001
EXPRESS_ID = 10007

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 = []
created_refunds = []
created_recharge_orders = []
created_files = []
temp_user_id = None
user_backup = {}
coupon_receive_backup = 0

# 1x1 PNG
PNG = bytes.fromhex(
    "89504e470d0a1a0a0000000d49484452000000010000000108060000001f15c489"
    "0000000d4944415408d76368f8ffff3f0005fe02fea735e1f90000000049454e44ae426082"
)


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, files=None, data=None):
    headers = {"platform": "MP-WEIXIN"}
    if token:
        headers["Access-Token"] = token
    if files is not None:
        resp = S.post(BASE + path, params=params or {}, files=files, data=data or {},
                      headers=headers, timeout=60)
    elif 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 upload(path, token=TOKEN, check_login=1, name="test.png"):
    files = {"file": (name, PNG, "image/png")}
    return call("POST", path, token=token, files=files, data={"checkLogin": str(check_login)})


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_id, order_goods_id)"""
    order_no = "TESTB3" + 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, 80.00, 0.00, '', '', 20, 0, "
        f"10, 0.00, 20, 20, 30, 0, 1, 1, 10, 0, '[]', 'H5', {USER_ID}, 0, 10001, "
        "UNIX_TIMESTAMP(), UNIX_TIMESTAMP())")
    created_orders.append(order_id)
    goods_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, 20, 1, 1, {order_id}, {USER_ID}, 0, "
        "10001, UNIX_TIMESTAMP())")
    return order_id, goods_id


def make_payment(client, method, is_default=1):
    return db_exec(
        "INSERT INTO yoshop_payment (client, method, is_must_template, template_id, is_enable, is_default, "
        f"others, store_id, create_time, update_time) VALUES ('{client}', '{method}', 0, 0, 1, {is_default}, "
        "'[]', 10001, UNIX_TIMESTAMP(), UNIX_TIMESTAMP())")


def cleanup():
    try:
        for order_id in created_orders:
            db_exec(f"DELETE FROM yoshop_order_goods WHERE order_id={order_id}")
            db_exec(f"DELETE FROM yoshop_order WHERE order_id={order_id}")
        for refund_id in created_refunds:
            db_exec(f"DELETE FROM yoshop_order_refund_image WHERE order_refund_id={refund_id}")
            db_exec(f"DELETE FROM yoshop_order_refund_address WHERE order_refund_id={refund_id}")
            db_exec(f"DELETE FROM yoshop_order_refund WHERE order_refund_id={refund_id}")
        for order_id in created_recharge_orders:
            db_exec(f"DELETE FROM yoshop_recharge_order_plan WHERE order_id={order_id}")
            db_exec(f"DELETE FROM yoshop_recharge_order WHERE order_id={order_id}")
        db_exec("DELETE FROM yoshop_payment WHERE store_id=10001 AND method IN ('wechat','balance') "
                "AND client='MP-WEIXIN'")
        db_exec(f"DELETE FROM yoshop_user_coupon WHERE user_id={USER_ID} AND coupon_id={COUPON_ID}")
        db_exec(f"UPDATE yoshop_coupon SET receive_num={coupon_receive_backup} WHERE coupon_id={COUPON_ID}")
        if temp_user_id:
            db_exec(f"DELETE FROM yoshop_user WHERE user_id={temp_user_id}")
        if user_backup:
            db_exec("UPDATE yoshop_user SET mobile='{}', nick_name='{}', avatar_id={} WHERE user_id={}".format(
                user_backup["mobile"], user_backup["nick_name"], user_backup["avatar_id"], USER_ID))
        for file_id in created_files:
            db_exec(f"DELETE FROM yoshop_upload_file WHERE file_id={file_id}")
        print("  CLEAN 测试数据已清理")
    except Exception as e:  # noqa: BLE001
        print("  CLEAN 失败:", e)


def main():
    global temp_user_id, coupon_receive_backup
    print("== 客户端 api 第 3 批（用户中心 / 优惠券 / 明细 / 售后 / 充值 / 上传） ==")
    redis_set("yoshop:api:token:" + TOKEN, USER_ID)
    row = db(f"SELECT mobile, nick_name, avatar_id FROM yoshop_user WHERE user_id={USER_ID}")[0]
    user_backup.update(mobile=row[0], nick_name=row[1], avatar_id=row[2])
    coupon_receive_backup = db(f"SELECT receive_num FROM yoshop_coupon WHERE coupon_id={COUPON_ID}")[0][0]

    # ---------------------------------------------------------------- 未登录
    print("-- 未登录校验")
    for path, label in [("/api/user/info", "user/info"), ("/api/user/assets", "user/assets"),
                        ("/api/myCoupon/list", "myCoupon/list"), ("/api/refund/list", "refund/list")]:
        r = get(path, token=None)
        check(f"未登录 {label} 返回 401", r.get("status") == 401, r.get("message"))

    # ---------------------------------------------------------------- 用户信息
    print("-- 用户中心")
    r = get("/api/user/info")
    info = field(r.get("data"), "userInfo") or {}
    check("user/info status=200", r.get("status") == 200, r.get("message"))
    check("user/info 手机号打码", info.get("mobile") == "185****1340", info.get("mobile"))
    check("user/info 含 avatarUrl / grade / balance 键",
          all(k in info for k in ("avatarUrl", "grade", "balance", "points")), list(info.keys()))
    check("user/info 隐藏 isDelete/storeId/createTime",
          not any(k in info for k in ("isDelete", "storeId", "createTime", "updateTime", "password")),
          [k for k in ("isDelete", "storeId", "createTime") if k in info])
    check("user/info 无 snake_case", not snake_keys(info), snake_keys(info)[:3])

    r = get("/api/user/assets")
    assets = field(r.get("data"), "assets") or {}
    check("user/assets 三个字段齐全", set(assets.keys()) == {"balance", "points", "coupon"}, assets.keys())
    check("user/assets balance 两位小数字符串",
          isinstance(assets.get("balance"), str) and re.fullmatch(r"-?\d+\.\d{2}", assets.get("balance", "")),
          assets.get("balance"))
    check("user/assets 无 snake_case", not snake_keys(assets), snake_keys(assets)[:3])

    # ---------------------------------------------------------------- 明细
    print("-- 余额 / 积分明细")
    r = get("/api/balance.log/list")
    logs = field(r.get("data"), "data") or []
    check("balance.log/list status=200", r.get("status") == 200, r.get("message"))
    check("balance.log/list 有数据且按时间倒序",
          len(logs) > 0 and (len(logs) < 2 or logs[0]["createTime"] >= logs[1]["createTime"]), len(logs))
    check("balance.log/list 隐藏 storeId 与 user",
          all("storeId" not in x and "user" not in x for x in logs[:3]), logs[:1])
    check("balance.log/list perPage=15", field(r.get("data"), "perPage") == 15, field(r.get("data"), "perPage"))

    r = get("/api/points.log/list")
    plogs = field(r.get("data"), "data") or []
    check("points.log/list status=200", r.get("status") == 200, r.get("message"))
    check("points.log/list 字段为 logId/value/describe",
          all({"logId", "value", "describe"} <= set(x.keys()) for x in plogs[:3]), plogs[:1])
    check("points.log/list 无 snake_case", not snake_keys(plogs), snake_keys(plogs)[:3])

    # ---------------------------------------------------------------- 修改资料
    print("-- 修改资料")
    r = post("/api/user/personal", {"form": {"nickName": "接口自测昵称"}})
    check("user/personal 修改成功", r.get("status") == 200 and r.get("message") == "恭喜您，信息修改成功", r)
    nick = db(f"SELECT nick_name, avatar_id FROM yoshop_user WHERE user_id={USER_ID}")[0]
    check("user/personal 昵称落库", nick[0] == "接口自测昵称", nick)
    check("user/personal 头像沿用原值", nick[1] == user_backup["avatar_id"], nick[1])

    # ---------------------------------------------------------------- 绑定手机号
    print("-- 绑定手机号")
    new_mobile = "19900002222"
    r = post("/api/user/bindMobile", {"form": {"mobile": new_mobile, "smsCode": "123456"}})
    check("bindMobile 无验证码时报错", r.get("status") == 500 and "短信验证码不存在" in (r.get("message") or ""), r)

    redis_set("yoshop:api:sms:" + new_mobile, json.dumps({"code": "654321", "times": 5}))
    r = post("/api/user/bindMobile", {"form": {"mobile": new_mobile, "smsCode": "000000"}})
    check("bindMobile 验证码错误时报错", r.get("status") == 500 and "验证码不正确" in (r.get("message") or ""), r)
    left = json.loads(db_redis("yoshop:api:sms:" + new_mobile) or "null")
    check("bindMobile 错误一次后剩余次数减一", left and left.get("times") == 4, left)

    r = post("/api/user/bindMobile", {"form": {"mobile": new_mobile, "smsCode": "654321"}})
    check("bindMobile 绑定成功", r.get("status") == 200 and r.get("message") == "恭喜您，手机号绑定成功", r)
    check("bindMobile 手机号落库", db(f"SELECT mobile FROM yoshop_user WHERE user_id={USER_ID}")[0][0] == new_mobile)
    check("bindMobile 验证码用后即焚", db_redis("yoshop:api:sms:" + new_mobile) is None)

    # 手机号已被占用
    temp_user_id = db_exec(
        "INSERT INTO yoshop_user (mobile, nick_name, avatar_id, gender, country, province, city, address_id, "
        "balance, points, pay_money, expend_money, grade_id, platform, last_login_time, is_delete, store_id, "
        "create_time, update_time) VALUES ('19900003333', '临时用户', 0, 0, '', '', '', 0, 0.00, 0, 0.00, "
        "0.00, 0, 'H5', 0, 0, 10001, UNIX_TIMESTAMP(), UNIX_TIMESTAMP())")
    redis_set("yoshop:api:sms:19900003333", json.dumps({"code": "111111", "times": 5}))
    r = post("/api/user/bindMobile", {"form": {"mobile": "19900003333", "smsCode": "111111"}})
    check("bindMobile 手机号已占用时报错",
          r.get("status") == 500 and "已绑定其他账户" in (r.get("message") or ""), r)

    # ---------------------------------------------------------------- 我的优惠券
    print("-- 我的优惠券")
    r = post("/api/myCoupon/receive", {"couponId": 999999})
    check("myCoupon/receive 券不存在时报错",
          r.get("status") == 500 and "优惠券不存在" in (r.get("message") or ""), r)

    r = post("/api/myCoupon/receive", {"couponId": COUPON_ID})
    check("myCoupon/receive 领取成功", r.get("status") == 200 and r.get("message") == "领取成功", r)
    recv = db(f"SELECT user_coupon_id, name, is_use, is_expire, start_time, end_time "
              f"FROM yoshop_user_coupon WHERE user_id={USER_ID} AND coupon_id={COUPON_ID}")
    check("领取记录落库且未使用未过期", len(recv) == 1 and recv[0][2] == 0 and recv[0][3] == 0, recv)
    check("领取后有效期为 7 天", len(recv) == 1 and recv[0][5] - recv[0][4] == 7 * 86400, recv)
    check("券已领取数量 +1",
          db(f"SELECT receive_num FROM yoshop_coupon WHERE coupon_id={COUPON_ID}")[0][0]
          == coupon_receive_backup + 1)

    r = post("/api/myCoupon/receive", {"couponId": COUPON_ID})
    check("myCoupon/receive 重复领取被拒",
          r.get("status") == 500 and "已领取该优惠券" in (r.get("message") or ""), r)

    r = get("/api/myCoupon/list")
    data = field(r.get("data"), "data") or []
    check("myCoupon/list 含刚领取的券", any(x.get("couponId") == COUPON_ID for x in data), len(data))
    check("myCoupon/list 隐藏 storeId/createTime/updateTime",
          all(not {"storeId", "createTime", "updateTime"} & set(x.keys()) for x in data), data[:1])
    check("myCoupon/list state 为正常", data and field(data[0].get("state"), "text") == "正常", data[:1])
    check("myCoupon/list discount 为 0-10 小数", all(isinstance(x.get("discount"), (int, float)) for x in data),
          [x.get("discount") for x in data[:2]])

    r = get("/api/myCoupon/list", {"dataType": "isUsable"})
    usable = field(r.get("data"), "data") or []
    check("myCoupon/list isUsable 含该券", any(x.get("couponId") == COUPON_ID for x in usable), len(usable))

    r = get("/api/myCoupon/list", {"dataType": "isUsable", "amount": 50})
    filtered = field(r.get("data"), "data") or []
    check("myCoupon/list amount=50 过滤掉满 100 的券",
          not any(x.get("couponId") == COUPON_ID for x in filtered), len(filtered))

    r = get("/api/user/assets")
    check("user/assets 可用券数量 +1", field(field(r.get("data"), "assets"), "coupon") == 1,
          field(field(r.get("data"), "assets"), "coupon"))

    # ---------------------------------------------------------------- 售后
    print("-- 售后服务")
    order_id, order_goods_id = make_order()
    r = get("/api/refund/goods", {"orderGoodsId": order_goods_id})
    goods = field(r.get("data"), "goods") or {}
    check("refund/goods status=200", r.get("status") == 200, r.get("message"))
    check("refund/goods 含商品图与金额", "goodsImage" in goods and goods.get("totalPayPrice") == "80.00",
          {k: goods.get(k) for k in ("goodsImage", "totalPayPrice")})
    check("refund/goods 无 snake_case", not snake_keys(goods), snake_keys(goods)[:3])

    r = get("/api/refund/goods", {"orderGoodsId": 99999999})
    check("refund/goods 不存在时报错", r.get("status") == 500 and "未找到订单商品" in (r.get("message") or ""), r)

    r = post("/api/refund/apply", {"orderGoodsId": order_goods_id, "form": {"content": "不想要了"}})
    check("refund/apply 缺少类型时报错", r.get("status") == 500 and "请选择售后类型" in (r.get("message") or ""), r)

    r = post("/api/refund/apply", {"orderGoodsId": order_goods_id,
                                   "form": {"type": 10, "content": "接口自测：商品有瑕疵"}})
    check("refund/apply 提交成功", r.get("status") == 200 and r.get("message") == "提交成功", r)
    refund_rows = db(f"SELECT order_refund_id, type, apply_desc, audit_status, status, is_user_send "
                     f"FROM yoshop_order_refund WHERE order_goods_id={order_goods_id}")
    check("售后单落库为待审核", len(refund_rows) == 1 and refund_rows[0][3] == 0 and refund_rows[0][4] == 0,
          refund_rows)
    refund_id = refund_rows[0][0]
    created_refunds.append(refund_id)

    r = get("/api/refund/goods", {"orderGoodsId": order_goods_id})
    check("已申请售后不可重复申请", r.get("status") == 500 and "已申请售后" in (r.get("message") or ""), r)

    r = get("/api/refund/detail", {"orderRefundId": refund_id})
    detail = field(r.get("data"), "detail") or {}
    check("refund/detail stateText=等待审核中", detail.get("stateText") == "等待审核中", detail.get("stateText"))
    check("refund/detail 含 orderGoods 且图片为空数组",
          isinstance(detail.get("orderGoods"), dict) and detail.get("images") == [], detail.get("images"))
    check("refund/detail 隐藏 storeId/updateTime",
          not {"storeId", "updateTime"} & set(detail.keys()), [k for k in ("storeId", "updateTime") if k in detail])
    check("refund/detail 无 snake_case", not snake_keys(detail), snake_keys(detail)[:3])

    r = get("/api/refund/list", {"state": 0})
    refund_list = field(r.get("data"), "data") or []
    check("refund/list 含该售后单", any(x.get("orderRefundId") == refund_id for x in refund_list), len(refund_list))
    check("refund/list 每项均带 stateText", all(x.get("stateText") for x in refund_list), refund_list[:1])
    check("refund/list 隐藏 storeId/updateTime/images",
          all(not {"storeId", "updateTime", "images"} & set(x.keys()) for x in refund_list), refund_list[:1])

    r = get("/api/refund/list", {"state": 20})
    check("refund/list state=20 不含进行中的单",
          not any(x.get("orderRefundId") == refund_id for x in (field(r.get("data"), "data") or [])), r.get("data"))

    # 未审核不允许发货
    r = post("/api/refund/delivery", {"orderRefundId": refund_id,
                                      "form": {"expressId": EXPRESS_ID, "expressNo": "SF0001"}})
    check("refund/delivery 未审核时报错", r.get("status") == 500 and "不合法" in (r.get("message") or ""), r)

    db_exec(f"UPDATE yoshop_order_refund SET audit_status=10 WHERE order_refund_id={refund_id}")
    r = post("/api/refund/delivery", {"orderRefundId": refund_id, "form": {"expressNo": "SF0001"}})
    check("refund/delivery 未选物流公司时报错",
          r.get("status") == 500 and "请选择物流公司" in (r.get("message") or ""), r)
    r = post("/api/refund/delivery", {"orderRefundId": refund_id, "form": {"expressId": EXPRESS_ID}})
    check("refund/delivery 未填单号时报错",
          r.get("status") == 500 and "请填写物流单号" in (r.get("message") or ""), r)

    r = post("/api/refund/delivery", {"orderRefundId": refund_id,
                                      "form": {"expressId": EXPRESS_ID, "expressNo": "SF0001"}})
    check("refund/delivery 发货成功", r.get("status") == 200 and r.get("message") == "操作成功", r)
    sent = db(f"SELECT is_user_send, express_id, express_no, send_time FROM yoshop_order_refund "
              f"WHERE order_refund_id={refund_id}")[0]
    check("售后单发货信息落库",
          sent[0] == 1 and str(sent[1]) == str(EXPRESS_ID) and sent[2] == "SF0001" and sent[3] > 0, sent)

    r = get("/api/refund/detail", {"orderRefundId": refund_id})
    detail = field(r.get("data"), "detail") or {}
    check("refund/detail 发货后 stateText 变化",
          detail.get("stateText") == "已发货，待平台确认", detail.get("stateText"))
    check("refund/detail 带物流公司名称",
          field(detail.get("express"), "expressName") == "中通快递", detail.get("express"))

    r = get("/api/refund/detail", {"orderRefundId": 99999999})
    check("refund/detail 不存在时报错", r.get("status") == 500 and "未找到该售后单" in (r.get("message") or ""), r)

    # ---------------------------------------------------------------- 充值
    print("-- 余额充值")
    r = get("/api/recharge/center", {"client": "MP-WEIXIN"})
    check("recharge/center 无支付方式时报错",
          r.get("status") == 500 and "没有可用的支付方式" in (r.get("message") or ""), r)

    make_payment("MP-WEIXIN", "wechat", 1)
    make_payment("MP-WEIXIN", "balance", 0)
    r = get("/api/recharge/center", {"client": "MP-WEIXIN"})
    center = r.get("data") or {}
    setting = field(center, "setting") or {}
    check("recharge/center status=200", r.get("status") == 200, r.get("message"))
    check("recharge/center 设置默认值（isCustom=1 / lowestMoney=10 / isMatchPlan=1）",
          setting.get("isCustom") == 1 and float(setting.get("lowestMoney")) == 10.0
          and setting.get("isMatchPlan") == 1, setting)
    check("recharge/center personal 为当前用户",
          field(field(center, "personal"), "userId") == USER_ID, field(center, "personal"))
    check("recharge/center planList 含套餐", any(p.get("planId") == PLAN_ID for p in
                                                (field(center, "planList") or [])), field(center, "planList"))
    check("recharge/center paymentMethods 排除余额支付",
          [m.get("method") for m in (field(center, "paymentMethods") or [])] == ["wechat"],
          field(center, "paymentMethods"))
    check("recharge/center 无 snake_case", not snake_keys(center), snake_keys(center)[:3])

    r = post("/api/recharge/submit", {"form": {"method": "balance", "client": "MP-WEIXIN"}})
    check("recharge/submit 缺套餐与金额时报错",
          r.get("status") == 500 and "请选择充值套餐或输入充值金额" in (r.get("message") or ""), r)

    r = post("/api/recharge/submit", {"form": {"method": "balance", "client": "MP-WEIXIN",
                                               "customMoney": "5"}})
    check("recharge/submit 低于最低金额时报错",
          r.get("status") == 500 and "最低充值金额不能低于10元" in (r.get("message") or ""), r)

    r = post("/api/recharge/submit", {"form": {"method": "balance", "client": "MP-WEIXIN",
                                               "customMoney": "100"}})
    check("recharge/submit 自定义金额下单成功",
          r.get("status") == 200 and r.get("message") == "下单成功" and field(r.get("data"), "payment") == [], r)
    row = db(f"SELECT order_id, order_no, recharge_type, plan_id, pay_price, gift_money, actual_money, "
             f"pay_status, platform FROM yoshop_recharge_order WHERE user_id={USER_ID} ORDER BY order_id DESC "
             "LIMIT 1")[0]
    created_recharge_orders.append(row[0])
    check("充值订单号以 RC 开头且 18 位以上", str(row[1]).startswith("RC") and len(str(row[1])) >= 18, row[1])
    check("自定义充值匹配套餐（赠 10）",
          row[2] == 10 and row[3] == PLAN_ID and float(row[5]) == 10.0 and float(row[6]) == 110.0, row)
    check("充值订单平台取请求头", row[8] == "MP-WEIXIN", row[8])
    snapshot = db(f"SELECT plan_id, plan_name, money, gift_money FROM yoshop_recharge_order_plan "
                  f"WHERE order_id={row[0]}")
    check("套餐快照落库", len(snapshot) == 1 and snapshot[0][0] == PLAN_ID, snapshot)

    r = post("/api/recharge/submit", {"form": {"method": "balance", "client": "MP-WEIXIN",
                                               "planId": PLAN_ID}})
    check("recharge/submit 指定套餐下单成功", r.get("status") == 200, r)
    row2 = db(f"SELECT order_id, recharge_type, plan_id, pay_price, gift_money, actual_money "
              f"FROM yoshop_recharge_order WHERE user_id={USER_ID} ORDER BY order_id DESC LIMIT 1")[0]
    created_recharge_orders.append(row2[0])
    check("套餐充值金额与赠送正确",
          row2[1] == 20 and float(row2[3]) == 100.0 and float(row2[4]) == 10.0 and float(row2[5]) == 110.0, row2)

    r = post("/api/recharge/submit", {"form": {"method": "wechat", "client": "MP-WEIXIN",
                                               "planId": PLAN_ID}})
    check("recharge/submit 微信支付未接入时明确报错",
          r.get("status") == 500 and "第三方支付" in (r.get("message") or ""), r)

    r = get("/api/recharge/tradeQuery", {"outTradeNo": "RC123456", "method": "wechat",
                                         "client": "MP-WEIXIN"})
    check("recharge/tradeQuery 未支付返回 isPay=false",
          r.get("status") == 200 and field(r.get("data"), "isPay") is False, r)
    check("recharge/tradeQuery 提示未支付",
          "订单未支付" in (r.get("message") or ""), r.get("message"))

    db_exec(f"UPDATE yoshop_recharge_order SET pay_status=20, pay_time=UNIX_TIMESTAMP(), pay_method='balance' "
            f"WHERE order_id={row[0]}")
    r = get("/api/recharge.order/list")
    orders = field(r.get("data"), "data") or []
    check("recharge.order/list 只列已支付订单",
          [o.get("orderId") for o in orders] == [row[0]], [o.get("orderId") for o in orders])
    check("recharge.order/list 隐藏 tradeId/createTime/storeId",
          all(not {"tradeId", "createTime", "storeId"} & set(o.keys()) for o in orders), orders[:1])
    check("recharge.order/list 金额为两位小数字符串",
          orders and orders[0].get("actualMoney") == "110.00", orders[:1])
    check("recharge.order/list 无 snake_case", not snake_keys(orders), snake_keys(orders)[:3])

    # ---------------------------------------------------------------- 上传
    print("-- 文件上传")
    r = upload("/api/upload/image", token=None, check_login=1)
    check("upload/image 需要登录时未登录返回 401", r.get("status") == 401, r)

    r = upload("/api/upload/image", check_login=1)
    file_info = field(r.get("data"), "fileInfo") or {}
    check("upload/image 上传成功", r.get("status") == 200 and r.get("message") == "文件上传成功", r)
    check("upload/image 返回 fileId 与 previewUrl",
          isinstance(file_info.get("fileId"), int) and str(file_info.get("previewUrl", "")).startswith("http"),
          file_info)
    check("upload/image 隐藏 filePath/fileName/storeId",
          not {"filePath", "fileName", "storeId", "uploaderId"} & set(file_info.keys()), list(file_info.keys()))
    if file_info.get("fileId"):
        created_files.append(file_info["fileId"])
        uploader = db(f"SELECT channel, uploader_id, file_type, is_delete FROM yoshop_upload_file "
                      f"WHERE file_id={file_info['fileId']}")[0]
        check("上传记录 channel=20 且记录上传者", uploader[0] == 20 and uploader[1] == USER_ID, uploader)

    r = upload("/api/upload/image", token=None, check_login=0)
    anon_info = field(r.get("data"), "fileInfo") or {}
    check("upload/image checkLogin=0 允许匿名上传", r.get("status") == 200, r)
    if anon_info.get("fileId"):
        created_files.append(anon_info["fileId"])
        check("匿名上传 uploader_id=0",
              db(f"SELECT uploader_id FROM yoshop_upload_file WHERE file_id={anon_info['fileId']}")[0][0] == 0)

    # 用上传的文件改头像
    if file_info.get("fileId"):
        r = post("/api/user/personal", {"form": {"avatarId": file_info["fileId"]}})
        check("user/personal 设置头像成功", r.get("status") == 200, r)
        r = get("/api/user/info")
        avatar_url = field(field(r.get("data"), "userInfo"), "avatarUrl")
        check("user/info 头像地址随之上线", isinstance(avatar_url, str) and avatar_url.startswith("http"),
              avatar_url)
        check("user/info avatar 对象非空", field(field(r.get("data"), "userInfo"), "avatar") is not None)

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


def db_redis(key):
    sock = socket.create_connection(("192.168.1.177", 6379), timeout=5)
    try:
        cmd(sock, "AUTH", REDIS_PWD)
        result = cmd(sock, "GET", key)
    finally:
        sock.close()
    return result if isinstance(result, str) else None


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