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"
)


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() == "0002_add_restore_audit_action"


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