MoviePilot/tests/test_plugin_identity_migration.py

"""插件身份表 Alembic 迁移测试。"""

import importlib
import os
import uuid

import pytest
import sqlalchemy as sa
from alembic.migration import MigrationContext
from alembic.operations import Operations
from sqlalchemy.dialects import postgresql
from sqlalchemy.schema import CreateTable

try:
    import psycopg2 as postgres_driver
    from psycopg2 import sql
    POSTGRESQL_DIALECT = "postgresql+psycopg2"
except ModuleNotFoundError:
    import psycopg as postgres_driver
    from psycopg import sql
    POSTGRESQL_DIALECT = "postgresql+psycopg"

from app.db.models.pluginidentity import PluginIdentity

MIGRATION = "database.versions.d2e4f6a8b0c1_3_0_9"


def _bind_migration(monkeypatch, connection):
    """把迁移绑定到隔离数据库连接。"""
    migration = importlib.import_module(MIGRATION)
    context = MigrationContext.configure(connection)
    monkeypatch.setattr(migration, "op", Operations(context))
    return migration


def test_plugin_identity_migration_upgrades_twice_and_downgrades(
    monkeypatch,
) -> None:
    """旧 SQLite schema 应可重复升级并完整删除 dormant 身份表。"""
    engine = sa.create_engine("sqlite://")
    with engine.begin() as connection:
        migration = _bind_migration(monkeypatch, connection)

        migration.upgrade()
        migration.upgrade()

        inspector = sa.inspect(connection)
        assert "pluginidentity" in inspector.get_table_names()
        columns = {
            column["name"] for column in inspector.get_columns("pluginidentity")
        }
        assert columns == {column.name for column in PluginIdentity.__table__.columns}
        unique_constraints = {
            constraint["name"]: tuple(constraint["column_names"])
            for constraint in inspector.get_unique_constraints("pluginidentity")
        }
        assert unique_constraints["uq_pluginidentity_normalized_plugin_id"] == (
            "normalized_plugin_id",
        )
        check_constraints = {
            constraint["name"]
            for constraint in inspector.get_check_constraints("pluginidentity")
        }
        assert check_constraints == {
            "ck_pluginidentity_normalized_plugin_id",
            "ck_pluginidentity_revision",
        }

        migration.downgrade()
        assert "pluginidentity" not in sa.inspect(connection).get_table_names()


def test_plugin_identity_migration_accepts_fresh_current_schema(monkeypatch) -> None:
    """create_all 已建当前表时重复升级不得创建冲突对象。"""
    engine = sa.create_engine("sqlite://")
    with engine.begin() as connection:
        PluginIdentity.__table__.create(connection)
        migration = _bind_migration(monkeypatch, connection)

        migration.upgrade()
        migration.upgrade()

        assert {
            column["name"]
            for column in sa.inspect(connection).get_columns("pluginidentity")
        } == {column.name for column in PluginIdentity.__table__.columns}


def test_plugin_identity_migration_matches_postgresql_identity() -> None:
    """独立 Alembic 路径应保留 PostgreSQL 循环 Identity 主键。"""
    migration = importlib.import_module(MIGRATION)
    metadata = sa.MetaData()
    table = sa.Table(
        "pluginidentity",
        metadata,
        migration._id_column("postgresql"),
        sa.PrimaryKeyConstraint("id"),
    )

    identity = table.c.id.identity
    assert identity is not None
    assert identity.start == 1
    assert identity.cycle is True
    ddl = str(CreateTable(table).compile(dialect=postgresql.dialect()))
    assert "GENERATED BY DEFAULT AS IDENTITY" in ddl
    assert "CYCLE" in ddl


def test_plugin_identity_migration_runs_on_postgresql(monkeypatch) -> None:
    """已配置的隔离 PostgreSQL 应执行真实建表、约束和回滚。"""
    prefix = "MOVIEPILOT_TEST_POSTGRESQL_"
    host = os.getenv(f"{prefix}HOST")
    database = os.getenv(f"{prefix}DATABASE")
    username = os.getenv(f"{prefix}USERNAME")
    if not host or not database or not username:
        pytest.skip("未配置隔离 PostgreSQL migration 测试库")

    port = os.getenv(f"{prefix}PORT", "5432")
    password = os.getenv(f"{prefix}PASSWORD", "")
    schema = f"plugin_identity_{uuid.uuid4().hex}"
    with postgres_driver.connect(
        host=host,
        port=port,
        dbname=database,
        user=username,
        password=password,
    ) as connection:
        connection.autocommit = True
        with connection.cursor() as cursor:
            cursor.execute(
                sql.SQL("CREATE SCHEMA {}").format(sql.Identifier(schema))
            )

    engine = None
    try:
        engine = sa.create_engine(
            sa.URL.create(
                POSTGRESQL_DIALECT,
                username=username,
                password=password,
                host=host,
                port=int(port),
                database=database,
            ),
            connect_args={"options": f"-csearch_path={schema}"},
        )
        with engine.begin() as connection:
            migration = _bind_migration(monkeypatch, connection)
            migration.upgrade()
            migration.upgrade()

            table = sa.Table(
                "pluginidentity",
                sa.MetaData(),
                autoload_with=connection,
            )
            inserted_id = connection.execute(
                table.insert().values(
                    plugin_id="DemoPlugin",
                    normalized_plugin_id="demoplugin",
                    trusted_source_type="unknown",
                    binding_basis="legacy_unbound",
                    payload_source_type="unknown",
                    revision=1,
                    created_at="2026-08-25T12:00:00+00:00",
                    updated_at="2026-08-25T12:00:00+00:00",
                ).returning(table.c.id)
            ).scalar_one()
            assert inserted_id == 1

            with pytest.raises(sa.exc.IntegrityError):
                with connection.begin_nested():
                    connection.execute(
                        table.insert().values(
                            plugin_id="UppercaseKey",
                            normalized_plugin_id="UppercaseKey",
                            trusted_source_type="unknown",
                            binding_basis="legacy_unbound",
                            payload_source_type="unknown",
                            revision=1,
                            created_at="2026-08-25T12:00:00+00:00",
                            updated_at="2026-08-25T12:00:00+00:00",
                        )
                    )

            migration.downgrade()
            assert "pluginidentity" not in sa.inspect(connection).get_table_names()
    finally:
        if engine is not None:
            engine.dispose()
        with postgres_driver.connect(
            host=host,
            port=port,
            dbname=database,
            user=username,
            password=password,
        ) as connection:
            connection.autocommit = True
            with connection.cursor() as cursor:
                cursor.execute(
                    sql.SQL("DROP SCHEMA IF EXISTS {} CASCADE").format(
                        sql.Identifier(schema)
                    )
                )