shop-java/tools/test_uniapp_contract.py

# -*- coding: utf-8 -*-
"""uniapp 客户端契约测试：按 uniapp 的真实提交/解包结构验证关键链路。

与 test_api_batch*.py 的区别：那些脚本用平铺 JSON，这里用 uniapp 实际发出的
`{form: {...}}` 包裹结构，并覆盖客户端对响应外层键的硬依赖。

覆盖：
  1. /api/setting/data 必须带 setting 包裹（登录页 SettingModel 读 result.data.setting）
  2. 图形验证码 → 发送短信 → 手机号登录（全部用 {form: ...} 包裹）
  3. 登录后的 token 能否访问 /api/user/info

前置：Java 服务运行在 18080，Redis / MySQL 可连（读 application.yml 取密码）。
"""
import json
import re
import socket
import sys
import requests
import pymysql

from redis_probe import cmd

BASE = "http://127.0.0.1:18080"
MOBILE = "13900002222"

YML = open("../yoshop-java/src/main/resources/application.yml", encoding="utf-8").read()
DB_PWD = re.findall(r"password:\s*(.+)", YML)[0].strip().strip("\"'")
REDIS_PWD = re.findall(r"password:\s*(.+)", YML)[1].strip().strip("\"'")

S = requests.Session()
S.trust_env = False
passed = failed = 0


def check(name, cond, detail=""):
    global passed, failed
    if cond:
        passed += 1
        print(f"  PASS  {name}")
    else:
        failed += 1
        print(f"  FAIL  {name}  {detail}")


def redis_cmd(*args):
    sock = socket.create_connection(("192.168.1.177", 6379), timeout=5)
    try:
        cmd(sock, "AUTH", REDIS_PWD)
        return cmd(sock, *args)
    finally:
        sock.close()


def main():
    # 1. 设置项结构
    r = S.get(BASE + "/api/setting/data", timeout=10).json()
    setting = (r.get("data") or {}).get("setting")
    check("setting/data 有 setting 包裹", isinstance(setting, dict), list(r.get("data") or {}))
    check("setting.register 存在", isinstance((setting or {}).get("register"), dict),
          (setting or {}).get("register"))
    check("registerMethod 有值", (setting or {}).get("register", {}).get("registerMethod") == 10,
          (setting or {}).get("register"))

    # 2. 图形验证码
    r = S.get(BASE + "/api/captcha/image", timeout=10).json()
    data = r.get("data") or {}
    code = json.loads(redis_cmd("GET", "yoshop:api:captcha:" + str(data.get("key"))))["code"]
    check("验证码返回 key/base64", bool(data.get("key")) and str(data.get("base64", "")).startswith("data:image"))

    # 3. 发送短信（uniapp 结构：{form: {...}}）
    r = S.post(BASE + "/api/captcha/sendSmsCaptcha",
               json={"form": {"mobile": MOBILE, "captchaCode": code, "captchaKey": data.get("key")}},
               timeout=10).json()
    check("form 包裹的发送短信成功", r.get("status") == 200, r)
    sms = json.loads(redis_cmd("GET", "yoshop:api:sms:" + MOBILE))["code"]

    # 4. 登录（uniapp 结构）
    r = S.post(BASE + "/api/passport/login",
               json={"form": {"mobile": MOBILE, "smsCode": sms, "isParty": False, "partyData": {}}},
               timeout=10).json()
    d = r.get("data") or {}
    check("form 包裹的登录成功", r.get("status") == 200 and r.get("message") == "登录成功", r)
    check("返回 userId/token", isinstance(d.get("userId"), int)
          and re.fullmatch(r"[0-9a-f]{32}", str(d.get("token"))) is not None, d)

    # 5. token 访问用户中心
    token = d.get("token")
    r = S.get(BASE + "/api/user/info", headers={"Access-Token": token}, timeout=10).json()
    info = (r.get("data") or {}).get("userInfo") or {}
    check("user/info 可访问", r.get("status") == 200, r)
    check("userInfo 有 userId/nickName", info.get("userId") == d.get("userId"), list(info)[:8])

    # 清理：删掉本次登录自动注册的用户与临时键
    conn = pymysql.connect(host="192.168.1.177", user="root", password=DB_PWD,
                           database="shop", charset="utf8mb4", autocommit=True)
    try:
        cur = conn.cursor()
        cur.execute("select user_id from yoshop_user where mobile=%s", (MOBILE,))
        row = cur.fetchone()
        if row:
            cur.execute("delete from yoshop_user where user_id=%s", (row[0],))
            print(f"  CLEAN 已删除测试用户 {row[0]}")
    finally:
        conn.close()
    for key in [f"yoshop:api:sms:{MOBILE}", f"yoshop:api:sms:send:{MOBILE}",
                "yoshop:api:token:" + str(token)]:
        redis_cmd("DEL", key)

    print(f"\n结果：PASS {passed} / FAIL {failed}")
    return 0 if failed == 0 else 1


if __name__ == "__main__":
    sys.exit(main())