1461a8d113
按 2026-11-02 需求补齐计划交付后端能力： - 角色授权：权限目录化，接口统一准入校验，未登记默认拒绝； 管理员判断改为按数据库角色重新判定，防止权限码持有者绕过 - 系统监控：单进程采集资源、请求与异常指标，分钟级聚合， 阈值配置化；敏感配置增加 repr 保护，Redis 密码 URL 转义 - 试用期计提：按回款日判定并写入快照，年度结清与调整单联动； 人员模型补试用期、离职日期与工资版本历史字段 - 计提模板：公式骨架受控，页面维护名称与参数绑定 - T+ 采集：自动化配置表与进度查询 内置角色禁停用保护对齐模块注释；核算规则版本提为常量并 补充批量核算失败堆栈日志。 影响范围：后端全部模块、数据库迁移 V22-V28 验证：uv run ruff check 通过；922 个单元测试全部通过； 集成测试因本机 MySQL 数据卷被其他容器锁定暂未运行
95 lines
4.0 KiB
Python
95 lines
4.0 KiB
Python
"""自动采集保存值由独立维护读取，关闭不破坏正在运行的任务。"""
|
|
from datetime import datetime
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.config import Settings
|
|
from app.core.exceptions import ConflictError
|
|
from app.models import OperationLog, TPlusAutomationConfig, TPlusSyncBatch, User
|
|
from app.schemas.tplus import TPlusAutomationUpdate
|
|
from app.services.tplus import automation, automation_config, task_service
|
|
from tests.unit.test_tplus_workbench import work_session as work_session
|
|
|
|
|
|
@pytest.fixture
|
|
def source_settings(monkeypatch):
|
|
settings = Settings(_env_file=None, app_secret_key="x" * 32, t_plus_enabled=True,
|
|
t_plus_database="test_source", t_plus_user="reader", t_plus_odbc_driver="unit",
|
|
t_plus_automation_enabled=True)
|
|
monkeypatch.setattr(task_service, "get_settings", lambda: settings)
|
|
return settings
|
|
|
|
|
|
def update(s, enabled, version=0):
|
|
actor = task_service.TaskActor(s.get(User, 1))
|
|
return automation_config.save(s, TPlusAutomationUpdate(
|
|
enabled=enabled, expected_version=version), actor)
|
|
|
|
|
|
def test_初次沿用环境且页面关闭后独立维护不再排队(work_session, source_settings):
|
|
s = work_session
|
|
initial = automation_config.configuration(s, source_settings)
|
|
assert initial == {"enabled": True, "version": 0, "editable": True,
|
|
"configuration_origin": "ENVIRONMENT"}
|
|
result = update(s, False)
|
|
assert result["enabled"] is False and result["version"] == 1
|
|
with Session(s.get_bind()) as other:
|
|
assert automation.tick(other, settings=source_settings) == {"status": "DISABLED", "queued": False}
|
|
assert s.scalar(select(TPlusSyncBatch)) is None
|
|
assert s.scalar(select(OperationLog).where(OperationLog.action == "修改自动采集")) is not None
|
|
|
|
|
|
def test_保存开启覆盖环境关闭且同周不重复(work_session, source_settings):
|
|
settings = source_settings.model_copy(update={"t_plus_automation_enabled": False})
|
|
update(work_session, True)
|
|
with Session(work_session.get_bind()) as other:
|
|
first = automation.tick(other, settings=settings, now=datetime(2026, 11, 2, 3))
|
|
assert first["queued"]
|
|
assert not automation.tick(other, settings=settings, now=datetime(2026, 11, 2, 4))["queued"]
|
|
|
|
|
|
def test_关闭不终止活动批次且仍可手动排队(work_session, source_settings):
|
|
s = work_session
|
|
s.add(TPlusSyncBatch(batch_no="running", source_database="test_source", sync_mode="FULL",
|
|
status="RUNNING", entity_types_json=["personnel"]))
|
|
s.commit()
|
|
update(s, False)
|
|
assert s.scalar(select(TPlusSyncBatch.status)) == "RUNNING"
|
|
assert automation.tick(s, settings=source_settings)["status"] == "DISABLED"
|
|
from uuid import uuid4
|
|
|
|
from app.schemas.tplus import TPlusCreateIn
|
|
|
|
queued = task_service.create_task(s, TPlusCreateIn(request_key=uuid4(), entity_types=["personnel"]))
|
|
assert queued.status == "QUEUED"
|
|
|
|
|
|
def test_旧版本不能覆盖新的开关(work_session, source_settings):
|
|
s = work_session
|
|
update(s, False)
|
|
with pytest.raises(ConflictError):
|
|
update(s, True, 0)
|
|
assert not bool(s.scalar(select(TPlusAutomationConfig.enabled)))
|
|
result = update(s, True, 1)
|
|
assert result["enabled"] and result["version"] == 2
|
|
|
|
|
|
def test_不同源的开关不互相覆盖(work_session, source_settings):
|
|
update(work_session, False)
|
|
other = source_settings.model_copy(update={"t_plus_database": "other_source"})
|
|
assert automation_config.configuration(work_session, other)["enabled"] is True
|
|
|
|
|
|
def test_没有采集权限不能保存(work_session, source_settings, monkeypatch):
|
|
from app.core.exceptions import ForbiddenError
|
|
|
|
def denied(*args):
|
|
raise ForbiddenError()
|
|
|
|
monkeypatch.setattr(automation_config, "require_permission", denied)
|
|
with pytest.raises(ForbiddenError):
|
|
update(work_session, False)
|
|
assert work_session.scalar(select(TPlusAutomationConfig)) is None
|