shop-java/tools/test_api_batch1.py

# -*- coding: utf-8 -*-
"""客户端 api 第 1 批接口冒烟：只读接口结构 + 关键字段断言。

用法：起 Java 18080 后执行
    python tools/test_api_batch1.py
环境变量：YOSHOP_API_BASE（默认 http://127.0.0.1:18080）、YOSHOP_TOKEN（已登录用户 token，可选）
"""
import json
import os
import re
import sys

import pymysql
import requests

from _root import ROOT

BASE = os.environ.get("YOSHOP_API_BASE", "http://127.0.0.1:18080")
TOKEN = os.environ.get("YOSHOP_TOKEN", "").strip()

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

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

passed = 0
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 get(path, params=None, token=None):
    headers = {"platform": "MP-WEIXIN"}
    if token:
        headers["Access-Token"] = token
    resp = S.get(BASE + path, params=params or {}, headers=headers, timeout=30)
    try:
        return resp.json()
    except Exception:
        return {"status": resp.status_code, "raw": resp.text[:200]}


def post(path, body=None, token=None):
    headers = {"platform": "MP-WEIXIN"}
    if token:
        headers["Access-Token"] = token
    resp = S.post(BASE + path, json=body or {}, headers=headers, timeout=30)
    try:
        return resp.json()
    except Exception:
        return {"status": resp.status_code, "raw": resp.text[:200]}


def snake_keys(obj, prefix="$"):
    """收集所有 snake_case 键"""
    found = []
    if isinstance(obj, dict):
        for k, v in obj.items():
            if isinstance(k, str) and re.fullmatch(r"[a-z0-9]+(_[a-z0-9]+)+", k):
                found.append(f"{prefix}.{k}")
            found.extend(snake_keys(v, f"{prefix}.{k}"))
    elif isinstance(obj, list):
        for i, v in enumerate(obj[:3]):
            found.extend(snake_keys(v, f"{prefix}[{i}]"))
    return found


def db(sql):
    conn = pymysql.connect(host="192.168.1.177", port=3306, user="root", password=PWD,
                           database="shop", charset="utf8mb4", autocommit=True)
    try:
        cur = conn.cursor()
        cur.execute(sql)
        return cur.fetchall()
    finally:
        conn.close()


