Files
PayrollManagementSystem/backend/tests/unit/test_models.py
T
liuwenxinandClaude Code 9964281c14 fix(approval): 版本号唯一键兜底 + 取号改当前读防并发重复
- V12 迁移:先重排存量重复 (order_type, ref_id) 分组的 version_no
  (V11 把存量一律置 1,而 V11 前重提是新建行),再加
  uk_approval_orders_order_type_ref_version;只重排真有重复的分组,
  避免改动前端已展示过的历史版本号
- _next_version 改锁读(select version_no ... FOR UPDATE,最大值在
  Python 侧算):请求级鉴权的普通 SELECT 已把 REPEATABLE READ 快照
  锚定在请求开始时刻,原 MAX 快照读在并发下会算出同一版本号;
  聚合 + FOR UPDATE 为 MySQL 明确禁止,故不锁聚合结果
- 编辑与月结提交路径把 1062 收敛为 400「提交冲突(版本号生成重复),
  请重试」,其余完整性错误原样上抛;submit_order 内不 catch/rollback
- 驳回清空方案的过时注释改为 V11 起的 append-only 语义(置历史不删行)
- init-dev.sql 同步唯一键与 alembic_version 7f4f7b918bc4,导出 V12 SQL
- 测试:集成 +5(锁读 SQL 形态、两会话快照回归、唯一键 1062、冲突 400、
  正常递增)+ 单元 +1;工作台假审批单改用不同 ref_id 适配唯一键
- 全量回归 897 passed;ruff 与前端 type-check 绿

Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-10-22 13:31:01 +08:00

180 lines
6.8 KiB
Python

"""ORM 模型元数据测试。
建表规范(表小写蛇形复数、字段中文 COMMENT、主键 id、索引 uk_/idx_ 前缀)
属于「红线」,靠人工 review 迟早会漏,这里用测试把它固定下来。
"""
import re
import pytest
from sqlalchemy.dialects import mysql
from app.models import Base
EXPECTED_TABLES = {
"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",
}
# 单复数同形的集合名词:本身即复数语义,不受「以 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")