PayrollManagementSystem/backend/tests/unit/test_models.py
liuwenxin 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
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""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")