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测试审计；迁移自身不修改历史审计。
    # 0015降级守卫要求IP条目/机构分配/IP审计全部清空
    # （test_ip_entries等文件残留的审计记录会触发该守卫）。
    engine = create_engine(database_url)
    with engine.begin() as connection:
        connection.execute(
            text(
                "DELETE FROM sys_audit_log "
                "WHERE action_type = 'RESTORE_DOCUMENT' "
                "OR action_type IN "
                "('CREATE_IP_ENTRY', 'EDIT_IP_ENTRY', 'DELETE_IP_ENTRY', "
                "'ASSIGN_ORG_IP')"
            )
        )
        connection.execute(text("DELETE FROM sys_org_ip_assignment"))
        connection.execute(text("DELETE FROM sys_ip_entry"))
        # 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")