shop-java/tools/test_final_writes.py

# -*- coding: utf-8 -*-
"""第 8 批写接口双端对照（评论编辑/删除、客户端设置更新、订单导出、小票打印）。

策略：同一初始数据下，先由 PHP 执行写操作并采样「响应 + 库内状态」，
恢复数据后由 Java 执行同一操作并再次采样，最后逐字段对照。

用法：python test_final_writes.py
前置：
  1. F:/shop/shop 的 PHP 8.2 本地站点运行在 8899（与 Java 共用 192.168.1.177/shop 库）；
  2. Java 服务运行在 18080；
  3. 评论接口依赖 tools/seed_comment_golden.py 的种子数据（脚本会自动重置）。
"""
import json
import re
import sys
from pathlib import Path

import openpyxl
import pymysql
import requests

PHP_BASE = "http://127.0.0.1:8899/index.php?s="
JAVA_BASE = "http://127.0.0.1:18080"
TOKEN = "testsession123"
STORE_ID = 10001
YML = Path(r"C:\Users\Administrator\WorkBuddy\2026-10-17-12-27-10\java-rewrite\yoshop-java\src\main\resources\application.yml")
JAVA_EXPORT_DIR = Path(r"C:\Users\Administrator\WorkBuddy\2026-10-17-12-27-10\java-rewrite\yoshop-java\runtime\downloads")
PHP_EXPORT_DIR = Path(r"F:\shop\shop\public\downloads")

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

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():
    pwd = re.findall(r"password:\s*(.+)", 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", charset="utf8mb4", autocommit=True,
                           cursorclass=pymysql.cursors.DictCursor)


def call(base, path, method="post", body=None):
    # PHP 用 ?s=/store/x/y&a=1 追加参数；Java 的路径是 /store/x/y?a=1
    java_path = path if "?" in path else path.replace("&", "?", 1)
    url = base + (java_path if base == JAVA_BASE else path)
    headers = {"Access-Token": TOKEN}
    if method == "post":
        r = SESSION.post(url, json=body, headers=headers, timeout=120)
    else:
        r = SESSION.get(url, headers=headers, timeout=120)
    try:
        return r.json()
    except Exception:
        return {"raw": r.text}


def reset_comments():
    import subprocess
    subprocess.run([sys.executable, str(Path(__file__).with_name("seed_comment_golden.py"))],
                   check=True, capture_output=True)


def comment_state(cur):
    cur.execute("select comment_id,score,content,is_picture,status,sort from yoshop_comment"
                " where comment_id in (10001,10002,10003) order by comment_id")
    comments = cur.fetchall()
    cur.execute("select comment_id,image_id from yoshop_comment_image"
                " where comment_id in (10001,10002,10003) order by comment_id,image_id")
    images = cur.fetchall()
    return {"comments": comments, "images": images}


def compare_pair(label, path, body, state_fn, method="post"):
    """PHP 先跑 → 采样 → 复位 → Java 跑 → 采样 → 对照"""
    reset_comments()
    php_resp = call(PHP_BASE, path, method, body)
    conn = db()
    cur = conn.cursor()
    php_state = state_fn(cur)
    conn.close()

    reset_comments()
    java_resp = call(JAVA_BASE, path, method, body)
    conn = db()
    cur = conn.cursor()
    java_state = state_fn(cur)
    conn.close()

    check(f"{label} status/message 一致",
          php_resp.get("status") == java_resp.get("status")
          and php_resp.get("message") == java_resp.get("message"),
          f"php={php_resp.get('status')}/{php_resp.get('message')!r} "
          f"java={java_resp.get('status')}/{java_resp.get('message')!r}")
    check(f"{label} 库内状态一致", php_state == java_state,
          f"\n    php={php_state}\n    java={java_state}")


# ------------------------------------------------------------------ 用例

