mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-09-03 22:51:47 +08:00
362 lines
13 KiB
Python
362 lines
13 KiB
Python
"""用户名唯一性 Alembic 迁移测试。"""
|
|
|
|
import importlib
|
|
|
|
import pytest
|
|
import sqlalchemy as sa
|
|
from alembic.migration import MigrationContext
|
|
from alembic.operations import Operations
|
|
from sqlalchemy.dialects import postgresql
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.schema import CreateTable
|
|
|
|
from app.db.engine import _register_sqlite_foreign_keys
|
|
from app.db.models.passkey import PassKey
|
|
from app.db.models.user import User
|
|
from app.db.models.userconfig import UserConfig
|
|
|
|
MIGRATION_MODULE = "database.versions.a9d4f2c7e6b1_3_0_18"
|
|
|
|
|
|
def _bind_migration(monkeypatch, connection):
|
|
"""把用户名迁移绑定到隔离数据库连接。"""
|
|
migration = importlib.import_module(MIGRATION_MODULE)
|
|
monkeypatch.setattr(
|
|
migration,
|
|
"op",
|
|
Operations(MigrationContext.configure(connection)),
|
|
)
|
|
return migration
|
|
|
|
|
|
def _legacy_metadata() -> tuple[sa.MetaData, sa.Table, sa.Table]:
|
|
"""构造允许重名且带用户外键的旧版最小表结构。"""
|
|
metadata = sa.MetaData()
|
|
users = sa.Table(
|
|
"user",
|
|
metadata,
|
|
sa.Column("id", sa.Integer(), primary_key=True),
|
|
sa.Column("name", sa.String(), nullable=False),
|
|
sa.Column("is_active", sa.Boolean(), nullable=True),
|
|
sa.Index("ix_user_name", "name"),
|
|
)
|
|
passkeys = sa.Table(
|
|
"passkey",
|
|
metadata,
|
|
sa.Column("id", sa.Integer(), primary_key=True),
|
|
sa.Column(
|
|
"user_id",
|
|
sa.Integer(),
|
|
sa.ForeignKey("user.id"),
|
|
nullable=False,
|
|
),
|
|
)
|
|
return metadata, users, passkeys
|
|
|
|
|
|
def _user_rows(connection, users: sa.Table) -> list[dict]:
|
|
"""按稳定身份返回用户快照。"""
|
|
return list(connection.execute(sa.select(users).order_by(users.c.id)).mappings())
|
|
|
|
|
|
def _indexes(connection) -> dict[str, dict]:
|
|
"""返回用户表的命名索引。"""
|
|
return {index["name"]: index for index in sa.inspect(connection).get_indexes("user")}
|
|
|
|
|
|
def test_user_model_declares_exact_unique_name_index() -> None:
|
|
"""当前用户聚合模型必须声明唯一身份和精确级联约束。"""
|
|
indexes = {index.name: index for index in User.__table__.indexes}
|
|
|
|
assert "ix_user_name" not in indexes
|
|
assert indexes["ux_user_name"].unique is True
|
|
assert tuple(column.name for column in indexes["ux_user_name"].columns) == ("name",)
|
|
user_config_fk = next(iter(UserConfig.__table__.foreign_keys))
|
|
assert user_config_fk.target_fullname == "user.name"
|
|
assert user_config_fk.onupdate == "CASCADE"
|
|
assert user_config_fk.ondelete == "CASCADE"
|
|
assert UserConfig.__table__.c.username.nullable is False
|
|
assert UserConfig.__table__.c.key.nullable is False
|
|
constraints = {constraint.name: constraint for constraint in UserConfig.__table__.constraints}
|
|
assert {column.name for column in constraints["uq_userconfig_username_key"].columns} == {"username", "key"}
|
|
passkey_fk = next(iter(PassKey.__table__.foreign_keys))
|
|
assert passkey_fk.target_fullname == "user.id"
|
|
assert passkey_fk.ondelete == "CASCADE"
|
|
|
|
|
|
def test_sqlite_connection_registration_enables_foreign_keys() -> None:
|
|
"""每条宿主 SQLite 连接都必须实际启用外键检查。"""
|
|
engine = sa.create_engine("sqlite://")
|
|
_register_sqlite_foreign_keys(engine)
|
|
with engine.connect() as connection:
|
|
assert connection.exec_driver_sql("PRAGMA foreign_keys").scalar_one() == 1
|
|
engine.dispose()
|
|
|
|
|
|
def test_user_child_constraints_compile_for_postgresql() -> None:
|
|
"""用户从属约束必须生成 PostgreSQL 可执行的级联 DDL。"""
|
|
user_config_ddl = str(
|
|
CreateTable(UserConfig.__table__).compile(
|
|
dialect=postgresql.dialect(),
|
|
)
|
|
)
|
|
passkey_ddl = str(
|
|
CreateTable(PassKey.__table__).compile(
|
|
dialect=postgresql.dialect(),
|
|
)
|
|
)
|
|
|
|
assert ('FOREIGN KEY(username) REFERENCES "user" (name) ON DELETE CASCADE ON UPDATE CASCADE') in user_config_ddl
|
|
assert ('FOREIGN KEY(user_id) REFERENCES "user" (id) ON DELETE CASCADE') in passkey_ddl
|
|
|
|
|
|
def test_user_child_migration_repairs_data_and_cascades(monkeypatch) -> None:
|
|
"""迁移清债后配置和 PassKey 不得脱离用户主体，并保持可逆。"""
|
|
metadata = sa.MetaData()
|
|
users = sa.Table(
|
|
"user",
|
|
metadata,
|
|
sa.Column("id", sa.Integer(), primary_key=True),
|
|
sa.Column("name", sa.String(), nullable=False),
|
|
sa.Column("is_active", sa.Boolean()),
|
|
sa.Index("ix_user_name", "name"),
|
|
)
|
|
configs = sa.Table(
|
|
"userconfig",
|
|
metadata,
|
|
sa.Column("id", sa.Integer(), primary_key=True),
|
|
sa.Column("username", sa.String(), nullable=True),
|
|
sa.Column("key", sa.String(), nullable=True),
|
|
sa.Column("value", sa.JSON()),
|
|
)
|
|
passkeys = sa.Table(
|
|
"passkey",
|
|
metadata,
|
|
sa.Column("id", sa.Integer(), primary_key=True),
|
|
sa.Column(
|
|
"user_id",
|
|
sa.Integer(),
|
|
sa.ForeignKey("user.id"),
|
|
nullable=False,
|
|
),
|
|
)
|
|
engine = sa.create_engine("sqlite://")
|
|
with engine.connect() as connection:
|
|
metadata.create_all(connection)
|
|
connection.execute(
|
|
users.insert().values(
|
|
id=1,
|
|
name="old",
|
|
is_active=True,
|
|
)
|
|
)
|
|
connection.execute(
|
|
configs.insert(),
|
|
[
|
|
{"id": 1, "username": "old", "key": "theme", "value": "first"},
|
|
{"id": 2, "username": "old", "key": "theme", "value": "duplicate"},
|
|
{"id": 3, "username": "ghost", "key": "theme", "value": "orphan"},
|
|
{"id": 4, "username": None, "key": "theme", "value": "null-user"},
|
|
{"id": 5, "username": "old", "key": None, "value": "null-key"},
|
|
],
|
|
)
|
|
connection.execute(
|
|
passkeys.insert(),
|
|
[
|
|
{"id": 11, "user_id": 1},
|
|
{"id": 12, "user_id": 999},
|
|
],
|
|
)
|
|
connection.commit()
|
|
migration = _bind_migration(monkeypatch, connection)
|
|
|
|
migration.upgrade()
|
|
connection.commit()
|
|
migration.upgrade()
|
|
connection.commit()
|
|
|
|
assert connection.execute(sa.select(configs.c.id, configs.c.username, configs.c.key)).all() == [
|
|
(1, "old", "theme")
|
|
]
|
|
assert connection.execute(sa.select(passkeys.c.id)).scalars().all() == [11]
|
|
inspector = sa.inspect(connection)
|
|
config_columns = {column["name"]: column for column in inspector.get_columns("userconfig")}
|
|
assert config_columns["username"]["nullable"] is False
|
|
assert config_columns["key"]["nullable"] is False
|
|
config_fk = inspector.get_foreign_keys("userconfig")[0]
|
|
assert config_fk["options"] == {
|
|
"ondelete": "CASCADE",
|
|
"onupdate": "CASCADE",
|
|
}
|
|
passkey_fk = inspector.get_foreign_keys("passkey")[0]
|
|
assert passkey_fk["options"] == {"ondelete": "CASCADE"}
|
|
unique = {item["name"]: item for item in inspector.get_unique_constraints("userconfig")}
|
|
assert unique["uq_userconfig_username_key"]["column_names"] == [
|
|
"username",
|
|
"key",
|
|
]
|
|
|
|
connection.exec_driver_sql("PRAGMA foreign_keys=ON")
|
|
connection.execute(users.update().values(name="new"))
|
|
assert connection.execute(sa.select(configs.c.username)).scalar_one() == "new"
|
|
with pytest.raises(IntegrityError):
|
|
with connection.begin_nested():
|
|
connection.execute(
|
|
configs.insert().values(
|
|
username="new",
|
|
key="theme",
|
|
value="duplicate",
|
|
)
|
|
)
|
|
with pytest.raises(IntegrityError):
|
|
with connection.begin_nested():
|
|
connection.execute(
|
|
configs.insert().values(
|
|
username="new",
|
|
key=None,
|
|
value="empty-key",
|
|
)
|
|
)
|
|
connection.execute(users.delete())
|
|
assert connection.execute(sa.select(configs.c.id)).all() == []
|
|
assert connection.execute(sa.select(passkeys.c.id)).all() == []
|
|
connection.commit()
|
|
|
|
migration.downgrade()
|
|
connection.commit()
|
|
downgraded = sa.inspect(connection)
|
|
columns = {column["name"]: column for column in downgraded.get_columns("userconfig")}
|
|
assert columns["username"]["nullable"] is True
|
|
assert columns["key"]["nullable"] is True
|
|
assert downgraded.get_foreign_keys("userconfig") == []
|
|
assert downgraded.get_foreign_keys("passkey")[0]["options"] == {}
|
|
engine.dispose()
|
|
|
|
|
|
def test_migration_preserves_rows_and_foreign_keys_across_replay(
|
|
monkeypatch,
|
|
) -> None:
|
|
"""升级应确定性修复重名并在降级、再升级时保留用户身份。"""
|
|
engine = sa.create_engine("sqlite://")
|
|
metadata, users, passkeys = _legacy_metadata()
|
|
with engine.begin() as connection:
|
|
connection.exec_driver_sql("PRAGMA foreign_keys = ON")
|
|
metadata.create_all(connection)
|
|
connection.execute(
|
|
users.insert(),
|
|
[
|
|
{"id": 1, "name": "alice", "is_active": True},
|
|
{"id": 2, "name": "alice", "is_active": True},
|
|
{"id": 3, "name": "alice", "is_active": None},
|
|
{
|
|
"id": 4,
|
|
"name": "alice__duplicate_2",
|
|
"is_active": True,
|
|
},
|
|
{"id": 5, "name": "bob", "is_active": True},
|
|
],
|
|
)
|
|
connection.execute(
|
|
passkeys.insert(),
|
|
[
|
|
{"id": 11, "user_id": 2},
|
|
{"id": 12, "user_id": 3},
|
|
],
|
|
)
|
|
migration = _bind_migration(monkeypatch, connection)
|
|
|
|
migration.upgrade()
|
|
upgraded_rows = _user_rows(connection, users)
|
|
migration.upgrade()
|
|
|
|
assert _user_rows(connection, users) == upgraded_rows
|
|
assert upgraded_rows == [
|
|
{"id": 1, "name": "alice", "is_active": True},
|
|
{
|
|
"id": 2,
|
|
"name": "alice__duplicate_2_1",
|
|
"is_active": False,
|
|
},
|
|
{
|
|
"id": 3,
|
|
"name": "alice__duplicate_3",
|
|
"is_active": False,
|
|
},
|
|
{
|
|
"id": 4,
|
|
"name": "alice__duplicate_2",
|
|
"is_active": True,
|
|
},
|
|
{"id": 5, "name": "bob", "is_active": True},
|
|
]
|
|
assert connection.execute(sa.select(passkeys.c.id, passkeys.c.user_id).order_by(passkeys.c.id)).all() == [
|
|
(11, 2),
|
|
(12, 3),
|
|
]
|
|
index = _indexes(connection)["ux_user_name"]
|
|
assert tuple(index["column_names"]) == ("name",)
|
|
assert index["unique"] == 1
|
|
assert "ix_user_name" not in _indexes(connection)
|
|
|
|
with pytest.raises(IntegrityError):
|
|
with connection.begin_nested():
|
|
connection.execute(
|
|
users.insert().values(
|
|
id=6,
|
|
name="alice",
|
|
is_active=True,
|
|
)
|
|
)
|
|
|
|
migration.downgrade()
|
|
downgraded_rows = _user_rows(connection, users)
|
|
assert downgraded_rows == upgraded_rows
|
|
assert "ux_user_name" not in _indexes(connection)
|
|
assert tuple(_indexes(connection)["ix_user_name"]["column_names"]) == ("name",)
|
|
connection.execute(users.insert().values(id=6, name="alice", is_active=True))
|
|
|
|
migration.upgrade()
|
|
|
|
assert _user_rows(connection, users)[-1] == {
|
|
"id": 6,
|
|
"name": "alice__duplicate_6",
|
|
"is_active": False,
|
|
}
|
|
assert connection.execute(sa.select(passkeys.c.user_id).order_by(passkeys.c.id)).scalars().all() == [2, 3]
|
|
assert _indexes(connection)["ux_user_name"]["unique"] == 1
|
|
|
|
|
|
def test_migration_repairs_malformed_canonical_index(monkeypatch) -> None:
|
|
"""升级必须替换同名但列或唯一语义错误的部分迁移索引。"""
|
|
engine = sa.create_engine("sqlite://")
|
|
metadata, users, _passkeys = _legacy_metadata()
|
|
with engine.begin() as connection:
|
|
metadata.create_all(connection)
|
|
connection.execute(
|
|
users.insert(),
|
|
[{"id": 1, "name": "alice", "is_active": True}],
|
|
)
|
|
connection.exec_driver_sql('CREATE INDEX ux_user_name ON "user" (is_active)')
|
|
migration = _bind_migration(monkeypatch, connection)
|
|
|
|
migration.upgrade()
|
|
|
|
index = _indexes(connection)["ux_user_name"]
|
|
assert tuple(index["column_names"]) == ("name",)
|
|
assert index["unique"] == 1
|
|
|
|
|
|
def test_migration_accepts_fresh_current_schema(monkeypatch) -> None:
|
|
"""当前模型已建表时，重复升级不得创建冲突索引。"""
|
|
engine = sa.create_engine("sqlite://")
|
|
with engine.begin() as connection:
|
|
User.__table__.create(connection)
|
|
migration = _bind_migration(monkeypatch, connection)
|
|
|
|
migration.upgrade()
|
|
migration.upgrade()
|
|
|
|
index = _indexes(connection)["ux_user_name"]
|
|
assert tuple(index["column_names"]) == ("name",)
|
|
assert index["unique"] == 1
|