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

from __future__ import annotations

import os
from pathlib import Path
from urllib.parse import urlparse

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

# alembic配置按测试文件位置解析，不依赖 pytest 启动目录。
ALEMBIC_INI = Path(__file__).resolve().parents[2] / "migrations" / "alembic.ini"


@pytest.mark.integration
def test_mysql_q3_upgrade_downgrade_upgrade_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(str(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'"
            )
        )
        # 0003明确阻止正文结果的有损降级；测试库先回到可安全降级状态。
        # 0019同样阻止已有提取失败记录的降级，一并清空。
        connection.execute(
            text(
                "UPDATE doc_document SET content_text = NULL, "
                "content_extract_status = 'UNSUPPORTED', "
                "content_extract_error = NULL, "
                "content_extracted_at = NULL, "
                "search_text = CONCAT_WS(' ', document_name, summary)"
            )
        )
    engine.dispose()
    command.downgrade(config, "0002_add_restore_audit_action")
    command.upgrade(config, "head")