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