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

from __future__ import annotations

import os
from urllib.parse import urlparse

import pytest
from alembic import command
from alembic.config import Config
from sqlalchemy import create_engine, text


@pytest.mark.integration
def test_mysql_upgrade_and_downgrade_when_test_database_is_configured(
    monkeypatch,
):
    database_url = os.environ.get("DMS_TEST_DATABASE_URL", "").strip()
    if not database_url:
        pytest.skip("未配置DMS_TEST_DATABASE_URL，未执行真实MySQL迁移")
    if not database_url.startswith("mysql+pymysql://"):
        pytest.fail("DMS_TEST_DATABASE_URL必须使用mysql+pymysql")

    database_name = urlparse(database_url).path.lstrip("/")
    if not database_name.endswith("_test"):
        pytest.fail("为避免误操作，MySQL测试库名称必须以_test结尾")

    monkeypatch.setenv("DMS_DATABASE_URL", database_url)
    config = Config("migrations/alembic.ini")
    command.upgrade(config, "head")
    # 仅清理独立测试库中的B8测试审计；迁移自身不修改历史审计。
    engine = create_engine(database_url)
    with engine.begin() as connection:
        connection.execute(
            text(
                "DELETE FROM sys_audit_log "
                "WHERE action_type = 'RESTORE_DOCUMENT'"
            )
        )
    engine.dispose()
    command.downgrade(config, "base")
    command.upgrade(config, "head")