diff --git a/backend/tests/integration/test_schema_sync.py b/backend/tests/integration/test_schema_sync.py index 322de41..c633331 100644 --- a/backend/tests/integration/test_schema_sync.py +++ b/backend/tests/integration/test_schema_sync.py @@ -28,10 +28,14 @@ _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"} +# ALTER TABLE 的列级变更(语义重放用):ADD COLUMN / DROP COLUMN; +# MODIFY 只改类型不改列名,对列集合无影响,无需处理 +_ALTER_ADD = re.compile(r"ALTER TABLE\s+`?(\w+)`?\s+ADD(?:\s+COLUMN)?\s+`?(\w+)`?", re.IGNORECASE) +_ALTER_DROP = re.compile(r"ALTER TABLE\s+`?(\w+)`?\s+DROP(?:\s+COLUMN)?\s+`?(\w+)`?", re.IGNORECASE) def _parse_tables(sql: str) -> dict[str, set[str]]: - """从 SQL 文本中提取 {表名: {列名集合}}。""" + """从 SQL 文本中提取 {表名: {列名集合}}(仅 CREATE TABLE 的静态解析)。""" tables: dict[str, set[str]] = {} for table_name, body in _CREATE_TABLE.findall(sql): tables[table_name] = { @@ -42,6 +46,27 @@ def _parse_tables(sql: str) -> dict[str, set[str]]: return tables +def _migration_final_state(paths: list[Path]) -> dict[str, set[str]]: + """按文件序重放迁移脚本,得到迁移链**终态**的 {表名: {列名集合}}。 + + 早前实现把全部脚本的 CREATE 列做并集:某表日后被 ALTER DROP COLUMN + 时,并集里旧列名永远存在,init-dev.sql 必须保留死列才能过测 —— 守卫 + 会主动背书一份与生产结构漂移的开发库 DDL。按序重放 ADD/DROP 后, + 比对基准才是终态(CHANGE RENAME 列名出现时重放会报差异,属人工核对 + 的正确时机:改名同时牵动 init-dev 与 ORM,须人来确认两处同步)。 + """ + merged: dict[str, set[str]] = {} + for path in paths: + sql = path.read_text(encoding="utf-8") + for table, columns in _parse_tables(sql).items(): + merged.setdefault(table, set()).update(columns) + for table, column in _ALTER_ADD.findall(sql): + merged.setdefault(table, set()).add(column) + for table, column in _ALTER_DROP.findall(sql): + merged.setdefault(table, set()).discard(column) + return merged + + def _alembic_head() -> str: """从迁移脚本推断当前 head 版本号。 @@ -65,15 +90,10 @@ def _alembic_head() -> str: @pytest.fixture(scope="module") def migration_tables() -> dict[str, set[str]]: - """合并解析 db/migration/ 下全部 V*.sql。""" + """按文件序重放 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 + return _migration_final_state(files) @pytest.fixture(scope="module")