.bin/git-rb-catchup

   1 #!/usr/bin/env python3
   2 # Distributed under the terms of the GNU General Public License v2 or later.
   3
   4 """Helper to automatically rebase onto latest commit possible.
   5
   6 Helpful when you have a branch tracking an old commit, and a lot of conflicting
   7 changes have landed in the latest branch, but you still want to update.
   8
   9 A single rebase to the latest commit will require addressing all the different
  10 changes at once which can be difficult, overwhelming, and error-prone.  Instead,
  11 if you rebased onto each intermediate conflicting point, you'd break up the work
  12 into smaller pieces, and be able to run tests to make sure things were still OK.
  13 """
  14
  15 import argparse
  16 import os
  17 from pathlib import Path
  18 import subprocess
  19 import sys
  20 from typing import List, Tuple, Union
  21
  22
  23 assert sys.version_info >= (3, 7), f'Need Python 3.7+, not {sys.version_info}'
  24
  25
  26 def git(args: List[str], **kwargs) -> subprocess.CompletedProcess:
  27     """Run git."""
  28     kwargs.setdefault('check', True)
  29     kwargs.setdefault('capture_output', True)
  30     kwargs.setdefault('encoding', 'utf-8')
  31     # pylint: disable=subprocess-run-check
  32     return subprocess.run(['git'] + args, **kwargs)
  33
  34
  35 def rebase(target: str) -> bool:
  36     """Try to rebase onto |target|."""
  37     try:
  38         git(['rebase', target])
  39         return True
  40     except KeyboardInterrupt:
  41         git(['rebase', '--abort'])
  42         print('aborted')
  43         sys.exit(1)
  44     except:
  45         git(['rebase', '--abort'])
  46         return False
  47
  48
  49 def rebase_bisect(lbranch: str,
  50                   rbranch: str,
  51                   behind: int,
  52                   leave_rebase: bool = False,
  53                   force_checkout: bool = False):
  54     """Try to rebase branch as close to |rbranch| as possible."""
  55     def attempt(pos: int) -> bool:
  56         target = f'{rbranch}~{pos}'
  57         print(f'Rebasing onto {target} ', end='')
  58         print('.', end='', flush=True)
  59         # Checking out these branches directly helps clobber orphaned files,
  60         # but is usually unnessary, and can slow down the overall process.
  61         if force_checkout:
  62             git(['checkout', '-f', target])
  63         print('.', end='', flush=True)
  64         if force_checkout:
  65             git(['checkout', '-f', lbranch])
  66         print('. ', end='', flush=True)
  67         ret = rebase(target)
  68         print('OK' if ret else 'failed')
  69         return ret
  70
  71     # "pmin" is the latest branch position while "pmax" is where we're now.
  72     pmin = 0
  73     pmax = behind
  74     old_mid = None
  75     first_fail = 0
  76     while True:
  77         mid = pmin + (pmax - pmin) // 2
  78         if mid == old_mid or mid < pmin or mid >= pmax:
  79             break
  80         if attempt(mid):
  81             pmax = mid
  82         else:
  83             first_fail = max(first_fail, mid)
  84             pmin = mid
  85         old_mid = mid
  86
  87     if pmin or pmax:
  88         last_target = f'{rbranch}~{first_fail}'
  89         result = git(['log', '-1', '--format=%s', last_target])
  90         subject = result.stdout.strip()
  91         if leave_rebase:
  92             print('Restarting', last_target)
  93             result = git(['rebase', last_target], check=False)
  94             print(result.stdout.strip())
  95         print(f'* Found first failure {last_target}: {subject}')
  96     else:
  97         print('All caught up!')
  98
  99
 100 def get_ahead_behind(lbranch: str, rbranch: str) -> Tuple[int, int]:
 101     """Return number of commits |lbranch| is ahead & behind relative to |rbranch|."""
 102     output = git(
 103         ['rev-list', '--first-parent', '--left-right', '--count',
 104          f'{lbranch}...{rbranch}']).stdout
 105     return [int(x) for x in output.split()]
 106
 107
 108 def get_tracking_branch(branch: str) -> Union[str, None]:
 109     """Return branch that |branch| is tracking."""
 110     merge = git(['config', '--local', f'branch.{branch}.merge']).stdout.strip()
 111     if not merge:
 112         return None
 113
 114     remote = git(['config', '--local', f'branch.{branch}.remote']).stdout.strip()
 115     if remote:
 116         if merge.startswith('refs/heads/'):
 117             merge = merge[11:]
 118         return f'{remote}/{merge}'
 119     else:
 120         return merge
 121
 122
 123 def get_local_branch() -> str:
 124     """Return the name of the local checked out branch."""
 125     return git(['branch', '--show-current']).stdout.strip()
 126
 127
 128 def get_parser() -> argparse.ArgumentParser:
 129     """Get CLI parser."""
 130     parser = argparse.ArgumentParser(
 131         description=__doc__,
 132         formatter_class=argparse.RawDescriptionHelpFormatter)
 133     parser.add_argument(
 134         '--skip-initial-rebase-latest', dest='initial_rebase',
 135         action='store_false', default=True,
 136         help='skip initial rebase attempt onto the latest branch')
 137     parser.add_argument(
 138         '--leave-at-last-failed-rebase', dest='leave_rebase',
 139         action='store_true', default=False,
 140         help='leave tree state at last failing rebase')
 141     parser.add_argument(
 142         '--checkout-before-rebase', dest='force_checkout',
 143         action='store_true', default=False,
 144         help='force checkout before rebasing to target (to cleanup orphans)')
 145     parser.add_argument(
 146         'branch', nargs='?',
 147         help='branch to rebase onto')
 148     return parser
 149
 150
 151 def main(argv: List[str]) -> int:
 152     """The main entry point for scripts."""
 153     parser = get_parser()
 154     opts = parser.parse_args(argv)
 155
 156     try:
 157         lbranch = get_local_branch()
 158     except subprocess.CalledProcessError as e:
 159         sys.exit(f'{os.path.basename(sys.argv[0])}: {Path.cwd()}:\n{e}\n{e.stderr.strip()}')
 160     print(f'Local branch resolved to "{lbranch}".')
 161     if not lbranch:
 162         print('Unable to resolve local branch', file=sys.stderr)
 163         return 1
 164
 165     if opts.branch:
 166         rbranch = opts.branch
 167     else:
 168         rbranch = get_tracking_branch(lbranch)
 169     print(f'Tracking branch resolved to "{rbranch}".')
 170
 171     ahead, behind = get_ahead_behind(lbranch, rbranch)
 172     print(f'Branch is {ahead} commits ahead and {behind} commits behind.')
 173     print('NB: Counts for the first parent in merge history, not all commits.')
 174
 175     if not behind:
 176         print('Up-to-date!')
 177     elif not ahead:
 178         print('Fast forwarding ...')
 179         git(['merge'])
 180     else:
 181         if opts.initial_rebase:
 182             print(f'Trying to rebase onto latest {rbranch} ... ',
 183                   end='', flush=True)
 184             if rebase(rbranch):
 185                 print('OK!')
 186                 return 0
 187             print('failed; falling back to bisect')
 188         rebase_bisect(lbranch, rbranch, behind, leave_rebase=opts.leave_rebase,
 189                       force_checkout=opts.force_checkout)
 190
 191     return 0
 192
 193
 194 if __name__ == '__main__':
 195     sys.exit(main(sys.argv[1:]))