1461a8d113
按 2026-11-02 需求补齐计划交付后端能力： - 角色授权：权限目录化，接口统一准入校验，未登记默认拒绝； 管理员判断改为按数据库角色重新判定，防止权限码持有者绕过 - 系统监控：单进程采集资源、请求与异常指标，分钟级聚合， 阈值配置化；敏感配置增加 repr 保护，Redis 密码 URL 转义 - 试用期计提：按回款日判定并写入快照，年度结清与调整单联动； 人员模型补试用期、离职日期与工资版本历史字段 - 计提模板：公式骨架受控，页面维护名称与参数绑定 - T+ 采集：自动化配置表与进度查询 内置角色禁停用保护对齐模块注释；核算规则版本提为常量并 补充批量核算失败堆栈日志。 影响范围：后端全部模块、数据库迁移 V22-V28 验证：uv run ruff check 通过；922 个单元测试全部通过； 集成测试因本机 MySQL 数据卷被其他容器锁定暂未运行
75 lines
2.9 KiB
Python
75 lines
2.9 KiB
Python
"""首页采集状态来自真实批次，关闭读取不抹掉历史成功记录。"""
|
|
|
|
from datetime import datetime
|
|
|
|
import pytest
|
|
from sqlalchemy import Column, DateTime, Integer, MetaData, String, Table, create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.config import Settings
|
|
from app.services import dashboard_service
|
|
|
|
|
|
@pytest.fixture
|
|
def sync_database(monkeypatch):
|
|
engine = create_engine("sqlite://")
|
|
metadata = MetaData()
|
|
# 只建 V14 已有列，证明读取状态不要求业务库先升级 V20。
|
|
table = Table(
|
|
"tplus_sync_batches", metadata,
|
|
Column("id", Integer, primary_key=True),
|
|
Column("sync_mode", String(16)),
|
|
Column("status", String(16)),
|
|
Column("finished_at", DateTime),
|
|
)
|
|
metadata.create_all(engine)
|
|
settings = Settings(_env_file=None, app_secret_key="x" * 32)
|
|
monkeypatch.setattr(dashboard_service, "get_settings", lambda: settings, raising=False)
|
|
with Session(engine) as session:
|
|
yield session, table, settings
|
|
engine.dispose()
|
|
|
|
|
|
def test_关闭采集仍显示历史成功且不将旧失败冒充当前失败(sync_database):
|
|
session, table, _ = sync_database
|
|
finished = datetime(2026, 10, 30, 16, 52, 3)
|
|
session.execute(table.insert(), [
|
|
{"id": 1, "sync_mode": "FULL", "status": "FAILED", "finished_at": finished},
|
|
{"id": 2, "sync_mode": "FULL", "status": "SUCCEEDED", "finished_at": finished},
|
|
{"id": 3, "sync_mode": "AUDIT", "status": "FAILED", "finished_at": finished},
|
|
])
|
|
result = dashboard_service._sync_status(session)
|
|
assert result["available"] is True
|
|
assert result["enabled"] is False and result["automation_enabled"] is False
|
|
assert result["latest_status"] == "SUCCEEDED"
|
|
assert result["last_success_at"] == finished
|
|
assert result["historical_failure_total"] == 1
|
|
|
|
|
|
@pytest.mark.parametrize("status", ["FAILED", "PARTIAL", "QUEUED", "RUNNING"])
|
|
def test_首页如实返回最近采集状态(sync_database, status):
|
|
session, table, settings = sync_database
|
|
settings.t_plus_enabled = True
|
|
session.execute(table.insert(), {"id": 1, "sync_mode": "FULL", "status": status})
|
|
result = dashboard_service._sync_status(session)
|
|
assert result["latest_status"] == status and result["enabled"] is True
|
|
assert result["last_success_at"] is None
|
|
|
|
|
|
def test_无记录不显示已采集(sync_database):
|
|
session, _, _ = sync_database
|
|
result = dashboard_service._sync_status(session)
|
|
assert result["available"] is True
|
|
assert result["latest_status"] is None and result["last_success_at"] is None
|
|
|
|
|
|
def test_未建同步表时明确返回不可查询():
|
|
engine = create_engine("sqlite://")
|
|
try:
|
|
with Session(engine) as session:
|
|
result = dashboard_service._sync_status(session)
|
|
assert result["available"] is False
|
|
assert result["last_success_at"] is None
|
|
finally:
|
|
engine.dispose()
|