Use MigrationLoader to order migrations (#208)

Co-authored-by: Adam Johnson <[email protected]>

HISTORY.rst

@@ -2,6 +2,9 @@
 History
 =======
 
+* Use Django’s ``MigrationLoader`` to find the latest migrations.
+  This also means django-linear-migrations operations detect migration conflicts.
+
 2.5.1 (2022-07-20)
 ------------------
 

README.rst

@@ -122,6 +122,7 @@ These are:
 * ``dlm.E002``: ``<app_label>``'s max_migration.txt contains multiple lines.
 * ``dlm.E003``: ``<app_label>``'s max_migration.txt points to non-existent migration '``<bad_migration_name>``'.
 * ``dlm.E004``: ``<app_label>``'s max_migration.txt contains '``<max_migration_name>``', but the latest migration is '``<real_max_migration_name>``'.
+* ``dlm.E005``: Conflicting migrations detected; multiple leaf nodes in the migration graph: ``<conflicting_migrations>``
 
 ``create_max_migration_files`` Command
 --------------------------------------

src/django_linear_migrations/apps.py

@@ -112,6 +112,20 @@ def names(self) -> set[str]:
         }
 
 
+def get_graph_plan(
+    loader: MigrationLoader, app_labels: Iterable[str] | None = None
+) -> list[tuple[str, str]]:
+    nodes = loader.graph.leaf_nodes()
+    if app_labels:
+        nodes = [
+            (app_label, name)
+            for app_label, name in loader.graph.leaf_nodes()
+            if app_label in app_labels
+        ]
+    plan: list[tuple[str, str]] = loader.graph._generate_plan(nodes, at_end=True)
+    return plan
+
+
 def check_max_migration_files(
     *, app_configs: Iterable[AppConfig] | None = None, **kwargs: object
 ) -> list[Error]:
