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