tests the migration executor full endtoend running bear in mind that if these are failing you should fix the other test failures first as they may be propagating into here tests running a simple set of migrations let s look at the plan first and make sure it s up to scratch were the tables there before alright let s try running it are the tables there now rebuild the graph to reflect the new db state alright let s undo what we did are the tables gone tests running a squashed migration from zero should ignore what it replaces check our leaf node is the squashed one check the plan were the tables there before alright let s try running it are the tables there now rebuild the graph to reflect the new db state alright let s undo what we did should also just use squashed are the tables gone migrate backward to a squashed migration unmigrate everything applying a nonatomic migration works as expected an atomic operation is properly rolled back inside a nonatomic migration record previous migration as successful rebuild the graph to reflect the new db state migrating backwards is also atomic replanning a full migration of a fullymigrated set doesn t perform spurious unmigrations and remigrations there was previously a bug where the executor just always performed the backwards plan for applied migrations which even for the most recent migration in an app might include other dependent apps and these were being unmigrated make the initial plan check it fakeapply all migrations rebuild the graph to reflect the new db state now plan a second time and make sure it s empty the resulting state should include applied migrations erase all the fake records although the migrationexecutor interfaces allows for mixed migration plans combined forwards and backwards migrations this is not supported prepare for mixed plan rebuild the graph to reflect the new db state generate mixed plan rebuild the graph to reflect the new db state are the tables gone tests detection of initial migrations already having been applied were the tables there before run it normally are the tables there now we shouldn t have faked that one rebuild the graph to reflect the new db state fakereverse that are the tables still there make sure that was faked finally migrate forwards this should fakeapply our initial migration applying the migration should raise a database level error because we haven t given the fakeinitial option reset the faked state allow faking of initial createmodel operations and migrate back to clean up the database regression test for 22325 references to a custom user model defined in the same app are not resolved correctly migrate forwards the softapplication detection works change tablenames to not return authuser during this as it wouldn t be there in a normal run and ensure migrations exists in the global app registry temporarily migrate back to clean up the database executor detectsoftapplied detects manytomanyfield tables from an addfield operation this checks the case of addfield in a migration with other operations 0001 and the case of addfield in its own migration 0002 from 0001 from 0002 create the tables for 0001 but make it look like the migration hasn t been applied table detection sees 0001 is applied but not 0002 create the tables for both migrations but make it look like neither has been applied table detection sees 0002 is applied leave the tables for 0001 except the manytomany table that missing table should cause detectsoftapplied to return false cleanup by removing the remaining tables 24123 all models of apps already applied which are unrelated to the first app being applied are part of the initial model state rebuild the graph to reflect the new db state migrate forwards this led to a lookup lookuperrors because lookuperrorb b2 is already applied rebuild the graph to reflect the new db state cleanup 24123 all models of apps being unapplied which are unrelated to the first app being unapplied are part of the initial model state rebuild the graph to reflect the new db state migrate backwards this led to a lookup lookuperrors because lookuperrorb b2 is not in the initial state unrelated to app c rebuild the graph to reflect the new db state cleanup 26647 unrelated applied migrations should be part of the final state in both directions migrate forward migrate backward 24129 tests callback process were the tables there before rebuild the graph to reflect the new db state apply initial migrations rebuild the graph to reflect the new db state apply pk type alteration rebuild the graph to reflect the new db state we can t simply unapply the migrations here because there is no implicit cast from varchar to int on the database level applying all replaced migrations marks replacement as applied 24628 place the database in a state where the replaced migrations are partially applied 0001 is applied 0002 is not use fake because we don t actually have the first migration applied so the second will fail and there s no need to actually createmodify tables here we re just testing the migrationrecord which works the same with or without fake because we ve now applied 0001 and 0002 both their squashed replacement should be marked as applied a new squash migration will be marked as applied even if all its replaced migrations were previously already applied 24628 record all replaced migrations as applied because 0001 and 0002 are both applied even though this migrate run didn t apply anything new their squashed replacement should be marked as applied when the feature is false the operation and the record won t be performed in a transaction and the test will systematically pass migrations are applied and recorded atomically class migrationmigrations migration operations migrations createmodel model id models autofieldprimarykeytrue executor migrationexecutorconnection with mock patch django db migrations executor migrationexecutor recordmigration as recordmigration recordmigration sideeffect runtimeerrorrecording migration failed with self assertraisesmessageruntimeerror recording migration failed executor applymigration projectstate migration0001initial recordmigration executor migratemigrations 0001initial the migration isn t recorded as applied since it failed migrationrecorder migrationrecorderconnection self assertis migrationrecorder migrationqs filter apprecordmigration name0001initial exists false self asserttablenotexistsrecordmigrationmodel def testmigrationsnotappliedondeferredsqlfailureself the migration isn t recorded as applied since it failed the djangomigrations table is not created if there are no migrations to record 0 queries since the query for hastable is being mocked really all we need is any object with a debuguseful repr def initself name self name name def reprself return ms self name class executorunittestssimpletestcase minimize unnecessary rollbacks in connected apps when you say manage py migrate appa 0001 rather than migrating to just after appa0001 in the linearized migration plan which could roll back migrations in other apps that depend on appa 0001 but don t need to be rolled back since we re not rolling back appa 0001 we migrate to just before appa0002 a1impl fakemigrationa1 a1 a 1 a2impl fakemigrationa2 a2 a 2 a3impl fakemigrationa3 a3 a 3 a4impl fakemigrationa4 a4 a 4 b1impl fakemigrationb1 b1 b 1 b2impl fakemigrationb2 b2 b 2 graph migrationgraph graph addnodea1 a1impl graph addnodea2 a2impl graph addnodea3 a3impl graph addnodea4 a4impl graph addnodeb1 b1impl graph addnodeb2 b2impl graph adddependencynone a2 a1 graph adddependencynone a3 a1 graph adddependencynone a4 a2 graph adddependencynone a4 a3 graph adddependencynone b2 b1 graph adddependencynone b1 a1 graph adddependencynone b2 a2 executor migrationexecutornone executor loader fakeloader graph a1 a1impl b1 b1impl a2 a2impl b2 b2impl a3 a3impl a4 a4impl plan executor migrationplana1 shouldberolledback b2impl a4impl a2impl a3impl exp m true for m in shouldberolledback self assertequalplan exp def testbackwardsnothingtodoself r if the current state satisfies the given target do nothing a 1 2 b 1 c 1 if a1 is applied already and a2 is not and we re asked to migrate to a1 don t apply or unapply b1 or c1 regardless of their current state tests the migration executor full end to end running bear in mind that if these are failing you should fix the other test failures first as they may be propagating into here tests running a simple set of migrations let s look at the plan first and make sure it s up to scratch were the tables there before alright let s try running it are the tables there now rebuild the graph to reflect the new db state alright let s undo what we did are the tables gone tests running a squashed migration from zero should ignore what it replaces check our leaf node is the squashed one check the plan were the tables there before alright let s try running it are the tables there now rebuild the graph to reflect the new db state alright let s undo what we did should also just use squashed are the tables gone migrate backward to a squashed migration unmigrate everything applying a non atomic migration works as expected an atomic operation is properly rolled back inside a non atomic migration record previous migration as successful rebuild the graph to reflect the new db state migrating backwards is also atomic re planning a full migration of a fully migrated set doesn t perform spurious unmigrations and remigrations there was previously a bug where the executor just always performed the backwards plan for applied migrations which even for the most recent migration in an app might include other dependent apps and these were being unmigrated make the initial plan check it fake apply all migrations rebuild the graph to reflect the new db state now plan a second time and make sure it s empty the resulting state should include applied migrations erase all the fake records although the migrationexecutor interfaces allows for mixed migration plans combined forwards and backwards migrations this is not supported prepare for mixed plan rebuild the graph to reflect the new db state generate mixed plan rebuild the graph to reflect the new db state are the tables gone tests detection of initial migrations already having been applied were the tables there before run it normally are the tables there now we shouldn t have faked that one rebuild the graph to reflect the new db state fake reverse that are the tables still there make sure that was faked finally migrate forwards this should fake apply our initial migration applying the migration should raise a database level error because we haven t given the fake initial option reset the faked state allow faking of initial createmodel operations and migrate back to clean up the database regression test for 22325 references to a custom user model defined in the same app are not resolved correctly migrate forwards the soft application detection works change table_names to not return auth_user during this as it wouldn t be there in a normal run and ensure migrations exists in the global app registry temporarily migrate back to clean up the database executor detect_soft_applied detects manytomanyfield tables from an addfield operation this checks the case of addfield in a migration with other operations 0001 and the case of addfield in its own migration 0002 from 0001 from 0002 create the tables for 0001 but make it look like the migration hasn t been applied table detection sees 0001 is applied but not 0002 create the tables for both migrations but make it look like neither has been applied table detection sees 0002 is applied leave the tables for 0001 except the many to many table that missing table should cause detect_soft_applied to return false cleanup by removing the remaining tables 24123 all models of apps already applied which are unrelated to the first app being applied are part of the initial model state rebuild the graph to reflect the new db state migrate forwards this led to a lookup lookuperrors because lookuperror_b b2 is already applied rebuild the graph to reflect the new db state cleanup 24123 all models of apps being unapplied which are unrelated to the first app being unapplied are part of the initial model state rebuild the graph to reflect the new db state migrate backwards this led to a lookup lookuperrors because lookuperror_b b2 is not in the initial state unrelated to app c rebuild the graph to reflect the new db state cleanup 26647 unrelated applied migrations should be part of the final state in both directions migrate forward migrate backward 24129 tests callback process were the tables there before rebuild the graph to reflect the new db state apply initial migrations rebuild the graph to reflect the new db state apply pk type alteration rebuild the graph to reflect the new db state we can t simply unapply the migrations here because there is no implicit cast from varchar to int on the database level applying all replaced migrations marks replacement as applied 24628 place the database in a state where the replaced migrations are partially applied 0001 is applied 0002 is not use fake because we don t actually have the first migration applied so the second will fail and there s no need to actually create modify tables here we re just testing the migrationrecord which works the same with or without fake because we ve now applied 0001 and 0002 both their squashed replacement should be marked as applied a new squash migration will be marked as applied even if all its replaced migrations were previously already applied 24628 record all replaced migrations as applied because 0001 and 0002 are both applied even though this migrate run didn t apply anything new their squashed replacement should be marked as applied when the feature is false the operation and the record won t be performed in a transaction and the test will systematically pass migrations are applied and recorded atomically the migration isn t recorded as applied since it failed migrations are not recorded if deferred sql application fails the migration isn t recorded as applied since it failed the django_migrations table is not created if there are no migrations to record 0 queries since the query for has_table is being mocked really all we need is any object with a debug useful repr more isolated unit tests for executor methods minimize unnecessary rollbacks in connected apps when you say manage py migrate appa 0001 rather than migrating to just after appa 0001 in the linearized migration plan which could roll back migrations in other apps that depend on appa 0001 but don t need to be rolled back since we re not rolling back appa 0001 we migrate to just before appa 0002 minimize rollbacks when target has multiple in app children a 1 3 2 4 b 1 2 if the current state satisfies the given target do nothing a 1 2 b 1 c 1 if a1 is applied already and a2 is not and we re asked to migrate to a1 don t apply or unapply b1 or c1 regardless of their current state
from unittest import mock