@@ -121,6 +135,33 @@ def check_max_migration_files(
     else:
         app_config_set = set()
 
+    migration_loader = MigrationLoader(None, ignore_no_migrations=True)
+    app_labels = [a.label for a in first_party_app_configs()]
+    conflicts = {
+        app_label: names
+        for app_label, names in migration_loader.detect_conflicts().items()
+        if app_label in app_labels
+    }
+    if conflicts:
+        conflict_msg = "".join(
+            f"\n* {app_label}: {', '.join(sorted(names))}"
+            for app_label, names in conflicts.items()
+        )
+        errors.append(
+            Error(
+                id="dlm.E005",
+                msg=(
+                    "Conflicting migrations detected - multiple leaf nodes "
+                    + f"detected for these apps:{conflict_msg}"
+                ),
+                hint=(
+                    "Fix the conflict, e.g. with './manage.py makemigrations --merge'."
+                ),
+            )
+        )
+        return errors
+
+    graph_plan = get_graph_plan(loader=migration_loader, app_labels=app_labels)
     for app_config in first_party_app_configs():
         # When only checking certain apps, skip the others
         if app_configs is not None and app_config not in app_config_set:
@@ -178,7 +219,9 @@ def check_max_migration_files(
             )
             continue
 
-        real_max_migration_name = max(migration_details.names)
+        real_max_migration_name = [
+            name for gp_app_label, name in graph_plan if gp_app_label == app_label
+        ][-1]
         if max_migration_name != real_max_migration_name:
             errors.append(
                 Error(

src/django_linear_migrations/management/commands/create_max_migration_files.py

@@ -6,8 +6,10 @@
 
 from django.apps import apps
 from django.core.management.commands.makemigrations import Command as BaseCommand
+from django.db.migrations.loader import MigrationLoader
 
 from django_linear_migrations.apps import first_party_app_configs
+from django_linear_migrations.apps import get_graph_plan
 from django_linear_migrations.apps import MigrationDetails
 
 
@@ -57,6 +59,8 @@ def handle(
             sys.exit(2)
 
         any_created = False
+        migration_loader = MigrationLoader(None, ignore_no_migrations=True)
+        graph_plan = get_graph_plan(loader=migration_loader, app_labels=labels)
         for app_config in first_party_app_configs():
             if labels and app_config.label not in labels:
                 continue
@@ -68,7 +72,9 @@ def handle(
             max_migration_txt = migration_details.dir / "max_migration.txt"
             if recreate or not max_migration_txt.exists():
                 if not dry_run:
-                    max_migration_name = max(migration_details.names)
+                    max_migration_name = [
+                        k[1] for k in graph_plan if k[0] == app_config.label
+                    ][-1]
                     max_migration_txt.write_text(max_migration_name + "\n")
                     self.stdout.write(
                         f"Created max_migration.txt for {app_config.label}."

tests/test_checks.py

@@ -2,15 +2,17 @@
 
 import sys
 import time
+from textwrap import dedent
 
 import pytest
 from django.test import override_settings
-from django.test import SimpleTestCase
+from django.test import TestCase
 
 from django_linear_migrations.apps import check_max_migration_files
+from tests.utils import empty_migration
 
 
-class CheckMaxMigrationFilesTests(SimpleTestCase):
+class CheckMaxMigrationFilesTests(TestCase):
     @pytest.fixture(autouse=True)
     def tmp_path_fixture(self, tmp_path):
         migrations_module_name = "migrations" + str(time.time()).replace(".", "")
@@ -54,7 +56,7 @@ def test_skipped_unspecified_app(self):
 
     def test_dlm_E001(self):
         (self.migrations_dir / "__init__.py").touch()
-        (self.migrations_dir / "0001_initial.py").touch()
+        (self.migrations_dir / "0001_initial.py").write_text(empty_migration)
 
         result = check_max_migration_files()
 
@@ -64,7 +66,7 @@ def test_dlm_E001(self):
 
     def test_dlm_E002(self):
         (self.migrations_dir / "__init__.py").touch()
-        (self.migrations_dir / "0001_initial.py").touch()
+        (self.migrations_dir / "0001_initial.py").write_text(empty_migration)
         (self.migrations_dir / "max_migration.txt").write_text("line1\nline2\n")
 
         result = check_max_migration_files()
@@ -75,7 +77,7 @@ def test_dlm_E002(self):
 
     def test_dlm_E003(self):
         (self.migrations_dir / "__init__.py").touch()
-        (self.migrations_dir / "0001_initial.py").touch()
+        (self.migrations_dir / "0001_initial.py").write_text(empty_migration)
         (self.migrations_dir / "max_migration.txt").write_text("0001_start\n")
 
         result = check_max_migration_files()
@@ -89,8 +91,16 @@ def test_dlm_E003(self):
 
     def test_dlm_E004(self):
         (self.migrations_dir / "__init__.py").touch()
-        (self.migrations_dir / "0001_initial.py").touch()
-        (self.migrations_dir / "0002_updates.py").touch()
+        (self.migrations_dir / "0001_initial.py").write_text(empty_migration)
+        (self.migrations_dir / "0002_updates.py").write_text(
+            dedent(
+                """
+                from django.db import migrations
+                class Migration(migrations.Migration):
+                    dependencies = [('testapp', '0001_initial')]
+                """
+            )
+        )
         (self.migrations_dir / "max_migration.txt").write_text("0001_initial\n")
 
         result = check_max_migration_files()
@@ -102,10 +112,50 @@ def test_dlm_E004(self):
             + " latest migration is '0002_updates'."
         )
 
+    def test_dlm_E005(self):
+        (self.migrations_dir / "__init__.py").touch()
+        (self.migrations_dir / "0001_initial.py").write_text(empty_migration)
+        (self.migrations_dir / "custom_name.py").write_text(
+            dedent(
+                """
+                from django.db import migrations
+                class Migration(migrations.Migration):
+                    dependencies = [('testapp', '0001_initial')]
+                """
+            )
+        )
+        (self.migrations_dir / "0002_updates.py").write_text(
+            dedent(
+                """
+                from django.db import migrations
+                class Migration(migrations.Migration):
+                    dependencies = [('testapp', '0001_initial')]
+                """
+            )
+        )
+        (self.migrations_dir / "max_migration.txt").write_text("0002_updates\n")
+
+        result = check_max_migration_files()
+        assert len(result) == 1
+        assert result[0].id == "dlm.E005"
+        assert result[0].msg == (
+            "Conflicting migrations detected - multiple leaf nodes "
+            + "detected for these apps:\n"
+            + "* testapp: 0002_updates, custom_name"
+        )
+
     def test_okay(self):
         (self.migrations_dir / "__init__.py").touch()
-        (self.migrations_dir / "0001_initial.py").touch()
-        (self.migrations_dir / "0002_updates.py").touch()
+        (self.migrations_dir / "0001_initial.py").write_text(empty_migration)
+        (self.migrations_dir / "0002_updates.py").write_text(
+            dedent(
+                """
+                from django.db import migrations
+                class Migration(migrations.Migration):
+                    dependencies = [('testapp', '0001_initial')]
+                """
+            )
+        )
         (self.migrations_dir / "max_migration.txt").write_text("0002_updates\n")
 
         result = check_max_migration_files()

tests/test_create_max_migration_files.py

@@ -3,12 +3,15 @@
 import sys
 import time
 from io import StringIO
+from textwrap import dedent
 
 import pytest
 from django.core.management import call_command
 from django.test import override_settings
 from django.test import TestCase
 
+from tests.utils import empty_migration
+
 
 class CreateMaxMigrationFilesTests(TestCase):
     @pytest.fixture(autouse=True)
@@ -77,7 +80,7 @@ def test_success_only_init(self):
     @override_settings(FIRST_PARTY_APPS=[])
     def test_success_setting_not_first_party(self):
         (self.migrations_dir / "__init__.py").touch()
-        (self.migrations_dir / "0001_initial.py").touch()
+        (self.migrations_dir / "0001_initial.py").write_text(empty_migration)
 
         out, err, returncode = self.call_command()
 
@@ -87,7 +90,7 @@ def test_success_setting_not_first_party(self):
 
     def test_success_dry_run(self):
         (self.migrations_dir / "__init__.py").touch()
-        (self.migrations_dir / "0001_initial.py").touch()
+        (self.migrations_dir / "0001_initial.py").write_text(empty_migration)
 
         out, err, returncode = self.call_command("--dry-run")
 
@@ -99,7 +102,7 @@ def test_success_dry_run(self):
 
     def test_success(self):
         (self.migrations_dir / "__init__.py").touch()
-        (self.migrations_dir / "0001_initial.py").touch()
+        (self.migrations_dir / "0001_initial.py").write_text(empty_migration)
 
         out, err, returncode = self.call_command()
 
@@ -111,7 +114,7 @@ def test_success(self):
 
     def test_success_already_exists(self):
         (self.migrations_dir / "__init__.py").touch()
-        (self.migrations_dir / "0001_initial.py").touch()
+        (self.migrations_dir / "0001_initial.py").write_text(empty_migration)
         (self.migrations_dir / "max_migration.txt").write_text("0001_initial\n")
 
         out, err, returncode = self.call_command()
@@ -122,7 +125,7 @@ def test_success_already_exists(self):
 
     def test_success_recreate(self):
         (self.migrations_dir / "__init__.py").touch()
-        (self.migrations_dir / "0001_initial.py").touch()
+        (self.migrations_dir / "0001_initial.py").write_text(empty_migration)
         (self.migrations_dir / "max_migration.txt").write_text("0001_initial\n")
 
         out, err, returncode = self.call_command("--recreate")
@@ -133,7 +136,7 @@ def test_success_recreate(self):
 
     def test_success_recreate_dry_run(self):
         (self.migrations_dir / "__init__.py").touch()
-        (self.migrations_dir / "0001_initial.py").touch()
+        (self.migrations_dir / "0001_initial.py").write_text(empty_migration)
         (self.migrations_dir / "max_migration.txt").write_text("0001_initial\n")
 
         out, err, returncode = self.call_command("--recreate", "--dry-run")
@@ -144,7 +147,7 @@ def test_success_recreate_dry_run(self):
 
     def test_success_specific_app_label(self):
         (self.migrations_dir / "__init__.py").touch()
-        (self.migrations_dir / "0001_initial.py").touch()
+        (self.migrations_dir / "0001_initial.py").write_text(empty_migration)
 
         out, err, returncode = self.call_command("testapp")
 
@@ -169,3 +172,33 @@ def test_success_ignored_app_label(self):
         assert out == "No max_migration.txt files need creating.\n"
         assert err == ""
         assert returncode == 0
+
+    def test_success_custom_migration_name(self):
+        (self.migrations_dir / "__init__.py").touch()
+        (self.migrations_dir / "0001_initial.py").write_text(empty_migration)
+        (self.migrations_dir / "custom_name.py").write_text(
+            dedent(
+                """
+                from django.db import migrations
+                class Migration(migrations.Migration):
+                    dependencies = [('testapp', '0001_initial')]
+                """
+            )
+        )
+        (self.migrations_dir / "0002_updates.py").write_text(
+            dedent(
+                """
+                from django.db import migrations
+                class Migration(migrations.Migration):
+                    dependencies = [('testapp', 'custom_name')]
+                """
+            )
+        )
+
+        out, err, returncode = self.call_command()
+
+        assert out == "Created max_migration.txt for testapp.\n"
+        assert err == ""
+        assert returncode == 0
+        max_migration_txt = self.migrations_dir / "max_migration.txt"
+        assert max_migration_txt.read_text() == "0002_updates\n"