shop-java/tools/test_goods_writes.py

# -*- coding: utf-8 -*-
"""商品写接口端到端自测：Java 写入 → PHP 读取 → 逐字段比对。

因为本地 Java 与线上 PHP 共用同一个库（192.168.1.177/shop），
所以「Java 新增/编辑的商品，PHP 端 detail 能读、且与 Java detail 完全一致」
是比单侧断言更强的正确性证明。

用法：python test_goods_writes.py
"""
import re
import json
import requests
import pymysql
from pathlib import Path

from compare_live import diff
from _root import ROOT

JAVA_BASE = "http://127.0.0.1:18080"
PHP_BASE = "https://shop.qinys.cn/index.php?s="
TOKEN = "testsession123"
STORE_ID = 10001

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

YML = Path(str(ROOT / "yoshop-java/src/main/resources/application.yml"))

passed = failed = 0


def check(name, cond, extra=""):
    global passed, failed
    if cond:
        passed += 1
        print(f"  [ok] {name}")
    else:
        failed += 1
        print(f"  [FAIL] {name} {extra}")


def db():
    text = YML.read_text(encoding="utf-8")
    pwd = re.findall(r"password:\s*(.+)", text)[0].strip().strip('"')
    # autocommit=True：Java 侧写入要立刻可见，避免 REPEATABLE READ 快照挡住新数据
    return pymysql.connect(host="192.168.1.177", port=3306, user="root",
                           password=pwd, database="shop", charset="utf8mb4",
                           autocommit=True, cursorclass=pymysql.cursors.DictCursor)


def php_token():
    src = Path(r"F:\import goods\upload_goods.py").read_text(encoding="utf-8")
    return re.search(r'TOKEN\s*=\s*"([^"]+)"', src).group(1)


def java(path, method="get", params=None, body=None):
    if method == "get":
        r = SESSION.get(JAVA_BASE + path, headers={"Access-Token": TOKEN}, params=params or {}, timeout=60)
    else:
        r = SESSION.post(JAVA_BASE + path, headers={"Access-Token": TOKEN}, json=body, timeout=60)
    return r.json()


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


def single_form(name, category_id, image_id):
    return {
        "goods_type": 10,
        "goods_name": name,
        "goods_no": "TEST-SINGLE",
        "selling_point": "自测卖点",
        "categoryIds": [category_id],
        "imagesIds": [image_id],
        "serviceIds": [],
        "video_id": 0,
        "video_cover_id": 0,
        "delivery_id": 10001,
        "is_ind_delivery_type": 0,
        "delivery_type": [10],
        "status": 10,
        "sort": 99,
        "spec_type": 10,
        "goods_price": 12.34,
        "line_price": 20,
        "stock_num": 50,
        "goods_weight": 0.5,
        "deduct_stock_type": 10,
        "is_restrict": 0,
        "content": "<p>自测商品详情</p>",
        "sales_initial": 3,
        "is_points_gift": 1,
        "is_points_discount": 1,
        "is_alone_points_discount": 0,
        "is_enable_grade": 1,
        "is_alone_grade": 0,
    }


def multi_form(name, category_id, image_id, price1=8.8, price2=18.8):
    return {
        **single_form(name, category_id, image_id),
        "spec_type": 20,
        "goods_price": None,
        "line_price": None,
        "stock_num": None,
        "specData": {
            "specList": [
                {"key": 0, "spec_name": "自测颜色", "valueList": [
                    {"key": 0, "groupKey": 0, "spec_value": "自测红"},
                    {"key": 1, "groupKey": 0, "spec_value": "自测蓝"},
                ]}
            ],
            "skuList": [
                {"image_id": 0, "goods_price": price1, "line_price": 0, "stock_num": 7,
                 "goods_weight": 0, "goods_sku_no": "T-R", "skuKeys": [{"groupKey": 0, "valueKey": 0}]},
                {"image_id": 0, "goods_price": price2, "line_price": 0, "stock_num": 9,
                 "goods_weight": 0, "goods_sku_no": "T-B", "skuKeys": [{"groupKey": 0, "valueKey": 1}]},
            ],
        },
    }


