diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index 39c4fe21..56b32c9d 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -15,6 +15,54 @@ os.environ.setdefault("CMA_DB_HOST", "localhost") os.environ.setdefault("CMA_DB_PORT", "3306") os.environ.setdefault("CMA_DB_NAME", "test") +# ── 测试期生产 Redis 隔离(project-20260918-P1-test-redis-token-fix)──────────── +# 机制(详见任务单/报告): +# app.auth_middleware 在 import 时构造 redis 客户端(127.0.0.1:6379 db=1 = 生产库), +# ping 成功即 _redis_available=True → create_token() 走 setex("cma:token:", 86400) +# 把测试签发的 token 写进生产 Redis;而 require_auth() 只按 token 里的 user_id 反查用户、 +# 不比对口令,测试库首个用户 id=1 与生产 users.id=1 撞号 —— 每跑一轮套件都会在生产 Redis +# 留下 24h 有效、可直接当生产会话使用的 token。 +# 处置(只改测试侧,不动 app/auth_middleware.py 的任何业务逻辑): +# ① 在导入任何 app 模块之前,把 redis.Redis 构造器替换成阻断器(计数 + 抛错)—— +# app 侧既有的 `except Exception → _redis=None / _redis_available=False` 降级分支 +# 自动生效,测试期对生产 Redis **零连接**(连 TCP 探测都不发生); +# ② 显式强制 _redis_available=False(进程内 _token_store 兜底,用例语义不变); +# ③ app.utils.cache 同样硬编码 db=1(测试夹具会调 delete("ai")),一并关闭, +# 使整轮测试对生产 Redis 零写命令; +# ④ pytest_sessionfinish 落盘隔离审计 JSON(构造尝试次数 / 通道状态)作为可复核证据。 +_REDIS_GUARD = {"constructor_attempts": 0, "callers": []} + + +class _BlockedProductionRedis: + """测试期阻断器:生产 Redis 客户端构造被计数并拒绝。 + + 抛出的异常与真实连接失败同形(Exception),由 app 侧既有 except 分支兜底降级, + 因此不会改变任何鉴权/缓存语义,只保证测试进程不碰生产 Redis。 + """ + + def __init__(self, *args, **kwargs): + import traceback + + _REDIS_GUARD["constructor_attempts"] += 1 + frames = [f for f in traceback.extract_stack()[:-1] if "conftest" not in f.filename] + if frames: + _REDIS_GUARD["callers"].append(f"{frames[-1].filename}:{frames[-1].lineno}") + raise RuntimeError( + "test-mode: production Redis (127.0.0.1:6379 db=1) blocked by backend/tests/conftest.py" + ) + + +def _install_production_redis_guard() -> None: + """替换 redis.Redis 构造器(进程级,覆盖整轮 pytest 会话)。""" + try: + import redis as redis_lib + except Exception: # redis 未安装:app 侧本来就会降级 + return + redis_lib.Redis = _BlockedProductionRedis + + +_install_production_redis_guard() + import pytest from typing import Generator from fastapi.testclient import TestClient @@ -71,6 +119,17 @@ event.listen(TEST_ENGINE, "connect", lambda dbapi_conn, rec: dbapi_conn.create_f db_module._engine = TEST_ENGINE db_module._SessionLocal = TEST_SESSION_LOCAL +# ── 测试期强制关闭 Redis 通道(生产 Redis 零写入硬隔离)────────────────────── +# 不修改 app/auth_middleware.py:只置其模块级开关,走既有进程内 _token_store 降级分支。 +# 导入点上方的 _BlockedProductionRedis 已生效 → 这两次 import 不会连生产 Redis。 +import app.auth_middleware as auth_middleware # noqa: E402 +from app.utils import cache as cache_util # noqa: E402 + +auth_middleware._redis_available = False +auth_middleware._redis = None # 导入期即被阻断,显式置空防后续被复活 +cache_util._available = False +cache_util._client = None + # 然后导入models(它们依赖Base) from app.models import ( User, StrategicMap, KPIDefinition, KPIValue, KPIAlert, @@ -101,6 +160,54 @@ def _disable_ai_suggestion_push(monkeypatch): monkeypatch.setattr(ai_analysis, "_push_decision_suggestion", lambda s: True) +@pytest.fixture(autouse=True) +def _enforce_production_redis_isolation(): + """每个用例前后强制断言:测试期 Redis 通道始终关闭(防用例把它翻回 True)。 + + project-20260918-P1-test-redis-token-fix:生产 Redis 写入的唯一入口是 + auth_middleware.create_token() 的 `if _redis_available` 分支 —— 用例级复检保证 + 「一轮全量测试 = 零生产 Redis 写命令」,任何把开关翻回 True 的用例都会当场暴露。 + """ + auth_middleware._redis_available = False + cache_util._available = False + yield + assert auth_middleware._redis_available is False, "测试期禁止开启生产 Redis token 通道" + assert cache_util._available is False, "测试期禁止开启生产 Redis 缓存通道" + + +@pytest.fixture +def production_redis_isolation_state() -> dict: + """暴露测试期 Redis 隔离状态(构造阻断计数 + 通道开关),供断言用例使用。""" + return _REDIS_GUARD + + +def pytest_sessionfinish(session, exitstatus): + """落盘本轮测试的 Redis 隔离审计(零生产 Redis 连接/写入的可复核证据)。""" + import json + import datetime as _dt + + audit = { + "generated_at": _dt.datetime.now().isoformat(timespec="seconds"), + "task": "project-20260918-P1-test-redis-token-fix", + "redis_constructor_attempts": _REDIS_GUARD["constructor_attempts"], + "redis_constructor_blocked": _REDIS_GUARD["constructor_attempts"], + "blocked_callsites": _REDIS_GUARD["callers"], + "auth_middleware_redis_available": bool(auth_middleware._redis_available), + "auth_middleware_redis_client_is_none": auth_middleware._redis is None, + "cache_available": bool(cache_util._available), + "cache_client_is_none": cache_util._client is None, + "token_store_size": len(auth_middleware._token_store), + "pytest_exitstatus": int(exitstatus), + "violation": bool(auth_middleware._redis_available or cache_util._available), + } + path = os.environ.get("CMA_TEST_REDIS_AUDIT", "/tmp/cma-test-redis-isolation-audit.json") + try: + with open(path, "w", encoding="utf-8") as fh: + json.dump(audit, fh, ensure_ascii=False, indent=2) + except Exception: + pass + + @pytest.fixture def db() -> Generator[Session, None, None]: """提供数据库 session""" diff --git a/backend/tests/test_redis_token_isolation_20260918.py b/backend/tests/test_redis_token_isolation_20260918.py new file mode 100644 index 00000000..0c21f8b5 --- /dev/null +++ b/backend/tests/test_redis_token_isolation_20260918.py @@ -0,0 +1,109 @@ +"""测试期生产 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"]))