1461a8d113
按 2026-11-02 需求补齐计划交付后端能力： - 角色授权：权限目录化，接口统一准入校验，未登记默认拒绝； 管理员判断改为按数据库角色重新判定，防止权限码持有者绕过 - 系统监控：单进程采集资源、请求与异常指标，分钟级聚合， 阈值配置化；敏感配置增加 repr 保护，Redis 密码 URL 转义 - 试用期计提：按回款日判定并写入快照，年度结清与调整单联动； 人员模型补试用期、离职日期与工资版本历史字段 - 计提模板：公式骨架受控，页面维护名称与参数绑定 - T+ 采集：自动化配置表与进度查询 内置角色禁停用保护对齐模块注释；核算规则版本提为常量并 补充批量核算失败堆栈日志。 影响范围：后端全部模块、数据库迁移 V22-V28 验证：uv run ruff check 通过；922 个单元测试全部通过； 集成测试因本机 MySQL 数据卷被其他容器锁定暂未运行
408 lines
18 KiB
Python
408 lines
18 KiB
Python
"""系统管理集成测试共用夹具（需要 MySQL 与 Redis）。
|
||
|
||
运行前置：``docker compose -f docker/docker-compose.dev.yml up -d``
|
||
|
||
数据卫生约定：
|
||
|
||
- 测试账号/角色统一 ``t_`` 前缀；工厂夹具清理自己创建的对象，另有
|
||
autouse 兜底按前缀全量清理（用例经**接口**创建的数据不在工厂名下，
|
||
失败中断时会残留并影响后续用例的 total 断言）
|
||
- M2 基础数据（人员姓名/产品名/公司名/测试岗位编码）同用 ``t_`` 前缀，
|
||
兜底清理按前缀删除；归属留痕对客户/人员 ID 无外键固化、不随主表删除
|
||
消失，必须显式清理，否则残留会影响后续用例对留痕条数的断言
|
||
- 操作日志的清理同时按 username 与 user_id —— 账号被删除后其 ``user_id``
|
||
会被 SET NULL，只能靠冗余的 username 追溯，清理逻辑必须覆盖两种状态
|
||
- M2 业务数据（业务/发货/回款）单据编号由系统生成（YW/SH/HK 前缀 + 当日
|
||
序号），**没有 t_ 前缀可用**，兜底清理按编号前缀全量清（三类单据表目前
|
||
只有测试会写）；外键均为 RESTRICT，删除顺序不能倒，且必须在客户/人员
|
||
清理之前（FK 会挡住删客户/删人员）；评级判定表则恢复种子基线 ——
|
||
评级用例会删改种子构造重叠/相邻场景，恢复保证跨文件幂等
|
||
"""
|
||
|
||
import os
|
||
import uuid
|
||
from decimal import Decimal
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
from sqlalchemy import delete, select, update
|
||
|
||
from app.core.constants import ROLE_ADMIN, ROLE_USER
|
||
from app.core.db import get_session_factory
|
||
from app.core.security import create_access_token, hash_password
|
||
from app.models import (
|
||
AdjustmentOrder,
|
||
ApprovalOrder,
|
||
Business,
|
||
BusinessItem,
|
||
BusinessRole,
|
||
BusinessStatusLog,
|
||
Customer,
|
||
CustomerRatingGrade,
|
||
MonthlySettlement,
|
||
OperationLog,
|
||
OwnershipChangeLog,
|
||
ParamChangeBatch,
|
||
ParamValue,
|
||
PerformanceLedger,
|
||
Personnel,
|
||
PersonnelPlan,
|
||
PersonnelRegion,
|
||
Product,
|
||
Receipt,
|
||
Role,
|
||
SettlementItem,
|
||
Shipment,
|
||
ShipmentItem,
|
||
SystemSetting,
|
||
User,
|
||
UserRole,
|
||
)
|
||
from app.models.customer import OWNER_TYPE_BUSINESS
|
||
|
||
# 评级判定表的种子基线（与 db/migration/V5 一致）：判定按 sort_order 升序
|
||
# 取首个 min_score ≤ 总分的等级，种子铺满 0-100
|
||
_RATING_SEEDS = [
|
||
("A", Decimal("90"), Decimal("100"), 1, "优秀"),
|
||
("B", Decimal("75"), Decimal("89"), 2, "良好"),
|
||
("C", Decimal("60"), Decimal("74"), 3, "合格"),
|
||
("D", Decimal("0"), Decimal("59"), 4, "待改进"),
|
||
]
|
||
|
||
TEST_PASSWORD = "TestPwd@2026"
|
||
# 模块级预计算一次：bcrypt（rounds=12）单次约 0.2–0.4 秒，逐用例重复哈希
|
||
# 会让整个集成套件多花 20 秒以上；这些测试账号全部走直签 JWT，
|
||
# 哈希值不被任何断言消费，复用一个哈希与每次重算等价
|
||
_TEST_PASSWORD_HASH = hash_password(TEST_PASSWORD)
|
||
|
||
|
||
def _purge_test_data() -> None:
|
||
"""按 `t_` 前缀清理测试残留（先日志后账号；角色先断关联再删）。
|
||
|
||
``t\\_%`` 中的下划线做了转义 —— 不转义时 `_` 是 LIKE 的单字符通配，
|
||
会把 tn... 这类无关账号一并卷入。
|
||
"""
|
||
from app.core.config import get_settings
|
||
from tests.isolation import ensure_test_database
|
||
|
||
ensure_test_database(get_settings().mysql_database, os.environ.get("PAYROLL_TEST_DATABASE"))
|
||
factory = get_session_factory()
|
||
ensure_test_database(factory.kw["bind"].url.database, os.environ.get("PAYROLL_TEST_DATABASE"))
|
||
with factory() as session:
|
||
from app.models import AuthorizationSetting, ProbationDeferral, ProbationPayment
|
||
|
||
session.execute(delete(ProbationPayment))
|
||
session.execute(delete(ProbationDeferral))
|
||
|
||
session.execute(update(AuthorizationSetting).values(maker_user_id=None, updated_by=None))
|
||
users = list(
|
||
session.execute(select(User.id, User.username).where(User.username.like("t\\_%"))).all()
|
||
)
|
||
ids = [row[0] for row in users]
|
||
names = [row[1] for row in users]
|
||
if ids:
|
||
session.execute(delete(OperationLog).where(OperationLog.user_id.in_(ids)))
|
||
if names:
|
||
session.execute(delete(OperationLog).where(OperationLog.username.in_(names)))
|
||
session.execute(delete(User).where(User.id.in_(ids)))
|
||
|
||
role_ids = list(
|
||
session.execute(select(Role.id).where(Role.role_code.like("t\\_%"))).scalars()
|
||
)
|
||
if role_ids:
|
||
session.execute(delete(UserRole).where(UserRole.role_id.in_(role_ids)))
|
||
session.execute(delete(Role).where(Role.id.in_(role_ids)))
|
||
|
||
# ---- M3-core：核算台账 / 月结 / 调整单（必须先于回款/业务/人员删除——
|
||
# ledger.receipt_id/personnel_id/settlement_id 外键 RESTRICT 会挡住后面
|
||
# 的单据与人员清理；仅测试写入，全量清）----
|
||
from app.models import PayrollAllocation, PayrollInput, PayrollMonth, TPlusAutomationState
|
||
session.execute(delete(TPlusAutomationState))
|
||
session.execute(delete(PayrollAllocation))
|
||
session.execute(delete(PayrollInput))
|
||
session.execute(delete(PayrollMonth))
|
||
session.execute(delete(PerformanceLedger))
|
||
session.execute(delete(SettlementItem))
|
||
session.execute(delete(MonthlySettlement))
|
||
session.execute(delete(AdjustmentOrder).where(AdjustmentOrder.order_no.like("TZ%")))
|
||
|
||
# ---- M2 业务数据（T11）：业务/发货/回款三类单据 ----
|
||
# 编号系统生成、无 t_ 前缀可用，按业务编号前缀全量清（三类单据表目前
|
||
# 只有测试会写）。外键均为 RESTRICT，删除顺序不能倒：回款 → 发货
|
||
# 明细 → 发货单 → 业务明细 → 状态留痕 → 业务维度归属留痕（无外键，
|
||
# 须显式清）→ 业务主表；且必须在下方客户/人员清理之前，否则 FK 挡住
|
||
# 删客户/删人员（businesses.customer_id / .owner_personnel_id RESTRICT）
|
||
business_ids = list(
|
||
session.execute(select(Business.id).where(Business.business_no.like("YW%"))).scalars()
|
||
)
|
||
shipment_ids = list(
|
||
session.execute(select(Shipment.id).where(Shipment.shipment_no.like("SH%"))).scalars()
|
||
)
|
||
if business_ids:
|
||
session.execute(delete(Receipt).where(Receipt.business_id.in_(business_ids)))
|
||
if shipment_ids:
|
||
session.execute(delete(ShipmentItem).where(ShipmentItem.shipment_id.in_(shipment_ids)))
|
||
session.execute(delete(Shipment).where(Shipment.id.in_(shipment_ids)))
|
||
if business_ids:
|
||
session.execute(delete(BusinessItem).where(BusinessItem.business_id.in_(business_ids)))
|
||
session.execute(
|
||
delete(BusinessStatusLog).where(BusinessStatusLog.business_id.in_(business_ids))
|
||
)
|
||
session.execute(
|
||
delete(OwnershipChangeLog).where(
|
||
OwnershipChangeLog.owner_type == OWNER_TYPE_BUSINESS,
|
||
OwnershipChangeLog.business_id.in_(business_ids),
|
||
)
|
||
)
|
||
session.execute(delete(Business).where(Business.id.in_(business_ids)))
|
||
|
||
# 评级判定表恢复种子基线：评级用例会删改种子构造重叠/相邻场景，
|
||
# 兜底恢复保证每个用例开始时都是纯种子（判定表无下游引用，全删全插安全）
|
||
session.execute(delete(CustomerRatingGrade))
|
||
session.add_all(
|
||
CustomerRatingGrade(grade_name=n, min_score=lo, max_score=hi, sort_order=so, remark=rm)
|
||
for n, lo, hi, so, rm in _RATING_SEEDS
|
||
)
|
||
|
||
# ---- M2 基础数据：客户/归属留痕/人员/产品/测试岗位 ----
|
||
# 外键均为 RESTRICT，删除顺序不能倒：留痕无外键先按 customer_id 清，
|
||
# 再删客户（子表）→ 人员（先置空 manager_id 自引用、清子表）→ 产品
|
||
personnel_ids = list(
|
||
session.execute(
|
||
select(Personnel.id).where(Personnel.personnel_name.like("t\\_%"))
|
||
).scalars()
|
||
)
|
||
customer_ids = list(
|
||
session.execute(
|
||
select(Customer.id).where(Customer.company_name.like("t\\_%"))
|
||
).scalars()
|
||
)
|
||
if customer_ids:
|
||
session.execute(
|
||
delete(OwnershipChangeLog).where(OwnershipChangeLog.customer_id.in_(customer_ids))
|
||
)
|
||
session.execute(delete(Customer).where(Customer.id.in_(customer_ids)))
|
||
# 客户被物理删除后其 customer_id 已无从收集，按操作人用户名兜底一条
|
||
session.execute(
|
||
delete(OwnershipChangeLog).where(OwnershipChangeLog.operator_name.like("t\\_%"))
|
||
)
|
||
if personnel_ids:
|
||
session.execute(
|
||
update(Personnel)
|
||
.where(Personnel.manager_id.in_(personnel_ids))
|
||
.values(manager_id=None)
|
||
)
|
||
session.execute(
|
||
delete(PersonnelRegion).where(PersonnelRegion.personnel_id.in_(personnel_ids))
|
||
)
|
||
session.execute(
|
||
delete(PersonnelPlan).where(PersonnelPlan.personnel_id.in_(personnel_ids))
|
||
)
|
||
session.execute(delete(Personnel).where(Personnel.id.in_(personnel_ids)))
|
||
session.execute(delete(Product).where(Product.product_name.like("t\\_%")))
|
||
# 人员已先删（business_role_id 外键 RESTRICT），直建的测试岗位此时可安全删
|
||
session.execute(delete(BusinessRole).where(BusinessRole.role_code.like("t\\_%")))
|
||
|
||
# ---- M3：审批单 / 二次分配 / 参数开关 / 系统设置基线 ----
|
||
# 审批单无 t_ 前缀可用，按 AP 编号前缀全量清（steps 外键 CASCADE 随删；
|
||
# ref_id 固化无外键，业务已在上一步删除，无悬挂影响）
|
||
session.execute(delete(ApprovalOrder).where(ApprovalOrder.order_no.like("AP%")))
|
||
# 开关参数恢复种子基线：测试提交的变更批次条目 id 均大于种子终值 28
|
||
# （V3 十九条 + V6 两条 + V7 调整单一条 + V8 经理基数开关一条；V8 另
|
||
# 插入两个审批开关的 10-19 生效行 id 23/24，V10 插入预警规则参数
|
||
# id 26/27/28，均属种子），删新条目与新批次后解析回退到种子值
|
||
stray_values = list(session.execute(select(ParamValue).where(ParamValue.id > 28)).scalars())
|
||
if stray_values:
|
||
stray_batches = {v.batch_id for v in stray_values}
|
||
session.execute(delete(ParamValue).where(ParamValue.id > 28))
|
||
session.execute(delete(ParamChangeBatch).where(ParamChangeBatch.id.in_(stray_batches)))
|
||
# 系统设置恢复种子单行（test_settings 会改公司名/传 logo——残留会让
|
||
# 后续 e2e 的品牌断言在污染状态下运行）；uploads 下的测试 logo 一并删
|
||
session.execute(
|
||
update(SystemSetting)
|
||
.where(SystemSetting.id == 1)
|
||
.values(company_name="衍水集团", logo_path=None)
|
||
)
|
||
session.commit()
|
||
logo_dir = get_settings().upload_dir / "logo"
|
||
expected_uploads = Path(__file__).resolve().parents[3] / ".ai" / "isolated-uploads" / get_settings().mysql_database
|
||
if not logo_dir.resolve().is_relative_to(expected_uploads.resolve()):
|
||
raise RuntimeError("测试文件清理目标不在本次隔离目录，拒绝清理")
|
||
if logo_dir.is_dir():
|
||
for f in logo_dir.iterdir():
|
||
f.unlink(missing_ok=True)
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _purge_after_each_test():
|
||
"""每个集成用例结束后兜底清理，避免残留影响后续用例的计数断言。"""
|
||
from app.core.config import get_settings
|
||
from tests.isolation import ensure_test_database
|
||
|
||
ensure_test_database(get_settings().mysql_database, os.environ.get("PAYROLL_TEST_DATABASE"))
|
||
ensure_test_database(get_session_factory().kw["bind"].url.database, os.environ.get("PAYROLL_TEST_DATABASE"))
|
||
yield
|
||
_purge_test_data()
|
||
|
||
|
||
@pytest.fixture
|
||
def make_user():
|
||
"""独立测试账号工厂：``make_user(["admin"])`` → dict（id/username/password）。
|
||
|
||
不复用真实 admin —— 测试会改密码、提 token_version、删除账号，
|
||
污染真实账号会让后续手工验证产生困惑。
|
||
"""
|
||
|
||
def _factory(role_codes: list[str] | None = None, username: str | None = None) -> dict:
|
||
codes = role_codes if role_codes is not None else [ROLE_USER]
|
||
name = username or f"t_{uuid.uuid4().hex[:8]}"
|
||
factory = get_session_factory()
|
||
with factory() as session:
|
||
roles = list(session.execute(select(Role).where(Role.role_code.in_(codes))).scalars())
|
||
assert len(roles) == len(codes), f"角色不存在：{codes}"
|
||
user = User(
|
||
username=name,
|
||
password=_TEST_PASSWORD_HASH,
|
||
real_name="测试账号",
|
||
status=1,
|
||
)
|
||
user.user_roles = [UserRole(role_id=role.id) for role in roles]
|
||
session.add(user)
|
||
session.commit()
|
||
return {"id": user.id, "username": name, "password": TEST_PASSWORD}
|
||
|
||
return _factory
|
||
|
||
|
||
@pytest.fixture
|
||
def auth_headers():
|
||
"""令牌直签工厂：``auth_headers(user)`` → Authorization 请求头。
|
||
|
||
绕开验证码与登录接口 —— 令牌只承载身份，授权判定以数据库中的角色为准
|
||
（`require_admin` 每次查库），直签不影响测试有效性，省去清理 Redis
|
||
限流键与黑名单。令牌版本变化（如重置密码、禁用）后旧令牌会失效，
|
||
再次调用本工厂可换新令牌。
|
||
"""
|
||
|
||
def _factory(user: dict) -> dict[str, str]:
|
||
factory = get_session_factory()
|
||
with factory() as session:
|
||
db_user = session.get(User, user["id"])
|
||
assert db_user is not None, "账号已被删除，无法签发令牌"
|
||
roles = [link.role.role_code for link in db_user.user_roles if link.role]
|
||
token, _, _ = create_access_token(
|
||
user_id=db_user.id,
|
||
token_version=db_user.token_version,
|
||
roles=roles,
|
||
)
|
||
return {"Authorization": f"Bearer {token}"}
|
||
|
||
return _factory
|
||
|
||
|
||
@pytest.fixture
|
||
def make_role():
|
||
"""独立测试角色工厂（ORM 直建，用后删除）。
|
||
|
||
清理时先删除 user_roles 关联（用例失败时可能残留绑定），再删角色。
|
||
"""
|
||
created: list[int] = []
|
||
|
||
def _factory(**overrides: object) -> dict:
|
||
code = str(overrides.pop("role_code", f"t_role_{uuid.uuid4().hex[:6]}"))
|
||
name = str(overrides.pop("role_name", "测试角色"))
|
||
factory = get_session_factory()
|
||
with factory() as session:
|
||
role = Role(
|
||
role_code=code,
|
||
role_name=name,
|
||
description=overrides.pop("description", None),
|
||
sort_order=overrides.pop("sort_order", 99),
|
||
status=overrides.pop("status", 1),
|
||
**overrides,
|
||
)
|
||
session.add(role)
|
||
session.commit()
|
||
created.append(role.id)
|
||
return {"id": role.id, "role_code": code, "role_name": name}
|
||
|
||
yield _factory
|
||
|
||
factory = get_session_factory()
|
||
with factory() as session:
|
||
session.execute(delete(UserRole).where(UserRole.role_id.in_(created)))
|
||
session.execute(delete(Role).where(Role.id.in_(created)))
|
||
session.commit()
|
||
|
||
|
||
@pytest.fixture
|
||
def admin(make_user, auth_headers, monkeypatch) -> dict:
|
||
"""管理员账号（含请求头）。"""
|
||
user = make_user([ROLE_ADMIN])
|
||
from app.models import AuthorizationSetting
|
||
with get_session_factory()() as session:
|
||
session.get(AuthorizationSetting, 1).maker_user_id = user["id"]
|
||
session.commit()
|
||
user["headers"] = auth_headers(user)
|
||
return user
|
||
|
||
|
||
@pytest.fixture
|
||
def normal(make_user, auth_headers) -> dict:
|
||
"""普通用户账号（含请求头）。"""
|
||
user = make_user([ROLE_USER])
|
||
user["headers"] = auth_headers(user)
|
||
return user
|
||
|
||
|
||
@pytest.fixture
|
||
def tplus_tasks(monkeypatch):
|
||
"""隔离测试任务，仅清理此夹具创建的批次和异常。"""
|
||
from app.core.config import Settings
|
||
from app.models import TPlusSyncBatch, TPlusSyncError
|
||
|
||
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)
|
||
keys = []
|
||
|
||
def payload():
|
||
key = str(uuid.uuid4())
|
||
keys.append(key)
|
||
return {"task_type": "READ", "entity_types": ["personnel"], "request_key": key}
|
||
|
||
yield payload
|
||
with get_session_factory()() as db:
|
||
numbers = list(
|
||
db.scalars(select(TPlusSyncBatch.batch_no).where(TPlusSyncBatch.request_key.in_(keys)))
|
||
)
|
||
if numbers:
|
||
db.execute(delete(TPlusSyncError).where(TPlusSyncError.batch_no.in_(numbers)))
|
||
db.execute(delete(TPlusSyncBatch).where(TPlusSyncBatch.batch_no.in_(numbers)))
|
||
db.commit()
|
||
|
||
|
||
@pytest.fixture
|
||
def legacy_calculation(monkeypatch):
|
||
"""明确启用旧引擎兼容回归；生产和正式工资测试默认关闭。"""
|
||
from app.core.config import get_settings
|
||
monkeypatch.setattr(get_settings(), "payroll_allow_legacy_calculation", True)
|
||
|
||
|
||
@pytest.fixture
|
||
def writer(normal, admin, monkeypatch):
|
||
"""指定一个普通角色账号为工资制作人，证明身份授权与管理员角色无关。"""
|
||
from app.models import AuthorizationSetting
|
||
with get_session_factory()() as session:
|
||
session.get(AuthorizationSetting, 1).maker_user_id = normal["id"]
|
||
role = session.scalar(select(Role).where(Role.role_code == "payroll_maker"))
|
||
session.add(UserRole(user_id=normal["id"], role_id=role.id))
|
||
session.commit()
|
||
return normal
|