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