def case_comment_edit():
    print("\n[case] 评价编辑：改分/内容/状态/排序 + 替换图片")
    compare_pair("comment.edit", "/store/goods.comment/edit&commentId=10002",
                 {"commentId": 10002,
                  "form": {"score": 30, "content": "改过的内容", "status": 0, "sort": 50,
                           "imageIds": [10003, 10004]}},
                 comment_state)


def case_comment_edit_clear_images():
    print("\n[case] 评价编辑：清空图片（is_picture 应变 0）")
    compare_pair("comment.edit.clear", "/store/goods.comment/edit&commentId=10001",
                 {"commentId": 10001,
                  "form": {"score": 10, "content": "很好", "status": 1, "sort": 0, "imageIds": []}},
                 comment_state)


def case_comment_delete():
    print("\n[case] 评价删除（软删除）")
    def state(cur):
        cur.execute("select comment_id,is_delete from yoshop_comment where comment_id=10003")
        return cur.fetchall()
    compare_pair("comment.delete", "/store/goods.comment/delete&commentId=10003",
                 {"commentId": 10003}, state)


def case_comment_missing():
    print("\n[case] 评价不存在：edit / delete")
    reset_comments()
    php_edit = call(PHP_BASE, "/store/goods.comment/edit&commentId=99999",
                    body={"commentId": 99999, "form": {"content": "x"}})
    java_edit = call(JAVA_BASE, "/store/goods.comment/edit&commentId=99999",
                     body={"commentId": 99999, "form": {"content": "x"}})
    check("comment.edit.missing 响应一致",
          php_edit.get("status") == java_edit.get("status")
          and php_edit.get("message") == java_edit.get("message"),
          f"php={php_edit} java={java_edit}")

    php_del = call(PHP_BASE, "/store/goods.comment/delete&commentId=99999", body={"commentId": 99999})
    java_del = call(JAVA_BASE, "/store/goods.comment/delete&commentId=99999", body={"commentId": 99999})
    check("comment.delete.missing 响应一致",
          php_del.get("status") == java_del.get("status")
          and php_del.get("message") == java_del.get("message"),
          f"php={php_del} java={java_del}")


def case_h5_setting():
    print("\n[case] H5 设置更新（新增 → 再更新）")
    def state(cur):
        cur.execute("select `key`,`describe`,`values`,store_id from yoshop_h5_setting order by `key`")
        return cur.fetchall()

    def clear():
        conn = db()
        conn.cursor().execute("delete from yoshop_h5_setting")
        conn.close()

    clear()
    php1 = call(PHP_BASE, "/store/client.h5.setting/update&key=basic",
                body={"key": "basic", "form": {"enabled": False, "baseUrl": "https://h5.example.com"}})
    conn = db()
    php_state = state(conn.cursor())
    conn.close()

    clear()
    java1 = call(JAVA_BASE, "/store/client.h5.setting/update?key=basic",
                 body={"key": "basic", "form": {"enabled": False, "baseUrl": "https://h5.example.com"}})
    conn = db()
    java_state = state(conn.cursor())
    conn.close()

    check("h5.update 首次响应一致",
          php1.get("status") == java1.get("status") and php1.get("message") == java1.get("message"),
          f"php={php1} java={java1}")
    check("h5.update 落库一致", php_state == java_state, f"\n    php={php_state}\n    java={java_state}")

    # 二次更新（覆盖同一条记录）
    php2 = call(PHP_BASE, "/store/client.h5.setting/update&key=basic",
                body={"key": "basic", "form": {"enabled": True, "baseUrl": "https://h5b.example.com"}})
    conn = db()
    php_state2 = state(conn.cursor())
    conn.close()
    clear()
    java2 = call(JAVA_BASE, "/store/client.h5.setting/update?key=basic",
                 body={"key": "basic", "form": {"enabled": True, "baseUrl": "https://h5b.example.com"}})
    conn = db()
    java_state2 = state(conn.cursor())
    conn.close()
    check("h5.update 二次覆盖一致",
          php2.get("status") == java2.get("status") and php_state2 == java_state2,
          f"\n    php={php2}\n    java={java2}\n    php_state={php_state2}\n    java_state={java_state2}")
    clear()


