108bf87225
补齐源快照、资料预演、正式导入及工资核算的依赖链路。 新增版本化输入、逐段 FIFO、生成锁和结构化当月调整， 并提供默认关闭的周期采集、有限重试及 Windows 托管。 原始客户资料、私有配置及首页改动不纳入本次提交。 BREAKING CHANGE: 升级前需应用新增迁移，并通过 PAYROLL_MAKER_USER_ID 绑定实际工资制作人，否则业务只读。 旧单批次核算默认关闭；历史兼容需显式启用 PAYROLL_ALLOW_LEGACY_CALCULATION。生产启用和真实金额待验收。
187 lines
7.1 KiB
Python
187 lines
7.1 KiB
Python
"""ORM 模型元数据测试。
|
||
|
||
建表规范（表小写蛇形复数、字段中文 COMMENT、主键 id、索引 uk_/idx_ 前缀）
|
||
属于「红线」，靠人工 review 迟早会漏，这里用测试把它固定下来。
|
||
"""
|
||
|
||
import re
|
||
|
||
import pytest
|
||
from sqlalchemy.dialects import mysql
|
||
|
||
from app.models import Base
|
||
|
||
EXPECTED_TABLES = {
|
||
"tplus_automation_states",
|
||
"payroll_months", "payroll_inputs", "payroll_allocations",
|
||
"tplus_source_heads", "tplus_facts", "tplus_source_links", "tplus_work_runs", "tplus_work_lines",
|
||
"users",
|
||
"roles",
|
||
"user_roles",
|
||
"permissions",
|
||
"role_permissions",
|
||
"login_logs",
|
||
"operation_logs",
|
||
"business_roles",
|
||
"system_settings",
|
||
"approval_orders",
|
||
"approval_steps",
|
||
"business_secondary_allocations",
|
||
"param_definitions",
|
||
"param_change_batches",
|
||
"param_values",
|
||
"personnel",
|
||
"personnel_regions",
|
||
"personnel_plans",
|
||
"products",
|
||
"customers",
|
||
"ownership_change_logs",
|
||
"businesses",
|
||
"business_items",
|
||
"business_status_logs",
|
||
"shipments",
|
||
"shipment_items",
|
||
"receipts",
|
||
"customer_rating_grades",
|
||
"performance_ledgers",
|
||
"monthly_settlements",
|
||
"settlement_items",
|
||
"adjustment_orders",
|
||
"tplus_sync_batches",
|
||
"tplus_sync_errors",
|
||
"tplus_sync_records",
|
||
"tplus_sync_watermarks",
|
||
}
|
||
|
||
# 单复数同形的集合名词：本身即复数语义，不受「以 s 结尾」检查约束
|
||
PLURAL_EXCEPTIONS = {"personnel"}
|
||
|
||
|
||
def _ddl(table_name: str) -> str:
|
||
"""返回某张表的建表 DDL 文本。"""
|
||
from sqlalchemy.schema import CreateTable
|
||
|
||
table = Base.metadata.tables[table_name]
|
||
return str(CreateTable(table).compile(dialect=mysql.dialect()))
|
||
|
||
|
||
class TestTableSet:
|
||
"""表集合必须完整 —— 漏一张就会让 alembic 生成「删除该表」的迁移。"""
|
||
|
||
def test_表集合与预期一致(self) -> None:
|
||
assert set(Base.metadata.tables) == EXPECTED_TABLES
|
||
|
||
def test_所有表名均为小写蛇形复数(self) -> None:
|
||
for name in Base.metadata.tables:
|
||
assert re.fullmatch(r"[a-z][a-z0-9_]*", name), f"表名 {name} 不符合小写蛇形"
|
||
# personnel 等集合名词单复数同形，本身即复数语义
|
||
if name not in PLURAL_EXCEPTIONS:
|
||
assert name.endswith("s"), f"表名 {name} 应为复数"
|
||
|
||
|
||
class TestNamingConvention:
|
||
"""约束命名：规范要求 uk_表_字段 / idx_表_字段。"""
|
||
|
||
@pytest.mark.parametrize("table_name", sorted(EXPECTED_TABLES))
|
||
def test_主键命名为pk_表名(self, table_name: str) -> None:
|
||
pk = Base.metadata.tables[table_name].primary_key
|
||
assert pk.name == f"pk_{table_name}"
|
||
|
||
def test_唯一键以uk_开头(self) -> None:
|
||
for table in Base.metadata.tables.values():
|
||
for constraint in table.constraints:
|
||
if constraint.__class__.__name__ == "UniqueConstraint":
|
||
assert constraint.name.startswith(f"uk_{table.name}_"), constraint.name
|
||
|
||
def test_索引以idx_开头(self) -> None:
|
||
for table in Base.metadata.tables.values():
|
||
for index in table.indexes:
|
||
assert index.name.startswith(f"idx_{table.name}_"), index.name
|
||
|
||
def test_外键以fk_开头(self) -> None:
|
||
for table in Base.metadata.tables.values():
|
||
for fk in table.foreign_keys:
|
||
assert fk.constraint.name.startswith(f"fk_{table.name}_"), fk.constraint.name
|
||
|
||
def test_审批单版本号唯一键覆盖三列(self) -> None:
|
||
"""并发提单算重版本号时由 DB 兜底（应用层取号走锁读，见
|
||
approval_service._next_version）；约束名必须与迁移 DDL 严格一致，
|
||
否则 alembic autogenerate 会误判成「约束缺失/多余」。"""
|
||
table = Base.metadata.tables["approval_orders"]
|
||
uks = [c for c in table.constraints if c.__class__.__name__ == "UniqueConstraint"]
|
||
hit = [
|
||
c
|
||
for c in uks
|
||
if {col.name for col in c.columns} == {"order_type", "ref_id", "version_no"}
|
||
]
|
||
assert hit, "审批单缺 (order_type, ref_id, version_no) 唯一键"
|
||
assert hit[0].name == "uk_approval_orders_order_type_ref_version"
|
||
|
||
|
||
class TestComments:
|
||
"""规范红线要求中文 COMMENT。"""
|
||
|
||
@pytest.mark.parametrize("table_name", sorted(EXPECTED_TABLES))
|
||
def test_表级中文注释存在(self, table_name: str) -> None:
|
||
assert Base.metadata.tables[table_name].comment, f"{table_name} 缺少表注释"
|
||
|
||
@pytest.mark.parametrize("table_name", sorted(EXPECTED_TABLES))
|
||
def test_每列都有中文注释(self, table_name: str) -> None:
|
||
table = Base.metadata.tables[table_name]
|
||
missing = [c.name for c in table.columns if not c.comment]
|
||
assert not missing, f"{table_name} 缺注释的列：{missing}"
|
||
|
||
@pytest.mark.parametrize("table_name", sorted(EXPECTED_TABLES))
|
||
def test_注释确实输出到DDL(self, table_name: str) -> None:
|
||
ddl = _ddl(table_name)
|
||
assert "COMMENT '" in ddl
|
||
assert f"COMMENT='{Base.metadata.tables[table_name].comment}'" in ddl
|
||
|
||
|
||
class TestMysqlTableOptions:
|
||
"""引擎与字符集：不显式指定则跟随服务端，跨环境可能不一致。"""
|
||
|
||
@pytest.mark.parametrize("table_name", sorted(EXPECTED_TABLES))
|
||
def test_显式指定引擎与字符集(self, table_name: str) -> None:
|
||
ddl = _ddl(table_name)
|
||
assert "ENGINE=InnoDB" in ddl
|
||
assert "CHARSET=utf8mb4" in ddl
|
||
assert "utf8mb4_general_ci" in ddl
|
||
|
||
@pytest.mark.parametrize("table_name", sorted(EXPECTED_TABLES))
|
||
def test_主键为无符号BIGINT(self, table_name: str) -> None:
|
||
assert "id BIGINT UNSIGNED NOT NULL" in _ddl(table_name)
|
||
|
||
|
||
class TestTimestampColumns:
|
||
"""更新时间必须由数据库维护 —— 这是 server_onupdate 做不到的。"""
|
||
|
||
@pytest.mark.parametrize("table_name", ["users", "roles", "permissions"])
|
||
def test_带时间戳的表含ON_UPDATE子句(self, table_name: str) -> None:
|
||
assert "ON UPDATE CURRENT_TIMESTAMP" in _ddl(table_name)
|
||
|
||
@pytest.mark.parametrize("table_name", ["login_logs", "operation_logs"])
|
||
def test_仅追加的日志表不含ON_UPDATE子句(self, table_name: str) -> None:
|
||
"""日志是不可变记录，没有"更新时间"语义。"""
|
||
assert "ON UPDATE CURRENT_TIMESTAMP" not in _ddl(table_name)
|
||
|
||
def test_关联表保留create_time便于审计(self) -> None:
|
||
for table_name in ("user_roles", "role_permissions"):
|
||
assert "create_time" in Base.metadata.tables[table_name].columns
|
||
|
||
|
||
class TestEnumColumns:
|
||
"""状态类字段的默认值。"""
|
||
|
||
def test_用户默认启用且令牌版本为1(self) -> None:
|
||
users = Base.metadata.tables["users"]
|
||
assert users.columns["status"].server_default.arg == "1"
|
||
assert users.columns["token_version"].server_default.arg == "1"
|
||
|
||
def test_角色编码唯一(self) -> None:
|
||
ddl = _ddl("roles")
|
||
assert "uk_roles_role_code UNIQUE (role_code)" in ddl
|
||
|
||
def test_用户名唯一(self) -> None:
|
||
assert "uk_users_username UNIQUE (username)" in _ddl("users")
|