ad5fe4a411
需求 2.5.1 要求记录用户管理操作，先落表结构： - operation_logs 表：冗余 username 保证账号删除后仍可追溯， user_id 外键 ON DELETE SET NULL，success 默认 '0' 与 login_logs 口径一致 - 截断常量（USERNAME/ERROR_MESSAGE/REQUEST_DATA）集中定义，供写入方引用 - 修复既有 V1 迁移与迁移模板的 downgrade：索引是外键唯一支撑索引， 先 drop_index 会报 errno 1553，改为只 drop_table - 同步 db/migration/V2 与 init-dev.sql，迁移版本号更新为 a45a8087831b Co-Authored-By: Claude Code <noreply@anthropic.com>
135 lines
5.6 KiB
Python
135 lines
5.6 KiB
Python
"""DDL 漂移检测。
|
||
|
||
结构定义同时存在于两处：
|
||
- `db/migration/V*.sql` —— 事实来源，生产迁移用
|
||
- `db/init/init-dev.sql` —— 开发环境重置脚本，Docker 首次启动唯一会自动执行的位置
|
||
|
||
两份 DDL 手工同步迟早会漂移，且后果隐蔽：开发库的表结构与迁移脚本不一致，
|
||
新人克隆下来建出的库和生产不一样，等到上线才发现。
|
||
本测试把「靠自觉」变成「可自动拦截」，成本极低。
|
||
|
||
**必须比对全部 V*.sql 的合集**：早期版本只解析 V1，新增 V2 之后，
|
||
两份 DDL 同时缺表仍会被判为「一致」，漂移守卫形同虚设。
|
||
|
||
注：本文件不需要容器，放在 integration 下是因为它校验的是交付产物而非应用代码。
|
||
"""
|
||
|
||
import re
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
PROJECT_ROOT = Path(__file__).resolve().parents[3]
|
||
MIGRATION_DIR = PROJECT_ROOT / "db" / "migration"
|
||
INIT_DEV_FILE = PROJECT_ROOT / "db" / "init" / "init-dev.sql"
|
||
ALEMBIC_VERSIONS_DIR = PROJECT_ROOT / "backend" / "alembic" / "versions"
|
||
|
||
_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
|
||
|
||
|
||
def _alembic_head() -> str:
|
||
"""从迁移脚本推断当前 head 版本号。
|
||
|
||
纯文本解析而非调用 alembic 运行时 —— 后者需要连数据库，会让本测试
|
||
从「秒级、无需容器」变成「必须起库」。
|
||
head 的定义：没有被任何其他迁移的 down_revision 引用过的那个 revision。
|
||
"""
|
||
revisions: set[str] = set()
|
||
referenced: set[str] = set()
|
||
for path in ALEMBIC_VERSIONS_DIR.glob("*.py"):
|
||
text = path.read_text(encoding="utf-8")
|
||
if found := re.search(r"^revision: str = ['\"](\w+)['\"]", text, re.MULTILINE):
|
||
revisions.add(found.group(1))
|
||
if found := re.search(r"^down_revision:.*?= ['\"](\w+)['\"]", text, re.MULTILINE):
|
||
referenced.add(found.group(1))
|
||
|
||
heads = revisions - referenced
|
||
assert len(heads) == 1, f"迁移链存在分叉或缺失，heads={sorted(heads)}"
|
||
return heads.pop()
|
||
|
||
|
||
@pytest.fixture(scope="module")
|
||
def migration_tables() -> dict[str, set[str]]:
|
||
"""合并解析 db/migration/ 下全部 V*.sql。"""
|
||
files = sorted(MIGRATION_DIR.glob("V*.sql"))
|
||
assert files, f"未找到迁移脚本：{MIGRATION_DIR}"
|
||
|
||
merged: dict[str, set[str]] = {}
|
||
for path in files:
|
||
for table, columns in _parse_tables(path.read_text(encoding="utf-8")).items():
|
||
merged.setdefault(table, set()).update(columns)
|
||
return merged
|
||
|
||
|
||
@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 list(MIGRATION_DIR.glob("V*.sql")), f"缺少迁移脚本：{MIGRATION_DIR}"
|
||
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 = "\n".join(
|
||
path.read_text(encoding="utf-8") for path in sorted(MIGRATION_DIR.glob("V*.sql"))
|
||
)
|
||
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_开发脚本标记的版本号等于当前head(self) -> None:
|
||
"""版本号必须等于 head，且不能只断言「存在」。
|
||
|
||
落后的后果：开发库重建后执行 `alembic upgrade head`，alembic 认为该迁移
|
||
尚未应用，重复建表并报 1050 Table already exists，只能手工补版本或 down -v。
|
||
"""
|
||
init_sql = INIT_DEV_FILE.read_text(encoding="utf-8")
|
||
found = re.search(
|
||
r"INSERT INTO alembic_version \(version_num\) VALUES \('(\w+)'\)", init_sql
|
||
)
|
||
assert found, "init-dev.sql 未写入 alembic_version"
|
||
assert found.group(1) == _alembic_head(), (
|
||
f"init-dev.sql 标记的版本号 {found.group(1)!r} 落后于当前 head {_alembic_head()!r}"
|
||
)
|