shop-java/tools/test_admin_api.py

# -*- coding: utf-8 -*-
"""admin 端 24 个接口的端到端验证。

覆盖：
  1. 登录：token 落 Redis（yoshop:admin:token:{token}，值为管理员 JSON、无 password、TTL 7 天）
  2. 未登录 / 错误 token 访问业务接口 → 401
  3. admin.user/detail、renew（改回原值，不污染数据）
  4. store/index、recycle、edit（原值回写）、recovery→move 回收站往返、add（免费版拒绝）
  5. store.api：树列表、新增、编辑、删除（子级保护）
  6. store.menu：树列表、详情、新增、编辑、set_apis（空值拒绝）、删除（子级保护）
  7. setting.cache/clear（只清 data，不清登录态）
  8. setting.science/info 结构
  9. setting.timer/test 与 yoshop_system_process 一致性
 10. logout 后 token 失效

前置：Java 服务在 18080；admin 账号口令临时改为 123456（脚本只管业务，口令由外部脚本改）。
"""
import json
import re
import socket
import sys

import pymysql
import requests

BASE = "http://127.0.0.1:18080"
REDIS = ("192.168.1.177", 6379, "@@Qinys12346..")
PREFIX = "yoshop:admin:token:"
STORE_PREFIX = "yoshop:store:token:"

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

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

passed = failed = 0


def check(name, cond, extra=""):
    global passed, failed
    if cond:
        passed += 1
        print("  [ok] %s" % name)
    else:
        failed += 1
        print("  [FAIL] %s %s" % (name, extra))


# ---------------------------------------------------------------- 最小 Redis 客户端
def resp(sock, *args):
    data = b"*%d\r\n" % len(args)
    for a in args:
        if isinstance(a, bytes):
            b = a
        elif isinstance(a, str):
            b = a.encode()
        else:
            b = str(a).encode()
        data += b"$%d\r\n%s\r\n" % (len(b), b)
    sock.sendall(data)
    return read(sock)


def read(sock):
    line = readline(sock)
    t, rest = line[:1], line[1:]
    if t in (b"+", b"-"):
        return rest.decode(errors="replace")
    if t == b":":
        return int(rest)
    if t == b"$":
        n = int(rest)
        if n == -1:
            return None
        buf = b""
        while len(buf) < n + 2:
            buf += sock.recv(n + 2 - len(buf))
        return buf[:n].decode(errors="replace")
    if t == b"*":
        n = int(rest)
        return None if n == -1 else [read(sock) for _ in range(n)]
    return line.decode(errors="replace")


def readline(sock):
    buf = b""
    while not buf.endswith(b"\r\n"):
        c = sock.recv(1)
        if not c:
            return buf
        buf += c
    return buf[:-2]


class Redis:
    def __enter__(self):
        self.s = socket.create_connection((REDIS[0], REDIS[1]), timeout=5)
        resp(self.s, "AUTH", REDIS[2])
        return self

    def __exit__(self, *a):
        self.s.close()

    def __call__(self, *args):
        return resp(self.s, *args)


# ---------------------------------------------------------------- HTTP
def api(method, path, token=None, body=None, params=None):
    headers = {"Content-Type": "application/json"}
    if token:
        headers["Access-Token"] = token
    url = BASE + path
    if method == "GET":
        r = S.get(url, headers=headers, params=params or {}, timeout=60)
    else:
        r = S.post(url, headers=headers, json=body or {}, params=params or {}, timeout=60)
    try:
        return r.json()
    except Exception:
        return {"status": -1, "message": r.text[:200], "data": []}


