90 lines
3.7 KiB
Python
90 lines
3.7 KiB
Python
|
"""DDL 漂移检测。
|
|||
|
|
|
|||
|
|
结构定义同时存在于两处：
|
|||
|
|
- `db/migration/V1__*.sql` —— 事实来源，生产迁移用
|
|||
|
|
- `db/init/init-dev.sql` —— 开发环境重置脚本，Docker 首次启动唯一会自动执行的位置
|
|||
|
|
|
|||
|
|
两份 DDL 手工同步迟早会漂移，且后果隐蔽：开发库的表结构与迁移脚本不一致，
|
|||
|
|
新人克隆下来建出的库和生产不一样，等到上线才发现。
|
|||
|
|
本测试把「靠自觉」变成「可自动拦截」，成本极低。
|
|||
|
|
|
|||
|
|
注：本文件不需要容器，放在 integration 下是因为它校验的是交付产物而非应用代码。
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
import re
|
|||
|
|
from pathlib import Path
|
|||
|
|
|
|||
|
|
import pytest
|
|||
|
|
|
|||
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[3]
|
|||
|
|
MIGRATION_FILE = PROJECT_ROOT / "db" / "migration" / "V1__init_rbac_and_login_logs.sql"
|
|||
|
|
INIT_DEV_FILE = PROJECT_ROOT / "db" / "init" / "init-dev.sql"
|
|||
|
|
|
|||
|
|
_CREATE_TABLE = re.compile(r"CREATE TABLE\s+`?(\w+)`?\s*\((.*?)\n\)", re.DOTALL)
|
|||
|
|
_COLUMN_LINE = re.compile(r"^\s{4}`?(\w+)`?\s", re.MULTILINE)
|
|||
|
|
# 这些开头的是约束定义，不是列定义
|
|||
|
|
_NON_COLUMN_KEYWORDS = {"CONSTRAINT", "PRIMARY", "UNIQUE", "FOREIGN", "KEY", "INDEX"}
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _parse_tables(sql: str) -> dict[str, set[str]]:
|
|||
|
|
"""从 SQL 文本中提取 {表名: {列名集合}}。"""
|
|||
|
|
tables: dict[str, set[str]] = {}
|
|||
|
|
for table_name, body in _CREATE_TABLE.findall(sql):
|
|||
|
|
tables[table_name] = {
|
|||
|
|
column
|
|||
|
|
for column in _COLUMN_LINE.findall(body)
|
|||
|
|
if column.upper() not in _NON_COLUMN_KEYWORDS
|
|||
|
|
}
|
|||
|
|
return tables
|
|||
|
|
|
|||
|
|
|
|||
|
|
@pytest.fixture(scope="module")
|
|||
|
|
def migration_tables() -> dict[str, set[str]]:
|
|||
|
|
return _parse_tables(MIGRATION_FILE.read_text(encoding="utf-8"))
|
|||
|
|
|
|||
|
|
|
|||
|
|
@pytest.fixture(scope="module")
|
|||
|
|
def init_dev_tables() -> dict[str, set[str]]:
|
|||
|
|
return _parse_tables(INIT_DEV_FILE.read_text(encoding="utf-8"))
|
|||
|
|
|
|||
|
|
|
|||
|
|
class TestSchemaSync:
|
|||
|
|
"""两份 DDL 的结构一致性。"""
|
|||
|
|
|
|||
|
|
def test_文件存在(self) -> None:
|
|||
|
|
assert MIGRATION_FILE.is_file(), f"缺少迁移脚本：{MIGRATION_FILE}"
|
|||
|
|
assert INIT_DEV_FILE.is_file(), f"缺少开发初始化脚本：{INIT_DEV_FILE}"
|
|||
|
|
|
|||
|
|
def test_表集合一致(self, migration_tables: dict, init_dev_tables: dict) -> None:
|
|||
|
|
assert set(init_dev_tables) == set(migration_tables), (
|
|||
|
|
"两份 DDL 的表集合不一致 —— 表结构变更时需同时维护 "
|
|||
|
|
"db/migration/V{n}__*.sql 与 db/init/init-dev.sql"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
def test_每张表的列集合一致(self, migration_tables: dict, init_dev_tables: dict) -> None:
|
|||
|
|
mismatched = {
|
|||
|
|
table: {
|
|||
|
|
"仅迁移脚本有": sorted(migration_tables[table] - init_dev_tables[table]),
|
|||
|
|
"仅初始化脚本有": sorted(init_dev_tables[table] - migration_tables[table]),
|
|||
|
|
}
|
|||
|
|
for table in set(migration_tables) & set(init_dev_tables)
|
|||
|
|
if migration_tables[table] != init_dev_tables[table]
|
|||
|
|
}
|
|||
|
|
assert not mismatched, f"以下表的列不一致：{mismatched}"
|
|||
|
|
|
|||
|
|
|
|||
|
|
class TestSeedData:
|
|||
|
|
"""字典数据在两处应保持同步。"""
|
|||
|
|
|
|||
|
|
@pytest.mark.parametrize("role_code", ["admin", "user"])
|
|||
|
|
def test_角色初始化数据在两处都存在(self, role_code: str) -> None:
|
|||
|
|
migration_sql = MIGRATION_FILE.read_text(encoding="utf-8")
|
|||
|
|
init_sql = INIT_DEV_FILE.read_text(encoding="utf-8")
|
|||
|
|
assert f"'{role_code}'" in migration_sql
|
|||
|
|
assert f"'{role_code}'" in init_sql
|
|||
|
|
|
|||
|
|
def test_开发脚本标记了迁移版本(self) -> None:
|
|||
|
|
"""不写 alembic_version 的话，在开发库上执行 upgrade head 会重复建表而失败。"""
|
|||
|
|
init_sql = INIT_DEV_FILE.read_text(encoding="utf-8")
|
|||
|
|
assert "INSERT INTO alembic_version" in init_sql
|