shop-java/tools/test_market_writes.py

# -*- coding: utf-8 -*-
"""营销与会员写接口端到端测试（优惠券 / 充值套餐 / 会员等级 / 会员充值 / 改等级 / 删除会员 / 领取记录）。

自带造数与清理：测试记录以 TESTJ 命名，跑完物理删除；用户余额、积分、等级恢复原值。
用法：python test_market_writes.py
"""
import json
import re
from pathlib import Path

import pymysql
import requests

JAVA = "http://127.0.0.1:18080"
TOKEN = "testsession123"
STORE_ID = 10001
ROOT = Path(r"C:\Users\Administrator\WorkBuddy\2026-10-17-12-27-10\java-rewrite")
YML = ROOT / "yoshop-java" / "src" / "main" / "resources" / "application.yml"

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

PASS = FAIL = 0
FAILS = []


def db():
    pwd = re.findall(r"password: *(.+)", YML.read_text(encoding="utf-8"))[0].strip().strip('"')
    return pymysql.connect(host="192.168.1.177", port=3306, user="root", password=pwd,
                           database="shop", autocommit=True)


CONN = db()


def query(sql, args=None):
    cur = CONN.cursor()
    cur.execute(sql, args or ())
    cols = [d[0] for d in cur.description]
    return [dict(zip(cols, row)) for row in cur.fetchall()]


def execute(sql, args=None):
    cur = CONN.cursor()
    cur.execute(sql, args or ())
    return cur.rowcount


def get(path, params=None):
    r = SESSION.get(JAVA + path, headers={"Access-Token": TOKEN}, params=params, timeout=60)
    return r.json()


def post(path, body):
    r = SESSION.post(JAVA + path, headers={"Access-Token": TOKEN}, json=body, timeout=60)
    return r.json()


def check(name, ok, detail=""):
    global PASS, FAIL
    if ok:
        PASS += 1
        print(f"  [ok] {name}")
    else:
        FAIL += 1
        FAILS.append(name)
        print(f"  [FAIL] {name} {detail}")


def eq(name, actual, expect):
    check(name, actual == expect, f"期望 {expect!r} 实际 {actual!r}")


