From 46c2a856c4d8952b7d2f02d7c3dcc154dc0bbdad Mon Sep 17 00:00:00 2001 From: InfinityPacer <160988576+InfinityPacer@users.noreply.github.com> Date: Wed, 12 Aug 2026 21:29:08 +0800 Subject: [PATCH] fix(database): harden fresh migration chain compatibility (#6278) --- database/versions/93f8cb6a4d1e_2_2_4.py | 45 +- database/versions/e6a1c4b8d2f0_2_2_13.py | 129 ++-- database/versions/f7b2d5c9a301_2_2_14.py | 42 +- tests/test_database_index_migration.py | 402 +++++++++++ ...base_media_identity_backfill_migrations.py | 683 ++++++++++++++++++ 5 files changed, 1234 insertions(+), 67 deletions(-) create mode 100644 tests/test_database_index_migration.py create mode 100644 tests/test_database_media_identity_backfill_migrations.py diff --git a/database/versions/93f8cb6a4d1e_2_2_4.py b/database/versions/93f8cb6a4d1e_2_2_4.py index 97c194f28..5aac9f739 100644 --- a/database/versions/93f8cb6a4d1e_2_2_4.py +++ b/database/versions/93f8cb6a4d1e_2_2_4.py @@ -145,6 +145,13 @@ DOWNGRADE_RESTORE_INDEXES = { def _load_schema_state(inspector: sa.Inspector): tables = set(inspector.get_table_names()) + table_columns = { + table_name: { + column["name"] + for column in inspector.get_columns(table_name) + } + for table_name in tables + } table_indexes = { table_name: { index["name"]: { @@ -155,7 +162,7 @@ def _load_schema_state(inspector: sa.Inspector): } for table_name in tables } - return tables, table_indexes + return tables, table_columns, table_indexes def _drop_index( @@ -215,10 +222,13 @@ def _create_index( index_name: str, columns: list[str], tables: set[str], + table_columns: dict[str, set[str]], table_indexes: dict[str, dict[str, dict[str, object]]], ) -> None: if table_name not in tables: return + if not set(columns).issubset(table_columns[table_name]): + return if index_name in table_indexes[table_name]: return if _has_index_signature(table_name, columns, tables, table_indexes, unique=False): @@ -231,8 +241,9 @@ def _create_index( def upgrade() -> None: + """以字段签名幂等替换 2.2.4 高频查询索引。""" inspector = sa.inspect(op.get_bind()) - tables, table_indexes = _load_schema_state(inspector) + tables, table_columns, table_indexes = _load_schema_state(inspector) for table_name, index_specs in REDUNDANT_ID_INDEXES.items(): for index_name, columns in index_specs: @@ -258,12 +269,20 @@ def upgrade() -> None: for table_name, index_specs in CREATE_INDEXES.items(): for index_name, columns in index_specs: - _create_index(table_name, index_name, columns, tables, table_indexes) + _create_index( + table_name, + index_name, + columns, + tables, + table_columns, + table_indexes, + ) def downgrade() -> None: + """移除组合索引并恢复适用于当前表结构的旧索引。""" inspector = sa.inspect(op.get_bind()) - tables, table_indexes = _load_schema_state(inspector) + tables, table_columns, table_indexes = _load_schema_state(inspector) for table_name, index_specs in CREATE_INDEXES.items(): for index_name, _ in index_specs: @@ -271,8 +290,22 @@ def downgrade() -> None: for table_name, index_specs in DOWNGRADE_RESTORE_INDEXES.items(): for index_name, columns in index_specs: - _create_index(table_name, index_name, columns, tables, table_indexes) + _create_index( + table_name, + index_name, + columns, + tables, + table_columns, + table_indexes, + ) for table_name, index_specs in REDUNDANT_ID_INDEXES.items(): for index_name, columns in index_specs: - _create_index(table_name, index_name, columns, tables, table_indexes) + _create_index( + table_name, + index_name, + columns, + tables, + table_columns, + table_indexes, + ) diff --git a/database/versions/e6a1c4b8d2f0_2_2_13.py b/database/versions/e6a1c4b8d2f0_2_2_13.py index 9e13649d6..b53c97e6f 100644 --- a/database/versions/e6a1c4b8d2f0_2_2_13.py +++ b/database/versions/e6a1c4b8d2f0_2_2_13.py @@ -36,55 +36,87 @@ def _has_column( ) +def _has_index( + inspector: sa.Inspector, + table_name: str, + index_name: str, +) -> bool: + """检查数据表是否已存在指定索引。""" + if table_name not in inspector.get_table_names(): + return False + return any( + index["name"] == index_name + for index in inspector.get_indexes(table_name) + ) + + +def _column_names(inspector: sa.Inspector, table_name: str) -> set[str]: + """读取数据表当前全部字段名。""" + if table_name not in inspector.get_table_names(): + return set() + return { + column["name"] + for column in inspector.get_columns(table_name) + } + + +def _ensure_column_and_index(column_name: str) -> None: + """独立补齐整理历史的规范身份字段及其索引。""" + table_name = "transferhistory" + index_name = f"ix_{table_name}_{column_name}" + inspector = sa.inspect(op.get_bind()) + if not _has_column(inspector, table_name, column_name): + op.add_column( + table_name, + sa.Column(column_name, sa.String(), nullable=True), + ) + + inspector = sa.inspect(op.get_bind()) + if not _has_index(inspector, table_name, index_name): + op.create_index(index_name, table_name, [column_name]) + + def upgrade() -> None: """升级整理历史数据源字段。""" - inspector = sa.inspect(op.get_bind()) - if not _has_column(inspector, "transferhistory", "media_source"): - op.add_column( - "transferhistory", - sa.Column("media_source", sa.String(), nullable=True), - ) - op.create_index( - "ix_transferhistory_media_source", - "transferhistory", - ["media_source"], - ) + _ensure_column_and_index("media_source") + _ensure_column_and_index("media_id") - inspector = sa.inspect(op.get_bind()) - if not _has_column(inspector, "transferhistory", "media_id"): - op.add_column( - "transferhistory", - sa.Column("media_id", sa.String(), nullable=True), - ) - op.create_index( - "ix_transferhistory_media_id", - "transferhistory", - ["media_id"], - ) - - transfer_history = sa.table( - "transferhistory", - sa.column("tmdbid", sa.Integer()), - sa.column("doubanid", sa.String()), + columns = _column_names(sa.inspect(op.get_bind()), "transferhistory") + table_columns = [ sa.column("media_source", sa.String()), sa.column("media_id", sa.String()), - ) + ] + if "tmdbid" in columns: + table_columns.append(sa.column("tmdbid", sa.Integer())) + if "doubanid" in columns: + table_columns.append(sa.column("doubanid", sa.String())) + transfer_history = sa.table("transferhistory", *table_columns) connection = op.get_bind() - connection.execute( - transfer_history.update() - .where(transfer_history.c.tmdbid.is_not(None)) - .where(transfer_history.c.media_id.is_(None)) - .values( - media_source="themoviedb", - media_id=sa.cast(transfer_history.c.tmdbid, sa.String()), + + if "tmdbid" in columns: + connection.execute( + transfer_history.update() + .where(transfer_history.c.tmdbid.is_not(None)) + .where(transfer_history.c.media_id.is_(None)) + .values( + media_source="themoviedb", + media_id=sa.cast(transfer_history.c.tmdbid, sa.String()), + ) ) - ) - connection.execute( + + if "doubanid" not in columns: + return + douban_update = ( transfer_history.update() - .where(transfer_history.c.tmdbid.is_(None)) .where(transfer_history.c.doubanid.is_not(None)) .where(transfer_history.c.media_id.is_(None)) - .values( + ) + if "tmdbid" in columns: + douban_update = douban_update.where( + transfer_history.c.tmdbid.is_(None) + ) + connection.execute( + douban_update.values( media_source="douban", media_id=transfer_history.c.doubanid, ) @@ -93,15 +125,12 @@ def upgrade() -> None: def downgrade() -> None: """回滚整理历史数据源字段。""" - inspector = sa.inspect(op.get_bind()) - if _has_column(inspector, "transferhistory", "media_id"): - op.drop_index("ix_transferhistory_media_id", table_name="transferhistory") - op.drop_column("transferhistory", "media_id") + for column_name in ("media_id", "media_source"): + index_name = f"ix_transferhistory_{column_name}" + inspector = sa.inspect(op.get_bind()) + if _has_index(inspector, "transferhistory", index_name): + op.drop_index(index_name, table_name="transferhistory") - inspector = sa.inspect(op.get_bind()) - if _has_column(inspector, "transferhistory", "media_source"): - op.drop_index( - "ix_transferhistory_media_source", - table_name="transferhistory", - ) - op.drop_column("transferhistory", "media_source") + inspector = sa.inspect(op.get_bind()) + if _has_column(inspector, "transferhistory", column_name): + op.drop_column("transferhistory", column_name) diff --git a/database/versions/f7b2d5c9a301_2_2_14.py b/database/versions/f7b2d5c9a301_2_2_14.py index 53e3b5362..46519b6a0 100644 --- a/database/versions/f7b2d5c9a301_2_2_14.py +++ b/database/versions/f7b2d5c9a301_2_2_14.py @@ -46,6 +46,17 @@ def _has_index( ) +def _column_names(table_name: str) -> set[str]: + """读取数据表当前全部字段名。""" + inspector = sa.inspect(op.get_bind()) + if table_name not in inspector.get_table_names(): + return set() + return { + column["name"] + for column in inspector.get_columns(table_name) + } + + def _add_columns(table_name: str, columns: Iterable[sa.Column]) -> None: """为指定表补充尚不存在的字段。""" for column in columns: @@ -61,22 +72,29 @@ def _create_index(table_name: str, index_name: str, columns: list[str]) -> None: op.create_index(index_name, table_name, columns) -def _backfill_media_identity(table_name: str, has_mediaid: bool = False) -> None: - """使用兼容 ID 幂等回填统一媒体身份。""" +def _backfill_media_identity(table_name: str) -> None: + """使用表中实际存在的兼容 ID 幂等回填统一媒体身份。""" + existing_columns = _column_names(table_name) columns = [ - sa.column("tmdbid", sa.Integer()), - sa.column("doubanid", sa.String()), - sa.column("bangumiid", sa.Integer()), - sa.column("anilistid", sa.Integer()), sa.column("media_source", sa.String()), sa.column("media_id", sa.String()), ] - if has_mediaid: - columns.append(sa.column("mediaid", sa.String())) + identity_columns = { + "mediaid": sa.String, + "tmdbid": sa.Integer, + "doubanid": sa.String, + "bangumiid": sa.Integer, + "anilistid": sa.Integer, + } + columns.extend( + sa.column(column_name, column_type()) + for column_name, column_type in identity_columns.items() + if column_name in existing_columns + ) table = sa.table(table_name, *columns) connection = op.get_bind() - if has_mediaid: + if "mediaid" in existing_columns: for prefix, source in ( ("tmdb", "themoviedb"), ("themoviedb", "themoviedb"), @@ -100,6 +118,8 @@ def _backfill_media_identity(table_name: str, has_mediaid: bool = False) -> None ("bangumi", "bangumiid"), ("anilist", "anilistid"), ): + if field not in existing_columns: + continue identity_column = table.c[field] connection.execute( table.update() @@ -162,8 +182,8 @@ def upgrade() -> None: ["type", "media_source", "media_id", "site"], ) - _backfill_media_identity("subscribe", has_mediaid=True) - _backfill_media_identity("subscribehistory", has_mediaid=True) + _backfill_media_identity("subscribe") + _backfill_media_identity("subscribehistory") _backfill_media_identity("downloadhistory") _backfill_media_identity("transferhistory") _backfill_media_identity("downloadfailure") diff --git a/tests/test_database_index_migration.py b/tests/test_database_index_migration.py new file mode 100644 index 000000000..0045ce7c4 --- /dev/null +++ b/tests/test_database_index_migration.py @@ -0,0 +1,402 @@ +import importlib +import os +from pathlib import Path +import subprocess +import sys +import uuid + +import psycopg2 +from psycopg2 import sql +import pytest +import sqlalchemy as sa +from alembic.migration import MigrationContext +from alembic.operations import Operations + + +MIGRATION_MODULE = "database.versions.93f8cb6a4d1e_2_2_4" +MEDIA_TABLES = ( + "subscribe", + "subscribehistory", + "downloadhistory", + "transferhistory", + "downloadfailure", + "mediaserveritem", +) +LEGACY_IDENTITY_COLUMNS = { + "tmdbid", + "imdbid", + "tvdbid", + "doubanid", + "bangumiid", + "anilistid", + "mediaid", +} +IDENTITY_INDEX_SIGNATURES = { + "subscribe": { + "ix_subscribe_media_identity": (("media_source", "media_id"), False), + }, + "subscribehistory": { + "ix_subscribehistory_media_identity": ( + ("media_source", "media_id"), False, + ), + }, + "downloadhistory": { + "ix_downloadhistory_media_identity": ( + ("media_source", "media_id"), False, + ), + }, + "transferhistory": { + "ix_transferhistory_media_identity": ( + ("media_source", "media_id"), False, + ), + }, + "downloadfailure": { + "ix_downloadfailure_media_identity_site": ( + ("type", "media_source", "media_id", "site"), False, + ), + }, + "mediaserveritem": { + "ix_mediaserveritem_media_identity_type": ( + ("media_source", "media_id", "item_type"), False, + ), + }, +} + + +CURRENT_SCHEMA_CHAIN_SCRIPT = """ +from app.testing.bootstrap import ensure_sites_stub + +# Alembic 会导入引用业务链的旧 revision；全新 CI 环境没有动态下发的 sites 模块。 +ensure_sites_stub() + +from alembic.config import Config +from alembic.script import ScriptDirectory +from sqlalchemy import inspect, text +from sqlalchemy.exc import IntegrityError + +from app.core.config import settings +from app.db import Engine +from app.db.init import init_db, update_db + +media_tables = {media_tables!r} +legacy_identity_columns = {legacy_identity_columns!r} +identity_index_signatures = {identity_index_signatures!r} + +config = Config() +config.set_main_option("script_location", str(settings.ROOT_PATH / "database")) +heads = ScriptDirectory.from_config(config).get_heads() +assert len(heads) == 1, heads + +init_db() +update_db() +update_db() + +with Engine.connect() as connection: + version = connection.execute( + text("SELECT version_num FROM alembic_version") + ).scalar_one() + inspector = inspect(connection) + assert version == heads[0], (version, heads) + + for table_name in media_tables: + columns = {{ + column["name"] + for column in inspector.get_columns(table_name) + }} + indexes = {{ + index["name"]: ( + tuple(index.get("column_names") or ()), + bool(index.get("unique")), + ) + for index in inspector.get_indexes(table_name) + }} + constraints = {{ + constraint["name"]: constraint.get("sqltext") or "" + for constraint in inspector.get_check_constraints(table_name) + }} + assert {{"media_source", "media_id"}}.issubset(columns), ( + table_name, + columns, + ) + assert legacy_identity_columns.isdisjoint(columns), ( + table_name, + columns, + ) + for index_name, signature in identity_index_signatures[table_name].items(): + assert indexes.get(index_name) == signature, ( + table_name, + index_name, + indexes, + ) + constraint_name = f"ck_{{table_name}}_media_identity" + assert constraint_name in constraints, ( + table_name, + constraints, + ) + normalized_sql = "".join( + constraints[constraint_name].lower().replace('"', '').split() + ) + for text_cast in ("::text[]", "::text", "::charactervarying"): + normalized_sql = normalized_sql.replace(text_cast, "") + for fragment in ( + "media_sourceisnull", + "media_idisnull", + "media_sourceisnotnull", + "media_idisnotnull", + "'themoviedb'", + "'anilist'", + ): + assert fragment in normalized_sql, ( + table_name, + constraints[constraint_name], + ) + assert any( + trim_form in normalized_sql + for trim_form in ( + "trim(media_id)", + "trim(bothfrommedia_id)", + ) + ), (table_name, constraints[constraint_name]) + assert "<>''" in normalized_sql, ( + table_name, + constraints[constraint_name], + ) + assert "<>'0'" in normalized_sql, ( + table_name, + constraints[constraint_name], + ) + + constraint_name = "ck_mediaserveritem_media_identity" + try: + with connection.begin_nested(): + connection.execute( + text( + "INSERT INTO mediaserveritem (media_source, media_id) " + "VALUES (:media_source, :media_id)" + ), + {{"media_source": "invalid_source", "media_id": "1"}}, + ) + except IntegrityError as error: + assert constraint_name in str(error.orig), str(error.orig) + else: + raise AssertionError("非法媒体身份未被具名检查约束拒绝") +""".format( + media_tables=MEDIA_TABLES, + legacy_identity_columns=LEGACY_IDENTITY_COLUMNS, + identity_index_signatures=IDENTITY_INDEX_SIGNATURES, +) + + +def _index_signatures( + connection, + table_name: str, +) -> dict[str, tuple[tuple[str, ...], bool]]: + """返回索引名称到字段顺序及唯一性的映射。""" + return { + index["name"]: ( + tuple(index.get("column_names") or ()), + bool(index.get("unique")), + ) + for index in sa.inspect(connection).get_indexes(table_name) + } + + +def _bind_migration(monkeypatch, connection): + """把历史 revision 绑定到当前 disposable connection。""" + migration = importlib.import_module(MIGRATION_MODULE) + context = MigrationContext.configure(connection) + monkeypatch.setattr(migration, "op", Operations(context)) + return migration + + +def _run_current_schema_chain( + repository: Path, + environment: dict[str, str], +) -> None: + """在隔离数据库中执行当前建表、完整升级及最终结构断言。""" + completed = subprocess.run( + [sys.executable, "-c", CURRENT_SCHEMA_CHAIN_SCRIPT], + cwd=repository, + env=environment, + capture_output=True, + text=True, + timeout=180, + check=False, + ) + + assert completed.returncode == 0, ( + f"stdout:\n{completed.stdout}\n" + f"stderr:\n{completed.stderr}" + ) + + +def test_index_migration_preserves_legacy_media_server_semantics( + monkeypatch, +) -> None: + """旧字段存在时应保持 2.2.4 的索引替换与回滚语义。""" + engine = sa.create_engine("sqlite://") + metadata = sa.MetaData() + media_server = sa.Table( + "mediaserveritem", + metadata, + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("tmdbid", sa.Integer()), + sa.Column("item_type", sa.String()), + ) + sa.Index("ix_mediaserveritem_id", media_server.c.id) + sa.Index("ix_mediaserveritem_tmdbid", media_server.c.tmdbid) + + with engine.begin() as connection: + metadata.create_all(connection) + migration = _bind_migration(monkeypatch, connection) + + migration.upgrade() + migration.upgrade() + + upgraded = _index_signatures(connection, "mediaserveritem") + assert upgraded.get("ix_mediaserveritem_tmdbid_item_type") == ( + ("tmdbid", "item_type"), False, + ) + assert "ix_mediaserveritem_tmdbid" not in upgraded + assert "ix_mediaserveritem_id" not in upgraded + + migration.downgrade() + + downgraded = _index_signatures(connection, "mediaserveritem") + assert "ix_mediaserveritem_tmdbid_item_type" not in downgraded + assert downgraded.get("ix_mediaserveritem_tmdbid") == ( + ("tmdbid",), False, + ) + assert downgraded.get("ix_mediaserveritem_id") == (("id",), False) + + +def test_index_migration_skips_only_indexes_with_missing_columns( + monkeypatch, +) -> None: + """当前 schema 应跳过旧字段索引，同时继续处理其他适用索引。""" + engine = sa.create_engine("sqlite://") + metadata = sa.MetaData() + media_server = sa.Table( + "mediaserveritem", + metadata, + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("media_source", sa.String()), + sa.Column("media_id", sa.String()), + sa.Column("item_type", sa.String()), + ) + sa.Index( + "ix_mediaserveritem_media_identity_type", + media_server.c.media_source, + media_server.c.media_id, + media_server.c.item_type, + ) + message = sa.Table( + "message", + metadata, + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("reg_time", sa.DateTime()), + ) + sa.Index("ix_message_reg_time", message.c.reg_time) + + with engine.begin() as connection: + metadata.create_all(connection) + migration = _bind_migration(monkeypatch, connection) + + migration.upgrade() + + media_indexes = _index_signatures(connection, "mediaserveritem") + message_indexes = _index_signatures(connection, "message") + assert "ix_mediaserveritem_tmdbid_item_type" not in media_indexes + assert media_indexes.get("ix_mediaserveritem_media_identity_type") == ( + ("media_source", "media_id", "item_type"), False, + ) + assert "ix_message_reg_time" not in message_indexes + assert message_indexes.get("ix_message_reg_time_id") == ( + ("reg_time", "id"), False, + ) + + migration.downgrade() + + media_indexes = _index_signatures(connection, "mediaserveritem") + message_indexes = _index_signatures(connection, "message") + assert "ix_mediaserveritem_tmdbid" not in media_indexes + assert media_indexes.get("ix_mediaserveritem_media_identity_type") == ( + ("media_source", "media_id", "item_type"), False, + ) + assert message_indexes.get("ix_message_reg_time") == ( + ("reg_time",), False, + ) + assert "ix_message_reg_time_id" not in message_indexes + + +def test_current_schema_reaches_current_alembic_head(tmp_path: Path) -> None: + """真实 fresh 启动链应到动态解析的唯一 head，且重复升级保持幂等。""" + repository = Path(__file__).resolve().parents[1] + environment = os.environ.copy() + environment.update({ + "CONFIG_DIR": str(tmp_path), + "DB_TYPE": "sqlite", + "SUPERUSER": "migration-test-admin", + "SUPERUSER_PASSWORD": "MigrationTestPassword123", + }) + _run_current_schema_chain(repository, environment) + + +def test_current_schema_reaches_current_alembic_head_on_postgresql( + tmp_path: Path, +) -> None: + """PostgreSQL fresh schema 应到唯一 head，不能被吞异常伪装成成功。""" + 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"p1_db1_{uuid.uuid4().hex}" + with psycopg2.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)) + ) + + repository = Path(__file__).resolve().parents[1] + environment = os.environ.copy() + environment.update({ + "CONFIG_DIR": str(tmp_path), + "DB_TYPE": "postgresql", + "DB_POSTGRESQL_HOST": host, + "DB_POSTGRESQL_PORT": port, + "DB_POSTGRESQL_DATABASE": database, + "DB_POSTGRESQL_USERNAME": username, + "DB_POSTGRESQL_PASSWORD": password, + "PGOPTIONS": f"-c search_path={schema}", + "SUPERUSER": "migration-test-admin", + "SUPERUSER_PASSWORD": "MigrationTestPassword123", + }) + try: + _run_current_schema_chain(repository, environment) + finally: + with psycopg2.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) + ) + ) diff --git a/tests/test_database_media_identity_backfill_migrations.py b/tests/test_database_media_identity_backfill_migrations.py new file mode 100644 index 000000000..d92432192 --- /dev/null +++ b/tests/test_database_media_identity_backfill_migrations.py @@ -0,0 +1,683 @@ +import importlib +import os +import uuid + +import psycopg2 +from psycopg2 import sql +import pytest +import sqlalchemy as sa +from alembic.migration import MigrationContext +from alembic.operations import Operations + + +E6_MIGRATION = "database.versions.e6a1c4b8d2f0_2_2_13" +F7_MIGRATION = "database.versions.f7b2d5c9a301_2_2_14" + +IDENTITY_COLUMN_TYPES = { + "mediaid": sa.String, + "tmdbid": sa.Integer, + "doubanid": sa.String, + "bangumiid": sa.Integer, + "anilistid": sa.Integer, + "media_source": sa.String, + "media_id": sa.String, +} +F7_INDEX_SIGNATURES = { + "subscribe": { + "ix_subscribe_anilistid": (("anilistid",), False), + "ix_subscribe_media_source": (("media_source",), False), + "ix_subscribe_media_id": (("media_id",), False), + "ix_subscribe_media_identity": ( + ("media_source", "media_id"), False, + ), + }, + "subscribehistory": { + "ix_subscribehistory_anilistid": (("anilistid",), False), + "ix_subscribehistory_media_source": (("media_source",), False), + "ix_subscribehistory_media_id": (("media_id",), False), + "ix_subscribehistory_media_identity": ( + ("media_source", "media_id"), False, + ), + }, + "downloadhistory": { + "ix_downloadhistory_bangumiid": (("bangumiid",), False), + "ix_downloadhistory_anilistid": (("anilistid",), False), + "ix_downloadhistory_media_source": (("media_source",), False), + "ix_downloadhistory_media_id": (("media_id",), False), + "ix_downloadhistory_media_identity": ( + ("media_source", "media_id"), False, + ), + }, + "transferhistory": { + "ix_transferhistory_bangumiid": (("bangumiid",), False), + "ix_transferhistory_anilistid": (("anilistid",), False), + "ix_transferhistory_media_identity": ( + ("media_source", "media_id"), False, + ), + }, + "downloadfailure": { + "ix_downloadfailure_media_identity_site": ( + ("type", "media_source", "media_id", "site"), False, + ), + }, +} + + +def _bind_migration(monkeypatch, connection, module_name: str): + """把历史 revision 绑定到当前 disposable connection。""" + migration = importlib.import_module(module_name) + context = MigrationContext.configure(connection) + monkeypatch.setattr(migration, "op", Operations(context)) + return migration + + +def _column_names(connection, table_name: str) -> set[str]: + """返回指定测试表的字段集合。""" + return { + column["name"] + for column in sa.inspect(connection).get_columns(table_name) + } + + +def _index_signatures( + connection, + table_name: str, +) -> dict[str, tuple[tuple[str, ...], bool]]: + """返回索引名称到字段顺序及唯一性的映射。""" + return { + index["name"]: ( + tuple(index.get("column_names") or ()), + bool(index.get("unique")), + ) + for index in sa.inspect(connection).get_indexes(table_name) + } + + +def _assert_index_signatures( + actual: dict[str, tuple[tuple[str, ...], bool]], + expected: dict[str, tuple[tuple[str, ...], bool]], +) -> None: + """断言关键索引的字段顺序和唯一性均符合迁移契约。""" + for index_name, signature in expected.items(): + assert actual.get(index_name) == signature, (index_name, actual) + + +def _rows(connection, table_name: str) -> list[dict]: + """按主键读取迁移后的测试数据。""" + table = sa.Table(table_name, sa.MetaData(), autoload_with=connection) + return list( + connection.execute(sa.select(table).order_by(table.c.id)).mappings() + ) + + +def _create_transferhistory( + connection, + columns: tuple[str, ...], + indexes: tuple[tuple[str, tuple[str, ...]], ...] = (), +) -> sa.Table: + """创建具有指定兼容字段和索引的最小整理历史表。""" + metadata = sa.MetaData() + table = sa.Table( + "transferhistory", + metadata, + sa.Column("id", sa.Integer(), primary_key=True), + *( + sa.Column(name, IDENTITY_COLUMN_TYPES[name]()) + for name in columns + ), + ) + for name, fields in indexes: + sa.Index(name, *(table.c[field] for field in fields)) + metadata.create_all(connection) + return table + + +def test_e6_backfill_preserves_priority_existing_identity_and_indexes( + monkeypatch, +) -> None: + """e6 应保持 TMDB 优先级、豆瓣兜底和已有规范身份。""" + engine = sa.create_engine("sqlite://") + with engine.begin() as connection: + table = _create_transferhistory( + connection, + ("tmdbid", "doubanid", "media_source", "media_id"), + (("ix_transferhistory_media_source", ("media_source",)),), + ) + connection.execute(table.insert(), [ + { + "id": 1, "tmdbid": 123, "doubanid": "ignored", + "media_source": None, "media_id": None, + }, + { + "id": 2, "tmdbid": None, "doubanid": "456", + "media_source": None, "media_id": None, + }, + { + "id": 3, + "tmdbid": 789, + "doubanid": "also-ignored", + "media_source": "plugin_source", + "media_id": "custom-1", + }, + ]) + migration = _bind_migration(monkeypatch, connection, E6_MIGRATION) + + migration.upgrade() + migration.upgrade() + + migrated_rows = _rows(connection, "transferhistory") + indexes = _index_signatures(connection, "transferhistory") + + assert (migrated_rows[0]["media_source"], migrated_rows[0]["media_id"]) == ( + "themoviedb", "123", + ) + assert (migrated_rows[1]["media_source"], migrated_rows[1]["media_id"]) == ( + "douban", "456", + ) + assert (migrated_rows[2]["media_source"], migrated_rows[2]["media_id"]) == ( + "plugin_source", "custom-1", + ) + _assert_index_signatures(indexes, { + "ix_transferhistory_media_source": (("media_source",), False), + "ix_transferhistory_media_id": (("media_id",), False), + }) + + +@pytest.mark.parametrize( + ("legacy_columns", "values", "expected_identity"), + ( + (("tmdbid",), {"tmdbid": 123}, ("themoviedb", "123")), + (("doubanid",), {"doubanid": "456"}, ("douban", "456")), + ((), {}, (None, None)), + ), +) +def test_e6_skips_only_missing_legacy_sources( + monkeypatch, + legacy_columns, + values, + expected_identity, +) -> None: + """e6 应仅执行物理存在的旧来源字段回填。""" + engine = sa.create_engine("sqlite://") + with engine.begin() as connection: + table = _create_transferhistory(connection, legacy_columns) + connection.execute(table.insert(), {"id": 1, **values}) + migration = _bind_migration(monkeypatch, connection, E6_MIGRATION) + + migration.upgrade() + migration.upgrade() + + migrated = _rows(connection, "transferhistory")[0] + + assert (migrated["media_source"], migrated["media_id"]) == expected_identity + + +def test_e6_retry_completes_interrupted_column_and_index_state( + monkeypatch, +) -> None: + """e6 重试应独立补齐缺失的 revision-owned 字段和索引。""" + engine = sa.create_engine("sqlite://") + with engine.begin() as connection: + table = _create_transferhistory( + connection, + ("tmdbid", "doubanid", "media_source"), + (("ix_transferhistory_media_source", ("media_source",)),), + ) + connection.execute( + table.insert(), + {"id": 1, "tmdbid": None, "doubanid": "456"}, + ) + migration = _bind_migration(monkeypatch, connection, E6_MIGRATION) + + migration.upgrade() + migration.upgrade() + + columns = _column_names(connection, "transferhistory") + indexes = _index_signatures(connection, "transferhistory") + migrated = _rows(connection, "transferhistory")[0] + + assert {"media_source", "media_id"}.issubset(columns) + _assert_index_signatures(indexes, { + "ix_transferhistory_media_source": (("media_source",), False), + "ix_transferhistory_media_id": (("media_id",), False), + }) + assert (migrated["media_source"], migrated["media_id"]) == ( + "douban", "456", + ) + + +def _create_f7_tables( + connection, + layouts: dict[str, tuple[str, ...]], + indexes: dict[str, tuple[tuple[str, tuple[str, ...]], ...]] | None = None, + transferhistory_identity: bool = True, +) -> dict[str, sa.Table]: + """创建五张具有不同历史字段状态的最小媒体表。""" + metadata = sa.MetaData() + indexes = indexes or {} + transferhistory_columns = ( + ("media_source", "media_id") if transferhistory_identity else () + ) + base_columns = { + "subscribe": (), + "subscribehistory": (), + "downloadhistory": (), + "transferhistory": transferhistory_columns, + "downloadfailure": (), + } + tables = {} + for table_name, required_columns in base_columns.items(): + column_names = tuple(dict.fromkeys(required_columns + layouts.get(table_name, ()))) + columns = [sa.Column("id", sa.Integer(), primary_key=True)] + if table_name == "downloadfailure": + columns.extend(( + sa.Column("type", sa.String()), + sa.Column("site", sa.Integer()), + )) + columns.extend( + sa.Column(name, IDENTITY_COLUMN_TYPES[name]()) + for name in column_names + ) + table = sa.Table(table_name, metadata, *columns) + for index_name, fields in indexes.get(table_name, ()): + sa.Index(index_name, *(table.c[field] for field in fields)) + tables[table_name] = table + metadata.create_all(connection) + return tables + + +def test_f7_backfill_uses_physical_columns_and_preserves_priority( + monkeypatch, +) -> None: + """f7 应先用带前缀 ID，再按四类来源字段顺序回填。""" + engine = sa.create_engine("sqlite://") + layouts = { + "subscribe": ( + "mediaid", "tmdbid", "doubanid", "bangumiid", "anilistid", + "media_source", "media_id", + ), + "subscribehistory": ("mediaid",), + "downloadhistory": ("doubanid",), + "transferhistory": ("tmdbid",), + "downloadfailure": (), + } + with engine.begin() as connection: + tables = _create_f7_tables(connection, layouts) + connection.execute(tables["subscribe"].insert(), [ + { + "id": 1, "mediaid": "douban:prefix-1", "tmdbid": 101, + "doubanid": "201", "bangumiid": 301, "anilistid": 401, + "media_source": None, "media_id": None, + }, + { + "id": 2, "mediaid": None, "tmdbid": 102, "doubanid": "202", + "bangumiid": 302, "anilistid": 402, + "media_source": None, "media_id": None, + }, + { + "id": 3, "mediaid": None, "tmdbid": None, + "doubanid": "203", "bangumiid": 303, + "anilistid": 403, + "media_source": None, "media_id": None, + }, + { + "id": 4, "mediaid": None, "tmdbid": None, + "doubanid": None, "bangumiid": 304, "anilistid": 404, + "media_source": None, "media_id": None, + }, + { + "id": 5, "mediaid": None, "tmdbid": None, + "doubanid": None, "bangumiid": None, "anilistid": 405, + "media_source": None, "media_id": None, + }, + { + "id": 6, "mediaid": "tmdb:106", "tmdbid": 106, + "doubanid": None, "bangumiid": None, "anilistid": None, + "media_source": "plugin_source", "media_id": "custom-6", + }, + ]) + connection.execute( + tables["subscribehistory"].insert(), + {"id": 1, "mediaid": "anilist:154587"}, + ) + connection.execute( + tables["downloadhistory"].insert(), + {"id": 1, "doubanid": "35209731"}, + ) + connection.execute( + tables["transferhistory"].insert(), + {"id": 1, "tmdbid": 209867}, + ) + connection.execute(tables["downloadfailure"].insert(), { + "id": 1, "type": "电视剧", "site": 1, + }) + migration = _bind_migration(monkeypatch, connection, F7_MIGRATION) + + migration.upgrade() + migration.upgrade() + + subscribe_rows = _rows(connection, "subscribe") + partial_rows = { + name: _rows(connection, name)[0] + for name in ( + "subscribehistory", "downloadhistory", "transferhistory", + "downloadfailure", + ) + } + + assert [ + (row["media_source"], row["media_id"]) + for row in subscribe_rows + ] == [ + ("douban", "prefix-1"), + ("themoviedb", "102"), + ("douban", "203"), + ("bangumi", "304"), + ("anilist", "405"), + ("plugin_source", "custom-6"), + ] + assert ( + partial_rows["subscribehistory"]["media_source"], + partial_rows["subscribehistory"]["media_id"], + ) == ("anilist", "154587") + assert ( + partial_rows["downloadhistory"]["media_source"], + partial_rows["downloadhistory"]["media_id"], + ) == ("douban", "35209731") + assert ( + partial_rows["transferhistory"]["media_source"], + partial_rows["transferhistory"]["media_id"], + ) == ("themoviedb", "209867") + assert partial_rows["downloadfailure"]["media_id"] is None + + +def test_f7_retry_completes_mixed_partial_table_states(monkeypatch) -> None: + """f7 重试应跨表补齐混合缺失的字段和 revision-owned 索引。""" + engine = sa.create_engine("sqlite://") + layouts = { + "subscribe": ( + "mediaid", "tmdbid", "anilistid", "media_source", "media_id", + ), + "subscribehistory": ("mediaid", "doubanid", "media_source"), + "downloadhistory": ("bangumiid", "media_id"), + "transferhistory": ("tmdbid", "bangumiid"), + "downloadfailure": ("anilistid", "media_source", "media_id"), + } + existing_indexes = { + "subscribe": (("ix_subscribe_anilistid", ("anilistid",)),), + "subscribehistory": ( + ("ix_subscribehistory_media_source", ("media_source",)), + ), + "downloadhistory": ( + ("ix_downloadhistory_bangumiid", ("bangumiid",)), + ), + "transferhistory": ( + ( + "ix_transferhistory_media_identity", + ("media_source", "media_id"), + ), + ), + "downloadfailure": ( + ( + "ix_downloadfailure_media_identity_site", + ("type", "media_source", "media_id", "site"), + ), + ), + } + with engine.begin() as connection: + tables = _create_f7_tables(connection, layouts, existing_indexes) + connection.execute(tables["subscribe"].insert(), { + "id": 1, + "mediaid": "douban:88", + "tmdbid": 11, + "media_source": "plugin_source", + "media_id": "preserved-1", + }) + connection.execute(tables["subscribehistory"].insert(), { + "id": 1, "mediaid": "tmdb:22", "doubanid": "33", + }) + connection.execute(tables["downloadhistory"].insert(), [ + {"id": 1, "bangumiid": 44, "media_id": None}, + {"id": 2, "bangumiid": 45, "media_id": "preserved-2"}, + ]) + connection.execute( + tables["transferhistory"].insert(), + {"id": 1, "tmdbid": 55, "bangumiid": 56}, + ) + connection.execute(tables["downloadfailure"].insert(), { + "id": 1, "type": "动画", "site": 1, "anilistid": 66, + }) + migration = _bind_migration(monkeypatch, connection, F7_MIGRATION) + + migration.upgrade() + migration.upgrade() + + columns = { + name: _column_names(connection, name) + for name in tables + } + indexes = { + name: _index_signatures(connection, name) + for name in tables + } + rows = {name: _rows(connection, name) for name in tables} + + assert {"anilistid", "media_source", "media_id"}.issubset(columns["subscribe"]) + assert {"anilistid", "media_source", "media_id"}.issubset( + columns["subscribehistory"] + ) + assert { + "bangumiid", "anilistid", "media_source", "media_id", + }.issubset(columns["downloadhistory"]) + assert {"bangumiid", "anilistid"}.issubset(columns["transferhistory"]) + assert { + "bangumiid", "anilistid", "media_source", "media_id", + }.issubset(columns["downloadfailure"]) + + for table_name, expected in F7_INDEX_SIGNATURES.items(): + _assert_index_signatures(indexes[table_name], expected) + + assert (rows["subscribe"][0]["media_source"], rows["subscribe"][0]["media_id"]) == ( + "plugin_source", "preserved-1", + ) + assert ( + rows["subscribehistory"][0]["media_source"], + rows["subscribehistory"][0]["media_id"], + ) == ("themoviedb", "22") + assert ( + rows["downloadhistory"][0]["media_source"], + rows["downloadhistory"][0]["media_id"], + ) == ("bangumi", "44") + assert rows["downloadhistory"][1]["media_source"] is None + assert rows["downloadhistory"][1]["media_id"] == "preserved-2" + assert ( + rows["transferhistory"][0]["media_source"], + rows["transferhistory"][0]["media_id"], + ) == ("themoviedb", "55") + assert ( + rows["downloadfailure"][0]["media_source"], + rows["downloadfailure"][0]["media_id"], + ) == ("anilist", "66") + + +def _exercise_e6_f7_round_trip(monkeypatch, engine: sa.Engine) -> None: + """验证两个历史 revision 可升级、降级并再次升级。""" + layouts = { + "subscribe": ("mediaid", "tmdbid", "doubanid", "bangumiid"), + "subscribehistory": ("mediaid", "tmdbid", "doubanid", "bangumiid"), + "downloadhistory": ("tmdbid", "doubanid"), + "transferhistory": ("tmdbid", "doubanid"), + "downloadfailure": ("tmdbid", "doubanid"), + } + with engine.begin() as connection: + tables = _create_f7_tables( + connection, + layouts, + transferhistory_identity=False, + ) + connection.execute(tables["subscribe"].insert(), { + "id": 1, "mediaid": "douban:88", "tmdbid": 11, + }) + connection.execute(tables["subscribehistory"].insert(), { + "id": 1, "mediaid": None, "tmdbid": 22, + }) + connection.execute(tables["downloadhistory"].insert(), { + "id": 1, "tmdbid": None, "doubanid": "33", + }) + connection.execute(tables["transferhistory"].insert(), { + "id": 1, "tmdbid": 55, "doubanid": "ignored", + }) + connection.execute(tables["downloadfailure"].insert(), { + "id": 1, "type": "动画", "site": 1, + "tmdbid": None, "doubanid": "66", + }) + + e6_migration = _bind_migration(monkeypatch, connection, E6_MIGRATION) + e6_migration.upgrade() + _assert_index_signatures( + _index_signatures(connection, "transferhistory"), + { + "ix_transferhistory_media_source": ( + ("media_source",), False, + ), + "ix_transferhistory_media_id": (("media_id",), False), + }, + ) + e6_migration.downgrade() + assert "media_source" not in _column_names(connection, "transferhistory") + assert "media_id" not in _column_names(connection, "transferhistory") + assert "ix_transferhistory_media_source" not in _index_signatures( + connection, "transferhistory" + ) + e6_migration.upgrade() + transfer_row = _rows(connection, "transferhistory")[0] + assert (transfer_row["media_source"], transfer_row["media_id"]) == ( + "themoviedb", "55", + ) + + pre_f7_columns = { + "subscribe": { + "id", "mediaid", "tmdbid", "doubanid", "bangumiid", + }, + "subscribehistory": { + "id", "mediaid", "tmdbid", "doubanid", "bangumiid", + }, + "downloadhistory": {"id", "tmdbid", "doubanid"}, + "transferhistory": { + "id", "tmdbid", "doubanid", "media_source", "media_id", + }, + "downloadfailure": { + "id", "type", "site", "tmdbid", "doubanid", + }, + } + for table_name, expected_columns in pre_f7_columns.items(): + assert _column_names(connection, table_name) == expected_columns + + f7_migration = _bind_migration(monkeypatch, connection, F7_MIGRATION) + f7_migration.upgrade() + for table_name, expected in F7_INDEX_SIGNATURES.items(): + _assert_index_signatures( + _index_signatures(connection, table_name), + expected, + ) + f7_migration.downgrade() + for table_name, expected_columns in pre_f7_columns.items(): + assert _column_names(connection, table_name) == expected_columns + for table_name, expected in F7_INDEX_SIGNATURES.items(): + indexes = _index_signatures(connection, table_name) + assert set(expected).isdisjoint(indexes), (table_name, indexes) + + f7_migration.upgrade() + for table_name, expected in F7_INDEX_SIGNATURES.items(): + _assert_index_signatures( + _index_signatures(connection, table_name), + expected, + ) + rows = { + table_name: _rows(connection, table_name)[0] + for table_name in layouts + } + + assert (rows["subscribe"]["media_source"], rows["subscribe"]["media_id"]) == ( + "douban", "88", + ) + assert ( + rows["subscribehistory"]["media_source"], + rows["subscribehistory"]["media_id"], + ) == ("themoviedb", "22") + assert ( + rows["downloadhistory"]["media_source"], + rows["downloadhistory"]["media_id"], + ) == ("douban", "33") + assert ( + rows["transferhistory"]["media_source"], + rows["transferhistory"]["media_id"], + ) == ("themoviedb", "55") + assert ( + rows["downloadfailure"]["media_source"], + rows["downloadfailure"]["media_id"], + ) == ("douban", "66") + + +def test_e6_f7_round_trip_on_sqlite(monkeypatch) -> None: + """SQLite 应支持两个 revision 的 upgrade/downgrade/re-upgrade。""" + _exercise_e6_f7_round_trip(monkeypatch, sa.create_engine("sqlite://")) + + +def test_e6_f7_round_trip_on_postgresql(monkeypatch) -> None: + """已配置的 PostgreSQL 15 应执行与 SQLite 相同的往返路径。""" + 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"p1_db1_roundtrip_{uuid.uuid4().hex}" + with psycopg2.connect( + host=host, + port=port, + dbname=database, + user=username, + password=password, + ) as connection: + assert connection.server_version // 10000 == 15, connection.server_version + 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+psycopg2", + username=username, + password=password, + host=host, + port=int(port), + database=database, + ), + connect_args={"options": f"-csearch_path={schema}"}, + ) + _exercise_e6_f7_round_trip(monkeypatch, engine) + finally: + if engine is not None: + engine.dispose() + with psycopg2.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) + ) + )