110 lines
4.9 KiB
Python
110 lines
4.9 KiB
Python
"""测试期生产 Redis 隔离 —— 回归断言（project-20260918-P1-test-redis-token-fix）
|
||
|
||
背景：测试套件通过真实 /api/cma/auth/login 签发 token，而 app.auth_middleware 在 import 时
|
||
连接 127.0.0.1:6379 db=1（生产 Redis）、ping 成功即 _redis_available=True → create_token()
|
||
把 token setex 进生产库（TTL 24h）。测试库首个用户 id=1 与生产 users.id=1 撞号，且
|
||
require_auth() 只按 user_id 反查用户，故每轮全量测试都会留下可直接使用的生产会话 token。
|
||
|
||
本文件断言隔离生效（可复现、无副作用）：
|
||
1) 测试期 auth_middleware / utils.cache 的 Redis 通道均关闭（客户端对象为 None）；
|
||
2) 生产 Redis 客户端构造被 conftest 阻断器全部拦下（计数 ≥ 2：auth_middleware + cache）；
|
||
3) 登录签发的 token 只落进程内 _token_store，且经 require_auth 仍可正常鉴权（语义不变）。
|
||
"""
|
||
import json
|
||
|
||
import pytest
|
||
|
||
from app import auth_middleware
|
||
from app.utils import cache as cache_util
|
||
from tests.conftest import create_test_user, get_token_for_user, auth_header
|
||
|
||
|
||
def test_auth_middleware_redis_channel_is_off_in_tests():
|
||
"""测试期 auth_middleware 必须走进程内 _token_store 降级分支（生产 Redis 零写入的前提）"""
|
||
assert auth_middleware._redis_available is False
|
||
assert auth_middleware._redis is None
|
||
|
||
|
||
def test_cache_redis_channel_is_off_in_tests():
|
||
"""测试夹具会调 cache.delete('ai')，同源 db=1 —— 一并关闭，保证整轮零写命令"""
|
||
assert cache_util._available is False
|
||
assert cache_util._client is None
|
||
|
||
|
||
def test_production_redis_constructor_is_blocked(production_redis_isolation_state):
|
||
"""redis.Redis 构造器被阻断器替换：导入期的两次构造尝试全部被拒（无生产连接）"""
|
||
state = production_redis_isolation_state
|
||
assert state["constructor_attempts"] >= 2, state
|
||
# 阻断即无客户端实例 → 不存在可用的生产 Redis 通道
|
||
assert auth_middleware._redis is None
|
||
assert cache_util._client is None
|
||
|
||
|
||
def test_login_token_lands_in_process_store_only(db, client):
|
||
"""登录 token 只落进程内 _token_store（不产生任何生产 Redis 键）"""
|
||
user = create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
|
||
assert token and len(token) == 64
|
||
assert token in auth_middleware._token_store
|
||
assert json.loads(auth_middleware._token_store[token]) == {
|
||
"user_id": user.id,
|
||
"entity_id": 1,
|
||
}
|
||
# token 只能从进程内存储解析（Redis 通道关闭）
|
||
assert auth_middleware._resolve_token_data(token) == {"user_id": user.id, "entity_id": 1}
|
||
assert auth_middleware._resolve_user_id(token) == user.id
|
||
|
||
|
||
def test_process_store_token_still_authenticates(db, client):
|
||
"""降级分支不影响鉴权语义：进程内 token 经 require_auth 正常通过（/me 200）"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
|
||
resp = client.get("/api/cma/auth/me", headers=auth_header(token))
|
||
assert resp.status_code == 200, resp.text
|
||
body = resp.json()
|
||
assert body["username"] == "testadmin"
|
||
assert body["role"] == "ceo"
|
||
assert body["entity_id"] == 1
|
||
|
||
# 无效 token 仍按原语义 401（隔离未放宽鉴权）
|
||
assert client.get("/api/cma/auth/me", headers=auth_header("0" * 64)).status_code == 401
|
||
|
||
|
||
def test_create_token_writes_process_store_not_redis(db):
|
||
"""create_token() 直接调用：写入进程内存储、解析一致、Redis 侧无客户端可写"""
|
||
user = create_test_user(db, username="redis_iso_user")
|
||
|
||
token = auth_middleware.create_token(user.id, 1)
|
||
assert token in auth_middleware._token_store
|
||
assert auth_middleware.get_token_entity_id(token) == 1
|
||
assert auth_middleware._redis is None, "测试期不应存在生产 Redis 客户端实例"
|
||
|
||
|
||
def test_token_store_is_process_local(db, client):
|
||
"""token 仅存于本进程：模拟另一个进程/worker 的独立存储看不到它（跨 worker 不可见即无共享 Redis）"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
|
||
fresh_store: dict = {} # 模拟另一个 worker 的独立进程内存储
|
||
assert token not in fresh_store
|
||
assert token in auth_middleware._token_store # 本进程可见
|
||
# 有 Redis 通道时 token 会在生产库跨 worker 共享 —— 这正是被本单关闭的路径
|
||
assert auth_middleware._redis is None
|
||
|
||
|
||
def test_production_redis_boundary_invariant():
|
||
"""不变量：只要 _redis_available 为 False，create_token 不可能触达 Redis（唯一写入分支）"""
|
||
assert auth_middleware._redis_available is False
|
||
# 观察既有代码结构：写入分支受 _redis_available 单点控制（本断言防未来改动绕开开关）
|
||
import inspect
|
||
|
||
src = inspect.getsource(auth_middleware.create_token)
|
||
assert "if _redis_available:" in src
|
||
assert src.count("_redis.setex(") == 1
|
||
|
||
|
||
if __name__ == "__main__": # pragma: no cover
|
||
raise SystemExit(pytest.main([__file__, "-q"]))
|