document-management/backend/tests/dms/test_migration.py

from __future__ import annotations

import importlib.util
from pathlib import Path

from alembic.config import Config
from alembic.script import ScriptDirectory


BACKEND_ROOT = Path(__file__).resolve().parents[2]
MIGRATIONS_ROOT = BACKEND_ROOT / "migrations"
REVISION_PATH = MIGRATIONS_ROOT / "versions" / "0001_initial_schema.py"
RESTORE_REVISION_PATH = (
    MIGRATIONS_ROOT / "versions" / "0002_add_restore_audit_action.py"
)
Q3_REVISION_PATH = (
    MIGRATIONS_ROOT
    / "versions"
    / "0003_q3_business_alignment_and_content_search.py"
)
REORDER_REVISION_PATH = (
    MIGRATIONS_ROOT / "versions" / "0009_reorder_audit_action.py"
)
ATTACHMENT_TYPE_REVISION_PATH = (
    MIGRATIONS_ROOT / "versions" / "0010_attachment_type_redefine.py"
)
REORDER_CATEGORY_REVISION_PATH = (
    MIGRATIONS_ROOT
    / "versions"
    / "0011_reorder_category_audit_action.py"
)
ROLE_MANAGEMENT_REVISION_PATH = (
    MIGRATIONS_ROOT / "versions" / "0012_role_management.py"
)
DOCUMENT_PASSWORD_LOCK_REVISION_PATH = (
    MIGRATIONS_ROOT / "versions" / "0013_document_password_lock.py"
)
DOCUMENT_RELATIONS_REVISION_PATH = (
    MIGRATIONS_ROOT / "versions" / "0014_document_relations.py"
)
LOGIN_IP_REVISION_PATH = (
    MIGRATIONS_ROOT / "versions" / "0015_login_ip_restriction.py"
)
PORTAL_USER_PROVISION_REVISION_PATH = (
    MIGRATIONS_ROOT / "versions" / "0016_portal_user_provision.py"
)


def _load_revision_module():
    spec = importlib.util.spec_from_file_location("dms_initial_migration", REVISION_PATH)
    assert spec is not None and spec.loader is not None
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def test_alembic_config_and_head_are_valid():
    config = Config(str(MIGRATIONS_ROOT / "alembic.ini"))
    script = ScriptDirectory.from_config(config)
    assert script.get_current_head() == "0016_portal_user_provision"


def test_initial_migration_exports_upgrade_and_downgrade():
    module = _load_revision_module()
    assert callable(module.upgrade)
    assert callable(module.downgrade)


def test_initial_migration_contains_seven_tables_and_generated_columns():
    source = REVISION_PATH.read_text(encoding="utf-8")
    for table_name in (
        "sys_organization",
        "sys_user",
        "doc_category",
        "doc_document",
        "doc_attachment_binding",
        "doc_permission",
        "sys_audit_log",
    ):
        assert f'"{table_name}"' in source
    assert source.count('sa.Computed("IF(is_deleted = 0, 1, NULL)"') == 2
    assert "uq_doc_attachment_binding_active" in source
    assert "uq_doc_permission_active" in source


def test_initial_migration_has_no_cascade():
    source = REVISION_PATH.read_text(encoding="utf-8")
    assert 'ondelete="CASCADE"' not in source
    assert source.count('ondelete="RESTRICT"') >= 8


def test_downgrade_uses_dependency_safe_order():
    source = REVISION_PATH.read_text(encoding="utf-8")
    positions = [
        source.index(f'op.drop_table("{name}")')
        for name in (
            "sys_audit_log",
            "doc_permission",
            "doc_attachment_binding",
            "doc_document",
            "doc_category",
            "sys_user",
            "sys_organization",
        )
    ]
    assert positions == sorted(positions)


def test_restore_migration_is_scoped_and_reversible():
    source = RESTORE_REVISION_PATH.read_text(encoding="utf-8")
    assert 'down_revision: str | None = "0001_initial_schema"' in source
    assert "RESTORE_DOCUMENT" in source
    assert "ix_doc_document_recycle_deleted" in source
    assert '["is_deleted", "deleted_at", "id"]' in source
    assert "drop_index" in source
    assert source.count("create_check_constraint") == 2
    assert "create_table" not in source
    assert "add_column" not in source


def test_q3_migration_is_single_scoped_revision_with_safe_downgrade():
    source = Q3_REVISION_PATH.read_text(encoding="utf-8")
    assert (
        'down_revision: str | None = "0002_add_restore_audit_action"'
        in source
    )
    assert "content_text" in source
    assert "content_extract_status" in source
    assert "content_extracted_at" in source
    assert "LONGTEXT" in source
    assert "auth_version = auth_version + 1" in source
    assert "role_code = 'USER'" in source
    assert "role_code IN ('USER', 'ADMIN')" in source
    assert "RuntimeError" in source
    assert "create_table" not in source


def test_reorder_audit_migration_is_scoped_and_reversible():
    source = REORDER_REVISION_PATH.read_text(encoding="utf-8")
    assert 'down_revision: str | None = "0008_document_sort_no"' in source
    assert "REORDER_DOCUMENT" in source
    assert "REORDER_ATTACHMENT" in source
    assert "audit_action_type" in source
    assert "RuntimeError" in source
    assert "create_table" not in source
    assert "add_column" not in source


