Mirrored_Repos/InvokeAI
1
0
Fork 0
mirror of https://github.com/invoke-ai/InvokeAI synced 2024-08-30 20:32:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(db): simplify migration registration validation

Browse Source 
With the previous change to assert that the to_version == from_version + 1, this validation can be simpler.
This commit is contained in:
psychedelicious 2023-12-12 09:52:03 +11:00
parent 55acc16b2d
commit afe4e55bf9
2 changed files with 14 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
invokeai/app/services/shared/sqlite_migrator/sqlite_migrator_common.py
View File

@ -65,10 +65,10 @@ class MigrationSet:
def register(self, migration: Migration) -> None:
"""Registers a migration."""
if any(m.from_version == migration.from_version for m in self._migrations):
raise MigrationVersionError(f"Migration from {migration.from_version} already registered")
if any(m.to_version == migration.to_version for m in self._migrations):
raise MigrationVersionError(f"Migration to {migration.to_version} already registered")
migration_from_already_registered = any(m.from_version == migration.from_version for m in self._migrations)
migration_to_already_registered = any(m.to_version == migration.to_version for m in self._migrations)
if migration_from_already_registered or migration_to_already_registered:
raise MigrationVersionError("Migration with from_version or to_version already registered")
self._migrations.add(migration)
def get(self, from_version: int) -> Optional[Migration]:

18
tests/test_sqlite_migrator.py
View File

@ -116,14 +116,16 @@ def test_migration_set_add_migration(migrator: SQLiteMigrator, migration_no_op:
def test_migration_set_may_not_register_dupes(
migrator: SQLiteMigrator, no_op_migrate_callback: MigrateCallback
) -> None:
migrate_1_to_2 = Migration(from_version=1, to_version=2, migrate=no_op_migrate_callback)
migrate_0_to_2 = Migration(from_version=0, to_version=2, migrate=no_op_migrate_callback)
migrate_1_to_3 = Migration(from_version=1, to_version=3, migrate=no_op_migrate_callback)
migrator._migration_set.register(migrate_1_to_2)
with pytest.raises(MigrationVersionError, match=r"Migration to 2 already registered"):
migrator._migration_set.register(migrate_0_to_2)
with pytest.raises(MigrationVersionError, match=r"Migration from 1 already registered"):
migrator._migration_set.register(migrate_1_to_3)
migrate_0_to_1_a = Migration(from_version=0, to_version=1, migrate=no_op_migrate_callback)
migrate_0_to_1_b = Migration(from_version=0, to_version=1, migrate=no_op_migrate_callback)
migrator._migration_set.register(migrate_0_to_1_a)
with pytest.raises(MigrationVersionError, match=r"Migration with from_version or to_version already registered"):
migrator._migration_set.register(migrate_0_to_1_b)
migrate_1_to_2_a = Migration(from_version=1, to_version=2, migrate=no_op_migrate_callback)
migrate_1_to_2_b = Migration(from_version=1, to_version=2, migrate=no_op_migrate_callback)
migrator._migration_set.register(migrate_1_to_2_a)
with pytest.raises(MigrationVersionError, match=r"Migration with from_version or to_version already registered"):
migrator._migration_set.register(migrate_1_to_2_b)
def test_migration_set_gets_migration(migration_no_op: Migration) -> None: