shop-java/tools/test_redis_auth.py

# -*- coding: utf-8 -*-
"""验证 /store/passport/login 的 Redis 登录态：登录写入 → 鉴权通过 → 登出删除 → 鉴权失效。

前置：Java 服务运行在 18080（yoshop.auth.mode=redis）。
"""
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:"

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:
        b = a.encode() if isinstance(a, str) else a
        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(username, password):
    return S.post(BASE + "/store/passport/login",
                  headers={"Content-Type": "application/json"},
                  data=json.dumps({"username": username, "password": password}),
                  timeout=30).json()


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


def main():
    with Redis() as r:
        print("== 1. 登录 ==")
        res = login("redistest_admin", "123456")
        check("登录成功", res.get("status") == 200 and res.get("message") == "登录成功", 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

        key = PREFIX + token
        print("== 2. Redis 落库 ==")
        raw = r("GET", key)
        check("key 存在", raw is not None, "key=%s" % key)
        ttl = r("TTL", key)
        check("TTL 为 7 天（604800，允许 5 秒误差）", isinstance(ttl, int) and 604795 <= ttl <= 604800, "ttl=%s" % ttl)
        try:
            val = json.loads(raw) if raw else {}
            # RedisConfig 的 value 序列化器是 Jackson2JsonRedisSerializer，
            # 写入 String 时会再包一层 JSON 编码（存储为 "{\"user\":...}"），
            # 这里解两层，兼容纯 JSON 与双层编码两种存法。
            if isinstance(val, str):
                val = json.loads(val)
        except Exception:
            val = {}
        check("value.user 有 store_user_id", isinstance(val.get("user"), dict) and "store_user_id" in val["user"],
              json.dumps(val, ensure_ascii=False)[:160])
        check("value.store_id 正确", val.get("store_id") == 10001, "store_id=%s" % val.get("store_id"))
        check("value.is_login 为 true", val.get("is_login") is True, "is_login=%r" % val.get("is_login"))

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

        print("== 4. 无 token 应 401 ==")
        no = api(None)
        check("无 token 返回 401", no.get("status") == 401 and no.get("message") == "请先登录后再访问",
              json.dumps(no, ensure_ascii=False)[:160])

        print("== 5. 登出 ==")
        out = S.post(BASE + "/store/passport/logout", headers={"Access-Token": token}, timeout=30).json()
        check("登出返回 200/操作成功", out.get("status") == 200 and out.get("message") == "操作成功",
              json.dumps(out, ensure_ascii=False)[:160])

        print("== 6. Redis 已清除 ==")
        check("key 已删除", r("GET", key) is None, "key=%s" % key)
        check("TTL 为 -2", r("TTL", key) == -2, "ttl=%s" % r("TTL", key))

        print("== 7. 登出后 token 失效 ==")
        after = api(token)
        check("旧 token 返回 401", after.get("status") == 401, json.dumps(after, ensure_ascii=False)[:160])

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


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