arcanis/ahriman
1
0
Fork 0
mirror of https://github.com/arcan1s/ahriman.git synced 2025-07-16 15:29:56 +00:00
Code Issues Projects Releases Wiki Activity

refactor: streamline migrations

Browse Source
This commit is contained in:
Evgenii Alekseev
2025-03-09 23:22:24 +02:00
parent aaab9069bf
commit 9a23f5c79d
2 changed files with 101 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files

113
tests/ahriman/core/database/migrations/test_migrations_init.py
View File

@ -19,17 +19,52 @@ def test_migrate(connection: Connection, configuration: Configuration, mocker: M
run_mock.assert_called_once_with()
def test_migration(migrations: Migrations) -> None:
def test_apply_migrations(migrations: Migrations, mocker: MockerFixture) -> None:
"""
must perform single migration
must apply list of migrations
"""
migrate_data_mock = MagicMock()
cursor = MagicMock()
migration = Migration(index=0, name="test", steps=["select 1"], migrate_data=migrate_data_mock)
migration = Migration(index=0, name="test", steps=["select 1"], migrate_data=MagicMock())
migrations.connection.cursor.return_value = cursor
migration_mock = mocker.patch("ahriman.core.database.migrations.Migrations.perform_migration")
migrations.migration(cursor, migration)
cursor.execute.assert_called_once_with("select 1")
migrate_data_mock.assert_called_once_with(migrations.connection, migrations.configuration)
migrations.apply_migrations([migration])
cursor.execute.assert_has_calls([
MockCall("begin exclusive"),
MockCall("commit"),
])
cursor.close.assert_called_once_with()
migration_mock.assert_called_once_with(cursor, migration)
def test_apply_migration_exception(migrations: Migrations, mocker: MockerFixture) -> None:
"""
must roll back and close cursor on exception during migration
"""
cursor = MagicMock()
mocker.patch("logging.Logger.info", side_effect=Exception())
migrations.connection.cursor.return_value = cursor
with pytest.raises(Exception):
migrations.apply_migrations([Migration(index=0, name="test", steps=["select 1"], migrate_data=MagicMock())])
cursor.execute.assert_has_calls([
MockCall("begin exclusive"),
MockCall("rollback"),
])
cursor.close.assert_called_once_with()
def test_apply_migration_sql_exception(migrations: Migrations) -> None:
"""
must close cursor on general migration error
"""
cursor = MagicMock()
cursor.execute.side_effect = Exception()
migrations.connection.cursor.return_value = cursor
with pytest.raises(Exception):
migrations.apply_migrations([Migration(index=0, name="test", steps=["select 1"], migrate_data=MagicMock())])
cursor.close.assert_called_once_with()
def test_migrations(migrations: Migrations) -> None:
@ -39,6 +74,19 @@ def test_migrations(migrations: Migrations) -> None:
assert migrations.migrations()
def test_perform_migration(migrations: Migrations) -> None:
"""
must perform single migration
"""
migrate_data_mock = MagicMock()
cursor = MagicMock()
migration = Migration(index=0, name="test", steps=["select 1"], migrate_data=migrate_data_mock)
migrations.perform_migration(cursor, migration)
cursor.execute.assert_called_once_with("select 1")
migrate_data_mock.assert_called_once_with(migrations.connection, migrations.configuration)
def test_run_skip(migrations: Migrations, mocker: MockerFixture) -> None:
"""
must skip migration if version is the same
@ -54,60 +102,15 @@ def test_run(migrations: Migrations, mocker: MockerFixture) -> None:
must run migration
"""
migration = Migration(index=0, name="test", steps=["select 1"], migrate_data=MagicMock())
cursor = MagicMock()
mocker.patch("ahriman.core.database.migrations.Migrations.user_version", return_value=0)
mocker.patch("ahriman.core.database.migrations.Migrations.migrations", return_value=[migration])
migrations.connection.cursor.return_value = cursor
migration_mock = mocker.patch("ahriman.core.database.migrations.Migrations.migration")
validate_mock = mocker.patch("ahriman.models.migration_result.MigrationResult.validate")
apply_mock = mocker.patch("ahriman.core.database.migrations.Migrations.apply_migrations")
migrations.run()
apply_mock.assert_called_once_with([migration])
validate_mock.assert_called_once_with()
cursor.execute.assert_has_calls([
MockCall("begin exclusive"),
MockCall("pragma user_version = 1"),
MockCall("commit"),
])
cursor.close.assert_called_once_with()
migration_mock.assert_called_once_with(cursor, migration)
def test_run_migration_exception(migrations: Migrations, mocker: MockerFixture) -> None:
"""
must roll back and close cursor on exception during migration
"""
cursor = MagicMock()
mocker.patch("logging.Logger.info", side_effect=Exception())
mocker.patch("ahriman.core.database.migrations.Migrations.user_version", return_value=0)
mocker.patch("ahriman.core.database.migrations.Migrations.migrations",
return_value=[Migration(index=0, name="test", steps=["select 1"], migrate_data=MagicMock())])
mocker.patch("ahriman.models.migration_result.MigrationResult.validate")
migrations.connection.cursor.return_value = cursor
with pytest.raises(Exception):
migrations.run()
cursor.execute.assert_has_calls([
MockCall("begin exclusive"),
MockCall("rollback"),
])
cursor.close.assert_called_once_with()
def test_run_sql_exception(migrations: Migrations, mocker: MockerFixture) -> None:
"""
must close cursor on general migration error
"""
cursor = MagicMock()
cursor.execute.side_effect = Exception()
mocker.patch("ahriman.core.database.migrations.Migrations.user_version", return_value=0)
mocker.patch("ahriman.core.database.migrations.Migrations.migrations",
return_value=[Migration(index=0, name="test", steps=["select 1"], migrate_data=MagicMock())])
mocker.patch("ahriman.models.migration_result.MigrationResult.validate")
migrations.connection.cursor.return_value = cursor
with pytest.raises(Exception):
migrations.run()
cursor.close.assert_called_once_with()
migrations.connection.execute.assert_called_once_with("pragma user_version = 1")
def test_user_version(migrations: Migrations) -> None: