Refs #29721 -- 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.
This commit is contained in:
Simon Charette 2021-01-20 20:59:20 -05:00 committed by Mariusz Felisiak
parent 7b3ec6bcc8
commit 533a583578
1 changed files with 20 additions and 4 deletions

View File

@ -1,11 +1,12 @@
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: