Refs -- Simplified migration used to test atomic recording.
This makes sure atomic recording of migration application is used when
the schema editor doesn't defer any statement.
from unittest import mock
from django.apps.registry import apps as global_apps
-from django.db import DatabaseError, connection
+from django.db import DatabaseError, connection, migrations, models
from django.db.migrations.exceptions import InvalidMigrationPlan
from django.db.migrations.executor import MigrationExecutor
from django.db.migrations.graph import MigrationGraph
from django.db.migrations.recorder import MigrationRecorder
+from django.db.migrations.state import ProjectState
from django.test import (
SimpleTestCase, modify_settings, override_settings, skipUnlessDBFeature,
)
@@ -655,18 +656,33 @@ class ExecutorTests(MigrationTestBase):
# When the feature is False, the operation and the record won't be
# performed in a transaction and the test will systematically pass.
@skipUnlessDBFeature('can_rollback_ddl')
- @override_settings(MIGRATION_MODULES={'migrations': 'migrations.test_migrations'})
def test_migrations_applied_and_recorded_atomically(self):
"""Migrations are applied and recorded atomically."""
+ class Migration(migrations.Migration):
+ operations = [
+ migrations.CreateModel('model', [
+ ('id', models.AutoField(primary_key=True)),
+ ]),
+ ]
+
executor = MigrationExecutor(connection)
with mock.patch('django.db.migrations.executor.MigrationExecutor.record_migration') as record_migration:
record_migration.side_effect = RuntimeError('Recording migration failed.')
with self.assertRaisesMessage(RuntimeError, 'Recording migration failed.'):
+ executor.apply_migration(
+ ProjectState(),
+ Migration('0001_initial', 'record_migration'),
+ )
executor.migrate([('migrations', '0001_initial')])
# The migration isn't recorded as applied since it failed.
migration_recorder = MigrationRecorder(connection)
- self.assertFalse(migration_recorder.migration_qs.filter(app='migrations', name='0001_initial').exists())
- self.assertTableNotExists('migrations_author')
+ self.assertIs(
+ migration_recorder.migration_qs.filter(
+ app='record_migration', name='0001_initial',
+ ).exists(),
+ False,
+ )
+ self.assertTableNotExists('record_migration_model')
class FakeLoader:
