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

"""db_upgrade 纯逻辑测试（不连接任何数据库）。"""
from alembic.config import Config
from alembic.script import ScriptDirectory

from dms.db_upgrade import _mask_password, resolve_migrations_dir, run_upgrade


def _script() -> ScriptDirectory:
    cfg = Config()
    cfg.set_main_option("script_location", str(resolve_migrations_dir()))
    return ScriptDirectory.from_config(cfg)


def test_resolve_migrations_dir_contains_env_and_versions():
    migrations = resolve_migrations_dir()
    assert (migrations / "env.py").exists()
    assert (migrations / "versions").is_dir()


def test_migration_chain_head_and_steps_from_0013():
    """远端生产库 0013 → head 恰好经过 0014~0038。"""
    script = _script()
    assert script.get_current_head() == "0038_session_jti"
    steps = [
        r.revision
        for r in reversed(
            list(script.walk_revisions("0013_document_password_lock", "head"))
        )
    ]
    assert steps == [
        "0013_document_password_lock",
        "0014_document_relations",
        "0015_login_ip_restriction",
        "0016_portal_user_provision",
        "0017_category_type_display_name",
        "0018_sub_plan_binding",
        "0019_content_extract_error",
        "0020_sub_plan_to_attachment",
        "0021_remove_assigned_scope",
        "0022_login_lockout",
        "0023_admin_permissions",
        "0024_manage_permission_matrix",
        "0025_move_audit_actions",
        "0026_borrow_permission",
        "0027_ui_theme",
        "0028_backup_audit_actions",
        "0029_system_management",
        "0030_role_edit_permission",
        "0031_user_sort_no",
        "0032_user_managed_org",
        "0033_role_upload_permission",
        "0034_role_anchor_self",
        "0035_drop_audit_action_check",
        "0036_notification",
        "0037_drop_audit_target_check",
        "0038_session_jti",
    ]


def test_run_upgrade_lists_pending_steps_excluding_current(monkeypatch, capsys):
    """0016 升级到 head：待执行清单含 0017~0038，当前版本不重复出现。"""
    monkeypatch.setenv("DMS_DATABASE_URL", "mysql+pymysql://u:secret@127.0.0.1:3306/dms")
    versions = iter(["0016_portal_user_provision", "0038_session_jti"])
    monkeypatch.setattr(
        "dms.db_upgrade._read_current_version", lambda url: next(versions)
    )
    monkeypatch.setattr("alembic.command.upgrade", lambda cfg, rev: None)
    assert run_upgrade(assume_yes=True) == 0
    out = capsys.readouterr().out
    assert "  - 0017_category_type_display_name" in out
    assert "  - 0018_sub_plan_binding" in out
    assert "  - 0019_content_extract_error" in out
    assert "  - 0020_sub_plan_to_attachment" in out
    assert "  - 0021_remove_assigned_scope" in out
    assert "  - 0022_login_lockout" in out
    assert "  - 0023_admin_permissions" in out
    assert "  - 0024_manage_permission_matrix" in out
    assert "  - 0025_move_audit_actions" in out
    assert "  - 0026_borrow_permission" in out
    assert "  - 0027_ui_theme" in out
    assert "  - 0028_backup_audit_actions" in out
    assert "  - 0029_system_management" in out
    assert "  - 0030_role_edit_permission" in out
    assert "  - 0031_user_sort_no" in out
    assert "  - 0032_user_managed_org" in out
    assert "  - 0033_role_upload_permission" in out
    assert "  - 0034_role_anchor_self" in out
    assert "  - 0035_drop_audit_action_check" in out
    assert "  - 0036_notification" in out
    assert "  - 0037_drop_audit_target_check" in out
    assert "  - 0038_session_jti" in out
    assert "  - 0016" not in out
    assert "secret" not in out  # 输出中的地址必须已脱敏


def test_mask_password():
    url = "mysql+pymysql://dms_app:secret-pwd@203.0.113.10:3306/dms?charset=utf8mb4"
    assert (
        _mask_password(url)
        == "mysql+pymysql://dms_app:***@203.0.113.10:3306/dms?charset=utf8mb4"
    )
    assert _mask_password("sqlite:///foo.db") == "sqlite:///foo.db"


def test_run_upgrade_requires_database_url(monkeypatch, capsys):
    """未配置 DMS_DATABASE_URL 时拒绝执行（fail-fast，绝不 fallback 到 ini 地址）。"""
    monkeypatch.delenv("DMS_DATABASE_URL", raising=False)
    assert run_upgrade(assume_yes=True) == 2
    assert "DMS_DATABASE_URL" in capsys.readouterr().out