# -*- 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 from _root import ROOT 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(str(ROOT / "yoshop-java/src/main/resources/application.yml")) JAVA_EXPORT_DIR = Path(str(ROOT / "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()