mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-30 20:05:42 +08:00
72 lines
2.4 KiB
Python
72 lines
2.4 KiB
Python
"""订阅搜索业务阶段字段的 Alembic 可逆迁移测试。"""
|
|
|
|
import importlib
|
|
|
|
import sqlalchemy as sa
|
|
from alembic.migration import MigrationContext
|
|
from alembic.operations import Operations
|
|
|
|
MIGRATION = "database.versions.f3c8a1d6b2e9_3_0_22"
|
|
|
|
|
|
def _bind_migration(monkeypatch, connection):
|
|
"""把 3.0.22 迁移绑定到隔离 SQLite 连接。"""
|
|
migration = importlib.import_module(MIGRATION)
|
|
monkeypatch.setattr(
|
|
migration,
|
|
"op",
|
|
Operations(MigrationContext.configure(connection)),
|
|
)
|
|
return migration
|
|
|
|
|
|
def _create_legacy_search_task(connection) -> None:
|
|
"""创建迁移前的最小订阅搜索任务表和一条排队记录。"""
|
|
metadata = sa.MetaData()
|
|
task = sa.Table(
|
|
"subscriptionsearchtask",
|
|
metadata,
|
|
sa.Column("id", sa.Integer(), primary_key=True),
|
|
sa.Column("task_id", sa.String(64), nullable=False),
|
|
sa.Column("state", sa.String(32), nullable=False),
|
|
)
|
|
metadata.create_all(connection)
|
|
connection.execute(task.insert(), {"id": 1, "task_id": "task-1", "state": "queued"})
|
|
|
|
|
|
def test_subscription_status_migration_upgrade_downgrade_reupgrade(monkeypatch) -> None:
|
|
"""存量任务应获得默认阶段，迁移可重复执行并完整回滚。"""
|
|
engine = sa.create_engine("sqlite://")
|
|
with engine.begin() as connection:
|
|
_create_legacy_search_task(connection)
|
|
migration = _bind_migration(monkeypatch, connection)
|
|
|
|
migration.upgrade()
|
|
migration.upgrade()
|
|
|
|
columns = {
|
|
column["name"]
|
|
for column in sa.inspect(connection).get_columns("subscriptionsearchtask")
|
|
}
|
|
assert {"phase", "current_site_id"}.issubset(columns)
|
|
row = connection.execute(
|
|
sa.text(
|
|
"SELECT phase, current_site_id FROM subscriptionsearchtask WHERE id = 1"
|
|
)
|
|
).mappings().one()
|
|
assert dict(row) == {"phase": "queued", "current_site_id": None}
|
|
|
|
migration.downgrade()
|
|
downgraded = {
|
|
column["name"]
|
|
for column in sa.inspect(connection).get_columns("subscriptionsearchtask")
|
|
}
|
|
assert "phase" not in downgraded
|
|
assert "current_site_id" not in downgraded
|
|
|
|
migration.upgrade()
|
|
assert connection.execute(
|
|
sa.text("SELECT phase FROM subscriptionsearchtask WHERE id = 1")
|
|
).scalar_one() == "queued"
|
|
engine.dispose()
|