def main():
    conn = db()
    cur = conn.cursor()
    token = php_token()

    # 依赖数据：一个分类 + 一张已有图片
    cur.execute("SELECT category_id FROM yoshop_category ORDER BY category_id LIMIT 1")
    category_id = cur.fetchone()["category_id"]
    cur.execute("SELECT file_id FROM yoshop_upload_file WHERE file_type=10 AND is_delete=0 ORDER BY file_id DESC LIMIT 1")
    image_id = cur.fetchone()["file_id"]
    print(f"依赖：categoryId={category_id} imageId={image_id}")

    # 清理历史残留（上次异常中断留下的同前缀商品）
    cur.execute("SELECT goods_id FROM yoshop_goods WHERE goods_name LIKE '接口自测商品-%'")
    for old in cur.fetchall():
        oid = old["goods_id"]
        for t in ("yoshop_goods_sku", "yoshop_goods_spec_rel", "yoshop_goods_category_rel", "yoshop_goods_image"):
            cur.execute(f"DELETE FROM {t} WHERE goods_id=%s", (oid,))
        cur.execute("DELETE FROM yoshop_goods WHERE goods_id=%s", (oid,))
    cur.execute("DELETE FROM yoshop_spec_value WHERE spec_value IN ('自测红','自测蓝')")
    cur.execute("DELETE FROM yoshop_spec WHERE spec_name='自测颜色'")

    name = f"接口自测商品-{int(__import__('time').time())}"

    # ---------------------------------------------------------------- 1. 新增（单规格）
    print("\n[1] 新增单规格商品")
    res = java("/store/goods/add", "post", body={"form": single_form(name, category_id, image_id)})
    check("add 返回 200/添加成功", res.get("status") == 200 and res.get("message") == "添加成功", res)
    cur.execute("SELECT * FROM yoshop_goods WHERE goods_name=%s ORDER BY goods_id DESC LIMIT 1", (name,))
    row = cur.fetchone()
    check("商品已落库", row is not None)
    goods_id = row["goods_id"]
    check("spec_type=10", row["spec_type"] == 10, row["spec_type"])
    check("goods_price_min/max=12.34", str(row["goods_price_min"]) == "12.34" and str(row["goods_price_max"]) == "12.34",
          (row["goods_price_min"], row["goods_price_max"]))
    check("stock_total=50", row["stock_total"] == 50, row["stock_total"])
    check("delivery_type 存 JSON", row["delivery_type"] == "[10]", row["delivery_type"])
    check("alone_grade_equity=[]", row["alone_grade_equity"] == "[]", row["alone_grade_equity"])
    cur.execute("SELECT * FROM yoshop_goods_sku WHERE goods_id=%s", (goods_id,))
    skus = cur.fetchall()
    check("生成 1 条 sku", len(skus) == 1, len(skus))
    check("sku goods_sku_id='0'", str(skus[0]["goods_sku_id"]) == "0", skus[0]["goods_sku_id"])
    check("sku 价格=12.34", str(skus[0]["goods_price"]) == "12.34", skus[0]["goods_price"])
    cur.execute("SELECT COUNT(*) c FROM yoshop_goods_category_rel WHERE goods_id=%s", (goods_id,))
    check("分类关联 1 条", cur.fetchone()["c"] == 1)
    cur.execute("SELECT COUNT(*) c FROM yoshop_goods_image WHERE goods_id=%s", (goods_id,))
    check("图片关联 1 条", cur.fetchone()["c"] == 1)

    # ---------------------------------------------------------------- 2. 读回一致性
    # 注意：本机 shop 库与线上 PHP 库并非强一致（PHP→本地单向下发），
    # 所以写操作用「Java 读回 == 库内实际值」自洽校验，不跨端比对。
    print("\n[2] Java 读回与库内数据一致（单规格）")
    j = java("/store/goods/detail", "get", {"goodsId": goods_id})
    info = j.get("data", {}).get("goodsInfo", {})
    check("detail 返回成功", j.get("status") == 200, j.get("message"))
    check("detail 名称一致", info.get("goods_name") == name, info.get("goods_name"))
    check("detail stock_total=50", info.get("stock_total") == 50, info.get("stock_total"))
    check("detail categoryIds", info.get("categoryIds") == [category_id], info.get("categoryIds"))
    check("detail skuList 1 条", len(info.get("skuList") or []) == 1)
    check("detail goods_image 非空", bool(info.get("goods_image")))
    check("detail isSpecLocked=false", info.get("isSpecLocked") is False)

    # ---------------------------------------------------------------- 3. 编辑为多规格
    print("\n[3] 编辑为多规格（sku 重建 + 规格入库）")
    res = java("/store/goods/edit", "post", body={"goodsId": goods_id,
                                                  "form": multi_form(name + "-多规格", category_id, image_id)})
    check("edit 返回 200/更新成功", res.get("status") == 200 and res.get("message") == "更新成功", res)
    cur.execute("SELECT * FROM yoshop_goods WHERE goods_id=%s", (goods_id,))
    row = cur.fetchone()
    check("spec_type=20", row["spec_type"] == 20, row["spec_type"])
    check("价格区间 8.80~18.80", (str(row["goods_price_min"]), str(row["goods_price_max"])) == ("8.80", "18.80"),
          (row["goods_price_min"], row["goods_price_max"]))
    check("库存合计 16", row["stock_total"] == 16, row["stock_total"])
    cur.execute("SELECT * FROM yoshop_goods_sku WHERE goods_id=%s ORDER BY id", (goods_id,))
    skus = cur.fetchall()
    check("sku 数量 2", len(skus) == 2, len(skus))
    # 只有 1 个规格组时组合就是单个规格值 ID（PHP implode 行为一致）
    check("goods_sku_id = 规格值 ID", str(skus[0]["goods_sku_id"]).isdigit(), skus[0]["goods_sku_id"])
    props = json.loads(skus[0]["goods_props"] or "[]")
    check("goods_props 结构正确", props and props[0]["group"]["name"] == "自测颜色", skus[0]["goods_props"])
    cur.execute("SELECT COUNT(*) c FROM yoshop_goods_spec_rel WHERE goods_id=%s", (goods_id,))
    check("规格关联 2 条", cur.fetchone()["c"] == 2)
    cur.execute("SELECT COUNT(*) c FROM yoshop_spec WHERE spec_name='自测颜色'")
    check("spec 组已创建", cur.fetchone()["c"] == 1)

    print("\n[4] Java 读回与库内数据一致（多规格）")
    j = java("/store/goods/detail", "get", {"goodsId": goods_id})
    info = j.get("data", {}).get("goodsInfo", {})
    check("detail 价格区间 8.80~18.80",
          (info.get("goods_price_min"), info.get("goods_price_max")) == ("8.80", "18.80"),
          (info.get("goods_price_min"), info.get("goods_price_max")))
    check("detail skuList 2 条", len(info.get("skuList") or []) == 2)
    check("detail specList 1 组", len(info.get("specList") or []) == 1)
    check("specList 结构正确",
          (info.get("specList") or [{}])[0].get("spec_name") == "自测颜色",
          info.get("specList"))
    check("sku 的 goods_props 为数组",
          isinstance((info.get("skuList") or [{}])[0].get("goods_props"), list),
          type((info.get("skuList") or [{}])[0].get("goods_props")).__name__)
    check("sku 的 spec_value_ids 为数组",
          isinstance((info.get("skuList") or [{}])[0].get("spec_value_ids"), list))

    # ---------------------------------------------------------------- 5. 上下架
    print("\n[5] 批量下架 / 上架")
    res = java("/store/goods/state", "post", body={"goodsIds": [goods_id], "state": False})
    check("state 返回 操作成功", res.get("status") == 200 and res.get("message") == "操作成功", res)
    cur.execute("SELECT status FROM yoshop_goods WHERE goods_id=%s", (goods_id,))
    check("下架后 status=20", cur.fetchone()["status"] == 20)
    java("/store/goods/state", "post", body={"goodsIds": [goods_id], "state": True})
    cur.execute("SELECT status FROM yoshop_goods WHERE goods_id=%s", (goods_id,))
    check("上架后 status=10", cur.fetchone()["status"] == 10)

    # ---------------------------------------------------------------- 6. 删除
    print("\n[6] 软删除")
    res = java("/store/goods/delete", "post", body={"goodsIds": [goods_id]})
    check("delete 返回 删除成功", res.get("status") == 200 and res.get("message") == "删除成功", res)
    cur.execute("SELECT is_delete FROM yoshop_goods WHERE goods_id=%s", (goods_id,))
    check("is_delete=1", cur.fetchone()["is_delete"] == 1)
    j = java("/store/goods/detail", "get", {"goodsId": goods_id})
    check("删除后 detail 报不存在", j.get("status") == 500 and "不存在" in str(j.get("message")), j.get("message"))
    jl = java("/store/goods/list", "get", {"page": 1})
    names = [g["goods_name"] for g in jl["data"]["list"]["data"]]
    check("删除后列表不可见", (name + "-多规格") not in names)

    # ---------------------------------------------------------------- 7. 清理
    print("\n[7] 清理测试数据")
    cur.execute("DELETE FROM yoshop_goods_sku WHERE goods_id=%s", (goods_id,))
    cur.execute("DELETE FROM yoshop_goods_spec_rel WHERE goods_id=%s", (goods_id,))
    cur.execute("DELETE FROM yoshop_goods_category_rel WHERE goods_id=%s", (goods_id,))
    cur.execute("DELETE FROM yoshop_goods_image WHERE goods_id=%s", (goods_id,))
    cur.execute("DELETE FROM yoshop_goods WHERE goods_id=%s", (goods_id,))
    # 测试规格组/规格值（仅本次创建的）
    cur.execute("DELETE FROM yoshop_spec_value WHERE spec_value IN ('自测红','自测蓝')")
    cur.execute("DELETE FROM yoshop_spec WHERE spec_name='自测颜色'")
    conn.commit()
    cur.execute("SELECT COUNT(*) c FROM yoshop_goods WHERE goods_id=%s", (goods_id,))
    check("测试商品已清理", cur.fetchone()["c"] == 0)
    conn.close()

    print(f"\n通过 {passed} / 失败 {failed}")


if __name__ == "__main__":
    main()