from django.apps.registry import apps as global_apps
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,
)
from django.test.utils import isolate_lru_cache

from .test_base import MigrationTestBase


@modify_settings(INSTALLED_APPS={"append": "migrations2"})
class ExecutorTests(MigrationTestBase):
    

    available_apps = [
        "migrations",
        "migrations2",
        "django.contrib.auth",
        "django.contrib.contenttypes",
    ]

    @override_settings(MIGRATION_MODULES={"migrations": "migrations.test_migrations"})
    def test_run(self):
        
        executor = MigrationExecutor(connection)
        
        plan = executor.migration_plan([("migrations", "0002_second")])
        self.assertEqual(
            plan,
            [
                (executor.loader.graph.nodes["migrations", "0001_initial"], False),
                (executor.loader.graph.nodes["migrations", "0002_second"], False),
            ],
        )
        
        self.assertTableNotExists("migrations_author")
        self.assertTableNotExists("migrations_book")
        
        executor.migrate([("migrations", "0002_second")])
        
        self.assertTableExists("migrations_author")
        self.assertTableExists("migrations_book")
        
        executor.loader.build_graph()
        
        plan = executor.migration_plan([("migrations", None)])
        self.assertEqual(
            plan,
            [
                (executor.loader.graph.nodes["migrations", "0002_second"], True),
                (executor.loader.graph.nodes["migrations", "0001_initial"], True),
            ],
        )
        executor.migrate([("migrations", None)])
        
        self.assertTableNotExists("migrations_author")
        self.assertTableNotExists("migrations_book")

    @override_settings(
        MIGRATION_MODULES={"migrations": "migrations.test_migrations_squashed"}
    )
    def test_run_with_squashed(self):
        
        executor = MigrationExecutor(connection)
        
        leaves = [
            key for key in executor.loader.graph.leaf_nodes() if key[0] == "migrations"
        ]
        self.assertEqual(leaves, [("migrations", "0001_squashed_0002")])
        
        plan = executor.migration_plan([("migrations", "0001_squashed_0002")])
        self.assertEqual(
            plan,
            [
                (
                    executor.loader.graph.nodes["migrations", "0001_squashed_0002"],
                    False,
                ),
            ],
        )
        
        self.assertTableNotExists("migrations_author")
        self.assertTableNotExists("migrations_book")
        
        executor.migrate([("migrations", "0001_squashed_0002")])
        
        self.assertTableExists("migrations_author")
        self.assertTableExists("migrations_book")
        
        executor.loader.build_graph()
        
        plan = executor.migration_plan([("migrations", None)])
        self.assertEqual(
            plan,
            [
                (executor.loader.graph.nodes["migrations", "0001_squashed_0002"], True),
            ],
        )
        executor.migrate([("migrations", None)])
        
        self.assertTableNotExists("migrations_author")
        self.assertTableNotExists("migrations_book")

    @override_settings(
        MIGRATION_MODULES={"migrations": "migrations.test_migrations_squashed"},
    )
    def test_migrate_backward_to_squashed_migration(self):
        executor = MigrationExecutor(connection)
        try:
            self.assertTableNotExists("migrations_author")
            self.assertTableNotExists("migrations_book")
            executor.migrate([("migrations", "0001_squashed_0002")])
            self.assertTableExists("migrations_author")
            self.assertTableExists("migrations_book")
            executor.loader.build_graph()
            
            executor.migrate([("migrations", "0001_initial")])
            self.assertTableExists("migrations_author")
            self.assertTableNotExists("migrations_book")
        finally:
            
            executor = MigrationExecutor(connection)
            executor.migrate([("migrations", None)])
            self.assertTableNotExists("migrations_author")
            self.assertTableNotExists("migrations_book")

    @override_settings(
        MIGRATION_MODULES={"migrations": "migrations.test_migrations_non_atomic"}
    )
    def test_non_atomic_migration(self):
        
        executor = MigrationExecutor(connection)
        with self.assertRaisesMessage(RuntimeError, "Abort migration"):
            executor.migrate([("migrations", "0001_initial")])
        self.assertTableExists("migrations_publisher")
        migrations_apps = executor.loader.project_state(
            ("migrations", "0001_initial")
        ).apps
        Publisher = migrations_apps.get_model("migrations", "Publisher")
        self.assertTrue(Publisher.objects.exists())
        self.assertTableNotExists("migrations_book")

    @override_settings(
        MIGRATION_MODULES={"migrations": "migrations.test_migrations_atomic_operation"}
    )
    def test_atomic_operation_in_non_atomic_migration(self):
        
        executor = MigrationExecutor(connection)
        with self.assertRaisesMessage(RuntimeError, "Abort migration"):
            executor.migrate([("migrations", "0001_initial")])
        migrations_apps = executor.loader.project_state(
            ("migrations", "0001_initial")
        ).apps
        Editor = migrations_apps.get_model("migrations", "Editor")
        self.assertFalse(Editor.objects.exists())
        
        executor.migrate([("migrations", "0001_initial")], fake=True)
        
        executor.loader.build_graph()
        
        with self.assertRaisesMessage(RuntimeError, "Abort migration"):
            executor.migrate([("migrations", None)])
        self.assertFalse(Editor.objects.exists())

    @override_settings(
        MIGRATION_MODULES={
            "migrations": "migrations.test_migrations",
            "migrations2": "migrations2.test_migrations_2",
        }
    )
    def test_empty_plan(self):
        
        
        executor = MigrationExecutor(connection)
        plan = executor.migration_plan(
            [
                ("migrations", "0002_second"),
                ("migrations2", "0001_initial"),
            ]
        )
        self.assertEqual(
            plan,
            [
                (executor.loader.graph.nodes["migrations", "0001_initial"], False),
                (executor.loader.graph.nodes["migrations", "0002_second"], False),
                (executor.loader.graph.nodes["migrations2", "0001_initial"], False),
            ],
        )
        
        executor.migrate(
            [("migrations", "0002_second"), ("migrations2", "0001_initial")], fake=True
        )
        
        executor.loader.build_graph()
        
        plan = executor.migration_plan(
            [
                ("migrations", "0002_second"),
                ("migrations2", "0001_initial"),
            ]
        )
        self.assertEqual(plan, [])
        
        state = executor.migrate(
            [
                ("migrations", "0002_second"),
                ("migrations2", "0001_initial"),
            ]
        )
        self.assertIn(("migrations", "book"), state.models)
        self.assertIn(("migrations", "author"), state.models)
        self.assertIn(("migrations2", "otherauthor"), state.models)
        
        executor.recorder.record_unapplied("migrations2", "0001_initial")
        executor.recorder.record_unapplied("migrations", "0002_second")
        executor.recorder.record_unapplied("migrations", "0001_initial")

    @override_settings(
        MIGRATION_MODULES={
            "migrations": "migrations.test_migrations",
            "migrations2": "migrations2.test_migrations_2_no_deps",
        }
    )
    def test_mixed_plan_not_supported(self):
        
        
        executor = MigrationExecutor(connection)
        plan = executor.migration_plan([("migrations", "0002_second")])
        self.assertEqual(
            plan,
            [
                (executor.loader.graph.nodes["migrations", "0001_initial"], False),
                (executor.loader.graph.nodes["migrations", "0002_second"], False),
            ],
        )
        executor.migrate(None, plan)
        
        executor.loader.build_graph()
        self.assertIn(
            ("migrations", "0001_initial"), executor.loader.applied_migrations
        )
        self.assertIn(("migrations", "0002_second"), executor.loader.applied_migrations)
        self.assertNotIn(
            ("migrations2", "0001_initial"), executor.loader.applied_migrations
        )

        
        plan = executor.migration_plan(
            [
                ("migrations", None),
                ("migrations2", "0001_initial"),
            ]
        )
        msg = (
            "Migration plans with both forwards and backwards migrations are "
            "not supported. Please split your migration process into separate "
            "plans of only forwards OR backwards migrations."
        )
        with self.assertRaisesMessage(InvalidMigrationPlan, msg) as cm:
            executor.migrate(None, plan)
        self.assertEqual(
            cm.exception.args[1],
            [
                (executor.loader.graph.nodes["migrations", "0002_second"], True),
                (executor.loader.graph.nodes["migrations", "0001_initial"], True),
                (executor.loader.graph.nodes["migrations2", "0001_initial"], False),
            ],
        )
        
        executor.loader.build_graph()
        executor.migrate(
            [
                ("migrations", None),
                ("migrations2", None),
            ]
        )
        
        self.assertTableNotExists("migrations_author")
        self.assertTableNotExists("migrations_book")
        self.assertTableNotExists("migrations2_otherauthor")

    @override_settings(MIGRATION_MODULES={"migrations": "migrations.test_migrations"})
    def test_soft_apply(self):
        
        state = {"faked": None}

        def fake_storer(phase, migration=None, fake=None):
            state["faked"] = fake

        executor = MigrationExecutor(connection, progress_callback=fake_storer)
        
        self.assertTableNotExists("migrations_author")
        self.assertTableNotExists("migrations_tribble")
        
        self.assertEqual(
            executor.migration_plan([("migrations", "0001_initial")]),
            [
                (executor.loader.graph.nodes["migrations", "0001_initial"], False),
            ],
        )
        executor.migrate([("migrations", "0001_initial")])
        
        self.assertTableExists("migrations_author")
        self.assertTableExists("migrations_tribble")
        
        self.assertIs(state["faked"], False)
        
        executor.loader.build_graph()
        
        executor.migrate([("migrations", None)], fake=True)
        
        self.assertTableExists("migrations_author")
        self.assertTableExists("migrations_tribble")
        
        self.assertIs(state["faked"], True)
        
        executor.loader.build_graph()
        self.assertEqual(
            executor.migration_plan([("migrations", "0001_initial")]),
            [
                (executor.loader.graph.nodes["migrations", "0001_initial"], False),
            ],
        )
        
        
        with self.assertRaises(DatabaseError):
            executor.migrate([("migrations", "0001_initial")])
        
        state = {"faked": None}
        
        executor.migrate([("migrations", "0001_initial")], fake_initial=True)
        self.assertIs(state["faked"], True)
        
        executor.loader.build_graph()
        executor.migrate([("migrations", None)])
        self.assertTableNotExists("migrations_author")
        self.assertTableNotExists("migrations_tribble")

    @override_settings(
        MIGRATION_MODULES={
            "migrations": "migrations.test_migrations_custom_user",
            "django.contrib.auth": "django.contrib.auth.migrations",
        },
        AUTH_USER_MODEL="migrations.Author",
    )
    def test_custom_user(self):
        
        with isolate_lru_cache(global_apps.get_swappable_settings_name):
            executor = MigrationExecutor(connection)
            self.assertTableNotExists("migrations_author")
            self.assertTableNotExists("migrations_tribble")
            
            executor.migrate([("migrations", "0001_initial")])
            self.assertTableExists("migrations_author")
            self.assertTableExists("migrations_tribble")
            
            
            
            
            old_table_names = connection.introspection.table_names
            connection.introspection.table_names = lambda c: [
                x for x in old_table_names(c) if x != "auth_user"
            ]
            migrations_apps = executor.loader.project_state(
                ("migrations", "0001_initial"),
            ).apps
            global_apps.get_app_config("migrations").models[
                "author"
            ] = migrations_apps.get_model("migrations", "author")
            try:
                migration = executor.loader.get_migration("auth", "0001_initial")
                self.assertIs(executor.detect_soft_applied(None, migration)[0], True)
            finally:
                connection.introspection.table_names = old_table_names
                del global_apps.get_app_config("migrations").models["author"]
                
                executor.loader.build_graph()
                executor.migrate([("migrations", None)])
                self.assertTableNotExists("migrations_author")
                self.assertTableNotExists("migrations_tribble")

    @override_settings(
        MIGRATION_MODULES={
            "migrations": "migrations.test_add_many_to_many_field_initial",
        },
    )
    def test_detect_soft_applied_add_field_manytomanyfield(self):
        
        tables = [
            
            "migrations_project",
            "migrations_task",
            "migrations_project_tasks",
            
            "migrations_task_projects",
        ]
        executor = MigrationExecutor(connection)
        
        
        executor.migrate([("migrations", "0001_initial")])
        executor.migrate([("migrations", None)], fake=True)
        for table in tables[:3]:
            self.assertTableExists(table)
        
        migration = executor.loader.get_migration("migrations", "0001_initial")
        self.assertIs(executor.detect_soft_applied(None, migration)[0], True)
        migration = executor.loader.get_migration("migrations", "0002_initial")
        self.assertIs(executor.detect_soft_applied(None, migration)[0], False)

        
        
        executor.loader.build_graph()
        executor.migrate([("migrations", "0001_initial")], fake=True)
        executor.migrate([("migrations", "0002_initial")])
        executor.loader.build_graph()
        executor.migrate([("migrations", None)], fake=True)
        
        migration = executor.loader.get_migration("migrations", "0002_initial")
        self.assertIs(executor.detect_soft_applied(None, migration)[0], True)

        
        
        with connection.schema_editor() as editor:
            for table in tables[2:]:
                editor.execute(editor.sql_delete_table % {"table": table})
        migration = executor.loader.get_migration("migrations", "0001_initial")
        self.assertIs(executor.detect_soft_applied(None, migration)[0], False)

        
        with connection.schema_editor() as editor:
            for table in tables[:2]:
                editor.execute(editor.sql_delete_table % {"table": table})
        for table in tables:
            self.assertTableNotExists(table)

    @override_settings(
        INSTALLED_APPS=[
            "migrations.migrations_test_apps.lookuperror_a",
            "migrations.migrations_test_apps.lookuperror_b",
            "migrations.migrations_test_apps.lookuperror_c",
        ]
    )
    def test_unrelated_model_lookups_forwards(self):
        
        try:
            executor = MigrationExecutor(connection)
            self.assertTableNotExists("lookuperror_a_a1")
            self.assertTableNotExists("lookuperror_b_b1")
            self.assertTableNotExists("lookuperror_c_c1")
            executor.migrate([("lookuperror_b", "0003_b3")])
            self.assertTableExists("lookuperror_b_b3")
            
            executor.loader.build_graph()

            
            
            executor.migrate(
                [
                    ("lookuperror_a", "0004_a4"),
                    ("lookuperror_c", "0003_c3"),
                ]
            )
            self.assertTableExists("lookuperror_a_a4")
            self.assertTableExists("lookuperror_c_c3")

            
            executor.loader.build_graph()
        finally:
            
            executor.migrate(
                [
                    ("lookuperror_a", None),
                    ("lookuperror_b", None),
                    ("lookuperror_c", None),
                ]
            )
            self.assertTableNotExists("lookuperror_a_a1")
            self.assertTableNotExists("lookuperror_b_b1")
            self.assertTableNotExists("lookuperror_c_c1")

    @override_settings(
        INSTALLED_APPS=[
            "migrations.migrations_test_apps.lookuperror_a",
            "migrations.migrations_test_apps.lookuperror_b",
            "migrations.migrations_test_apps.lookuperror_c",
        ]
    )
    def test_unrelated_model_lookups_backwards(self):
        
        try:
            executor = MigrationExecutor(connection)
            self.assertTableNotExists("lookuperror_a_a1")
            self.assertTableNotExists("lookuperror_b_b1")
            self.assertTableNotExists("lookuperror_c_c1")
            executor.migrate(
                [
                    ("lookuperror_a", "0004_a4"),
                    ("lookuperror_b", "0003_b3"),
                    ("lookuperror_c", "0003_c3"),
                ]
            )
            self.assertTableExists("lookuperror_b_b3")
            self.assertTableExists("lookuperror_a_a4")
            self.assertTableExists("lookuperror_c_c3")
            
            executor.loader.build_graph()

            
            
            executor.migrate([("lookuperror_a", None)])

            
            executor.loader.build_graph()
        finally:
            
            executor.migrate([("lookuperror_b", None), ("lookuperror_c", None)])
            self.assertTableNotExists("lookuperror_a_a1")
            self.assertTableNotExists("lookuperror_b_b1")
            self.assertTableNotExists("lookuperror_c_c1")

    @override_settings(
        INSTALLED_APPS=[
            "migrations.migrations_test_apps.mutate_state_a",
            "migrations.migrations_test_apps.mutate_state_b",
        ]
    )
    def test_unrelated_applied_migrations_mutate_state(self):
        
        executor = MigrationExecutor(connection)
        executor.migrate(
            [
                ("mutate_state_b", "0002_add_field"),
            ]
        )
        
        executor.loader.build_graph()
        state = executor.migrate(
            [
                ("mutate_state_a", "0001_initial"),
            ]
        )
        self.assertIn("added", state.models["mutate_state_b", "b"].fields)
        executor.loader.build_graph()
        
        state = executor.migrate(
            [
                ("mutate_state_a", None),
            ]
        )
        self.assertIn("added", state.models["mutate_state_b", "b"].fields)
        executor.migrate(
            [
                ("mutate_state_b", None),
            ]
        )

    @override_settings(MIGRATION_MODULES={"migrations": "migrations.test_migrations"})
    def test_process_callback(self):
        
        call_args_list = []

        def callback(*args):
            call_args_list.append(args)

        executor = MigrationExecutor(connection, progress_callback=callback)
        
        self.assertTableNotExists("migrations_author")
        self.assertTableNotExists("migrations_tribble")
        executor.migrate(
            [
                ("migrations", "0001_initial"),
                ("migrations", "0002_second"),
            ]
        )
        
        executor.loader.build_graph()

        executor.migrate(
            [
                ("migrations", None),
                ("migrations", None),
            ]
        )
        self.assertTableNotExists("migrations_author")
        self.assertTableNotExists("migrations_tribble")

        migrations = executor.loader.graph.nodes
        expected = [
            ("render_start",),
            ("render_success",),
            ("apply_start", migrations["migrations", "0001_initial"], False),
            ("apply_success", migrations["migrations", "0001_initial"], False),
            ("apply_start", migrations["migrations", "0002_second"], False),
            ("apply_success", migrations["migrations", "0002_second"], False),
            ("render_start",),
            ("render_success",),
            ("unapply_start", migrations["migrations", "0002_second"], False),
            ("unapply_success", migrations["migrations", "0002_second"], False),
            ("unapply_start", migrations["migrations", "0001_initial"], False),
            ("unapply_success", migrations["migrations", "0001_initial"], False),
        ]
        self.assertEqual(call_args_list, expected)

    @override_settings(
        INSTALLED_APPS=[
            "migrations.migrations_test_apps.alter_fk.author_app",
            "migrations.migrations_test_apps.alter_fk.book_app",
        ]
    )
    def test_alter_id_type_with_fk(self):
        try:
            executor = MigrationExecutor(connection)
            self.assertTableNotExists("author_app_author")
            self.assertTableNotExists("book_app_book")
            
            executor.migrate(
                [
                    ("author_app", "0001_initial"),
                    ("book_app", "0001_initial"),
                ]
            )
            self.assertTableExists("author_app_author")
            self.assertTableExists("book_app_book")
            
            executor.loader.build_graph()

            
            executor.migrate([("author_app", "0002_alter_id")])

            
            executor.loader.build_graph()
        finally:
            
            
            with connection.schema_editor() as editor:
                editor.execute(editor.sql_delete_table % {"table": "book_app_book"})
                editor.execute(editor.sql_delete_table % {"table": "author_app_author"})
            self.assertTableNotExists("author_app_author")
            self.assertTableNotExists("book_app_book")
            executor.migrate([("author_app", None)], fake=True)

    @override_settings(
        MIGRATION_MODULES={"migrations": "migrations.test_migrations_squashed"}
    )
    def test_apply_all_replaced_marks_replacement_as_applied(self):
        
        recorder = MigrationRecorder(connection)
        
        
        recorder.record_applied("migrations", "0001_initial")
        executor = MigrationExecutor(connection)
        
        
        
        
        executor.migrate([("migrations", "0002_second")], fake=True)

        
        
        self.assertIn(
            ("migrations", "0001_squashed_0002"),
            recorder.applied_migrations(),
        )

    @override_settings(
        MIGRATION_MODULES={"migrations": "migrations.test_migrations_squashed"}
    )
    def test_migrate_marks_replacement_applied_even_if_it_did_nothing(self):
        
        recorder = MigrationRecorder(connection)
        
        recorder.record_applied("migrations", "0001_initial")
        recorder.record_applied("migrations", "0002_second")
        executor = MigrationExecutor(connection)
        executor.migrate([("migrations", "0001_squashed_0002")])

        
        
        
        self.assertIn(
            ("migrations", "0001_squashed_0002"),
            recorder.applied_migrations(),
        )

    @override_settings(
        MIGRATION_MODULES={"migrations": "migrations.test_migrations_squashed"}
    )
    def test_migrate_marks_replacement_unapplied(self):
        executor = MigrationExecutor(connection)
        executor.migrate([("migrations", "0001_squashed_0002")])
        try:
            self.assertIn(
                ("migrations", "0001_squashed_0002"),
                executor.recorder.applied_migrations(),
            )
        finally:
            executor.loader.build_graph()
            executor.migrate([("migrations", None)])
            self.assertNotIn(
                ("migrations", "0001_squashed_0002"),
                executor.recorder.applied_migrations(),
            )

    
    
    @skipUnlessDBFeature("can_rollback_ddl")
    def test_migrations_applied_and_recorded_atomically(self):
        

        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")])
        
        migration_recorder = MigrationRecorder(connection)
        self.assertIs(
            migration_recorder.migration_qs.filter(
                app="record_migration",
                name="0001_initial",
            ).exists(),
            False,
        )
        self.assertTableNotExists("record_migration_model")

    def test_migrations_not_applied_on_deferred_sql_failure(self):
        

        class DeferredSQL:
            def __str__(self):
                raise DatabaseError("Failed to apply deferred SQL")

        class Migration(migrations.Migration):
            atomic = False

            def apply(self, project_state, schema_editor, collect_sql=False):
                schema_editor.deferred_sql.append(DeferredSQL())

        executor = MigrationExecutor(connection)
        with self.assertRaisesMessage(DatabaseError, "Failed to apply deferred SQL"):
            executor.apply_migration(
                ProjectState(),
                Migration("0001_initial", "deferred_sql"),
            )
        
        migration_recorder = MigrationRecorder(connection)
        self.assertIs(
            migration_recorder.migration_qs.filter(
                app="deferred_sql",
                name="0001_initial",
            ).exists(),
            False,
        )

    @mock.patch.object(MigrationRecorder, "has_table", return_value=False)
    def test_migrate_skips_schema_creation(self, mocked_has_table):
        
        executor = MigrationExecutor(connection)
        
        with self.assertNumQueries(0):
            executor.migrate([], plan=[])