def test_attachment_type_redefine_migration_maps_legacy_values():
    source = ATTACHMENT_TYPE_REVISION_PATH.read_text(encoding="utf-8")
    assert 'down_revision: str | None = "0009_reorder_audit_action"' in source
    assert "SUPPORT_ACHIEVEMENT" in source
    assert "TRAINING_MATERIAL" in source
    assert "TYPICAL_CASE" in source
    assert "document_attachment_type" in source
    assert "UPDATE doc_document SET attachment_type = 'SUPPORT_ACHIEVEMENT'" in source
    assert "RuntimeError" in source
    assert "create_table" not in source
    assert "add_column" not in source


def test_reorder_category_audit_migration_is_scoped_and_reversible():
    source = REORDER_CATEGORY_REVISION_PATH.read_text(encoding="utf-8")
    assert (
        'down_revision: str | None = "0010_attachment_type_redefine"'
        in source
    )
    assert "REORDER_CATEGORY" in source
    assert "audit_action_type" in source
    assert "RuntimeError" in source
    assert "create_table" not in source
    assert "add_column" not in source


def test_role_management_migration_is_scoped_and_reversible():
    source = ROLE_MANAGEMENT_REVISION_PATH.read_text(encoding="utf-8")
    assert (
        'down_revision: str | None = "0011_reorder_category_audit_action"'
        in source
    )
    for table_name in ("sys_role", "sys_role_dept", "sys_role_permission"):
        assert f'"{table_name}"' in source
    assert "uq_sys_role_default_active" in source
    assert "IF(is_default = 1 AND is_deleted = 0, 1, NULL)" in source
    assert "can_config_permission" in source
    assert "can_bind_attachment" in source
    assert "can_unbind_attachment" in source
    assert "fk_sys_user_role_id_sys_role" in source
    assert "CHANGE_ROLE_PERMISSION" in source
    assert "'ROLE'" in source
    assert "RuntimeError" in source
    assert source.count("op.create_table") == 3
    # 降级按依赖序先删矩阵/部门，再删角色本体
    positions = [
        source.index(f'op.drop_table("{name}")')
        for name in ("sys_role_permission", "sys_role_dept", "sys_role")
    ]
    assert positions == sorted(positions)


def test_document_password_lock_migration_is_scoped_and_reversible():
    source = DOCUMENT_PASSWORD_LOCK_REVISION_PATH.read_text(encoding="utf-8")
    assert (
        'down_revision: str | None = "0012_role_management"' in source
    )
    assert '"password_hash"' in source
    for column in (
        "protect_view",
        "protect_download",
        "protect_edit",
        "protect_delete",
        "protect_config_permission",
        "protect_bind_attachment",
        "protect_unbind_attachment",
    ):
        assert f'"{column}"' in source
    assert "document_password_main_only" in source
    assert "(password_hash IS NULL) OR (document_type = 'MAIN')" in source
    assert "DOCUMENT_PASSWORD_SET" in source
    assert "DOCUMENT_PASSWORD_REMOVE" in source
    assert "RuntimeError" in source
    assert "create_table" not in source


def test_document_relations_migration_is_scoped_and_reversible():
    source = DOCUMENT_RELATIONS_REVISION_PATH.read_text(encoding="utf-8")
    assert (
        'down_revision: str | None = "0013_document_password_lock"' in source
    )
    assert 'op.create_table(\n        "doc_document_relation"' in source
    assert '"relation_type"' in source
    assert "sa.String(length=64)" in source
    assert "uq_doc_document_relation_active" in source
    assert "IF(is_deleted = 0, 1, NULL)" in source
    assert "document_relation_no_self" in source
    assert "document_id <> related_document_id" in source
    assert source.count("ondelete=\"RESTRICT\"") == 2
    # 审计复用EDIT_DOCUMENT，不扩展sys_audit_log的CHECK约束
    assert "audit_action_type" not in source
    assert "RuntimeError" in source
    assert source.count("op.create_table") == 1
    assert "add_column" not in source


def test_login_ip_restriction_migration_is_scoped_and_reversible():
    source = LOGIN_IP_REVISION_PATH.read_text(encoding="utf-8")
    assert (
        'down_revision: str | None = "0014_document_relations"' in source
    )
    for table_name in ("sys_ip_entry", "sys_org_ip_assignment"):
        assert f'"{table_name}"' in source
    assert '"ip_value"' in source
    assert "uq_sys_ip_entry_ip_value_active" in source
    assert "uq_sys_org_ip_assignment_active" in source
    assert "IF(is_deleted = 0, 1, NULL)" in source
    assert "fk_sys_org_ip_assignment_organization_id_sys_organization" in source
    assert "fk_sys_org_ip_assignment_ip_entry_id_sys_ip_entry" in source
    assert "CREATE_IP_ENTRY" in source
    assert "ASSIGN_ORG_IP" in source
    assert '"IP_ENTRY"' in source
    assert "RuntimeError" in source
    assert source.count("op.create_table") == 2
    # 降级按依赖序先删分配表，再删IP条目表
    positions = [
        source.index(f'op.drop_table("{name}")')
        for name in ("sys_org_ip_assignment", "sys_ip_entry")
    ]
    assert positions == sorted(positions)


def test_portal_user_provision_migration_is_scoped_and_reversible():
    source = PORTAL_USER_PROVISION_REVISION_PATH.read_text(encoding="utf-8")
    assert (
        'down_revision: str | None = "0015_login_ip_restriction"' in source
    )
    assert '"external_id"' in source
    assert "sa.String(length=128)" in source
    assert "uq_sys_user_external_id_active" in source
    assert "ix_sys_user_external_active" in source
    assert "IF(is_deleted = 0, 1, NULL)" in source
    # 降级保护：存在门户用户时拒绝有损降级
    assert "external_id IS NOT NULL" in source
    assert "RuntimeError" in source
    assert "create_table" not in source