def main():
    # ---------------------------------------------------------------- 基线
    users = query("select user_id, balance, points, grade_id from yoshop_user where user_id=10001")
    if not users:
        print("缺少测试会员 10001，终止")
        return
    base = users[0]
    base_balance = base["balance"]
    max_balance_log = (query("select coalesce(max(log_id),0) m from yoshop_user_balance_log")[0]["m"])
    max_points_log = (query("select coalesce(max(log_id),0) m from yoshop_user_points_log")[0]["m"])
    max_grade_log = (query("select coalesce(max(log_id),0) m from yoshop_user_grade_log")[0]["m"])
    print(f"基线：余额 {base_balance} 积分 {base['points']} 等级 {base['grade_id']}")

    coupon_ids = []
    plan_ids = []
    grade_ids = []
    test_user = 10999

    try:
        # ------------------------------------------------------------ 优惠券
        print("\n[优惠券] 新增（满减 · 领取后生效 · 全部商品）")
        res = post("/store/market.coupon/add", {"form": {
            "name": "TESTJ券A", "coupon_type": 10, "reduce_price": 10, "min_price": 100,
            "expire_type": 10, "expire_day": 7, "apply_range": 10, "total_num": -1,
            "status": 1, "describe": "自动化用例", "sort": 100}})
        eq("add 返回成功", res.get("status"), 200)
        eq("add 文案", res.get("message"), "添加成功")
        rows = query("select * from yoshop_coupon where name='TESTJ券A'")
        check("库内已写入", len(rows) == 1)
        a = rows[0]
        coupon_ids.append(a["coupon_id"])
        eq("start_time 清零", a["start_time"], 0)
        eq("end_time 清零", a["end_time"], 0)
        eq("expire_day 保留", a["expire_day"], 7)
        eq("apply_range_config 空数组", a["apply_range_config"], "[]")
        eq("store_id", a["store_id"], STORE_ID)
        eq("reduce_price", str(a["reduce_price"]), "10.00")
        check("create_time 已写", a["create_time"] > 0)

        detail = get("/store/market.coupon/detail", {"couponId": a["coupon_id"]})["data"]["detail"]
        eq("detail 折扣为 0", detail["discount"], 0)
        eq("detail 开始时间为 null", detail["start_time"], None)
        eq("detail 适用范围空数组", detail["apply_range_config"], [])
        eq("detail state 正常", detail["state"], {"text": "正常", "value": 1})

        print("\n[优惠券] 新增（折扣 · 固定时间 · 指定商品）")
        res = post("/store/market.coupon/add", {"form": {
            "name": "TESTJ券B", "coupon_type": 20, "discount": 9.9, "min_price": 50,
            "expire_type": 20, "betweenTime": ["2026-10-01", "2026-10-31"],
            "apply_range": 20, "apply_range_config": {"goodsIds": ["10004"]},
            "total_num": 5, "status": 1, "describe": "", "sort": 90}})
        eq("add 返回成功", res.get("status"), 200)
        rows = query("select * from yoshop_coupon where name='TESTJ券B'")
        check("库内已写入", len(rows) == 1)
        b = rows[0]
        coupon_ids.append(b["coupon_id"])
        eq("discount 乘 10 入库", b["discount"], 99)
        eq("expire_day 置 1", b["expire_day"], 1)
        eq("apply_range_config 存对象", json.loads(b["apply_range_config"]), {"goodsIds": ["10004"]})
        check("end_time 补足当天 23:59:59", (b["end_time"] - b["start_time"]) % 86400 == 86399)
        detail = get("/store/market.coupon/detail", {"couponId": b["coupon_id"]})["data"]["detail"]
        eq("detail 折扣率 9.9", detail["discount"], 9.9)
        eq("detail 开始时间 Y/m/d", detail["start_time"], "2026/10/01")
        eq("detail 结束时间 Y/m/d", detail["end_time"], "2026/10/31")

        print("\n[优惠券] 固定时间缺少区间")
        res = post("/store/market.coupon/add", {"form": {
            "name": "TESTJ券C", "coupon_type": 10, "reduce_price": 5, "min_price": 10,
            "expire_type": 20, "apply_range": 10, "total_num": -1, "status": 1, "sort": 100}})
        eq("返回失败", res.get("status"), 500)
        eq("提示有效期", res.get("message"), "请选择有效期范围")
        eq("未写入脏数据", len(query("select coupon_id from yoshop_coupon where name='TESTJ券C'")), 0)

        print("\n[优惠券] 编辑")
        res = post("/store/market.coupon/edit", {"couponId": a["coupon_id"], "form": {
            "name": "TESTJ券A改", "coupon_type": 10, "reduce_price": 20, "min_price": 200,
            "expire_type": 10, "expire_day": 3, "apply_range": 20,
            "apply_range_config": {"goodsIds": ["10004", "10005"]},
            "total_num": 100, "status": 1, "describe": "改过", "sort": 88}})
        eq("edit 返回成功", res.get("status"), 200)
        eq("edit 文案", res.get("message"), "更新成功")
        a2 = query("select * from yoshop_coupon where coupon_id=%s", (a["coupon_id"],))[0]
        eq("名称已改", a2["name"], "TESTJ券A改")
        eq("减免金额已改", str(a2["reduce_price"]), "20.00")
        eq("适用范围已改", a2["apply_range"], 20)
        eq("适用范围配置已改", json.loads(a2["apply_range_config"]), {"goodsIds": ["10004", "10005"]})
        eq("排序已改", a2["sort"], 88)

        print("\n[优惠券] 编辑为全部商品 → 配置清空")
        post("/store/market.coupon/edit", {"couponId": a["coupon_id"], "form": {
            "name": "TESTJ券A改", "coupon_type": 10, "reduce_price": 20, "min_price": 200,
            "expire_type": 10, "expire_day": 3, "apply_range": 10,
            "apply_range_config": {"goodsIds": ["10004"]},
            "total_num": 100, "status": 1, "describe": "改过", "sort": 88}})
        a3 = query("select apply_range_config from yoshop_coupon where coupon_id=%s", (a["coupon_id"],))[0]
        eq("配置被清空", a3["apply_range_config"], "[]")

        print("\n[优惠券] 编辑不存在的券")
        res = post("/store/market.coupon/edit", {"couponId": 999999, "form": {"name": "x"}})
        eq("返回失败", res.get("status"), 500)

        print("\n[优惠券] 删除")
        res = post("/store/market.coupon/delete", {"couponId": a["coupon_id"]})
        eq("delete 返回成功", res.get("status"), 200)
        eq("软删除标记", query("select is_delete from yoshop_coupon where coupon_id=%s",
                            (a["coupon_id"],))[0]["is_delete"], 1)
        listed = get("/store/market.coupon/list", {"page": 1, "search": "TESTJ券A"})["data"]["list"]["data"]
        eq("列表不再出现", len(listed), 0)

        print("\n[优惠券] 删除不存在的券")
        eq("返回失败", post("/store/market.coupon/delete", {"couponId": 999999}).get("status"), 500)

        # ---------------------------------------------------------- 充值套餐
        print("\n[充值套餐] 新增 / 编辑 / 删除")
        res = post("/store/market.recharge.plan/add", {"form": {
            "plan_name": "TESTJ套餐A", "money": 100, "gift_money": 10, "sort": 100}})
        eq("add 返回成功", res.get("status"), 200)
        rows = query("select * from yoshop_recharge_plan where plan_name='TESTJ套餐A'")
        check("库内已写入", len(rows) == 1)
        plan = rows[0]
        plan_ids.append(plan["plan_id"])
        eq("充值金额", str(plan["money"]), "100.00")
        eq("赠送金额", str(plan["gift_money"]), "10.00")
        eq("store_id", plan["store_id"], STORE_ID)

        res = post("/store/market.recharge.plan/edit", {"planId": plan["plan_id"], "form": {
            "plan_name": "TESTJ套餐A改", "money": 200, "gift_money": 30, "sort": 66}})
        eq("edit 返回成功", res.get("status"), 200)
        plan2 = query("select * from yoshop_recharge_plan where plan_id=%s", (plan["plan_id"],))[0]
        eq("名称已改", plan2["plan_name"], "TESTJ套餐A改")
        eq("金额已改", str(plan2["money"]), "200.00")
        eq("排序已改", plan2["sort"], 66)

        res = post("/store/market.recharge.plan/delete", {"planId": plan["plan_id"]})
        eq("delete 返回成功", res.get("status"), 200)
        eq("软删除标记", query("select is_delete from yoshop_recharge_plan where plan_id=%s",
                            (plan["plan_id"],))[0]["is_delete"], 1)
        eq("删除不存在的套餐", post("/store/market.recharge.plan/delete", {"planId": 999999}).get("status"), 500)

        # ---------------------------------------------------------- 会员等级
        print("\n[会员等级] 新增 / 权重重复 / 编辑 / 删除")
        res = post("/store/user.grade/add", {"form": {
            "name": "TESTJ等级", "weight": 9,
            "upgrade": {"expend_money": 50000}, "equity": {"discount": "9.5"}, "status": 1}})
        eq("add 返回成功", res.get("status"), 200)
        rows = query("select * from yoshop_user_grade where name='TESTJ等级'")
        check("库内已写入", len(rows) == 1)
        grade = rows[0]
        grade_ids.append(grade["grade_id"])
        eq("权重", grade["weight"], 9)
        eq("升级条件 JSON", json.loads(grade["upgrade"]), {"expend_money": 50000})
        eq("等级权益 JSON", json.loads(grade["equity"]), {"discount": "9.5"})
        eq("store_id", grade["store_id"], STORE_ID)

        res = post("/store/user.grade/add", {"form": {"name": "TESTJ等级2", "weight": 9,
                                                      "upgrade": {}, "equity": {}, "status": 1}})
        eq("权重重复被拦", res.get("status"), 500)
        eq("权重重复提示", res.get("message"), "等级权重已存在")

        res = post("/store/user.grade/edit", {"gradeId": grade["grade_id"], "form": {
            "name": "TESTJ等级改", "weight": 8,
            "upgrade": {"expend_money": 60000}, "equity": {"discount": "9.4"}, "status": 1}})
        eq("edit 返回成功", res.get("status"), 200)
        g2 = query("select * from yoshop_user_grade where grade_id=%s", (grade["grade_id"],))[0]
        eq("名称已改", g2["name"], "TESTJ等级改")
        eq("权重已改", g2["weight"], 8)

        res = post("/store/user.grade/edit", {"gradeId": grade["grade_id"], "form": {
            "name": "TESTJ等级改", "weight": 1,
            "upgrade": {"expend_money": 60000}, "equity": {"discount": "9.4"}, "status": 1}})
        eq("编辑撞已有权重被拦", res.get("status"), 500)

        # 有会员占用时不允许删除：先造一个挂在该等级下的会员
        execute("delete from yoshop_user where user_id=%s", (test_user,))
        execute("""insert into yoshop_user
             (user_id,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 (%s,'19900000000','TESTJ会员',0,0,'','','',0,0,0,0,0,%s,'H5',0,0,%s,0,0)""",
                (test_user, grade["grade_id"], STORE_ID))
        res = post("/store/user.grade/delete", {"gradeId": grade["grade_id"]})
        eq("等级下有会员时禁止删除", res.get("status"), 500)
        eq("提示文案", res.get("message"), "该会员等级下存在用户，不允许删除")
        eq("未被删除", query("select is_delete from yoshop_user_grade where grade_id=%s",
                          (grade["grade_id"],))[0]["is_delete"], 0)

        execute("update yoshop_user set grade_id=0 where user_id=%s", (test_user,))
        res = post("/store/user.grade/delete", {"gradeId": grade["grade_id"]})
        eq("无会员时可删除", res.get("status"), 200)
        eq("软删除标记", query("select is_delete from yoshop_user_grade where grade_id=%s",
                            (grade["grade_id"],))[0]["is_delete"], 1)

        # ------------------------------------------------------------ 会员充值
        print("\n[会员充值] 余额 · 增加")
        before = query("select balance from yoshop_user where user_id=10001")[0]["balance"]
        res = post("/store/user/recharge", {"userId": 10001, "target": "balance",
                                            "form": {"balance": {"mode": "inc", "money": 50, "remark": "TESTJ充值"}}})
        eq("返回成功", res.get("status"), 200)
        eq("文案", res.get("message"), "操作成功")
        after = query("select balance from yoshop_user where user_id=10001")[0]["balance"]
        eq("余额增加 50", str(after - before), "50.00")
        logs = query("""select * from yoshop_user_balance_log where user_id=10001
                        and log_id>%s order by log_id desc limit 1""", (max_balance_log,))
        eq("写入余额流水", len(logs), 1)
        eq("流水场景=管理员操作", logs[0]["scene"], 30)
        eq("流水金额", str(logs[0]["money"]), "50.00")
        check("流水描述含管理员", logs[0]["describe"].startswith("后台管理员 [")
              and logs[0]["describe"].endswith("] 操作"), logs[0]["describe"])
        eq("流水备注", logs[0]["remark"], "TESTJ充值")

        print("\n[会员充值] 余额 · 减少")
        before = query("select balance from yoshop_user where user_id=10001")[0]["balance"]
        res = post("/store/user/recharge", {"userId": 10001, "target": "balance",
                                            "form": {"balance": {"mode": "dec", "money": 20, "remark": ""}}})
        eq("返回成功", res.get("status"), 200)
        after = query("select balance from yoshop_user where user_id=10001")[0]["balance"]
        eq("余额减少 20", str(before - after), "20.00")
        logs = query("""select money from yoshop_user_balance_log where user_id=10001
                        and log_id>%s order by log_id desc limit 1""", (max_balance_log,))
        eq("流水记录负数", str(logs[0]["money"]), "-20.00")

        print("\n[会员充值] 余额 · 最终金额")
        res = post("/store/user/recharge", {"userId": 10001, "target": "balance",
                                            "form": {"balance": {"mode": "final", "money": 888, "remark": ""}}})
        eq("返回成功", res.get("status"), 200)
        eq("余额变为 888", str(query("select balance from yoshop_user where user_id=10001")[0]["balance"]), "888.00")

        print("\n[会员充值] 金额非法")
        res = post("/store/user/recharge", {"userId": 10001, "target": "balance",
                                            "form": {"balance": {"mode": "inc", "money": "", "remark": ""}}})
        eq("空金额被拦", res.get("status"), 500)
        eq("提示文案", res.get("message"), "请输入正确的金额")
        res = post("/store/user/recharge", {"userId": 10001, "target": "unknown", "form": {}})
        eq("非法 target 被拦", res.get("status"), 500)

        print("\n[会员充值] 积分 · 增加 / 最终值")
        res = post("/store/user/recharge", {"userId": 10001, "target": "points",
                                            "form": {"points": {"mode": "inc", "value": 30, "remark": "TESTJ积分"}}})
        eq("返回成功", res.get("status"), 200)
        eq("积分 +30", query("select points from yoshop_user where user_id=10001")[0]["points"], 30)
        logs = query("""select * from yoshop_user_points_log where user_id=10001
                        and log_id>%s order by log_id desc limit 1""", (max_points_log,))
        eq("写入积分流水", len(logs), 1)
        eq("流水积分值", logs[0]["value"], 30)
        eq("流水描述", logs[0]["describe"].startswith("后台管理员 ["), True)
        res = post("/store/user/recharge", {"userId": 10001, "target": "points",
                                            "form": {"points": {"mode": "final", "value": 0, "remark": ""}}})
        eq("返回成功", res.get("status"), 200)
        eq("积分归零", query("select points from yoshop_user where user_id=10001")[0]["points"], 0)
        res = post("/store/user/recharge", {"userId": 10001, "target": "points",
                                            "form": {"points": {"mode": "inc", "value": "", "remark": ""}}})
        eq("空积分被拦", res.get("status"), 500)
        eq("提示文案", res.get("message"), "请输入正确的积分数量")

        # ------------------------------------------------------------ 改等级
        print("\n[会员等级变更]")
        res = post("/store/user/grade", {"userId": 10001, "form": {"grade_id": 10002, "remark": "TESTJ调级"}})
        eq("返回成功", res.get("status"), 200)
        eq("等级已改", query("select grade_id from yoshop_user where user_id=10001")[0]["grade_id"], 10002)
        glog = query("""select * from yoshop_user_grade_log where user_id=10001
                        and log_id>%s order by log_id desc limit 1""", (max_grade_log,))
        eq("写入等级变更记录", len(glog), 1)
        eq("变更前等级", glog[0]["old_grade_id"], 0)
        eq("变更后等级", glog[0]["new_grade_id"], 10002)
        eq("变更类型=管理员", glog[0]["change_type"], 10)
        eq("变更备注", glog[0]["remark"], "TESTJ调级")
        post("/store/user/grade", {"userId": 10001, "form": {"grade_id": 0, "remark": ""}})
        eq("已恢复原等级", query("select grade_id from yoshop_user where user_id=10001")[0]["grade_id"], 0)

        # ------------------------------------------------------------ 删除会员
        print("\n[删除会员]")
        execute("""insert into yoshop_user_oauth (user_id,oauth_type,oauth_id,unionid,store_id,is_delete,
                   create_time,update_time) values (%s,'TESTJ','oauth-test','',%s,0,0,0)""",
                (test_user, STORE_ID))
        res = post("/store/user/delete", {"userId": test_user})
        eq("返回成功", res.get("status"), 200)
        eq("文案", res.get("message"), "删除成功")
        eq("会员标记删除", query("select is_delete from yoshop_user where user_id=%s",
                            (test_user,))[0]["is_delete"], 1)
        eq("三方绑定同步标记", query("select is_delete from yoshop_user_oauth where user_id=%s",
                              (test_user,))[0]["is_delete"], 1)
        eq("重复删除被拦", post("/store/user/delete", {"userId": test_user}).get("status"), 500)

        # -------------------------------------------------------- 领取记录列表
        print("\n[领取记录] 造一条领取记录后读取")
        execute("""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 (10002,'TESTJ领取券',20,0.00,99,50.00,20,1,1790784000,1793462399,10,'[]','',
                     0,0,10001,%s,1792000000,1792000000)""", (STORE_ID,))
        uc = query("select user_coupon_id from yoshop_user_coupon where name='TESTJ领取券'")[0]["user_coupon_id"]
        res = get("/store/market.coupon/receive", {"page": 1, "couponName": "TESTJ领取券"})
        data = res["data"]["list"]
        eq("总数", data["total"], 1)
        row = data["data"][0]
        eq("折扣率 9.9", row["discount"], 9.9)
        eq("开始时间 Y/m/d", row["start_time"], "2026/10/01")
        eq("state 正常", row["state"], {"text": "正常", "value": 1})
        check("带会员信息", isinstance(row.get("user"), dict) and row["user"]["user_id"] == 10001)
        check("会员含头像键", "avatar" in row["user"] and "avatar_url" in row["user"])
        check("会员无 grade 键", "grade" not in row["user"])

        res = get("/store/market.coupon/receive", {"page": 1, "nickName": "185****1340"})
        check("按昵称可检索", res["data"]["list"]["total"] >= 1)
        res = get("/store/market.coupon/receive", {"page": 1, "nickName": "不存在的昵称XYZ"})
        eq("昵称无匹配为空", res["data"]["list"]["total"], 0)

    finally:
        # ------------------------------------------------------------ 清理
        # 只删本次新建的记录（本地存在与线上对照的种子数据，不能按名称前缀删）
        print("\n[清理]")
        if coupon_ids:
            execute("delete from yoshop_coupon where coupon_id in (%s)"
                    % ",".join(str(int(i)) for i in coupon_ids))
        if plan_ids:
            execute("delete from yoshop_recharge_plan where plan_id in (%s)"
                    % ",".join(str(int(i)) for i in plan_ids))
        if grade_ids:
            execute("delete from yoshop_user_grade where grade_id in (%s)"
                    % ",".join(str(int(i)) for i in grade_ids))
        execute("delete from yoshop_user_coupon where name='TESTJ领取券'")
        execute("delete from yoshop_user_balance_log where log_id>%s", (max_balance_log,))
        execute("delete from yoshop_user_points_log where log_id>%s", (max_points_log,))
        execute("delete from yoshop_user_grade_log where log_id>%s", (max_grade_log,))
        execute("delete from yoshop_user_oauth where user_id=%s", (test_user,))
        execute("delete from yoshop_user where user_id=%s", (test_user,))
        execute("update yoshop_user set balance=%s, points=%s, grade_id=%s where user_id=10001",
                (base_balance, base["points"], base["grade_id"]))
        restored = query("select balance, points, grade_id from yoshop_user where user_id=10001")[0]
        eq("余额已恢复", str(restored["balance"]), str(base_balance))
        eq("积分已恢复", restored["points"], base["points"])
        eq("等级已恢复", restored["grade_id"], base["grade_id"])

    print(f"\n通过 {PASS} / 失败 {FAIL}")
    if FAILS:
        print("失败项：" + "、".join(FAILS))


if __name__ == "__main__":
    main()