test(cma): force production Redis off during tests (P1 token leak isolation)

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:<hex>", 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"""