.bin/git-rb-all

   1 #!/usr/bin/env python3
   2
   3 """Helper to rebase all local branches."""
   4
   5 import argparse
   6 import functools
   7 import os
   8 from pathlib import Path
   9 import re
  10 import subprocess
  11 import sys
  12
  13
  14 # Not sure if newer features are used.
  15 assert sys.version_info >= (3, 6), f'Python 3.6+ required but found {sys.version}'
  16
  17
  18 PROG = os.path.basename(__file__)
  19 DEBUG = False
  20
  21
  22 class Terminal:
  23     """Terminal escape sequences."""
  24
  25     CSI_PREFIX = '\033['
  26     SGR_SUFFIX = 'm'
  27     NORMAL = ''
  28     BOLD = '1'
  29     _BLACK, _RED, _GREEN, _YELLOW, _BLUE, _MAGENTA, _CYAN, _WHITE = range(0, 8)
  30     _FG = 30
  31     _BG = 40
  32     FG_BLACK = str(_FG + _BLACK)
  33     FG_RED = str(_FG + _RED)
  34     FG_GREEN = str(_FG + _GREEN)
  35     FG_YELLOW = str(_FG + _YELLOW)
  36     FG_BLUE = str(_FG + _BLUE)
  37     FG_MAGENTA = str(_FG + _MAGENTA)
  38     FG_CYAN = str(_FG + _CYAN)
  39     FG_WHITE = str(_FG + _WHITE)
  40
  41
  42 class Color:
  43     """Helper colors."""
  44
  45     _combine = lambda *args: Terminal.CSI_PREFIX + ';'.join(args) + Terminal.SGR_SUFFIX
  46     NORMAL = _combine(Terminal.NORMAL)
  47     GOOD = _combine(Terminal.FG_GREEN)
  48     WARN = _combine(Terminal.FG_YELLOW)
  49     BAD = _combine(Terminal.FG_RED)
  50     HILITE = _combine(Terminal.BOLD, Terminal.FG_CYAN)
  51     BRACKET = _combine(Terminal.BOLD, Terminal.FG_BLUE)
  52
  53     @classmethod
  54     def good(cls, msg):
  55         return cls.GOOD + msg + cls.NORMAL
  56
  57     @classmethod
  58     def bad(cls, msg):
  59         return cls.BAD + msg + cls.NORMAL
  60
  61
  62 def dbg(msg):
  63     """Print a debug |msg|."""
  64     if DEBUG:
  65         print(msg, file=sys.stderr)
  66
  67
  68 def fatal(msg):
  69     """Show an error |msg| then exit."""
  70     print(Color.bad(f'{PROG}: error: {msg}'), file=sys.stderr)
  71     sys.exit(1)
  72
  73
  74 def git(args, **kwargs):
  75     """Run git."""
  76     kwargs.setdefault('check', True)
  77     kwargs.setdefault('stdout', subprocess.PIPE)
  78     kwargs.setdefault('stderr', subprocess.STDOUT)
  79     kwargs.setdefault('encoding', 'utf-8')
  80     if DEBUG:
  81         print('+', 'git', *args)
  82     return subprocess.run(['git'] + args, **kwargs)
  83
  84
  85 @functools.lru_cache(maxsize=None)
  86 def checkout_is_dirty():
  87     """Determine whether the checkout is dirty (e.g. modified files)."""
  88     output = git(['status', '--porcelain']).stdout
  89     return any(x for x in output.splitlines() if '?' not in x[0:2])
  90
  91
  92 @functools.lru_cache(maxsize=None)
  93 def rebase_inprogress():
  94     """Determine whether a rebase is already in progress."""
  95     output = git(['rev-parse', '--git-path', 'rebase-merge']).stdout.strip()
  96     if Path(output).exists():
  97         return True
  98
  99     output = git(['rev-parse', '--git-path', 'rebase-apply']).stdout.strip()
 100     if Path(output).exists():
 101         return True
 102
 103     return False
 104
 105
 106 @functools.lru_cache(maxsize=None)
 107 def cherry_pick_inprogress():
 108     """Determine whether a cherry-pick is in progress."""
 109     output = git(['rev-parse', '--git-path', 'CHERRY_PICK_HEAD']).stdout.strip()
 110     return Path(output).exists()
 111
 112
 113 class AppendOption(argparse.Action):
 114     """Append the command line option (with no arguments) to dest.
 115
 116     parser.add_argument('-b', '--barg', dest='out', action='append_option')
 117     options = parser.parse_args(['-b', '--barg'])
 118     options.out == ['-b', '--barg']
 119     """
 120
 121     def __init__(self, option_strings, dest, **kwargs):
 122         if 'nargs' in kwargs:
 123             raise ValueError('nargs is not supported for append_option action')
 124         super().__init__(option_strings, dest, nargs=0, **kwargs)
 125
 126     def __call__(self, parser, namespace, values, option_string=None):
 127         if getattr(namespace, self.dest, None) is None:
 128             setattr(namespace, self.dest, [])
 129         getattr(namespace, self.dest).append(option_string)
 130
 131
 132 def get_parser():
 133     """Get CLI parser."""
 134     parser = argparse.ArgumentParser(description=__doc__)
 135     parser.add_argument(
 136         '--catchup', action='store_true',
 137         help='run git-rb-catchup when rebasing')
 138     parser.add_argument(
 139         '-d', '--debug', action='store_true',
 140         help='enable debug output')
 141     parser.add_argument(
 142         '-q', '--quiet', dest='git_options', action=AppendOption, default=['-q'],
 143         help='passthru to git rebase')
 144     return parser
 145
 146
 147 def main(argv):
 148     """The main entry point for scripts."""
 149     parser = get_parser()
 150     opts = parser.parse_args(argv)
 151
 152     global DEBUG
 153     DEBUG = opts.debug
 154
 155     # Switch to the top dir in case the working dir doesn't exist in every branch.
 156     try:
 157         topdir = git(['rev-parse', '--show-toplevel'], stderr=subprocess.PIPE).stdout.strip()
 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     os.chdir(topdir)
 161
 162     # Example output:
 163     #  ||m|refs/remotes/origin/master|ahead 2, behind 203
 164     # *||master|refs/remotes/origin/master|ahead 1
 165     #  |/usr/local/src/gnu/gdb/build/release|release||behind 10
 166     #  ||s-stash||
 167     state = git(
 168         ['for-each-ref', '--format=%(HEAD)|%(worktreepath)|%(refname:short)|%(upstream)|%(upstream:track,nobracket)',
 169          'HEAD', 'refs/heads/*']).stdout.splitlines()
 170
 171     curr_state = None
 172     branch_width = 0
 173     local_count = 0
 174     for line in state:
 175         head, worktreepath, branch, tracking, ahead_behind = line.split('|')
 176         branch_width = max(branch_width, len(branch))
 177         if head == '*':
 178             curr_state = branch
 179             local_count += 1
 180         elif not (worktreepath and worktreepath != topdir):
 181             local_count += 1
 182         else:
 183             dbg(f'{worktreepath}:{branch}: Skipping branch checked out in diff worktree')
 184     # If there are no branches to rebase, go silently.
 185     if not local_count:
 186         return 0
 187     if not curr_state:
 188         # Are we in a detached head state?
 189         if not git(['symbolic-ref', '-q', 'HEAD'], check=False).returncode:
 190             fatal('unable to resolve current branch')
 191         curr_state = git(['rev-parse', 'HEAD']).stdout.strip()
 192
 193     switched_head = False
 194     branches = {}
 195     for line in state:
 196         _, worktreepath, branch, tracking, ahead_behind = line.split('|')
 197
 198         # If it's a branch in another worktree, ignore it.
 199         if worktreepath and worktreepath != topdir:
 200             continue
 201
 202         print(f'{Color.BRACKET}### {Color.GOOD}{branch:{branch_width}}{Color.NORMAL} ',
 203               end='', flush=True)
 204         if not tracking:
 205             print(f'{Color.WARN}skipping{Color.NORMAL} due to missing merge branch')
 206             continue
 207
 208         m = re.match(r'ahead ([0-9]+)', ahead_behind)
 209         ahead = int(m.group(1)) if m else 0
 210         m = re.search(r'behind ([0-9]+)', ahead_behind)
 211         behind = int(m.group(1)) if m else 0
 212         if not behind:
 213             print(Color.good('up-to-date'))
 214             continue
 215         elif not ahead:
 216             # If we haven't switched the checkout, update-ref on current HEAD
 217             # will get us into a dirty checkout, so use merge to handle it.
 218             if switched_head or curr_state != branch:
 219                 git(['update-ref', f'refs/heads/{branch}', tracking])
 220                 print('fast forwarded [updated ref]')
 221             else:
 222                 result = git(['merge', '-q', '--ff-only'], check=False)
 223                 if result.returncode:
 224                     print(Color.bad('unable to merge') + '\n' + result.stdout.strip())
 225                 else:
 226                     print('fast forwarded [merged]')
 227             continue
 228
 229         # Skip this ref if tree is in a bad state.
 230         if rebase_inprogress():
 231             print(Color.bad('skipping due to active rebase'))
 232             continue
 233         if cherry_pick_inprogress():
 234             print(Color.bad('skipping due to active cherry-pick'))
 235             continue
 236         if checkout_is_dirty():
 237             print(Color.bad('unable to rebase: tree is dirty'))
 238             continue
 239
 240         print(f'rebasing [{ahead_behind}] ', end='', flush=True)
 241         result = git(['checkout', '-q', branch], check=False)
 242         if result.returncode:
 243             print(Color.bad('unable to checkout') + '\n' + result.stdout.strip())
 244             continue
 245         switched_head = True
 246         if opts.catchup:
 247             print()
 248             result = git(['rb-catchup'], capture_output=False, check=False)
 249         else:
 250             result = git(['rebase'] + opts.git_options, check=False)
 251             if result.returncode:
 252                 git(['rebase', '--abort'])
 253                 print(Color.bad('failed') + '\n' + result.stdout.strip())
 254             else:
 255                 print(Color.good('OK'))
 256
 257     if switched_head:
 258         git(['checkout', '-q', curr_state])
 259     return 0
 260
 261
 262 if __name__ == '__main__':
 263     sys.exit(main(sys.argv[1:]))