class FakeLoader:
    def __init__(self, graph, applied):
        self.graph = graph
        self.applied_migrations = applied
        self.replace_migrations = True


class FakeMigration:
    

    def __init__(self, name):
        self.name = name

    def __repr__(self):
        return "M<%s>" % self.name


class ExecutorUnitTests(SimpleTestCase):
    

    def test_minimize_rollbacks(self):
        
        a1_impl = FakeMigration("a1")
        a1 = ("a", "1")
        a2_impl = FakeMigration("a2")
        a2 = ("a", "2")
        b1_impl = FakeMigration("b1")
        b1 = ("b", "1")
        graph = MigrationGraph()
        graph.add_node(a1, a1_impl)
        graph.add_node(a2, a2_impl)
        graph.add_node(b1, b1_impl)
        graph.add_dependency(None, b1, a1)
        graph.add_dependency(None, a2, a1)

        executor = MigrationExecutor(None)
        executor.loader = FakeLoader(
            graph,
            {
                a1: a1_impl,
                b1: b1_impl,
                a2: a2_impl,
            },
        )

        plan = executor.migration_plan({a1})

        self.assertEqual(plan, [(a2_impl, True)])

    def test_minimize_rollbacks_branchy(self):
        r
        a1_impl = FakeMigration("a1")
        a1 = ("a", "1")
        a2_impl = FakeMigration("a2")
        a2 = ("a", "2")
        a3_impl = FakeMigration("a3")
        a3 = ("a", "3")
        a4_impl = FakeMigration("a4")
        a4 = ("a", "4")
        b1_impl = FakeMigration("b1")
        b1 = ("b", "1")
        b2_impl = FakeMigration("b2")
        b2 = ("b", "2")
        graph = MigrationGraph()
        graph.add_node(a1, a1_impl)
        graph.add_node(a2, a2_impl)
        graph.add_node(a3, a3_impl)
        graph.add_node(a4, a4_impl)
        graph.add_node(b1, b1_impl)
        graph.add_node(b2, b2_impl)
        graph.add_dependency(None, a2, a1)
        graph.add_dependency(None, a3, a1)
        graph.add_dependency(None, a4, a2)
        graph.add_dependency(None, a4, a3)
        graph.add_dependency(None, b2, b1)
        graph.add_dependency(None, b1, a1)
        graph.add_dependency(None, b2, a2)

        executor = MigrationExecutor(None)
        executor.loader = FakeLoader(
            graph,
            {
                a1: a1_impl,
                b1: b1_impl,
                a2: a2_impl,
                b2: b2_impl,
                a3: a3_impl,
                a4: a4_impl,
            },
        )

        plan = executor.migration_plan({a1})

        should_be_rolled_back = [b2_impl, a4_impl, a2_impl, a3_impl]
        exp = [(m, True) for m in should_be_rolled_back]
        self.assertEqual(plan, exp)

    def test_backwards_nothing_to_do(self):
        r
        a1_impl = FakeMigration("a1")
        a1 = ("a", "1")
        a2_impl = FakeMigration("a2")
        a2 = ("a", "2")
        b1_impl = FakeMigration("b1")
        b1 = ("b", "1")
        c1_impl = FakeMigration("c1")
        c1 = ("c", "1")
        graph = MigrationGraph()
        graph.add_node(a1, a1_impl)
        graph.add_node(a2, a2_impl)
        graph.add_node(b1, b1_impl)
        graph.add_node(c1, c1_impl)
        graph.add_dependency(None, a2, a1)
        graph.add_dependency(None, b1, a1)
        graph.add_dependency(None, c1, a1)

        executor = MigrationExecutor(None)
        executor.loader = FakeLoader(
            graph,
            {
                a1: a1_impl,
                b1: b1_impl,
            },
        )

        plan = executor.migration_plan({a1})

        self.assertEqual(plan, [])