Detect if this is a merge or rebase in rebase_migration command

CHANGELOG.rst

@@ -2,6 +2,11 @@
 Changelog
 =========
 
+2.8.0 (2023-05-28)
+------------------
+
+* Improve `rebase_migration` command to handle both rebasing and merging of the base branch
+
 2.7.0 (2023-02-25)
 ------------------
 

README.rst

@@ -150,7 +150,8 @@ Following a conflicted “rebase” operation in Git, run it with the name of th
     $ python manage.py rebase_migration <app_label>
 
 The command will use the conflict information in the ``max_migration.txt`` file to determine which migration to rebase.
-It will then rename the migration, edit it to depend on the new migration in your main branch, and update ``max_migration.txt``.
+It will automatically detect whether it's a merge or rebase operation by checking for the existence of the ``MERGE_HEAD`` file in the ``.git`` directory.
+The command then rename the migration, edit it to depend on the new migration in your main branch, and update ``max_migration.txt``.
 If Black is installed, it will format the updated migration file with it, like Django’s built-in migration commands (from version 4.1+).
 See below for some examples and caveats.
 

src/django_linear_migrations/management/commands/rebase_migration.py

@@ -184,7 +184,28 @@ def find_migration_names(max_migration_lines: list[str]) -> tuple[str, str] | No
         return None
     if not lines[-1].startswith(">>>>>>>"):
         return None
-    return lines[1].strip(), lines[-2].strip()
+    migration_names = (lines[1].strip(), lines[-2].strip())
+    if is_merge_in_progress():
+        # During the merge 'ours' and 'theirs' are swapped in comparison with rebase
+        migration_names = (migration_names[1], migration_names[0])
+    return migration_names
+
+
+def is_merge_in_progress() -> bool:
+    try:
+        result = subprocess.run(
+            ["git", "rev-parse", "--git-dir"],
+            capture_output=True,
+            check=True,
+            text=True,
+        )
+    except (FileNotFoundError, subprocess.SubprocessError):
+        # Either `git` is not available or there is no git repository, fall back to
+        # default behaviour
+        return False
+
+    git_dir = result.stdout.strip()
+    return Path(git_dir).joinpath("MERGE_HEAD").exists()
 
 
 def migration_applied(app_label: str, migration_name: str) -> bool:

tests/test_rebase_migration.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import subprocess
 import sys
 import time
 from functools import partial
@@ -438,7 +439,7 @@ def test_none_when_no_second_marker(self):
         result = module.find_migration_names(["<<<<<<<", "0002_author_nicknames"])
         assert result is None
 
-    def test_works_with_two_way_merge(self):
+    def test_works_with_two_way_merge_during_rebase(self):
         result = module.find_migration_names(
             [
                 "<<<<<<<",
@@ -450,7 +451,7 @@ def test_works_with_two_way_merge(self):
         )
         assert result == ("0002_author_nicknames", "0002_longer_titles")
 
-    def test_works_with_three_way_merge(self):
+    def test_works_with_three_way_merge_during_rebase(self):
         result = module.find_migration_names(
             [
                 "<<<<<<<",
@@ -464,6 +465,104 @@ def test_works_with_three_way_merge(self):
         )
         assert result == ("0002_author_nicknames", "0002_longer_titles")
 
+    def test_works_with_two_way_merge_during_merge(self):
+        with mock.patch.object(module, "is_merge_in_progress", return_value=True):
+            result = module.find_migration_names(
+                [
+                    "<<<<<<<",
+                    "0002_longer_titles",
+                    "=======",
+                    "0002_author_nicknames",
+                    ">>>>>>>",
+                ]
+            )
+        assert result == ("0002_author_nicknames", "0002_longer_titles")
+
+    def test_works_with_three_way_merge_during_merge(self):
+        with mock.patch.object(module, "is_merge_in_progress", return_value=True):
+            result = module.find_migration_names(
+                [
+                    "<<<<<<<",
+                    "0002_longer_titles",
+                    "|||||||",
+                    "0001_initial",
+                    "=======",
+                    "0002_author_nicknames",
+                    ">>>>>>>",
+                ]
+            )
+        assert result == ("0002_author_nicknames", "0002_longer_titles")
+
+
+class IsMergeInProgressTests(SimpleTestCase):
+    git_command = ["git", "rev-parse", "--git-dir"]
+
+    def setUp(self) -> None:
+        subprocess_run_patch = mock.patch(
+            "django_linear_migrations.management.commands.rebase_migration"
+            ".subprocess.run"
+        )
+        self.mock_subprocess_run = subprocess_run_patch.start()
+
+        self.addCleanup(subprocess_run_patch.stop)
+
+    @pytest.fixture(autouse=True)
+    def tmp_path_fixture(self, tmp_path):
+        git_dir_name = ".git" + str(time.time()).replace(".", "")
+        self.git_dir_path = tmp_path / git_dir_name
+        self.git_dir_path.mkdir()
+
+    def test_true_when_git_repository_exists_and_merge_in_progress(self):
+        with open(self.git_dir_path.joinpath("MERGE_HEAD"), "w"):
+            pass
+        self.mock_subprocess_run.return_value = subprocess.CompletedProcess(
+            args=self.git_command,
+            returncode=0,
+            stdout=str(self.git_dir_path),
+        )
+
+        result = module.is_merge_in_progress()
+
+        assert result is True
+        self.mock_subprocess_run.assert_called_once_with(
+            self.git_command,
+            capture_output=True,
+            check=True,
+            text=True,
+        )
+
+    def test_false_when_git_repository_exists_and_not_merge_in_progress(self):
+        self.mock_subprocess_run.return_value = subprocess.CompletedProcess(
+            args=self.git_command,
+            returncode=0,
+            stdout=str(self.git_dir_path),
+        )
+
+        result = module.is_merge_in_progress()
+
+        assert result is False
+
+    def test_false_when_repository_not_exists(self):
+        # subprocess.run raises SubprocessError with 128 code when there is
+        # no git repository
+        self.mock_subprocess_run.side_effect = subprocess.SubprocessError(
+            f"Command '{self.git_command}' returned non-zero exit status 128"
+        )
+
+        result = module.is_merge_in_progress()
+
+        assert result is False
+
+    def test_false_when_git_command_is_not_available(self):
+        # subprocess.run raises FailNotFound error when `git` command is not found
+        self.mock_subprocess_run.side_effect = FileNotFoundError(
+            "No such file or directory: 'git'"
+        )
+
+        result = module.is_merge_in_progress()
+
+        assert result is False
+
 
 class MigrationAppliedTests(TestCase):
     def test_table_does_not_exist(self):