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、0015，且两者 upgrade 不含数据删除。"""
    script = _script()
    assert script.get_current_head() == "0015_login_ip_restriction"
    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",
    ]


def test_run_upgrade_lists_pending_steps_excluding_current(monkeypatch, capsys):
    """0013 升级到 head：待执行清单只含 0014/0015，当前版本不重复出现。"""
    monkeypatch.setenv("DMS_DATABASE_URL", "mysql+pymysql://u:secret@127.0.0.1:3306/dms")
    versions = iter(["0013_document_password_lock", "0015_login_ip_restriction"])
    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 "  - 0014_document_relations" in out
    assert "  - 0015_login_ip_restriction" in out
    assert "  - 0013" not in out
    assert "secret" not in out  # 输出中的地址必须已脱敏


def test_mask_password():
    url = "mysql+pymysql://dms_app:asdqwe123!@82.156.83.248:3306/dms?charset=utf8mb4"
    assert (
        _mask_password(url)
        == "mysql+pymysql://dms_app:***@82.156.83.248: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