shop-java/tools/smoke_camel.py

# -*- coding: utf-8 -*-
"""驼峰改造冒烟：核心读接口出参不得出现 snake_case key；写接口用驼峰入参。"""
import re
import sys

sys.path.insert(0, "tools")
from test_redis_auth import login, S, BASE, Redis, PREFIX


def check_snake(obj, path="$"):
    hits = []
    if isinstance(obj, dict):
        for k, v in obj.items():
            if re.match(r'^[a-z]+(_[a-z0-9]+)+$', k):
                hits.append("%s.%s" % (path, k))
            hits += check_snake(v, path + "." + k)
    elif isinstance(obj, list):
        for i, v in enumerate(obj[:3]):
            hits += check_snake(v, "%s[%d]" % (path, i))
    return hits


tok = login("redistest_admin", "123456")["data"]["token"]
H = {"Access-Token": tok}


def get(p):
    return S.get(BASE + p, headers=H, timeout=30).json()


def post(p, b):
    return S.post(BASE + p, headers=H, json=b, timeout=30).json()


fails = []
CASES = [
    "/store/goods/list?page=1&pageSize=5",
    "/store/order/list?dataType=order&page=1&pageSize=5",
    "/store/order/list?dataType=delivery&page=1&pageSize=5",
    "/store/order/list?dataType=receipt&page=1&pageSize=5",
    "/store/order/list?dataType=complete&page=1&pageSize=5",
    "/store/store.user/list?page=1",
    "/store/files/list?page=1",
    "/store/goods.comment/list?page=1",
    "/store/market.coupon/list?page=1",
    "/store/page/list",
    "/store/setting/detail",
    "/store/statistics.data/data",
    "/store/statistics.data/survey",
    "/store/category/list",
    "/store/client.h5.setting/detail?key=basic",
    "/store/client.wxapp.setting/detail?key=basic",
]
# goods/detail 用列表里真实存在的商品 ID（10001/10002 已软删）
first = get("/store/goods/list?page=1&pageSize=1")
lst = first.get("data", {}).get("list", {})
real_gid = (lst.get("data") or [{}])[0].get("goodsId") if isinstance(lst, dict) else None
if real_gid:
    CASES.insert(1, "/store/goods/detail?goodsId=%s" % real_gid)

for p in CASES:
    r = get(p)
    hits = check_snake(r)
    tag = "ok" if (r.get("status") == 200 and not hits) else "FAIL"
    print("[%s] %-46s status=%s snake=%s" % (tag, p, r.get("status"), hits[:4] if hits else "无"))
    if tag == "FAIL":
        fails.append(p)

# 写接口：驼峰入参
r5 = post("/store/category/add", {"form": {"name": "冒烟分类", "parentId": 0, "sort": 99, "imageId": 0}})
print("[--] category.add(驼峰入参):", r5["status"], r5["message"])
if r5["status"] != 200:
    fails.append("category.add")

with Redis() as rd:
    rd("DEL", PREFIX + tok)

print("\n冒烟结果:", "全部通过" if not fails else "FAIL: %s" % fails)
sys.exit(1 if fails else 0)