import compileall import os from importlib import import_module from django.db import connection, connections from django.db.migrations.exceptions import ( AmbiguityError, InconsistentMigrationHistory, NodeNotFoundError, ) from django.db.migrations.loader import MigrationLoader from django.db.migrations.recorder import MigrationRecorder from django.test import TestCase, modify_settings, override_settings from .test_base import MigrationTestBase class RecorderTests(TestCase): """ Tests recording migrations as applied or not. """ databases = {"default", "other"} def test_apply(self): """ Tests marking migrations as applied/unapplied. """ recorder = MigrationRecorder(connection) self.assertEqual( {(x, y) for (x, y) in recorder.applied_migrations() if x == "myapp"}, set(), ) recorder.record_applied("myapp", "0432_ponies") self.assertEqual( {(x, y) for (x, y) in recorder.applied_migrations() if x == "myapp"}, {("myapp", "0432_ponies")}, ) # That should not affect records of another database recorder_other = MigrationRecorder(connections["other"]) self.assertEqual( {(x, y) for (x, y) in recorder_other.applied_migrations() if x == "myapp"}, set(), ) recorder.record_unapplied("myapp", "0432_ponies") self.assertEqual( {(x, y) for (x, y) in recorder.applied_migrations() if x == "myapp"}, set(), ) class LoaderTests(TestCase): """ Tests the disk and database loader, and running through migrations in memory. """ def setUp(self): self.applied_records = [] def tearDown(self): # Unapply records on databases that don't roll back changes after each # test method. if not connection.features.supports_transactions: for recorder, app, name in self.applied_records: recorder.record_unapplied(app, name) def record_applied(self, recorder, app, name): recorder.record_applied(app, name) self.applied_records.append((recorder, app, name)) @override_settings(MIGRATION_MODULES={"migrations": "migrations.test_migrations"}) @modify_settings(INSTALLED_APPS={"append": "basic"}) def test_load(self): """ Makes sure the loader can load the migrations for the test apps, and then render them out to a new Apps. """ # Load and test the plan migration_loader = MigrationLoader(connection) self.assertEqual( migration_loader.graph.forwards_plan(("migrations", "0002_second")), [ ("migrations", "0001_initial"), ("migrations", "0002_second"), ], ) # Now render it out! project_state = migration_loader.project_state(("migrations", "0002_second")) self.assertEqual(len(project_state.models), 2) author_state = project_state.models["migrations", "author"] self.assertEqual( list(author_state.fields), ["id", "name", "slug", "age", "rating"] ) book_state = project_state.models["migrations", "book"] self.assertEqual(list(book_state.fields), ["id", "author"]) # Ensure we've included unmigrated apps in there too self.assertIn("basic", project_state.real_apps) @override_settings( MIGRATION_MODULES={ "migrations": "migrations.test_migrations", "migrations2": "migrations2.test_migrations_2", } ) @modify_settings(INSTALLED_APPS={"append": "migrations2"}) def test_plan_handles_repeated_migrations(self): """ _generate_plan() doesn't readd migrations already in the plan (#29180). """ migration_loader = MigrationLoader(connection) nodes = [("migrations", "0002_second"), ("migrations2", "0001_initial")] self.assertEqual( migration_loader.graph._generate_plan(nodes, at_end=True), [ ("migrations", "0001_initial"), ("migrations", "0002_second"), ("migrations2", "0001_initial"), ], ) @override_settings( MIGRATION_MODULES={"migrations": "migrations.test_migrations_unmigdep"} ) def test_load_unmigrated_dependency(self): """ The loader can load migrations with a dependency on an unmigrated app. """ # Load and test the plan migration_loader = MigrationLoader(connection) self.assertEqual( migration_loader.graph.forwards_plan(("migrations", "0001_initial")), [ ("contenttypes", "0001_initial"), ("auth", "0001_initial"), ("migrations", "0001_initial"), ], ) # Now render it out! project_state = migration_loader.project_state(("migrations", "0001_initial")) self.assertEqual( len([m for a, m in project_state.models if a == "migrations"]), 1 ) book_state = project_state.models["migrations", "book"] self.assertEqual(list(book_state.fields), ["id", "user"]) @override_settings( MIGRATION_MODULES={"migrations": "migrations.test_migrations_run_before"} ) def test_run_before(self): """ Makes sure the loader uses Migration.run_before. """ # Load and test the plan migration_loader = MigrationLoader(connection) self.assertEqual( migration_loader.graph.forwards_plan(("migrations", "0002_second")), [ ("migrations", "0001_initial"), ("migrations", "0003_third"), ("migrations", "0002_second"), ], ) @override_settings( MIGRATION_MODULES={ "migrations": "migrations.test_migrations_first", "migrations2": "migrations2.test_migrations_2_first", } ) @modify_settings(INSTALLED_APPS={"append": "migrations2"}) def test_first(self): """ Makes sure the '__first__' migrations build correctly. """ migration_loader = MigrationLoader(connection) self.assertEqual( migration_loader.graph.forwards_plan(("migrations", "second")), [ ("migrations", "thefirst"), ("migrations2", "0001_initial"), ("migrations2", "0002_second"), ("migrations", "second"), ], ) @override_settings(MIGRATION_MODULES={"migrations": "migrations.test_migrations"}) def test_name_match(self): "Tests prefix name matching" migration_loader = MigrationLoader(connection) self.assertEqual( migration_loader.get_migration_by_prefix("migrations", "0001").name, "0001_initial", ) msg = "There is more than one migration for 'migrations' with the prefix '0'" with self.assertRaisesMessage(AmbiguityError, msg): migration_loader.get_migration_by_prefix("migrations", "0") msg = "There is no migration for 'migrations' with the prefix 'blarg'" with self.assertRaisesMessage(KeyError, msg): migration_loader.get_migration_by_prefix("migrations", "blarg") def test_load_import_error(self): with override_settings( MIGRATION_MODULES={"migrations": "import_error_package"} ): with self.assertRaises(ImportError): MigrationLoader(connection) def test_load_module_file(self): with override_settings( MIGRATION_MODULES={"migrations": "migrations.faulty_migrations.file"} ): loader = MigrationLoader(connection) self.assertIn( "migrations", loader.unmigrated_apps, "App with migrations module file not in unmigrated apps.", ) def test_load_empty_dir(self): with override_settings( MIGRATION_MODULES={"migrations": "migrations.faulty_migrations.namespace"} ): loader = MigrationLoader(connection) self.assertIn( "migrations", loader.unmigrated_apps, "App missing __init__.py in migrations module not in unmigrated apps.", ) @override_settings( INSTALLED_APPS=["migrations.migrations_test_apps.migrated_app"], ) def test_marked_as_migrated(self): """ Undefined MIGRATION_MODULES implies default migration module. """ migration_loader = MigrationLoader(connection) self.assertEqual(migration_loader.migrated_apps, {"migrated_app"}) self.assertEqual(migration_loader.unmigrated_apps, set()) @override_settings( INSTALLED_APPS=["migrations.migrations_test_apps.migrated_app"], MIGRATION_MODULES={"migrated_app": None}, ) def test_marked_as_unmigrated(self): """ MIGRATION_MODULES allows disabling of migrations for a particular app. """ migration_loader = MigrationLoader(connection) self.assertEqual(migration_loader.migrated_apps, set()) self.assertEqual(migration_loader.unmigrated_apps, {"migrated_app"}) @override_settings( INSTALLED_APPS=["migrations.migrations_test_apps.migrated_app"], MIGRATION_MODULES={"migrated_app": "missing-module"}, ) def test_explicit_missing_module(self): """ If a MIGRATION_MODULES override points to a missing module, the error raised during the importation attempt should be propagated unless `ignore_no_migrations=True`. """ with self.assertRaisesMessage(ImportError, "missing-module"): migration_loader = MigrationLoader(connection) migration_loader = MigrationLoader(connection, ignore_no_migrations=True) self.assertEqual(migration_loader.migrated_apps, set()) self.assertEqual(migration_loader.unmigrated_apps, {"migrated_app"}) @override_settings( MIGRATION_MODULES={"migrations": "migrations.test_migrations_squashed"} ) def test_loading_squashed(self): "Tests loading a squashed migration" migration_loader = MigrationLoader(connection) recorder = MigrationRecorder(connection) self.addCleanup(recorder.flush) # Loading with nothing applied should just give us the one node self.assertEqual( len([x for x in migration_loader.graph.nodes if x[0] == "migrations"]), 1, ) # However, fake-apply one migration and it should now use the old two self.record_applied(recorder, "migrations", "0001_initial") migration_loader.build_graph() self.assertEqual( len([x for x in migration_loader.graph.nodes if x[0] == "migrations"]), 2, ) @override_settings( MIGRATION_MODULES={"migrations": "migrations.test_migrations_squashed_complex"} ) def test_loading_squashed_complex(self): "Tests loading a complex set of squashed migrations" loader = MigrationLoader(connection) recorder = MigrationRecorder(connection) self.addCleanup(recorder.flush) def num_nodes(): plan = set(loader.graph.forwards_plan(("migrations", "7_auto"))) return len(plan - loader.applied_migrations.keys()) # Empty database: use squashed migration loader.build_graph() self.assertEqual(num_nodes(), 5) # Starting at 1 or 2 should use the squashed migration too self.record_applied(recorder, "migrations", "1_auto") loader.build_graph() self.assertEqual(num_nodes(), 4) self.record_applied(recorder, "migrations", "2_auto") loader.build_graph() self.assertEqual(num_nodes(), 3) # However, starting at 3 to 5 cannot use the squashed migration self.record_applied(recorder, "migrations", "3_auto") loader.build_graph() self.assertEqual(num_nodes(), 4) self.record_applied(recorder, "migrations", "4_auto") loader.build_graph() self.assertEqual(num_nodes(), 3) # Starting at 5 to 7 we are past the squashed migrations. self.record_applied(recorder, "migrations", "5_auto") loader.build_graph() self.assertEqual(num_nodes(), 2) self.record_applied(recorder, "migrations", "6_auto") loader.build_graph() self.assertEqual(num_nodes(), 1) self.record_applied(recorder, "migrations", "7_auto") loader.build_graph() self.assertEqual(num_nodes(), 0) @override_settings( MIGRATION_MODULES={ "app1": "migrations.test_migrations_squashed_complex_multi_apps.app1", "app2": "migrations.test_migrations_squashed_complex_multi_apps.app2", } ) @modify_settings( INSTALLED_APPS={ "append": [ "migrations.test_migrations_squashed_complex_multi_apps.app1", "migrations.test_migrations_squashed_complex_multi_apps.app2", ] } ) def test_loading_squashed_complex_multi_apps(self): loader = MigrationLoader(connection) loader.build_graph() plan = set(loader.graph.forwards_plan(("app1", "4_auto"))) expected_plan = { ("app1", "1_auto"), ("app2", "1_squashed_2"), ("app1", "2_squashed_3"), ("app1", "4_auto"), } self.assertEqual(plan, expected_plan) @override_settings( MIGRATION_MODULES={ "app1": "migrations.test_migrations_squashed_complex_multi_apps.app1", "app2": "migrations.test_migrations_squashed_complex_multi_apps.app2", } ) @modify_settings( INSTALLED_APPS={ "append": [ "migrations.test_migrations_squashed_complex_multi_apps.app1", "migrations.test_migrations_squashed_complex_multi_apps.app2", ] } ) def test_loading_squashed_complex_multi_apps_partially_applied(self): loader = MigrationLoader(connection) recorder = MigrationRecorder(connection) self.record_applied(recorder, "app1", "1_auto") self.record_applied(recorder, "app1", "2_auto") loader.build_graph() plan = set(loader.graph.forwards_plan(("app1", "4_auto"))) plan = plan - loader.applied_migrations.keys() expected_plan = { ("app2", "1_squashed_2"), ("app1", "3_auto"), ("app1", "4_auto"), } self.assertEqual(plan, expected_plan) @override_settings( MIGRATION_MODULES={ "migrations": "migrations.test_migrations_squashed_erroneous" } ) def test_loading_squashed_erroneous(self): "Tests loading a complex but erroneous set of squashed migrations" loader = MigrationLoader(connection) recorder = MigrationRecorder(connection) self.addCleanup(recorder.flush) def num_nodes(): plan = set(loader.graph.forwards_plan(("migrations", "7_auto"))) return len(plan - loader.applied_migrations.keys()) # Empty database: use squashed migration loader.build_graph() self.assertEqual(num_nodes(), 5) # Starting at 1 or 2 should use the squashed migration too self.record_applied(recorder, "migrations", "1_auto") loader.build_graph() self.assertEqual(num_nodes(), 4) self.record_applied(recorder, "migrations", "2_auto") loader.build_graph() self.assertEqual(num_nodes(), 3) # However, starting at 3 or 4, nonexistent migrations would be needed. msg = ( "Migration migrations.6_auto depends on nonexistent node " "('migrations', '5_auto'). Django tried to replace migration " "migrations.5_auto with any of [migrations.3_squashed_5] but wasn't able " "to because some of the replaced migrations are already applied." ) self.record_applied(recorder, "migrations", "3_auto") with self.assertRaisesMessage(NodeNotFoundError, msg): loader.build_graph() self.record_applied(recorder, "migrations", "4_auto") with self.assertRaisesMessage(NodeNotFoundError, msg): loader.build_graph() # Starting at 5 to 7 we are passed the squashed migrations self.record_applied(recorder, "migrations", "5_auto") loader.build_graph() self.assertEqual(num_nodes(), 2) self.record_applied(recorder, "migrations", "6_auto") loader.build_graph() self.assertEqual(num_nodes(), 1) self.record_applied(recorder, "migrations", "7_auto") loader.build_graph() self.assertEqual(num_nodes(), 0) @override_settings( MIGRATION_MODULES={"migrations": "migrations.test_migrations"}, INSTALLED_APPS=["migrations"], ) def test_check_consistent_history(self): loader = MigrationLoader(connection=None) loader.check_consistent_history(connection) recorder = MigrationRecorder(connection) self.record_applied(recorder, "migrations", "0002_second") msg = ( "Migration migrations.0002_second is applied before its dependency " "migrations.0001_initial on database 'default'." ) with self.assertRaisesMessage(InconsistentMigrationHistory, msg): loader.check_consistent_history(connection) @override_settings( MIGRATION_MODULES={"migrations": "migrations.test_migrations_squashed_extra"}, INSTALLED_APPS=["migrations"], ) def test_check_consistent_history_squashed(self): """ MigrationLoader.check_consistent_history() should ignore unapplied squashed migrations that have all of their `replaces` applied. """ loader = MigrationLoader(connection=None) recorder = MigrationRecorder(connection) self.record_applied(recorder, "migrations", "0001_initial") self.record_applied(recorder, "migrations", "0002_second") loader.check_consistent_history(connection) self.record_applied(recorder, "migrations", "0003_third") loader.check_consistent_history(connection) @override_settings( MIGRATION_MODULES={ "app1": "migrations.test_migrations_squashed_ref_squashed.app1", "app2": "migrations.test_migrations_squashed_ref_squashed.app2", } ) @modify_settings( INSTALLED_APPS={ "append": [ "migrations.test_migrations_squashed_ref_squashed.app1", "migrations.test_migrations_squashed_ref_squashed.app2", ] } ) def test_loading_squashed_ref_squashed(self): "Tests loading a squashed migration with a new migration referencing it" r""" The sample migrations are structured like this: app_1 1 --> 2 ---------------------*--> 3 *--> 4 \ / / *-------------------*----/--> 2_sq_3 --* \ / / =============== \ ============= / == / ====================== app_2 *--> 1_sq_2 --* / \ / *--> 1 --> 2 --* Where 2_sq_3 is a replacing migration for 2 and 3 in app_1, as 1_sq_2 is a replacing migration for 1 and 2 in app_2. """ loader = MigrationLoader(connection) recorder = MigrationRecorder(connection) self.addCleanup(recorder.flush) # Load with nothing applied: both migrations squashed. loader.build_graph() plan = set(loader.graph.forwards_plan(("app1", "4_auto"))) plan = plan - loader.applied_migrations.keys() expected_plan = { ("app1", "1_auto"), ("app2", "1_squashed_2"), ("app1", "2_squashed_3"), ("app1", "4_auto"), } self.assertEqual(plan, expected_plan) # Load with nothing applied and migrate to a replaced migration. # Not possible if loader.replace_migrations is True (default). loader.build_graph() msg = "Node ('app1', '3_auto') not a valid node" with self.assertRaisesMessage(NodeNotFoundError, msg): loader.graph.forwards_plan(("app1", "3_auto")) # Possible if loader.replace_migrations is False. loader.replace_migrations = False loader.build_graph() plan = set(loader.graph.forwards_plan(("app1", "3_auto"))) plan = plan - loader.applied_migrations.keys() expected_plan = { ("app1", "1_auto"), ("app2", "1_auto"), ("app2", "2_auto"), ("app1", "2_auto"), ("app1", "3_auto"), } self.assertEqual(plan, expected_plan) loader.replace_migrations = True # Fake-apply a few from app1: unsquashes migration in app1. self.record_applied(recorder, "app1", "1_auto") self.record_applied(recorder, "app1", "2_auto") loader.build_graph() plan = set(loader.graph.forwards_plan(("app1", "4_auto"))) plan = plan - loader.applied_migrations.keys() expected_plan = { ("app2", "1_squashed_2"), ("app1", "3_auto"), ("app1", "4_auto"), } self.assertEqual(plan, expected_plan) # Fake-apply one from app2: unsquashes migration in app2 too. self.record_applied(recorder, "app2", "1_auto") loader.build_graph() plan = set(loader.graph.forwards_plan(("app1", "4_auto"))) plan = plan - loader.applied_migrations.keys() expected_plan = { ("app2", "2_auto"), ("app1", "3_auto"), ("app1", "4_auto"), } self.assertEqual(plan, expected_plan) @override_settings( MIGRATION_MODULES={"migrations": "migrations.test_migrations_private"} ) def test_ignore_files(self): """Files prefixed with underscore, tilde, or dot aren't loaded.""" loader = MigrationLoader(connection) loader.load_disk() migrations = [ name for app, name in loader.disk_migrations if app == "migrations" ] self.assertEqual(migrations, ["0001_initial"]) @override_settings( MIGRATION_MODULES={ "migrations": "migrations.test_migrations_namespace_package" }, ) def test_loading_namespace_package(self): """Migration directories without an __init__.py file are ignored.""" loader = MigrationLoader(connection) loader.load_disk() migrations = [ name for app, name in loader.disk_migrations if app == "migrations" ] self.assertEqual(migrations, []) @override_settings(MIGRATION_MODULES={"migrations": "migrations.test_migrations"}) def test_loading_package_without__file__(self): """ To support frozen environments, MigrationLoader loads migrations from regular packages with no __file__ attribute. """ test_module = import_module("migrations.test_migrations") loader = MigrationLoader(connection) # __file__ == __spec__.origin or the latter is None and former is # undefined. module_file = test_module.__file__ module_origin = test_module.__spec__.origin module_has_location = test_module.__spec__.has_location try: del test_module.__file__ test_module.__spec__.origin = None test_module.__spec__.has_location = False loader.load_disk() migrations = [ name for app, name in loader.disk_migrations if app == "migrations" ] self.assertCountEqual(migrations, ["0001_initial", "0002_second"]) finally: test_module.__file__ = module_file test_module.__spec__.origin = module_origin test_module.__spec__.has_location = module_has_location class PycLoaderTests(MigrationTestBase): def test_valid(self): """ To support frozen environments, MigrationLoader loads .pyc migrations. """ with self.temporary_migration_module( module="migrations.test_migrations" ) as migration_dir: # Compile .py files to .pyc files and delete .py files. compileall.compile_dir(migration_dir, force=True, quiet=1, legacy=True) for name in os.listdir(migration_dir): if name.endswith(".py"): os.remove(os.path.join(migration_dir, name)) loader = MigrationLoader(connection) self.assertIn(("migrations", "0001_initial"), loader.disk_migrations) def test_invalid(self): """ MigrationLoader reraises ImportErrors caused by "bad magic number" pyc files with a more helpful message. """ with self.temporary_migration_module( module="migrations.test_migrations_bad_pyc" ) as migration_dir: # The -tpl suffix is to avoid the pyc exclusion in MANIFEST.in. os.rename( os.path.join(migration_dir, "0001_initial.pyc-tpl"), os.path.join(migration_dir, "0001_initial.pyc"), ) msg = ( r"Couldn't import '\w+.migrations.0001_initial' as it appears " "to be a stale .pyc file." ) with self.assertRaisesRegex(ImportError, msg): MigrationLoader(connection)