PayrollManagementSystem/backend/tests/integration/test_schema_sync.py

"""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"^[ \t]+`?(\w+)`?[ \t]", 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 文本中提取 {表名: {列名集合}}（仅 CREATE TABLE 的静态解析）。"""
    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 _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):
            # ADD CONSTRAINT ... UNIQUE 的「CONSTRAINT」不是列（与 _parse_tables
            # 的约束关键词同一份清单过滤），V8 台账唯一键三列扩展即此形态
            if column.upper() not in _NON_COLUMN_KEYWORDS:
                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 版本号。

    纯文本解析而非调用 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(?:\s*:\s*str)?\s*=\s*['\"](\w+)['\"]", text, re.MULTILINE):
            revisions.add(found.group(1))
        if found := re.search(r"^down_revision(?:\s*:.*?)?\s*=\s*['\"](\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}"
    return _migration_final_state(files)


@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", "marketing_ops", "finance", "leader"])
    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}"
        )