shop-java/tools/test_setting_writes.py

# -*- coding: utf-8 -*-
"""设置类写接口端到端测试（配送模板 / 物流公司 / 打印机 / 支付设置 / 支付模板）。

自带造数与清理：所有测试记录以 TESTJ 命名，跑完物理删除，payment 表恢复为空。
用法：python test_setting_writes.py
"""
import json
import re
import sys
from decimal import Decimal
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 post_multipart(path, data, files=None):
    r = SESSION.post(JAVA + path, headers={"Access-Token": TOKEN}, data=data, files=files, 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(f"{name} :: {detail}")
        print(f"  [FAIL] {name} :: {detail}")


def cleanup():
    execute("delete from yoshop_delivery where name like 'TESTJ%%'")
    execute("delete from yoshop_delivery_rule where delivery_id not in (select delivery_id from yoshop_delivery)")
    execute("delete from yoshop_express where express_name like 'TESTJ%%'")
    execute("delete from yoshop_printer where printer_name like 'TESTJ%%'")
    execute("delete from yoshop_payment")
    execute("delete from yoshop_payment_template where name like 'TESTJ%%'")
    for f in (ROOT / "yoshop-java" / "runtime" / "uploads" / "payment").rglob("*.pem"):
        if f.name.startswith("apiclientCert-"):
            f.unlink(missing_ok=True)


# ---------------------------------------------------------------- 配送模板

def test_delivery():
    print("\n== 配送模板 ==")
    rules = [
        {"region": [2, 783], "region_text": [{"name": "北京市", "citys": []}, {"name": "上海市", "citys": []}],
         "first": 2, "first_fee": "4.50", "additional": 1, "additional_fee": "2.00"},
        {"region": [648], "region_text": [{"name": "辽宁省", "citys": []}],
         "first": 1, "first_fee": 0, "additional": 0, "additional_fee": 0},
    ]
    res = post("/store/setting.delivery/add",
               {"form": {"name": "TESTJ配送模板", "method": 20, "sort": 99, "rules": rules}})
    check("add 返回添加成功", res.get("status") == 200 and res.get("message") == "添加成功", res)

    rows = query("select * from yoshop_delivery where name='TESTJ配送模板'")
    check("add 落库 1 条", len(rows) == 1, rows)
    if not rows:
        return
    rec = rows[0]
    did = rec["delivery_id"]
    check("add method/sort/store_id", rec["method"] == 20 and rec["sort"] == 99
          and rec["store_id"] == STORE_ID and rec["is_delete"] == 0, rec)
    check("add 写入 create_time/update_time", rec["create_time"] > 0 and rec["update_time"] > 0, rec)

    rule_rows = query("select * from yoshop_delivery_rule where delivery_id=%s order by rule_id", (did,))
    check("add 写入 2 条区域", len(rule_rows) == 2, len(rule_rows))
    if len(rule_rows) == 2:
        check("add region 存 JSON 数组", rule_rows[0]["region"] == "[2,783]", rule_rows[0]["region"])
        check("add region_text 存 JSON", rule_rows[0]["region_text"] == '[{"name":"北京市","citys":[]},{"name":"上海市","citys":[]}]',
              rule_rows[0]["region_text"])
        check("add first/first_fee", float(rule_rows[0]["first"]) == 2.0
              and Decimal(rule_rows[0]["first_fee"]) == Decimal("4.50"), rule_rows[0])

    detail = get("/store/setting.delivery/detail", {"deliveryId": did})
    check("detail 返回 rule 数组", isinstance(detail["data"]["detail"].get("rule"), list)
          and len(detail["data"]["detail"]["rule"]) == 2, detail.get("data"))
    check("detail rule.first 为数字", detail["data"]["detail"]["rule"][0]["first"] == 2, detail)
    check("detail rule.first_fee 为字符串", detail["data"]["detail"]["rule"][0]["first_fee"] == "4.50", detail)

    res = post("/store/setting.delivery/add", {"form": {"name": "TESTJ空规则", "method": 10, "sort": 1}})
    check("add 无 rules 被拦截", res.get("status") == 500 and "请选择可配送区域" in (res.get("message") or ""), res)

    new_rule = [{"region": [2], "region_text": [{"name": "北京市", "citys": []}],
                 "first": 3, "first_fee": "9.90", "additional": 2, "additional_fee": "1.00"}]
    res = post("/store/setting.delivery/edit",
               {"deliveryId": did,
                "form": {"name": "TESTJ配送模板-改", "method": 10, "sort": 98, "rules": new_rule}})
    check("edit 返回更新成功", res.get("status") == 200 and res.get("message") == "更新成功", res)
    rec = query("select * from yoshop_delivery where delivery_id=%s", (did,))[0]
    check("edit 主表已更新", rec["name"] == "TESTJ配送模板-改" and rec["method"] == 10 and rec["sort"] == 98, rec)
    rule_rows = query("select * from yoshop_delivery_rule where delivery_id=%s", (did,))
    check("edit 区域被整体替换为 1 条", len(rule_rows) == 1, len(rule_rows))
    if rule_rows:
        check("edit 新区域内容正确", rule_rows[0]["region"] == "[2]"
              and Decimal(rule_rows[0]["first_fee"]) == Decimal("9.90"), rule_rows[0])

    res = post("/store/setting.delivery/edit",
               {"deliveryId": did, "form": {"name": "TESTJ配送模板-改", "method": 10, "rules": []}})
    check("edit 无 rules 被拦截", res.get("status") == 500 and "请选择可配送区域" in (res.get("message") or ""), res)

    res = post("/store/setting.delivery/delete", {"deliveryId": did})
    check("delete 返回删除成功", res.get("status") == 200 and res.get("message") == "删除成功", res)
    rec = query("select is_delete from yoshop_delivery where delivery_id=%s", (did,))[0]
    check("delete 置 is_delete=1", rec["is_delete"] == 1, rec)
    detail = get("/store/setting.delivery/detail", {"deliveryId": did})
    check("delete 后 detail 为 null", detail["data"]["detail"] is None, detail.get("data"))

    res = post("/store/setting.delivery/delete", {"deliveryId": 10001})
    check("delete 被商品使用的模板被拦截",
          res.get("status") == 500 and "该模板被6个商品使用，不允许删除" in (res.get("message") or ""), res)


# ---------------------------------------------------------------- 物流公司

def test_express():
    print("\n== 物流公司 ==")
    res = post("/store/setting.express/add",
               {"form": {"express_name": "TESTJ快递", "kuaidi100_code": "testj", "kdniao_code": "TESTJ", "sort": 88}})
    check("add 返回添加成功", res.get("status") == 200 and res.get("message") == "添加成功", res)
    rows = query("select * from yoshop_express where express_name='TESTJ快递'")
    check("add 落库 1 条", len(rows) == 1, rows)
    if not rows:
        return
    eid = rows[0]["express_id"]
    rec = rows[0]
    check("add 字段正确", rec["kuaidi100_code"] == "testj" and rec["kdniao_code"] == "TESTJ"
          and rec["sort"] == 88 and rec["store_id"] == STORE_ID and rec["is_delete"] == 0, rec)

    res = post("/store/setting.express/edit",
               {"expressId": eid,
                "form": {"express_name": "TESTJ快递-改", "kuaidi100_code": "testj2", "kdniao_code": "TESTJ2", "sort": 66}})
    check("edit 返回更新成功", res.get("status") == 200 and res.get("message") == "更新成功", res)
    rec = query("select * from yoshop_express where express_id=%s", (eid,))[0]
    check("edit 字段已更新", rec["express_name"] == "TESTJ快递-改" and rec["kuaidi100_code"] == "testj2"
          and rec["kdniao_code"] == "TESTJ2" and rec["sort"] == 66, rec)

    res = post("/store/setting.express/delete", {"expressId": eid})
    check("delete 返回删除成功", res.get("status") == 200 and res.get("message") == "删除成功", res)
    rec = query("select is_delete from yoshop_express where express_id=%s", (eid,))[0]
    check("delete 置 is_delete=1", rec["is_delete"] == 1, rec)


# ------------------------------------------------------------------ 打印机

def test_printer():
    print("\n== 打印机 ==")
    config = {"USER": "testuser", "UKEY": "testukey", "SN": "123456789"}
    res = post("/store/setting.printer/add",
               {"form": {"printer_name": "TESTJ打印机", "printer_type": "FEI_E_YUN",
                         "printer_config": config, "print_times": 2, "sort": 77}})
    check("add 返回添加成功", res.get("status") == 200 and res.get("message") == "添加成功", res)
    rows = query("select * from yoshop_printer where printer_name='TESTJ打印机'")
    check("add 落库 1 条", len(rows) == 1, rows)
    if not rows:
        return
    pid = rows[0]["printer_id"]
    rec = rows[0]
    check("add printer_config 存 JSON", json.loads(rec["printer_config"]) == config, rec["printer_config"])
    check("add 其他字段", rec["printer_type"] == "FEI_E_YUN" and rec["print_times"] == 2
          and rec["sort"] == 77 and rec["store_id"] == STORE_ID, rec)

    all_data = get("/store/setting.printer/all")["data"]["list"]
    hit = [p for p in all_data if p["printer_id"] == pid]
    check("all 输出 printer_config 为对象", len(hit) == 1 and hit[0]["printer_config"] == config, hit)

    res = post("/store/setting.printer/edit",
               {"printerId": pid,
                "form": {"printer_name": "TESTJ打印机-改", "printer_type": "PRINT_CENTER",
                         "printer_config": {"deviceNo": "D1", "key": "K1"}, "print_times": 3, "sort": 55}})
    check("edit 返回更新成功", res.get("status") == 200 and res.get("message") == "更新成功", res)
    rec = query("select * from yoshop_printer where printer_id=%s", (pid,))[0]
    check("edit 字段已更新", rec["printer_name"] == "TESTJ打印机-改" and rec["printer_type"] == "PRINT_CENTER"
          and rec["print_times"] == 3 and rec["sort"] == 55, rec)
    check("edit 配置已替换", json.loads(rec["printer_config"]) == {"deviceNo": "D1", "key": "K1"}, rec["printer_config"])

    res = post("/store/setting.printer/delete", {"printerId": pid})
    check("delete 返回删除成功", res.get("status") == 200 and res.get("message") == "删除成功", res)
    rec = query("select is_delete from yoshop_printer where printer_id=%s", (pid,))[0]
    check("delete 置 is_delete=1", rec["is_delete"] == 1, rec)


# ---------------------------------------------------------------- 支付设置

def test_payment():
    print("\n== 支付设置 ==")
    before = get("/store/setting.payment/options")["data"]["options"]
    check("options 默认 4 个客户端", list(before.keys()) == ["MP-WEIXIN", "H5", "WXOFFICIAL", "APP"], list(before.keys()))
    check("options 方法数 2/3/1/3", [len(before[k]["methods"]) for k in before] == [2, 3, 1, 3],
          {k: len(v["methods"]) for k, v in before.items()})
    check("options 默认仅余额启用",
          [m["method"] for m in before["MP-WEIXIN"]["methods"] if m["is_enable"]] == ["balance"], before["MP-WEIXIN"])

    form = json.loads(json.dumps(before))
    form["MP-WEIXIN"]["methods"][0]["is_enable"] = True
    form["MP-WEIXIN"]["methods"][0]["template_id"] = 0
    form["H5"]["methods"][1]["is_enable"] = True
    form["H5"]["methods"][1]["template_id"] = 0
    res = post("/store/setting.payment/update", {"form": form})
    check("update 返回更新成功", res.get("status") == 200 and res.get("message") == "更新成功", res)

    rows = query("select * from yoshop_payment where store_id=%s", (STORE_ID,))
    check("update 落库 9 条", len(rows) == 9, len(rows))
    wechat = [r for r in rows if r["client"] == "MP-WEIXIN" and r["method"] == "wechat"]
    check("update wechat 已启用", len(wechat) == 1 and wechat[0]["is_enable"] == 1, wechat)
    check("update others 存空数组", all(r["others"] == "[]" for r in rows), [r["others"] for r in rows][:3])
    check("update is_must_template 落库", wechat[0]["is_must_template"] == 1 if wechat else False, wechat)

    after = get("/store/setting.payment/options")["data"]["options"]
    check("options 读回微信已启用", after["MP-WEIXIN"]["methods"][0]["is_enable"] is True, after["MP-WEIXIN"]["methods"][0])
    check("options 读回支付宝已启用", after["H5"]["methods"][1]["is_enable"] is True, after["H5"]["methods"][1])
    check("options 读回仍保留 key 字段", after["MP-WEIXIN"]["methods"][0].get("key") == 1, after["MP-WEIXIN"]["methods"][0])

    # 还原为空（PHP 侧初始即无记录）
    execute("delete from yoshop_payment")
    restored = get("/store/setting.payment/options")["data"]["options"]
    check("清空后恢复默认", restored["MP-WEIXIN"]["methods"][0]["is_enable"] is False, restored["MP-WEIXIN"]["methods"][0])


# ---------------------------------------------------------------- 支付模板

def test_payment_template():
    print("\n== 支付模板 ==")
    cfg = {"wechat": {
        "mchType": "normal", "version": "v2",
        "normal": {"appId": "wxtest123", "mchId": "1900000109", "apiKey": "k" * 16,
                   "signatureMethod": "platformCert", "publicKeyId": "", "publicKey": "",
                   "apiclientCert": "", "apiclientKey": ""},
        "provider": {"spAppId": "", "spMchId": "", "spApiKey": "", "subAppId": "", "subMchId": "",
                     "spSignatureMethod": "platformCert", "spPublicKeyId": "", "spPublicKey": "",
                     "spApiclientCert": "", "spApiclientKey": ""},
    }}
    cert = b"-----BEGIN CERTIFICATE-----\nTESTJ\n-----END CERTIFICATE-----\n"
    res = post_multipart("/store/setting.payment.template/add",
                         data={"name": "TESTJ支付模板", "sort": 90, "method": "wechat",
                               "remarks": "auto", "config": json.dumps(cfg, ensure_ascii=False)},
                         files={"apiclientCert": ("apiclientCert.pem", cert, "application/x-pem-file")})
    check("add 返回添加成功", res.get("status") == 200 and res.get("message") == "添加成功", res)

    rows = query("select * from yoshop_payment_template where name='TESTJ支付模板'")
    check("add 落库 1 条", len(rows) == 1, rows)
    if not rows:
        return
    tid = rows[0]["template_id"]
    rec = rows[0]
    saved = json.loads(rec["config"])
    check("add config 仅保留当前支付方式", list(saved.keys()) == ["wechat"], list(saved.keys()))
    cert_name = saved["wechat"]["normal"]["apiclientCert"]
    check("add 证书文件名规则", bool(re.fullmatch(r"apiclientCert-[0-9a-f]{8}\.pem", cert_name or "")), cert_name)
    check("add 保留原配置值", saved["wechat"]["normal"]["appId"] == "wxtest123"
          and saved["wechat"]["provider"]["spAppId"] == "", saved["wechat"]["normal"])
    cert_path = ROOT / "yoshop-java" / "runtime" / "uploads" / "payment" / "wechat" / str(STORE_ID) / cert_name
    check("add 证书文件已落盘", cert_path.exists() and cert_path.read_bytes() == cert, str(cert_path))
    check("add wechatpay_serial 为空(v2)", rec["wechatpay_serial"] == "", rec["wechatpay_serial"])
    check("add sort/remarks/store_id", rec["sort"] == 90 and rec["remarks"] == "auto" and rec["store_id"] == STORE_ID, rec)

    detail = get("/store/setting.payment.template/detail", {"templateId": tid})["data"]["detail"]
    check("detail 返回 config 合并默认", list(detail["config"].keys()) == ["wechat", "alipay"], list(detail["config"].keys()))
    check("detail 保留证书文件名", detail["config"]["wechat"]["normal"]["apiclientCert"] == cert_name, detail["config"]["wechat"]["normal"])

    # 前端编辑页会用 detail 回填 config（含已保存的证书文件名），这里保持一致
    res = post_multipart("/store/setting.payment.template/edit",
                         data={"templateId": tid, "name": "TESTJ支付模板-改", "sort": 91,
                               "method": "wechat", "remarks": "auto2",
                               "config": json.dumps(detail["config"], ensure_ascii=False)})
    check("edit 返回更新成功", res.get("status") == 200 and res.get("message") == "更新成功", res)
    rec = query("select * from yoshop_payment_template where template_id=%s", (tid,))[0]
    check("edit 基础字段已更新", rec["name"] == "TESTJ支付模板-改" and rec["sort"] == 91 and rec["remarks"] == "auto2", rec)
    saved = json.loads(rec["config"])
    check("edit 未传文件时保留原证书名", saved["wechat"]["normal"]["apiclientCert"] == cert_name, saved["wechat"]["normal"])

    execute("insert into yoshop_payment (client,method,is_must_template,template_id,is_enable,is_default,others,store_id,create_time,update_time)"
            " values ('H5','wechat',1,%s,1,0,'[]',%s,0,0)", (tid, STORE_ID))
    res = post_multipart("/store/setting.payment.template/delete", data={"templateId": tid})
    check("delete 被引用时被拦截",
          res.get("status") == 500 and "当前支付模板已被引用" in (res.get("message") or ""), res)
    execute("delete from yoshop_payment where template_id=%s", (tid,))
    res = post_multipart("/store/setting.payment.template/delete", data={"templateId": tid})
    check("delete 解除引用后成功", res.get("status") == 200 and res.get("message") == "删除成功", res)
    rec = query("select is_delete from yoshop_payment_template where template_id=%s", (tid,))[0]
    check("delete 置 is_delete=1", rec["is_delete"] == 1, rec)
    cert_path.unlink(missing_ok=True)


def main():
    print("清理历史测试数据…")
    cleanup()
    test_delivery()
    test_express()
    test_printer()
    test_payment()
    test_payment_template()
    print("\n清理测试数据…")
    cleanup()
    print(f"\n通过 {PASS} / 失败 {FAIL}")
    if FAILS:
        print("失败明细：")
        for f in FAILS:
            print("  -", f)
        sys.exit(1)


if __name__ == "__main__":
    main()