def main():
    conn = pymysql.connect(host="192.168.1.177", port=3306, user="root", password=DB_PWD,
                           database="shop", autocommit=True, charset="utf8mb4")
    cur = conn.cursor()
    cur.execute("select store_id, remark, sort, is_recycle from yoshop_store where store_id = 10001")
    store_snapshot = cur.fetchone()
    cur.execute("select store_user_id, user_name from yoshop_store_user where store_id = 10001 and is_super = 1")
    user_snapshot = cur.fetchone()

    temp_ids = {"api": [], "menu": []}
    try:
        with Redis() as r:
            print("== 1. 登录与登录态 ==")
            res = api("POST", "/admin/passport/login", body={"username": "admin", "password": "123456"})
            check("登录成功", res.get("status") == 200 and res.get("message") == "登录成功",
                  json.dumps(res, ensure_ascii=False))
            data = res.get("data") or {}
            token = data.get("token")
            check("返回 userId / token", data.get("userId") == 10001 and isinstance(token, str) and len(token) == 32,
                  json.dumps(data, ensure_ascii=False))
            if not token:
                return 1

            key = PREFIX + token
            raw = r("GET", key)
            val = json.loads(raw) if raw else {}
            if isinstance(val, str):
                val = json.loads(val)
            check("value 顶层是管理员字段", val.get("adminUserId") == 10001 and val.get("userName") == "admin",
                  json.dumps(val, ensure_ascii=False)[:160])
            check("不含 password", "password" not in val, sorted(val.keys()))
            ttl = r("TTL", key)
            check("TTL = 604800", isinstance(ttl, int) and 604795 <= ttl <= 604800, "ttl=%s" % ttl)

            print("== 2. 未登录 / 错误 token ==")
            no = api("GET", "/admin/store/index")
            check("无 token → 401", no.get("status") == 401 and no.get("message") == "请先登录后再访问",
                  json.dumps(no, ensure_ascii=False)[:160])
            bad = api("GET", "/admin/store/index", token="0" * 32)
            check("错误 token → 401", bad.get("status") == 401, json.dumps(bad, ensure_ascii=False)[:160])

            print("== 3. admin.user/detail、renew ==")
            detail = api("GET", "/admin/admin.user/detail", token=token)
            ui = (detail.get("data") or {}).get("userInfo") or {}
            check("detail 返回当前管理员", detail.get("status") == 200 and ui.get("adminUserId") == 10001,
                  json.dumps(detail, ensure_ascii=False)[:200])
            check("detail 不含 password", "password" not in ui, sorted(ui.keys()))
            renew = api("POST", "/admin/admin.user/renew", token=token,
                        body={"form": {"userName": "admin", "password": "", "passwordConfirm": ""}})
            check("renew 空密码不修改 → 200", renew.get("status") == 200, json.dumps(renew, ensure_ascii=False)[:160])
            mismatch = api("POST", "/admin/admin.user/renew", token=token,
                           body={"form": {"userName": "admin", "password": "a123456", "passwordConfirm": "b123456"}})
            check("两次密码不一致 → 确认密码不正确", mismatch.get("status") == 500
                  and mismatch.get("message") == "确认密码不正确", json.dumps(mismatch, ensure_ascii=False)[:160])
            still = api("POST", "/admin/passport/login", body={"username": "admin", "password": "123456"})
            check("口令未被改坏（123456 仍可登录）", still.get("status") == 200,
                  json.dumps(still, ensure_ascii=False)[:160])
            if still.get("status") == 200:
                r("DEL", PREFIX + still["data"]["token"])

            print("== 4. 商城管理 ==")
            idx = api("GET", "/admin/store/index", token=token)
            lst = (idx.get("data") or {}).get("list") or {}
            rows = lst.get("data") or []
            check("store/index 200 且分页结构正确",
                  idx.get("status") == 200 and lst.get("perPage") == 15 and isinstance(rows, list),
                  json.dumps(idx, ensure_ascii=False)[:200])
            first = rows[0] if rows else {}
            check("列表字段驼峰且带 superUser",
                  first.get("storeId") == 10001 and (first.get("superUser") or {}).get("userName") == "admin"
                  and "logoImage" not in first, json.dumps(first, ensure_ascii=False)[:240])

            add = api("POST", "/admin/store/add", token=token)
            check("store/add 免费版拒绝", add.get("status") == 500
                  and add.get("message") == "很抱歉，免费版暂不支持多开商城", json.dumps(add, ensure_ascii=False)[:160])

            edit = api("POST", "/admin/store/edit", token=token,
                       body={"storeId": 10001, "form": {"userName": user_snapshot[1],
                                                        "password": "", "passwordConfirm": "",
                                                        "remark": store_snapshot[1], "sort": store_snapshot[2]}})
            check("store/edit 原值回写 → 200", edit.get("status") == 200, json.dumps(edit, ensure_ascii=False)[:200])

            rec = api("POST", "/admin/store/recovery", token=token, body={"storeId": 10001})
            cur.execute("select is_recycle from yoshop_store where store_id = 10001")
            recycled = cur.fetchone()[0]
            check("recovery 移入回收站（is_recycle=1）", rec.get("status") == 200 and recycled == 1,
                  "status=%s is_recycle=%s" % (rec.get("status"), recycled))
            rec_list = api("GET", "/admin/store/recycle", token=token)
            rec_rows = ((rec_list.get("data") or {}).get("list") or {}).get("data") or []
            check("回收站列表能看到该商城", any(x.get("storeId") == 10001 for x in rec_rows),
                  json.dumps(rec_rows, ensure_ascii=False)[:160])
            mv = api("POST", "/admin/store/move", token=token, body={"storeId": 10001})
            cur.execute("select is_recycle from yoshop_store where store_id = 10001")
            moved = cur.fetchone()[0]
            check("move 移出回收站（is_recycle=0）", mv.get("status") == 200 and moved == 0,
                  "status=%s is_recycle=%s" % (mv.get("status"), moved))

            print("== 5. API 权限点 ==")
            tree = api("GET", "/admin/store.api/index", token=token)
            tree_rows = (tree.get("data") or {}).get("list") or []
            check("store.api/index 返回树", tree.get("status") == 200 and isinstance(tree_rows, list) and tree_rows,
                  json.dumps(tree, ensure_ascii=False)[:200])
            check("根节点有 children", any("children" in x for x in tree_rows), "根节点数=%d" % len(tree_rows))

            add_api = api("POST", "/admin/store.api/add", token=token,
                         body={"form": {"name": "临时权限A", "url": "tmp/a", "parentId": 0, "sort": 900}})
            cur.execute("select api_id from yoshop_store_api where url = 'tmp/a' order by api_id desc limit 1")
            row = cur.fetchone()
            api_id = row[0] if row else None
            if api_id:
                temp_ids["api"].append(api_id)
            check("store.api/add 成功", add_api.get("status") == 200 and api_id is not None,
                  json.dumps(add_api, ensure_ascii=False)[:160])

            child = api("POST", "/admin/store.api/add", token=token,
                        body={"form": {"name": "临时权限A-子", "url": "tmp/a-child", "parentId": api_id, "sort": 901}})
            cur.execute("select api_id from yoshop_store_api where url = 'tmp/a-child' order by api_id desc limit 1")
            row = cur.fetchone()
            child_id = row[0] if row else None
            if child_id:
                temp_ids["api"].append(child_id)
            check("新增子权限成功", child.get("status") == 200 and child_id is not None,
                  json.dumps(child, ensure_ascii=False)[:160])

            edit_api = api("POST", "/admin/store.api/edit", token=token,
                           body={"apiId": api_id, "form": {"name": "临时权限A改", "parentId": 0, "sort": 902}})
            check("store.api/edit 成功", edit_api.get("status") == 200, json.dumps(edit_api, ensure_ascii=False)[:160])

            cycle = api("POST", "/admin/store.api/edit", token=token,
                        body={"apiId": api_id, "form": {"name": "临时权限A改", "parentId": child_id}})
            check("上级选到自己的子级被拒", cycle.get("status") == 500
                  and cycle.get("message") == "上级权限不允许设置为当前子权限",
                  json.dumps(cycle, ensure_ascii=False)[:160])

            del_parent = api("POST", "/admin/store.api/delete", token=token, body={"apiId": api_id})
            check("有子级时删除被拒", del_parent.get("status") == 500
                  and del_parent.get("message") == "当前权限下存在子权限，请先删除",
                  json.dumps(del_parent, ensure_ascii=False)[:160])

            del_child = api("POST", "/admin/store.api/delete", token=token, body={"apiId": child_id})
            del_parent2 = api("POST", "/admin/store.api/delete", token=token, body={"apiId": api_id})
            check("删子级 + 删父级成功", del_child.get("status") == 200 and del_parent2.get("status") == 200,
                  "%s / %s" % (del_child.get("status"), del_parent2.get("status")))
            if del_parent2.get("status") == 200:
                temp_ids["api"] = []

            print("== 6. 后台菜单 ==")
            menus = api("GET", "/admin/store.menu/index", token=token)
            menu_rows = (menus.get("data") or {}).get("list") or []
            check("store.menu/index 返回树", menus.get("status") == 200 and bool(menu_rows),
                  json.dumps(menus, ensure_ascii=False)[:200])
            check("菜单带 apiIds / 不含 menuApi",
                  any("apiIds" in x for x in menu_rows) and all("menuApi" not in x for x in menu_rows),
                  sorted(menu_rows[0].keys()) if menu_rows else "")

            first_menu_id = menu_rows[0]["menuId"]
            info = api("GET", "/admin/store.menu/info", token=token, params={"menuId": first_menu_id})
            info_row = (info.get("data") or {}).get("info") or {}
            check("store.menu/info 返回详情", info.get("status") == 200 and info_row.get("menuId") == first_menu_id,
                  json.dumps(info, ensure_ascii=False)[:200])
            check("详情不含 apiIds/children", "apiIds" not in info_row and "children" not in info_row,
                  sorted(info_row.keys()))

            add_menu = api("POST", "/admin/store.menu/add", token=token,
                           body={"form": {"name": "临时菜单A", "path": "/tmp-a", "type": 10, "isPage": 1,
                                          "parentId": 0, "sort": 900}})
            cur.execute("select menu_id from yoshop_store_menu where path = '/tmp-a' order by menu_id desc limit 1")
            row = cur.fetchone()
            menu_id = row[0] if row else None
            if menu_id:
                temp_ids["menu"].append(menu_id)
            check("store.menu/add 成功", add_menu.get("status") == 200 and menu_id is not None,
                  json.dumps(add_menu, ensure_ascii=False)[:160])

            edit_menu = api("POST", "/admin/store.menu/edit", token=token,
                            body={"menuId": menu_id, "form": {"name": "临时菜单A改", "parentId": 0, "sort": 901}})
            check("store.menu/edit 成功", edit_menu.get("status") == 200, json.dumps(edit_menu, ensure_ascii=False)[:160])

            empty_apis = api("POST", "/admin/store.menu/set_apis", token=token,
                             body={"menuId": menu_id, "form": {"apiIds": []}})
            check("set_apis 空数组被拒", empty_apis.get("status") == 500
                  and empty_apis.get("message") == "API权限不能为空",
                  json.dumps(empty_apis, ensure_ascii=False)[:160])

            bind = api("POST", "/admin/store.menu/set_apis", token=token,
                       body={"menuId": menu_id, "form": {"apiIds": [1, 2]}})
            cur.execute("select api_id from yoshop_store_menu_api where menu_id = %s order by api_id", (menu_id,))
            bound = [x[0] for x in cur.fetchall()]
            check("set_apis 绑定成功", bind.get("status") == 200 and bound == [1, 2],
                  "status=%s bound=%s" % (bind.get("status"), bound))

            del_menu = api("POST", "/admin/store.menu/delete", token=token, body={"menuId": menu_id})
            cur.execute("select count(*) from yoshop_store_menu_api where menu_id = %s", (menu_id,))
            left = cur.fetchone()[0]
            check("删除菜单并清关联", del_menu.get("status") == 200 and left == 0,
                  "status=%s left=%s" % (del_menu.get("status"), left))
            if del_menu.get("status") == 200:
                temp_ids["menu"] = []

            print("== 7. 清理缓存 ==")
            clear = api("POST", "/admin/setting.cache/clear", token=token, body={"form": {"item": ["data"]}})
            check("cache/clear 返回 200", clear.get("status") == 200, json.dumps(clear, ensure_ascii=False)[:160])
            check("非强制模式不清登录态", r("GET", key) is not None, "key=%s" % key)

            print("== 8. 环境检测 ==")
            sci = api("GET", "/admin/setting.science/info", token=token)
            info_ = (sci.get("data") or {}).get("scienceInfo") or {}
            groups = ["server", "phpinfo", "writeable"]
            check("science/info 三段结构", sci.get("status") == 200 and all(g in info_ for g in groups),
                  sorted(info_.keys()))
            items = (info_.get("server") or []) + (info_.get("writeable") or [])
            check("检测项形状 {name,key,value,status,remark}",
                  items and all({"name", "key", "value", "status", "remark"} <= set(x.keys()) for x in items),
                  json.dumps(items[:1], ensure_ascii=False)[:200])

            print("== 9. 定时任务探活 ==")
            cur.execute("select last_working_time from yoshop_system_process where `key` = 'timer'")
            row = cur.fetchone()
            last = row[0] if row else 0
            alive = __import__("time").time() - last <= 10
            timer = api("POST", "/admin/setting.timer/test", token=token)
            expect = "恭喜您，定时任务已开启" if alive else "很抱歉，定时任务未开启"
            check("timer/test 结果与 system_process 一致", timer.get("message") == expect,
                  "expect=%s got=%s" % (expect, json.dumps(timer, ensure_ascii=False)[:160]))

            print("== 10. 登出 ==")
            out = api("POST", "/admin/passport/logout", token=token)
            check("logout 200", out.get("status") == 200, json.dumps(out, ensure_ascii=False)[:160])
            check("Redis 登录态已删除", r("GET", key) is None, "key=%s" % key)
            after = api("GET", "/admin/store/index", token=token)
            check("旧 token → 401", after.get("status") == 401, json.dumps(after, ensure_ascii=False)[:160])

            print("== 11. 商家 token 不能访问 admin ==")
            store_key = r("KEYS", STORE_PREFIX + "*")
            if store_key:
                fake = store_key[0].decode().replace(STORE_PREFIX, "") if isinstance(store_key[0], bytes) \
                    else str(store_key[0]).replace(STORE_PREFIX, "")
                cross = api("GET", "/admin/store/index", token=fake)
                check("商家 token 访问 /admin → 401", cross.get("status") == 401,
                      json.dumps(cross, ensure_ascii=False)[:160])
            else:
                print("  [skip] 当前无商家登录态可借用")
    finally:
        # 还原商城数据
        cur.execute("update yoshop_store set remark = %s, sort = %s, is_recycle = %s where store_id = 10001",
                    (store_snapshot[1], store_snapshot[2], store_snapshot[3]))
        cur.execute("update yoshop_store_user set user_name = %s where store_user_id = %s",
                    (user_snapshot[1], user_snapshot[0]))
        # 清理临时数据
        for api_id in temp_ids["api"]:
            cur.execute("delete from yoshop_store_menu_api where api_id = %s", (api_id,))
            cur.execute("delete from yoshop_store_api where api_id = %s", (api_id,))
        for menu_id in temp_ids["menu"]:
            cur.execute("delete from yoshop_store_menu_api where menu_id = %s", (menu_id,))
            cur.execute("delete from yoshop_store_menu where menu_id = %s", (menu_id,))
        cur.execute("delete from yoshop_store_api where url in ('tmp/a', 'tmp/a-child')")
        cur.execute("delete from yoshop_store_menu where path = '/tmp-a'")
        conn.close()

    print("\n通过 %d / 失败 %d" % (passed, failed))
    return 1 if failed else 0


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