Use MigrationLoader to order migrations

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

@@ -16,6 +16,8 @@
 from django.core.checks import register
 from django.core.checks import Tags
 from django.core.signals import setting_changed
+from django.db import connections
+from django.db import DEFAULT_DB_ALIAS
 from django.db.migrations.loader import MigrationLoader
 from django.dispatch import receiver
 from django.utils.functional import cached_property
@@ -58,6 +60,17 @@ def first_party_app_configs() -> Generator[AppConfig, None, None]:
             yield app_config
 
 
+def get_graph_plan(
+    loader: MigrationLoader,
+    app_names: Iterable[str] | None = None,
+) -> list[tuple[str, str]]:
+    nodes = loader.graph.leaf_nodes()
+    if app_names:
+        nodes = [key for key in loader.graph.leaf_nodes() if key[0] in app_names]
+    plan: list[tuple[str, str]] = loader.graph._generate_plan(nodes, at_end=True)
+    return plan
+
+
 class MigrationDetails:
     migrations_module_name: str | None
     migrations_module: ModuleType | None
@@ -121,6 +134,34 @@ def check_max_migration_files(
     else:
         app_config_set = set()
 
+    migration_loader = MigrationLoader(
+        connections[DEFAULT_DB_ALIAS], ignore_no_migrations=True
+    )
+    app_names = [a.label for a in first_party_app_configs()]
+    conflicts = migration_loader.detect_conflicts()
+    if app_names:
+        conflicts = {
+            app_label: conflict
+            for app_label, conflict in conflicts.items()
+            if app_label in app_names
+        }
+    if conflicts:
+        conflict_msg = "; ".join(
+            f"{', '.join(names)} in {app}" for app, names in conflicts.items()
+        )
+        errors.append(
+            Error(
+                id="dlm.E005",
+                msg=(
+                    "Conflicting migrations detected;"
+                    + f" multiple leaf nodes in the migration graph: {conflict_msg}."
+                ),
+                hint="Run 'python manage.py makemigrations --merge'",
+            )
+        )
+        return errors
+
+    graph_plan = get_graph_plan(loader=migration_loader, app_names=app_names)
     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,7 @@ def check_max_migration_files(
             )
             continue
 
-        real_max_migration_name = max(migration_details.names)
+        real_max_migration_name = [k[1] for k in graph_plan if k[0] == 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,12 @@
 
 from django.apps import apps
 from django.core.management.commands.makemigrations import Command as BaseCommand
+from django.db import connections
+from django.db import DEFAULT_DB_ALIAS
+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 +61,10 @@ def handle(
             sys.exit(2)
 
         any_created = False
+        migration_loader = MigrationLoader(
+            connections[DEFAULT_DB_ALIAS], ignore_no_migrations=True
+        )
+        graph_plan = get_graph_plan(loader=migration_loader, app_names=labels)
         for app_config in first_party_app_configs():
             if labels and app_config.label not in labels:
                 continue
@@ -68,7 +76,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"
+            + " in the migration graph: 0002_updates, custom_name in testapp."
+        )
+
     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"