feat: configurable conflict resolver for rebase conflicts

PyGitUp/git_wrapper.py

@@ -8,7 +8,7 @@
 """
 
 
-__all__ = ['GitWrapper', 'GitError']
+__all__ = ['GitWrapper', 'GitError', 'UnresolvedConflictError']
 
 ###############################################################################
 # IMPORTS
@@ -144,11 +144,9 @@ def stash():
 
             stashed[0] = True
 
-        stash.suppress_pop = False
-
         yield stash
 
-        if stashed[0] and not stash.suppress_pop:
+        if stashed[0]:
             print(colored('unstashing', 'magenta'))
             try:
                 self._run('stash', 'pop')
@@ -346,3 +344,19 @@ def __init__(self, current_branch, target_branch, **kwargs):
             current_branch, target_branch
         )
         GitError.__init__(self, message, **kwargs)
+
+
+class UnresolvedConflictError(GitError):
+    """
+    Rebase conflict could not be resolved. Repo left in conflicted state.
+    """
+
+    def __init__(self, branch_name, target_branch, repo_path, **kwargs):
+        kwargs.pop('message', None)
+        message = (
+            f"Failed to resolve rebase conflicts for {branch_name} "
+            f"onto {target_branch}.\n"
+            f"The repo at {repo_path} is left in a conflicted state.\n"
+            f"Resolve manually, then run: git rebase --continue"
+        )
+        GitError.__init__(self, message, **kwargs)

PyGitUp/gitup.py

@@ -38,7 +38,8 @@
 
 # PyGitUp libs
 from PyGitUp.utils import execute, uniq, find
-from PyGitUp.git_wrapper import GitWrapper, GitError, RebaseError
+from PyGitUp.git_wrapper import GitWrapper, GitError, RebaseError, \
+    UnresolvedConflictError
 
 ON_WINDOWS = sys.platform == 'win32'
 
@@ -124,6 +125,7 @@ class GitUp:
         'rebase.arguments': None,
         'rebase.auto': True,
         'rebase.log-hook': None,
+        'rebase.conflict-resolver': None,
         'updates.check': True,
         'push.auto': False,
         'push.tags': False,
@@ -336,7 +338,15 @@ def rebase_all_branches(self):
                 else:
                     stasher()
                     self.git.checkout(branch.name)
-                    self.git.rebase(target)
+                    try:
+                        self.git.rebase(target)
+                    except RebaseError:
+                        if self._try_resolve_conflicts(
+                            branch.name, target.name,
+                            self.repo.working_dir
+                        ):
+                            continue
+                        raise
 
             if (self.repo.head.is_detached  # Only on Travis CI,
                     # we get a detached head after doing our rebase *confused*.
@@ -464,9 +474,58 @@ def _rebase_in_worktree(self, branch, target, worktree_path,
                 try:
                     worktree_git.rebase(target)
                 except RebaseError:
-                    stash.suppress_pop = True
+                    if self._try_resolve_conflicts(
+                        branch.name, target.name, worktree_path
+                    ):
+                        return
                     raise
 
+    def _try_resolve_conflicts(self, branch_name, target_name, repo_path):
+        """
+        Invoke the configured conflict resolver command.
+
+        Returns True if the resolver succeeded and rebase completed.
+        Returns False if no resolver is configured.
+        Raises UnresolvedConflictError if the resolver failed.
+        """
+        resolver_command = self.settings['rebase.conflict-resolver']
+        if not resolver_command:
+            return False
+
+        print(colored('invoking conflict resolver...', 'yellow'))
+
+        env = os.environ.copy()
+        env['GITUP_BRANCH'] = branch_name
+        env['GITUP_TARGET'] = target_name
+        env['GITUP_REPO_PATH'] = repo_path
+
+        result = subprocess.run(
+            resolver_command, shell=True, cwd=repo_path, env=env
+        )
+
+        if result.returncode != 0:
+            raise UnresolvedConflictError(
+                branch_name, target_name, repo_path
+            )
+
+        # Verify rebase completed
+        git_dir = subprocess.run(
+            ['git', 'rev-parse', '--git-dir'],
+            cwd=repo_path, capture_output=True, text=True
+        ).stdout.strip()
+
+        if not os.path.isabs(git_dir):
+            git_dir = os.path.join(repo_path, git_dir)
+
+        if (os.path.isdir(os.path.join(git_dir, 'rebase-merge')) or
+                os.path.isdir(os.path.join(git_dir, 'rebase-apply'))):
+            raise UnresolvedConflictError(
+                branch_name, target_name, repo_path
+            )
+
+        print(colored('conflict resolved', 'green'))
+        return True
+
     def fetch(self):
         """
         Fetch the recent refs from the remotes.

PyGitUp/tests/test_conflict_resolver.py