def case_wxapp_setting():
    print("\n[case] 小程序设置更新（enableShipping=false）")
    def state(cur):
        cur.execute("select `key`,`describe`,`values`,store_id from yoshop_wxapp_setting order by `key`")
        return cur.fetchall()

    def clear():
        conn = db()
        conn.cursor().execute("delete from yoshop_wxapp_setting")
        conn.close()

    form = {"enabled": True, "app_id": "wxtest123", "app_secret": "secret456", "enableShipping": False}
    clear()
    php1 = call(PHP_BASE, "/store/client.wxapp.setting/update&key=basic",
                body={"key": "basic", "form": form})
    conn = db()
    php_state = state(conn.cursor())
    conn.close()

    clear()
    java1 = call(JAVA_BASE, "/store/client.wxapp.setting/update?key=basic",
                 body={"key": "basic", "form": form})
    conn = db()
    java_state = state(conn.cursor())
    conn.close()

    check("wxapp.update 响应一致",
          php1.get("status") == java1.get("status") and php1.get("message") == java1.get("message"),
          f"php={php1} java={java1}")
    check("wxapp.update 落库一致", php_state == java_state, f"\n    php={php_state}\n    java={java_state}")
    clear()


def export_case_rows(base, export_dir, label):
    """执行一次导出，返回 (响应, 记录行, 单元格矩阵)"""
    resp = call(base, "/store/order.export/exportOrder&dataType=all"
                      "&columns%5B%5D=order_no&columns%5B%5D=goods_detail&columns%5B%5D=pay_price"
                      "&columns%5B%5D=create_time&columns%5B%5D=order_status&columns%5B%5D=pay_method"
                      "&columns%5B%5D=receipt_address&columns%5B%5D=update_price&columns%5B%5D=is_comment",
                method="get")
    conn = db()
    cur = conn.cursor()
    cur.execute("select id,start_time,end_time,file_path,status,store_id from yoshop_order_export"
                " order by id desc limit 1")
    record = cur.fetchone()
    conn.close()
    rows = []
    if record:
        path = export_dir / record["file_path"].replace("downloads/", "")
        if path.exists():
            wb = openpyxl.load_workbook(path)
            ws = wb.active
            rows = [[("" if c is None else str(c)) for c in row]
                    for row in ws.iter_rows(values_only=True)]
            # 导出文件保留（文件名带时间戳，互不覆盖；沙箱禁止删除，交由人工清理）
        # 记录里带时间戳，清理掉便于重复执行
        conn = db()
        conn.cursor().execute("delete from yoshop_order_export where id=%s", (record["id"],))
        conn.close()
    print(f"    {label} 导出响应={resp.get('status')}/{resp.get('message')} 记录={record}")
    return resp, record, rows


def case_export():
    print("\n[case] 订单导出：Excel 内容 + 导出记录")
    php_resp, php_rec, php_rows = export_case_rows(PHP_BASE, PHP_EXPORT_DIR, "PHP")
    java_resp, java_rec, java_rows = export_case_rows(JAVA_BASE, JAVA_EXPORT_DIR, "Java")

    check("exportOrder 响应一致",
          php_resp.get("status") == java_resp.get("status")
          and php_resp.get("message") == java_resp.get("message"),
          f"php={php_resp} java={java_resp}")
    check("导出记录字段一致（file_path 归一化时间戳）",
          php_rec is not None and java_rec is not None
          and re.sub(r"order-\d+", "order-TS", php_rec["file_path"])
          == re.sub(r"order-\d+", "order-TS", java_rec["file_path"])
          and php_rec["status"] == java_rec["status"]
          and php_rec["store_id"] == java_rec["store_id"]
          and php_rec["start_time"] == java_rec["start_time"]
          and php_rec["end_time"] == java_rec["end_time"],
          f"\n    php={php_rec}\n    java={java_rec}")
    check("Excel 单元格完全一致", php_rows == java_rows,
          f"\n    php={php_rows}\n    java={java_rows}")

    # 无数据分支：筛选一个不存在的会员
    php_empty = call(PHP_BASE, "/store/order.export/exportOrder&dataType=all&userId=99999"
                               "&columns%5B%5D=order_no", method="get")
    java_empty = call(JAVA_BASE, "/store/order.export/exportOrder?dataType=all&userId=99999"
                                 "&columns%5B%5D=order_no", method="get")
    check("exportOrder 无数据提示一致",
          php_empty.get("status") == java_empty.get("status")
          and php_empty.get("message") == java_empty.get("message"),
          f"php={php_empty} java={java_empty}")


