shop-java/tools/test_token_user.py

# -*- coding: utf-8 -*-
"""验证「登录后把 StoreUser 写进 Redis，以 token 为 key，凭 token 取用户信息」。

覆盖点：
  1. 登录落库：key = yoshop:store:token:{token}，value 就是用户 JSON（顶层字段，无 user/store_id/is_login 包裹，无 password）
  2. TTL = 7 天
  3. /store/store.user/info 的用户信息来自缓存（篡改缓存里的 userName，接口跟着变，说明没回查数据库）
  4. 带 token 访问业务接口正常
  5. /store/store.user/renew 改资料后缓存被刷新，且不延长登录时长
  6. 登出后 key 删除，旧 token 再访问 401

前置：Java 服务运行在 18080（yoshop.auth.mode=redis），库内有临时账号 camel_test_admin / 123456。
"""
import json
import socket
import sys

import requests

BASE = "http://127.0.0.1:18080"
REDIS = ("192.168.1.177", 6379, "@@Qinys12346..")
PREFIX = "yoshop:store:token:"
USER_NAME = "camel_test_admin"
PASSWORD = "123456"

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()
    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()
    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)


def login():
    return S.post(BASE + "/store/passport/login",
                  headers={"Content-Type": "application/json"},
                  data=json.dumps({"username": USER_NAME, "password": PASSWORD}),
                  timeout=30).json()


def get(path, token):
    headers = {"Access-Token": token} if token else {}
    return S.get(BASE + path, headers=headers, timeout=30).json()


def post(path, token, body=None):
    headers = {"Content-Type": "application/json"}
    if token:
        headers["Access-Token"] = token
    return S.post(BASE + path, headers=headers, json=body or {}, timeout=30).json()


def main():
    with Redis() as r:
        print("== 1. 登录 ==")
        res = login()
        check("登录成功", res.get("status") == 200, json.dumps(res, ensure_ascii=False))
        token = (res.get("data") or {}).get("token")
        check("返回 token", isinstance(token, str) and len(token) == 32, str(token))
        if not token:
            return 1

        key = PREFIX + token
        print("== 2. 登录态落 Redis：token 即 key，value 即用户 ==")
        raw = r("GET", key)
        check("key = token", raw is not None, "key=%s" % key)
        ttl = r("TTL", key)
        check("TTL = 604800（允许 5 秒误差）", isinstance(ttl, int) and 604795 <= ttl <= 604800, "ttl=%s" % ttl)
        try:
            val = json.loads(raw) if raw else {}
        except Exception:
            val = {}
        if isinstance(val, str):
            val = json.loads(val)
        check("value 顶层就是用户字段", val.get("storeUserId") == 10992 and val.get("userName") == USER_NAME,
              json.dumps(val, ensure_ascii=False)[:200])
        check("不再有 user / store_id / is_login 包裹",
              not ({"user", "store_id", "is_login"} & set(val.keys())), sorted(val.keys()))
        check("关键字段齐全",
              all(k in val for k in ("storeUserId", "userName", "realName", "isSuper", "storeId", "createTime")),
              sorted(val.keys()))
        check("password 不落缓存", "password" not in val, sorted(val.keys()))

        print("== 3. 用户信息取自缓存（不回查数据库）==")
        info = get("/store/store.user/info", token)
        user_info = (info.get("data") or {}).get("userInfo") or {}
        check("info 返回 200", info.get("status") == 200, json.dumps(info, ensure_ascii=False)[:160])
        check("userInfo 为当前登录用户", user_info.get("storeUserId") == 10992 and user_info.get("userName") == USER_NAME,
              json.dumps(user_info, ensure_ascii=False)[:200])
        check("roles.isSuper = 0", ((info.get("data") or {}).get("roles") or {}).get("isSuper") == 0,
              json.dumps(info.get("data"), ensure_ascii=False)[:160])

        sentinel = dict(val, userName="from_redis_only")
        r("SET", key, json.dumps(sentinel, ensure_ascii=False))
        hit = ((get("/store/store.user/info", token).get("data") or {}).get("userInfo") or {})
        check("篡改缓存后接口跟着变（证明按 token 读缓存）", hit.get("userName") == "from_redis_only",
              json.dumps(hit, ensure_ascii=False)[:160])
        # 还原缓存
        r("SET", key, json.dumps(val, ensure_ascii=False))
        r("EXPIRE", key, 604800)

        print("== 4. 带 token 访问业务接口 ==")
        goods = get("/store/goods/list", token)
        check("/store/goods/list 返回 200", goods.get("status") == 200, json.dumps(goods, ensure_ascii=False)[:160])

        print("== 5. 改资料后缓存同步刷新，且不延长登录时长 ==")
        before = r("TTL", key)
        renewed = post("/store/store.user/renew", token, {"form": {"realName": "驼峰测试改"}})
        check("renew 返回 200", renewed.get("status") == 200, json.dumps(renewed, ensure_ascii=False)[:160])
        fresh = json.loads(r("GET", key) or "{}")
        if isinstance(fresh, str):
            fresh = json.loads(fresh)
        check("缓存 realName 已更新", fresh.get("realName") == "驼峰测试改",
              json.dumps(fresh, ensure_ascii=False)[:200])
        after = r("TTL", key)
        check("TTL 未被延长（沿用剩余有效期）", isinstance(after, int) and after <= before and after > before - 20,
              "before=%s after=%s" % (before, after))
        # 改回去，保持库内数据原样
        post("/store/store.user/renew", token, {"form": {"realName": "驼峰测试"}})

        print("== 6. 登出 ==")
        out = post("/store/passport/logout", token)
        check("登出返回 200", out.get("status") == 200, json.dumps(out, ensure_ascii=False)[:160])
        check("key 已删除", r("GET", key) is None, "key=%s" % key)
        check("TTL 为 -2", r("TTL", key) == -2, "ttl=%s" % r("TTL", key))
        after_logout = get("/store/goods/list", token)
        check("旧 token 再访问返回 401", after_logout.get("status") == 401,
              json.dumps(after_logout, ensure_ascii=False)[:160])

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


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