def main():
    print("== 客户端 api 第 1 批（基础 / 内容 / 商品 / 营销 只读） ==")

    # ---------- 基础信息
    r = get("/api/store/data")
    data = r.get("data") or {}
    check("store/data status=200", r.get("status") == 200, r.get("message"))
    check("store/data 含 storeInfo/client/setting/clientData",
          all(k in data for k in ("storeInfo", "client", "setting", "clientData")), list(data))
    check("store/data client=MP-WEIXIN", data.get("client") == "MP-WEIXIN", data.get("client"))
    store_info = data.get("storeInfo") or {}
    check("storeInfo 隐藏 sort/isRecycle/isDelete", not any(
        k in store_info for k in ("sort", "isRecycle", "isDelete")), list(store_info))
    check("storeInfo 含 imageUrl 键", "imageUrl" in store_info, list(store_info))
    check("clientData.h5.setting 存在", "setting" in ((data.get("clientData") or {}).get("h5") or {}))

    r = get("/api/setting/data")
    setting = r.get("data") or {}
    check("setting/data 六个公共项", set(setting) == {
        "appTheme", "pageCategoryTemplate", "points", "recharge", "register", "customer"}, list(setting))
    check("setting/data register 有 registerMethod",
          "registerMethod" in (setting.get("register") or {}), setting.get("register"))

    # ---------- 地区 / 分类 / 内容
    r = get("/api/region/all")
    region_all = (r.get("data") or {}).get("list") or {}
    check("region/all 非空", len(region_all) > 3000, len(region_all))
    r = get("/api/region/tree")
    region_tree = (r.get("data") or {}).get("list") or {}
    check("region/tree 省级 34 条", len(region_tree) >= 30, len(region_tree))

    r = get("/api/category/list")
    cats = (r.get("data") or {}).get("list") or []
    check("category/list 是数组", isinstance(cats, list), type(cats))
    check("category/list 有子分类或至少 1 条", len(cats) >= 1, len(cats))
    if cats:
        keys = set(cats[0])
        check("分类隐藏 storeId/createTime/updateTime",
              not ({"storeId", "createTime", "updateTime"} & keys), keys)

    # 文章：库里现有文章已被软删除，临时插一条用于验证，跑完删除
    article_id = None
    try:
        db("INSERT INTO yoshop_article (title, show_type, category_id, image_id, content, sort, status, "
           "virtual_views, actual_views, is_delete, store_id, create_time, update_time) "
           "VALUES ('api自测文章', 10, 0, 0, '<p>正文</p>', 100, 1, 0, 0, 0, 10001, "
           "UNIX_TIMESTAMP(), UNIX_TIMESTAMP())")
        article_id = db("SELECT MAX(article_id) FROM yoshop_article")[0][0]

        r = get("/api/article.category/list")
        check("article.category/list status=200", r.get("status") == 200, r.get("message"))
        r = get("/api/article/list")
        page_list = (r.get("data") or {}).get("list") or {}
        check("article/list status=200 且分页结构", r.get("status") == 200
              and all(k in page_list for k in ("total", "perPage", "currentPage", "lastPage")),
              r.get("message"))
        check("article/list 命中临时文章", any(a.get("articleId") == article_id for a in (page_list.get("data") or [])))
        r = get("/api/article/detail", {"articleId": article_id})
        article = (r.get("data") or {}).get("detail") or {}
        check("article/detail 返回文章", article.get("articleId") == article_id,
              article.get("message") or r.get("message"))
        check("文章隐藏 virtualViews/actualViews",
              "actualViews" not in article and "virtualViews" not in article)
        check("文章 content 已反转义", "<" not in str(article.get("content")), article.get("content"))
        views = db(f"SELECT actual_views FROM yoshop_article WHERE article_id={article_id}")[0][0]
        check("文章阅读数 +1", views == 1, views)
    finally:
        if article_id:
            db(f"DELETE FROM yoshop_article WHERE article_id={article_id}")

    r = get("/api/help/list")
    check("help/list status=200", r.get("status") == 200, r.get("message"))

    # ---------- 页面
    r = get("/api/page/detail")
    page_data = (r.get("data") or {}).get("pageData") or {}
    check("page/detail 默认首页有 page/items", "page" in page_data and "items" in page_data, list(page_data))
    items = page_data.get("items") or []
    check("页面组件已去掉 defaultData/demo",
          all("defaultData" not in it and "demo" not in it for it in items))

    # ---------- 商品
    r = get("/api/goods/list")
    page = (r.get("data") or {}).get("list") or {}
    goods_list = page.get("data") or []
    check("goods/list 有数据", len(goods_list) > 0, r.get("message"))
    check("goods/list 分页 15", page.get("perPage") == 15, page.get("perPage"))
    if goods_list:
        g = goods_list[0]
        check("列表隐藏 content/goodsImages/images",
              not ({"content", "goodsImages", "images"} & set(g)), set(g) & {"content", "goodsImages", "images"})
        check("列表含 goodsImage/goodsSales/isUserGrade",
              all(k in g for k in ("goodsImage", "goodsSales", "isUserGrade")), list(g))

    goods_ids = [g["goodsId"] for g in goods_list[:3]]
    r = get("/api/goods/detail", {"goodsId": 10004})
    detail = (r.get("data") or {}).get("detail") or {}
    check("goods/detail 多规格含 skuList/specList",
          bool(detail.get("skuList")) and bool(detail.get("specList")), list(detail))
    check("详情隐藏 images", "images" not in detail, "images" in detail)
    check("详情含 goodsImages", "goodsImages" in detail, list(detail))
    check("详情 content 已反转义（不含 <）", "<" not in str(detail.get("content")))

    r = get("/api/goods/detail", {"goodsId": 999999})
    check("goods/detail 不存在时 status=500", r.get("status") == 500, r.get("status"))

    r = get("/api/goods/basic", {"goodsId": 10003})
    basic = (r.get("data") or {}).get("detail") or {}
    check("goods/basic 不含 skuList/video", "skuList" not in basic and "video" not in basic, list(basic))
    check("goods/basic 含 content", "content" in basic)

    r = get("/api/goods/spec_data", {"goodsId": 10004})
    spec = (r.get("data") or {}).get("specData") or {}
    check("goods/spec_data 含 skuList/specList", bool(spec.get("skuList")) and bool(spec.get("specList")))
    sku0 = (spec.get("skuList") or [{}])[0]
    check("skuList 隐藏 image/storeId/createTime", not ({"image", "storeId", "createTime"} & set(sku0)), set(sku0))
    check("skuList 含 imageUrl", "imageUrl" in sku0, list(sku0))

    sku_id = sku0.get("goodsSkuId")
    if sku_id:
        r = get("/api/goods/sku_info", {"goodsId": 10004, "goodsSkuId": sku_id})
        sku_info = (r.get("data") or {}).get("skuInfo") or {}
        check("goods/sku_info 命中 SKU", sku_info.get("goodsSkuId") == sku_id, sku_info)

    r = get("/api/goods.service/list", {"goodsId": 10004})
    check("goods.service/list status=200", r.get("status") == 200, r.get("message"))
    services = (r.get("data") or {}).get("list") or []
    if services:
        check("服务承诺隐藏 status/sort/isDefault",
              not ({"status", "sort", "isDefault"} & set(services[0])), set(services[0]))

    # ---------- 评价 / 优惠券 / 快递
    for path, key in [("/api/comment/list", "list"), ("/api/comment/list_rows", "list")]:
        r = get(path, {"goodsId": 10003})
        check(f"{path} status=200", r.get("status") == 200, r.get("message"))
    r = get("/api/comment/total", {"goodsId": 10003})
    total = r.get("data") or {}
    check("comment/total 四个统计项", set(total) == {"all", "praise", "review", "negative"}, total)

    r = get("/api/coupon/list")
    coupons = (r.get("data") or {}).get("list") or []
    check("coupon/list 有数据", len(coupons) >= 1, r.get("message"))
    if coupons:
        c = coupons[0]
        check("优惠券隐藏 totalNum/receiveNum/status",
              not ({"totalNum", "receiveNum", "status"} & set(c)), set(c))
        check("未登录不返回 isReceive", "isReceive" not in c, list(c))

    r = get("/api/express/list")
    express = (r.get("data") or {}).get("list") or []
    check("express/list 有数据", len(express) > 0, r.get("message"))
    if express:
        check("快递隐藏 kuaidi100Code/kdniaoCode",
              not ({"kuaidi100Code", "kdniaoCode"} & set(express[0])), set(express[0]))

    # ---------- 登录态相关（带 token 才有 isReceive / 会员价）
    if TOKEN:
        r = get("/api/coupon/list", token=TOKEN)
        coupons = (r.get("data") or {}).get("list") or []
        if coupons:
            check("登录后 coupon/list 返回 isReceive", "isReceive" in coupons[0], list(coupons[0]))
    else:
        print("  SKIP  带 token 的 isReceive 检查（未设置 YOSHOP_TOKEN）")

    # ---------- 驼峰检查
    for path, params in [("/api/goods/list", {}), ("/api/store/data", {}),
                         ("/api/goods/detail", {"goodsId": 10004}),
                         ("/api/category/list", {})]:
        r = get(path, params)
        bad = snake_keys(r.get("data"))
        check(f"{path} 无 snake_case 键", not bad, bad[:5])

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


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