def case_printer():
    print("\n[case] 小票打印：打印机不存在 / 驱动返回错误")
    php = call(PHP_BASE, "/store/order.event/printer&orderId=10001", body={"orderId": 10001, "form": {"printerId": 99999}})
    java = call(JAVA_BASE, "/store/order.event/printer?orderId=10001", body={"orderId": 10001, "form": {"printerId": 99999}})
    check("printer 不存在响应一致",
          php.get("status") == java.get("status") and php.get("message") == java.get("message"),
          f"php={php} java={java}")

    # 真实调用飞鹅云接口（测试账号必然被拒），验证两端驱动错误透传一致
    conn = db()
    cur = conn.cursor()
    cur.execute("select count(*) c from yoshop_printer where printer_id=10001")
    has_printer = cur.fetchone()["c"] > 0
    conn.close()
    if not has_printer:
        print("  [skip] 未找到 10001 测试打印机，跳过驱动错误对照")
        return
    php2 = call(PHP_BASE, "/store/order.event/printer&orderId=10001", body={"orderId": 10001, "form": {"printerId": 10001}})
    java2 = call(JAVA_BASE, "/store/order.event/printer?orderId=10001", body={"orderId": 10001, "form": {"printerId": 10001}})
    check("printer 驱动失败响应一致（含 data 结构）",
          php2.get("status") == java2.get("status")
          and php2.get("message") == java2.get("message")
          and php2.get("data") == java2.get("data"),
          f"php={php2} java={java2}")


def case_export_between_time():
    print("\n[case] 订单导出：带 betweenTime 时落库起止日期")
    def run(base, export_dir, label):
        resp = call(base, "/store/order.export/exportOrder&dataType=all"
                          "&betweenTime%5B%5D=2026-10-01&betweenTime%5B%5D=2026-10-17"
                          "&columns%5B%5D=order_no&columns%5B%5D=create_time", method="get")
        conn = db()
        cur = conn.cursor()
        cur.execute("select id,start_time,end_time,status,store_id,file_path from yoshop_order_export"
                    " order by id desc limit 1")
        rec = cur.fetchone()
        cur.execute("delete from yoshop_order_export where id=%s", (rec["id"],))
        conn.close()
        print(f"    {label} 记录={rec}")
        return resp, rec

    php_resp, php_rec = run(PHP_BASE, PHP_EXPORT_DIR, "PHP")
    java_resp, java_rec = run(JAVA_BASE, JAVA_EXPORT_DIR, "Java")
    check("betweenTime 导出响应一致",
          php_resp.get("status") == java_resp.get("status")
          and php_resp.get("message") == java_resp.get("message"),
          f"php={php_resp} java={java_resp}")
    check("betweenTime 起止时间落库一致",
          php_rec and java_rec
          and php_rec["start_time"] == java_rec["start_time"]
          and php_rec["end_time"] == java_rec["end_time"]
          and php_rec["status"] == java_rec["status"],
          f"\n    php={php_rec}\n    java={java_rec}")


def main():
    case_comment_edit()
    case_comment_edit_clear_images()
    case_comment_delete()
    case_comment_missing()
    case_h5_setting()
    case_wxapp_setting()
    case_export()
    case_export_between_time()
    case_printer()
    reset_comments()
    print(f"\n通过 {passed} / 失败 {failed}")


if __name__ == "__main__":
    main()