@@ -0,0 +1,251 @@
+# System imports
+import os
+import platform
+import stat
+from os.path import join
+
+import pytest
+from git import *
+from PyGitUp.git_wrapper import RebaseError, UnresolvedConflictError
+from PyGitUp.tests import basepath, write_file, init_master, update_file, \
+    testfile_name
+
+pytestmark = pytest.mark.skipif(
+    platform.system() == 'Windows',
+    reason="resolver scripts here are bash scripts; on Windows the "
+           "resolver command runs via cmd.exe (not bash), so a bare "
+           ".sh path won't execute there",
+)
+
+test_name_success = 'conflict_resolve_success'
+test_name_fail = 'conflict_resolve_fail'
+test_name_noresolver = 'conflict_no_resolver'
+test_name_context = 'conflict_resolve_context'
+test_name_worktree = 'conflict_resolve_worktree'
+
+repo_path_success = join(basepath, test_name_success + os.sep)
+repo_path_fail = join(basepath, test_name_fail + os.sep)
+repo_path_noresolver = join(basepath, test_name_noresolver + os.sep)
+repo_path_context = join(basepath, test_name_context + os.sep)
+repo_path_worktree = join(basepath, test_name_worktree + os.sep)
+worktree_path_worktree = join(basepath, test_name_worktree + '-wt' + os.sep)
+
+
+def setup_conflict_repo(test_name):
+    """Set up a repo with a rebase conflict."""
+    master_path, master = init_master(test_name)
+
+    # Prepare master repo
+    master.git.checkout(b=test_name)
+
+    # Clone to test repo
+    path = join(basepath, test_name)
+    master.clone(path, b=test_name)
+    repo = Repo(path, odbt=GitCmdObjectDB)
+    assert repo.working_dir == path
+
+    # Modify file in master
+    update_file(master, test_name)
+
+    # Modify same file in our repo (conflicting change)
+    contents = 'completely changed!'
+    repo_file = join(path, testfile_name)
+    write_file(repo_file, contents)
+    repo.index.add([repo_file])
+    repo.index.commit(test_name)
+
+    # Modify file in master again
+    update_file(master, test_name)
+
+    return master, repo
+
+
+def make_resolver_script(basedir, script_content):
+    """Write a resolver shell script and return its path."""
+    script_path = join(basedir, 'resolver.sh')
+    write_file(script_path, script_content)
+    os.chmod(script_path, stat.S_IRWXU)
+    return script_path
+
+
+def make_context_capturing_script(basedir, marker_path):
+    """
+    A resolver script that records its cwd and the GITUP_* env vars it
+    was invoked with, then resolves the conflict as usual.
+    """
+    return make_resolver_script(basedir, (
+        '#!/bin/bash\n'
+        'echo "$PWD" > "{marker}"\n'
+        'echo "$GITUP_BRANCH" >> "{marker}"\n'
+        'echo "$GITUP_TARGET" >> "{marker}"\n'
+        'echo "$GITUP_REPO_PATH" >> "{marker}"\n'
+        'git checkout --theirs .\n'
+        'git add -A\n'
+        'GIT_EDITOR=true git rebase --continue\n'
+    ).format(marker=marker_path))
+
+
+def read_marker(marker_path):
+    with open(marker_path) as f:
+        lines = [line.strip() for line in f.readlines()]
+    return lines
+
+
+def setup_worktree_conflict(test_name, worktree_path):
+    """
+    Set up a repo with a branch checked out in a worktree that has a
+    conflicting change relative to its upstream.
+    """
+    master_path, master = init_master(test_name)
+    master.git.checkout(b=test_name)
+
+    path = join(basepath, test_name)
+    master.clone(path, b=test_name)
+    repo = Repo(path, odbt=GitCmdObjectDB)
+    assert repo.working_dir == path
+
+    wt_branch = test_name + '-wt'
+    repo.git.branch(wt_branch, 'origin/' + test_name)
+    repo.git.worktree('add', worktree_path, wt_branch)
+    repo.git.branch('--set-upstream-to', 'origin/' + test_name, wt_branch)
+
+    # Conflicting change in the worktree branch. GitPython can't reliably
+    # access a worktree's git dir directly (see test_worktree.py), so use
+    # the git CLI via a plain Git() instance pointed at the worktree.
+    wt_git = Git(worktree_path)
+    contents = 'completely changed in worktree!'
+    wt_file = join(worktree_path, testfile_name)
+    write_file(wt_file, contents)
+    wt_git.add(testfile_name)
+    wt_git.commit(m=test_name)
+
+    # Diverge master with a conflicting change to the same file
+    update_file(master, test_name)
+
+    return master, repo, wt_branch
+
+
+def setup_module():
+    global master_success, repo_success
+    global master_fail, repo_fail
+    global master_noresolver, repo_noresolver
+    global master_context, repo_context
+    global master_worktree, repo_worktree, worktree_branch
+
+    master_success, repo_success = setup_conflict_repo(test_name_success)
+    master_fail, repo_fail = setup_conflict_repo(test_name_fail)
+    master_noresolver, repo_noresolver = setup_conflict_repo(
+        test_name_noresolver
+    )
+    master_context, repo_context = setup_conflict_repo(test_name_context)
+    master_worktree, repo_worktree, worktree_branch = (
+        setup_worktree_conflict(test_name_worktree, worktree_path_worktree)
+    )
+
+
+def test_resolver_succeeds():
+    """Resolver fixes conflicts and completes rebase."""
+    os.chdir(repo_path_success)
+
+    script = make_resolver_script(repo_path_success, (
+        '#!/bin/bash\n'
+        'git checkout --theirs .\n'
+        'git add -A\n'
+        'GIT_EDITOR=true git rebase --continue\n'
+    ))
+
+    from PyGitUp.gitup import GitUp
+    gitup = GitUp(testing=True)
+    gitup.settings['rebase.conflict-resolver'] = script
+    gitup.run()
+
+    assert 'rebasing' in gitup.states
+
+
+def test_resolver_fails():
+    """Resolver exits non-zero; UnresolvedConflictError is raised."""
+    os.chdir(repo_path_fail)
+
+    script = make_resolver_script(repo_path_fail, (
+        '#!/bin/bash\n'
+        'exit 1\n'
+    ))
+
+    from PyGitUp.gitup import GitUp
+    gitup = GitUp(testing=True)
+    gitup.settings['rebase.conflict-resolver'] = script
+
+    with pytest.raises(UnresolvedConflictError):
+        gitup.run()
+
+
+def test_no_resolver():
+    """Without a resolver, RebaseError is raised as before."""
+    os.chdir(repo_path_noresolver)
+
+    from PyGitUp.gitup import GitUp
+    gitup = GitUp(testing=True)
+    gitup.settings['rebase.conflict-resolver'] = None
+
+    with pytest.raises(RebaseError):
+        gitup.run()
+
+
+def test_resolver_runs_in_repo_with_context():
+    """
+    The resolver runs with cwd set to the repo (not inherited from the
+    calling process) and with GITUP_BRANCH/GITUP_TARGET/GITUP_REPO_PATH
+    set correctly.
+    """
+    os.chdir(repo_path_context)
+
+    marker = join(basepath, 'resolver-context.txt')
+    script = make_context_capturing_script(repo_path_context, marker)
+
+    from PyGitUp.gitup import GitUp
+    gitup = GitUp(testing=True)
+    gitup.settings['rebase.conflict-resolver'] = script
+
+    # Move outside the repo before running: if the resolver's cwd were
+    # ever accidentally inherited from the calling process instead of
+    # passed explicitly, this would make that bug visible.
+    os.chdir(basepath)
+
+    gitup.run()
+
+    assert 'rebasing' in gitup.states
+
+    pwd, branch, target, repo_path = read_marker(marker)
+    assert os.path.realpath(pwd) == os.path.realpath(
+        repo_path_context.rstrip(os.sep)
+    )
+    assert branch == test_name_context
+    assert target == 'origin/' + test_name_context
+    assert os.path.realpath(repo_path) == os.path.realpath(
+        repo_path_context.rstrip(os.sep)
+    )
+
+
+def test_resolver_in_worktree():
+    """The resolver is also invoked for branches rebased via a worktree."""
+    os.chdir(repo_path_worktree)
+
+    marker = join(basepath, 'resolver-worktree-context.txt')
+    script = make_context_capturing_script(worktree_path_worktree, marker)
+
+    from PyGitUp.gitup import GitUp
+    gitup = GitUp(testing=True)
+    gitup.settings['rebase.conflict-resolver'] = script
+    gitup.run()
+
+    assert 'rebasing' in gitup.states
+
+    pwd, branch, target, repo_path = read_marker(marker)
+    assert os.path.realpath(pwd) == os.path.realpath(
+        worktree_path_worktree.rstrip(os.sep)
+    )
+    assert branch == worktree_branch
+    assert target == 'origin/' + test_name_worktree
+    assert os.path.realpath(repo_path) == os.path.realpath(
+        worktree_path_worktree.rstrip(os.sep)
+    )

PyGitUp/tests/test_overwrite_untracked_error.py

@@ -38,6 +38,7 @@ def test_fast_forwarded():
 
     from PyGitUp.gitup import GitUp
     gitup = GitUp(testing=True)
+    gitup.settings['rebase.conflict-resolver'] = None
 
     with pytest.raises(RebaseError):
         gitup.run()