1461a8d113
按 2026-11-02 需求补齐计划交付后端能力： - 角色授权：权限目录化，接口统一准入校验，未登记默认拒绝； 管理员判断改为按数据库角色重新判定，防止权限码持有者绕过 - 系统监控：单进程采集资源、请求与异常指标，分钟级聚合， 阈值配置化；敏感配置增加 repr 保护，Redis 密码 URL 转义 - 试用期计提：按回款日判定并写入快照，年度结清与调整单联动； 人员模型补试用期、离职日期与工资版本历史字段 - 计提模板：公式骨架受控，页面维护名称与参数绑定 - T+ 采集：自动化配置表与进度查询 内置角色禁停用保护对齐模块注释；核算规则版本提为常量并 补充批量核算失败堆栈日志。 影响范围：后端全部模块、数据库迁移 V22-V28 验证：uv run ruff check 通过；922 个单元测试全部通过； 集成测试因本机 MySQL 数据卷被其他容器锁定暂未运行
76 lines
2.5 KiB
Python
76 lines
2.5 KiB
Python
"""测试共用夹具。"""
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
|
||
|
||
@pytest.fixture
|
||
def tplus_session(monkeypatch):
|
||
"""T+ 管理逻辑的真实内存事务；MySQL 并发锁另做集成验证。"""
|
||
from sqlalchemy import DefaultClause, Integer, MetaData, create_engine, text
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.config import Settings
|
||
from app.models import TPlusSyncBatch, TPlusSyncError, TPlusSyncRecord
|
||
|
||
engine = create_engine("sqlite:///:memory:")
|
||
metadata = MetaData()
|
||
for model in (TPlusSyncBatch, TPlusSyncError, TPlusSyncRecord):
|
||
table = model.__table__.to_metadata(metadata)
|
||
table.c.id.type = Integer()
|
||
table.c.update_time.server_default = DefaultClause(text("CURRENT_TIMESTAMP"))
|
||
metadata.create_all(engine)
|
||
settings = Settings(
|
||
_env_file=None,
|
||
app_secret_key="x" * 32,
|
||
t_plus_enabled=True,
|
||
t_plus_database="unit_tplus",
|
||
t_plus_user="reader",
|
||
t_plus_odbc_driver="unit",
|
||
)
|
||
monkeypatch.setattr("app.services.tplus.task_service.get_settings", lambda: settings)
|
||
with Session(engine, expire_on_commit=False) as session:
|
||
yield session
|
||
engine.dispose()
|
||
|
||
|
||
@pytest.fixture(scope="session")
|
||
def client() -> TestClient:
|
||
"""构造应用测试客户端（会话级复用，避免重复构建应用）。
|
||
|
||
注意：测试环境未必有 MySQL/Redis 在跑，健康检查会如实返回 false，
|
||
因此断言应聚焦于响应结构而非依赖的连通性。
|
||
"""
|
||
from unittest.mock import patch
|
||
|
||
from app.main import app
|
||
|
||
# 测试客户端不启动后台采集，避免单测通过环境配置写入现有业务库。
|
||
with (
|
||
patch.object(app.state.monitor, "start"),
|
||
patch.object(app.state.monitor, "stop"),
|
||
TestClient(app) as test_client,
|
||
):
|
||
yield test_client
|
||
|
||
|
||
@pytest.fixture
|
||
def fake_redis(monkeypatch: pytest.MonkeyPatch):
|
||
"""把 Redis 客户端替换为进程内假实现。
|
||
|
||
验证码的 GETDEL、限流的 INCR/EXPIRE/TTL、黑名单的 SETEX——都只用到
|
||
基础字符串命令，fakeredis 能完整覆盖，因此这些逻辑无需起容器即可测试，
|
||
单测得以保持秒级。
|
||
|
||
替换的是 `app.core.redis_client` 的模块级单例，各 service 通过
|
||
`get_redis()` 取用，故一处替换即全局生效。
|
||
"""
|
||
import fakeredis
|
||
|
||
from app.core import redis_client
|
||
|
||
fake = fakeredis.FakeRedis(decode_responses=True)
|
||
monkeypatch.setattr(redis_client, "_client", fake)
|
||
yield fake
|
||
fake.close()
|