mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-08-15 02:54:20 +08:00
107 lines
3.6 KiB
Python
107 lines
3.6 KiB
Python
import importlib
|
|
|
|
import sqlalchemy as sa
|
|
from alembic.migration import MigrationContext
|
|
from alembic.operations import Operations
|
|
from sqlalchemy.dialects import postgresql
|
|
from sqlalchemy.schema import CreateTable
|
|
|
|
from app.db.models.agenttask import AgentTask
|
|
from app.db.models.agenttaskrun import AgentTaskRun
|
|
|
|
|
|
MIGRATION = "database.versions.f4c8d2a7b1e6_3_0_6"
|
|
|
|
|
|
def _bind_migration(monkeypatch, connection):
|
|
"""把迁移绑定到隔离数据库连接。"""
|
|
migration = importlib.import_module(MIGRATION)
|
|
context = MigrationContext.configure(connection)
|
|
monkeypatch.setattr(migration, "op", Operations(context))
|
|
return migration
|
|
|
|
|
|
def _legacy_agent_task(metadata: sa.MetaData) -> None:
|
|
"""建立迁移前的最小 AgentTask 表。"""
|
|
sa.Table(
|
|
"agenttask",
|
|
metadata,
|
|
sa.Column("id", sa.Integer(), primary_key=True),
|
|
sa.Column("last_status", sa.String(), nullable=False),
|
|
)
|
|
|
|
|
|
def test_agent_task_run_migration_upgrades_legacy_schema_and_downgrades(
|
|
monkeypatch,
|
|
) -> None:
|
|
"""旧 SQLite schema 应可重复升级并完整回滚新增结构。"""
|
|
engine = sa.create_engine("sqlite://")
|
|
metadata = sa.MetaData()
|
|
_legacy_agent_task(metadata)
|
|
with engine.begin() as connection:
|
|
metadata.create_all(connection)
|
|
migration = _bind_migration(monkeypatch, connection)
|
|
|
|
migration.upgrade()
|
|
migration.upgrade()
|
|
|
|
inspector = sa.inspect(connection)
|
|
assert "agenttaskrun" in inspector.get_table_names()
|
|
assert "last_run_id" in {
|
|
column["name"] for column in inspector.get_columns("agenttask")
|
|
}
|
|
indexes = {
|
|
index["name"]: (tuple(index["column_names"]), index["unique"])
|
|
for index in inspector.get_indexes("agenttaskrun")
|
|
}
|
|
assert indexes["ix_agenttaskrun_run_id"] == (("run_id",), 1)
|
|
assert indexes["ix_agenttaskrun_task_started"] == (
|
|
("task_id", "started_at", "id"),
|
|
0,
|
|
)
|
|
|
|
migration.downgrade()
|
|
inspector = sa.inspect(connection)
|
|
assert "agenttaskrun" not in inspector.get_table_names()
|
|
assert "last_run_id" not in {
|
|
column["name"] for column in inspector.get_columns("agenttask")
|
|
}
|
|
|
|
|
|
def test_agent_task_run_migration_accepts_fresh_current_schema(monkeypatch) -> None:
|
|
"""create_all 已建立当前结构时，迁移重复升级不得创建冲突对象。"""
|
|
engine = sa.create_engine("sqlite://")
|
|
with engine.begin() as connection:
|
|
AgentTask.__table__.create(connection)
|
|
AgentTaskRun.__table__.create(connection)
|
|
migration = _bind_migration(monkeypatch, connection)
|
|
|
|
migration.upgrade()
|
|
migration.upgrade()
|
|
|
|
inspector = sa.inspect(connection)
|
|
assert "agenttaskrun" in inspector.get_table_names()
|
|
assert {
|
|
column["name"] for column in inspector.get_columns("agenttaskrun")
|
|
} == {column.name for column in AgentTaskRun.__table__.columns}
|
|
assert len(inspector.get_indexes("agenttaskrun")) == 2
|
|
|
|
|
|
def test_agent_task_run_migration_matches_postgresql_identity() -> None:
|
|
"""独立 Alembic 路径应与 PostgreSQL create_all 使用相同 Identity。"""
|
|
migration = importlib.import_module(MIGRATION)
|
|
metadata = sa.MetaData()
|
|
table = sa.Table(
|
|
"agenttaskrun",
|
|
metadata,
|
|
migration._id_column("postgresql"),
|
|
)
|
|
|
|
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
|