316 lines
12 KiB
Python
316 lines
12 KiB
Python
"""
|
||
管理会计OS 测试配置
|
||
|
||
使用 SQLite 内存数据库进行测试，避免依赖外部 MySQL。
|
||
测试前自动建表，测试后自动清理。
|
||
|
||
重要：此文件在pytest收集测试时最先加载，确保环境变量在app模块导入前注入。
|
||
"""
|
||
import os
|
||
|
||
# 必须在任何app模块导入之前设置环境变量（通过 pytest.ini 的 python_files 保证加载顺序）
|
||
os.environ.setdefault("CMA_DB_USER", "test")
|
||
os.environ.setdefault("CMA_DB_PASS", "test")
|
||
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
|
||
from sqlalchemy import create_engine
|
||
from sqlalchemy.orm import sessionmaker, Session
|
||
from sqlalchemy.pool import StaticPool
|
||
|
||
# 在 import app 模块之前就先覆盖掉 database.py 的 engine
|
||
# 方式：直接 monkey-patch database 模块
|
||
from app import database as db_module
|
||
from app.database import Base
|
||
|
||
# SQLite 内存引擎
|
||
TEST_ENGINE = create_engine(
|
||
"sqlite:///:memory:",
|
||
connect_args={"check_same_thread": False},
|
||
poolclass=StaticPool,
|
||
)
|
||
TEST_SESSION_LOCAL = sessionmaker(autocommit=False, autoflush=False, bind=TEST_ENGINE)
|
||
|
||
|
||
# MySQL-only 的 date_format() 在 SQLite 下注册等价实现（仅测试库）
|
||
# 生产用 MySQL 原生函数；此处仅为让测试能跑通 expenses 月度累计校验/stats 统计
|
||
def _sqlite_date_format(dt_val, fmt):
|
||
if dt_val is None:
|
||
return None
|
||
import datetime as _dt
|
||
if isinstance(dt_val, str):
|
||
for f in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d", "%Y-%m"):
|
||
try:
|
||
dt_val = _dt.datetime.strptime(str(dt_val)[:19], f)
|
||
break
|
||
except ValueError:
|
||
continue
|
||
else:
|
||
return None
|
||
if isinstance(dt_val, _dt.datetime):
|
||
d = dt_val
|
||
elif isinstance(dt_val, _dt.date):
|
||
d = _dt.datetime(dt_val.year, dt_val.month, dt_val.day)
|
||
else:
|
||
return None
|
||
return {
|
||
"%Y": f"{d.year:04d}",
|
||
"%Y-%m": f"{d.year:04d}-{d.month:02d}",
|
||
"%Y-%m-%d": f"{d.year:04d}-{d.month:02d}-{d.day:02d}",
|
||
}.get(fmt)
|
||
|
||
|
||
from sqlalchemy import event # noqa: E402
|
||
event.listen(TEST_ENGINE, "connect", lambda dbapi_conn, rec: dbapi_conn.create_function("date_format", 2, _sqlite_date_format))
|
||
|
||
# 替换 database 模块的全局引擎
|
||
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,
|
||
ActionPlan, OrgNode, StrategicMapVersion,
|
||
)
|
||
import hashlib
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def setup_db():
|
||
"""每个测试函数自动初始化和清理数据库"""
|
||
from app.utils import cache as cache_util
|
||
cache_util.delete("ai") # 清AI分析缓存，防测试间Redis污染（dashboard-analysis缓存全局共享）
|
||
Base.metadata.create_all(bind=TEST_ENGINE)
|
||
yield
|
||
Base.metadata.drop_all(bind=TEST_ENGINE)
|
||
cache_util.delete("ai")
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _disable_ai_suggestion_push(monkeypatch):
|
||
"""R1触达修复(2026-08-31): 测试库把企微推送替换为 no-op，防测试建议推真实企微群
|
||
|
||
生产环境真实推送（8800 relay）；测试只验证推送逻辑（决策类推/预警不推/幂等）不打真实企微。
|
||
测试类如需断言推送内容，可自行 monkeypatch.setattr 覆盖本 no-op。
|
||
"""
|
||
from app.api import ai_analysis
|
||
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"""
|
||
session = TEST_SESSION_LOCAL()
|
||
try:
|
||
yield session
|
||
finally:
|
||
session.close()
|
||
|
||
|
||
@pytest.fixture
|
||
def client(db) -> Generator[TestClient, None, None]:
|
||
"""提供测试 HTTP 客户端"""
|
||
from app.main import app
|
||
from app.models import Entity
|
||
|
||
# 账套模式：确保测试库存在 entity_id=1 的active实体
|
||
ent = db.query(Entity).filter(Entity.id == 1).first()
|
||
if not ent:
|
||
db.add(Entity(id=1, name="测试企业", short_name="测试", status="active"))
|
||
db.commit()
|
||
|
||
# 重写依赖，使用测试数据库
|
||
app.dependency_overrides[db_module.get_db] = lambda: db
|
||
|
||
with TestClient(app) as c:
|
||
yield c
|
||
|
||
app.dependency_overrides.clear()
|
||
|
||
|
||
# ── 测试数据工厂 ──
|
||
|
||
# 测试夹具口令（仅内存 SQLite 测试库使用；不是生产默认口令——生产凭据只存 600 密钥文件
|
||
# .secrets/cma-admin.env）。可用 $CMA_TEST_USER_PASSWORD 覆盖；用例一律从这里导入，禁止写字面量。
|
||
TEST_USER_PASSWORD = os.environ.get("CMA_TEST_USER_PASSWORD", "test-fixture-pass")
|
||
|
||
def create_test_user(db: Session, **kwargs) -> User:
|
||
"""创建测试用户"""
|
||
defaults = {
|
||
"username": "testadmin",
|
||
"password_hash": hashlib.sha256(TEST_USER_PASSWORD.encode()).hexdigest(),
|
||
"name": "测试管理员",
|
||
"role": "ceo",
|
||
}
|
||
defaults.update(kwargs)
|
||
user = User(**defaults)
|
||
db.add(user)
|
||
db.commit()
|
||
db.refresh(user)
|
||
return user
|
||
|
||
|
||
def get_token_for_user(client: TestClient, username: str = "testadmin", password: str = TEST_USER_PASSWORD) -> str:
|
||
"""获取测试用户的token（账套模式：需entity_id）"""
|
||
resp = client.post("/api/cma/auth/login", json={
|
||
"username": username,
|
||
"password": password,
|
||
"entity_id": 1,
|
||
})
|
||
if resp.status_code != 200:
|
||
raise RuntimeError(f"登录失败: {resp.status_code} {resp.text[:300]}")
|
||
data = resp.json()
|
||
return data.get("token") or data.get("access_token")
|
||
|
||
|
||
def auth_header(token: str) -> dict:
|
||
return {"Authorization": f"Bearer {token}"}
|
||
|
||
|
||
def create_test_kpi(db: Session, **kwargs) -> KPIDefinition:
|
||
"""创建测试KPI"""
|
||
defaults = {
|
||
"kpi_code": "TEST_001",
|
||
"kpi_name": "测试KPI",
|
||
"dimension": "finance",
|
||
"target_value": 100.0,
|
||
"unit": "%",
|
||
"status": "active",
|
||
}
|
||
defaults.update(kwargs)
|
||
kpi = KPIDefinition(**defaults)
|
||
db.add(kpi)
|
||
db.commit()
|
||
db.refresh(kpi)
|
||
return kpi
|
||
|
||
|
||
def create_test_map(db: Session, **kwargs) -> StrategicMap:
|
||
"""创建测试战略地图"""
|
||
defaults = {
|
||
"title": "测试地图",
|
||
"status": "draft",
|
||
"dimensions": [
|
||
{"key": "finance", "name": "财务维度", "icon": "💰", "color": "#409eff", "objectives": []},
|
||
{"key": "customer", "name": "客户维度", "icon": "🤝", "color": "#67c23a", "objectives": []},
|
||
],
|
||
"canvas_data": {"connections": []},
|
||
}
|
||
defaults.update(kwargs)
|
||
m = StrategicMap(**defaults)
|
||
db.add(m)
|
||
db.commit()
|
||
db.refresh(m)
|
||
return m
|