tools/patman/gitutil.py

   1 # Copyright (c) 2011 The Chromium OS Authors.
   2 #
   3 # SPDX-License-Identifier:      GPL-2.0+
   4 #
   5
   6 import command
   7 import re
   8 import os
   9 import series
  10 import subprocess
  11 import sys
  12 import terminal
  13
  14 import checkpatch
  15 import settings
  16
  17
  18 def CountCommitsToBranch():
  19     """Returns number of commits between HEAD and the tracking branch.
  20
  21     This looks back to the tracking branch and works out the number of commits
  22     since then.
  23
  24     Return:
  25         Number of patches that exist on top of the branch
  26     """
  27     pipe = [['git', 'log', '--no-color', '--oneline', '--no-decorate',
  28              '@{upstream}..'],
  29             ['wc', '-l']]
  30     stdout = command.RunPipe(pipe, capture=True, oneline=True).stdout
  31     patch_count = int(stdout)
  32     return patch_count
  33
  34 def GetUpstream(git_dir, branch):
  35     """Returns the name of the upstream for a branch
  36
  37     Args:
  38         git_dir: Git directory containing repo
  39         branch: Name of branch
  40
  41     Returns:
  42         Name of upstream branch (e.g. 'upstream/master') or None if none
  43     """
  44     try:
  45         remote = command.OutputOneLine('git', '--git-dir', git_dir, 'config',
  46                                        'branch.%s.remote' % branch)
  47         merge = command.OutputOneLine('git', '--git-dir', git_dir, 'config',
  48                                       'branch.%s.merge' % branch)
  49     except:
  50         return None
  51
  52     if remote == '.':
  53         return merge
  54     elif remote and merge:
  55         leaf = merge.split('/')[-1]
  56         return '%s/%s' % (remote, leaf)
  57     else:
  58         raise ValueError, ("Cannot determine upstream branch for branch "
  59                 "'%s' remote='%s', merge='%s'" % (branch, remote, merge))
  60
  61
  62 def GetRangeInBranch(git_dir, branch, include_upstream=False):
  63     """Returns an expression for the commits in the given branch.
  64
  65     Args:
  66         git_dir: Directory containing git repo
  67         branch: Name of branch
  68     Return:
  69         Expression in the form 'upstream..branch' which can be used to
  70         access the commits. If the branch does not exist, returns None.
  71     """
  72     upstream = GetUpstream(git_dir, branch)
  73     if not upstream:
  74         return None
  75     return '%s%s..%s' % (upstream, '~' if include_upstream else '', branch)
  76
  77 def CountCommitsInBranch(git_dir, branch, include_upstream=False):
  78     """Returns the number of commits in the given branch.
  79
  80     Args:
  81         git_dir: Directory containing git repo
  82         branch: Name of branch
  83     Return:
  84         Number of patches that exist on top of the branch, or None if the
  85         branch does not exist.
  86     """
  87     range_expr = GetRangeInBranch(git_dir, branch, include_upstream)
  88     if not range_expr:
  89         return None
  90     pipe = [['git', '--git-dir', git_dir, 'log', '--oneline', '--no-decorate',
  91              range_expr],
  92             ['wc', '-l']]
  93     result = command.RunPipe(pipe, capture=True, oneline=True)
  94     patch_count = int(result.stdout)
  95     return patch_count
  96
  97 def CountCommits(commit_range):
  98     """Returns the number of commits in the given range.
  99
 100     Args:
 101         commit_range: Range of commits to count (e.g. 'HEAD..base')
 102     Return:
 103         Number of patches that exist on top of the branch
 104     """
 105     pipe = [['git', 'log', '--oneline', '--no-decorate', commit_range],
 106             ['wc', '-l']]
 107     stdout = command.RunPipe(pipe, capture=True, oneline=True).stdout
 108     patch_count = int(stdout)
 109     return patch_count
 110
 111 def Checkout(commit_hash, git_dir=None, work_tree=None, force=False):
 112     """Checkout the selected commit for this build
 113
 114     Args:
 115         commit_hash: Commit hash to check out
 116     """
 117     pipe = ['git']
 118     if git_dir:
 119         pipe.extend(['--git-dir', git_dir])
 120     if work_tree:
 121         pipe.extend(['--work-tree', work_tree])
 122     pipe.append('checkout')
 123     if force:
 124         pipe.append('-f')
 125     pipe.append(commit_hash)
 126     result = command.RunPipe([pipe], capture=True, raise_on_error=False)
 127     if result.return_code != 0:
 128         raise OSError, 'git checkout (%s): %s' % (pipe, result.stderr)
 129
 130 def Clone(git_dir, output_dir):
 131     """Checkout the selected commit for this build
 132
 133     Args:
 134         commit_hash: Commit hash to check out
 135     """
 136     pipe = ['git', 'clone', git_dir, '.']
 137     result = command.RunPipe([pipe], capture=True, cwd=output_dir)
 138     if result.return_code != 0:
 139         raise OSError, 'git clone: %s' % result.stderr
 140
 141 def Fetch(git_dir=None, work_tree=None):
 142     """Fetch from the origin repo
 143
 144     Args:
 145         commit_hash: Commit hash to check out
 146     """
 147     pipe = ['git']
 148     if git_dir:
 149         pipe.extend(['--git-dir', git_dir])
 150     if work_tree:
 151         pipe.extend(['--work-tree', work_tree])
 152     pipe.append('fetch')
 153     result = command.RunPipe([pipe], capture=True)
 154     if result.return_code != 0:
 155         raise OSError, 'git fetch: %s' % result.stderr
 156
 157 def CreatePatches(start, count, series):
 158     """Create a series of patches from the top of the current branch.
 159
 160     The patch files are written to the current directory using
 161     git format-patch.
 162
 163     Args:
 164         start: Commit to start from: 0=HEAD, 1=next one, etc.
 165         count: number of commits to include
 166     Return:
 167         Filename of cover letter
 168         List of filenames of patch files
 169     """
 170     if series.get('version'):
 171         version = '%s ' % series['version']
 172     cmd = ['git', 'format-patch', '-M', '--signoff']
 173     if series.get('cover'):
 174         cmd.append('--cover-letter')
 175     prefix = series.GetPatchPrefix()
 176     if prefix:
 177         cmd += ['--subject-prefix=%s' % prefix]
 178     cmd += ['HEAD~%d..HEAD~%d' % (start + count, start)]
 179
 180     stdout = command.RunList(cmd)
 181     files = stdout.splitlines()
 182
 183     # We have an extra file if there is a cover letter
 184     if series.get('cover'):
 185        return files[0], files[1:]
 186     else:
 187        return None, files
 188
 189 def ApplyPatch(verbose, fname):
 190     """Apply a patch with git am to test it
 191
 192     TODO: Convert these to use command, with stderr option
 193
 194     Args:
 195         fname: filename of patch file to apply
 196     """
 197     col = terminal.Color()
 198     cmd = ['git', 'am', fname]
 199     pipe = subprocess.Popen(cmd, stdout=subprocess.PIPE,
 200             stderr=subprocess.PIPE)
 201     stdout, stderr = pipe.communicate()
 202     re_error = re.compile('^error: patch failed: (.+):(\d+)')
 203     for line in stderr.splitlines():
 204         if verbose:
 205             print line
 206         match = re_error.match(line)
 207         if match:
 208             print checkpatch.GetWarningMsg(col, 'warning', match.group(1),
 209                                            int(match.group(2)), 'Patch failed')
 210     return pipe.returncode == 0, stdout
 211
 212 def ApplyPatches(verbose, args, start_point):
 213     """Apply the patches with git am to make sure all is well
 214
 215     Args:
 216         verbose: Print out 'git am' output verbatim
 217         args: List of patch files to apply
 218         start_point: Number of commits back from HEAD to start applying.
 219             Normally this is len(args), but it can be larger if a start
 220             offset was given.
 221     """
 222     error_count = 0
 223     col = terminal.Color()
 224
 225     # Figure out our current position
 226     cmd = ['git', 'name-rev', 'HEAD', '--name-only']
 227     pipe = subprocess.Popen(cmd, stdout=subprocess.PIPE)
 228     stdout, stderr = pipe.communicate()
 229     if pipe.returncode:
 230         str = 'Could not find current commit name'
 231         print col.Color(col.RED, str)
 232         print stdout
 233         return False
 234     old_head = stdout.splitlines()[0]
 235     if old_head == 'undefined':
 236         str = "Invalid HEAD '%s'" % stdout.strip()
 237         print col.Color(col.RED, str)
 238         return False
 239
 240     # Checkout the required start point
 241     cmd = ['git', 'checkout', 'HEAD~%d' % start_point]
 242     pipe = subprocess.Popen(cmd, stdout=subprocess.PIPE,
 243             stderr=subprocess.PIPE)
 244     stdout, stderr = pipe.communicate()
 245     if pipe.returncode:
 246         str = 'Could not move to commit before patch series'
 247         print col.Color(col.RED, str)
 248         print stdout, stderr
 249         return False
 250
 251     # Apply all the patches
 252     for fname in args:
 253         ok, stdout = ApplyPatch(verbose, fname)
 254         if not ok:
 255             print col.Color(col.RED, 'git am returned errors for %s: will '
 256                     'skip this patch' % fname)
 257             if verbose:
 258                 print stdout
 259             error_count += 1
 260             cmd = ['git', 'am', '--skip']
 261             pipe = subprocess.Popen(cmd, stdout=subprocess.PIPE)
 262             stdout, stderr = pipe.communicate()
 263             if pipe.returncode != 0:
 264                 print col.Color(col.RED, 'Unable to skip patch! Aborting...')
 265                 print stdout
 266                 break
 267
 268     # Return to our previous position
 269     cmd = ['git', 'checkout', old_head]
 270     pipe = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
 271     stdout, stderr = pipe.communicate()
 272     if pipe.returncode:
 273         print col.Color(col.RED, 'Could not move back to head commit')
 274         print stdout, stderr
 275     return error_count == 0
 276
 277 def BuildEmailList(in_list, tag=None, alias=None, raise_on_error=True):
 278     """Build a list of email addresses based on an input list.
 279
 280     Takes a list of email addresses and aliases, and turns this into a list
 281     of only email address, by resolving any aliases that are present.
 282
 283     If the tag is given, then each email address is prepended with this
 284     tag and a space. If the tag starts with a minus sign (indicating a
 285     command line parameter) then the email address is quoted.
 286
 287     Args:
 288         in_list:        List of aliases/email addresses
 289         tag:            Text to put before each address
 290         alias:          Alias dictionary
 291         raise_on_error: True to raise an error when an alias fails to match,
 292                 False to just print a message.
 293
 294     Returns:
 295         List of email addresses
 296
 297     >>> alias = {}
 298     >>> alias['fred'] = ['f.bloggs@napier.co.nz']
 299     >>> alias['john'] = ['j.bloggs@napier.co.nz']
 300     >>> alias['mary'] = ['Mary Poppins <m.poppins@cloud.net>']
 301     >>> alias['boys'] = ['fred', ' john']
 302     >>> alias['all'] = ['fred ', 'john', '   mary   ']
 303     >>> BuildEmailList(['john', 'mary'], None, alias)
 304     ['j.bloggs@napier.co.nz', 'Mary Poppins <m.poppins@cloud.net>']
 305     >>> BuildEmailList(['john', 'mary'], '--to', alias)
 306     ['--to "j.bloggs@napier.co.nz"', \
 307 '--to "Mary Poppins <m.poppins@cloud.net>"']
 308     >>> BuildEmailList(['john', 'mary'], 'Cc', alias)
 309     ['Cc j.bloggs@napier.co.nz', 'Cc Mary Poppins <m.poppins@cloud.net>']
 310     """
 311     quote = '"' if tag and tag[0] == '-' else ''
 312     raw = []
 313     for item in in_list:
 314         raw += LookupEmail(item, alias, raise_on_error=raise_on_error)
 315     result = []
 316     for item in raw:
 317         if not item in result:
 318             result.append(item)
 319     if tag:
 320         return ['%s %s%s%s' % (tag, quote, email, quote) for email in result]
 321     return result
 322
 323 def EmailPatches(series, cover_fname, args, dry_run, raise_on_error, cc_fname,
 324         self_only=False, alias=None, in_reply_to=None):
 325     """Email a patch series.
 326
 327     Args:
 328         series: Series object containing destination info
 329         cover_fname: filename of cover letter
 330         args: list of filenames of patch files
 331         dry_run: Just return the command that would be run
 332         raise_on_error: True to raise an error when an alias fails to match,
 333                 False to just print a message.
 334         cc_fname: Filename of Cc file for per-commit Cc
 335         self_only: True to just email to yourself as a test
 336         in_reply_to: If set we'll pass this to git as --in-reply-to.
 337             Should be a message ID that this is in reply to.
 338
 339     Returns:
 340         Git command that was/would be run
 341
 342     # For the duration of this doctest pretend that we ran patman with ./patman
 343     >>> _old_argv0 = sys.argv[0]
 344     >>> sys.argv[0] = './patman'
 345
 346     >>> alias = {}
 347     >>> alias['fred'] = ['f.bloggs@napier.co.nz']
 348     >>> alias['john'] = ['j.bloggs@napier.co.nz']
 349     >>> alias['mary'] = ['m.poppins@cloud.net']
 350     >>> alias['boys'] = ['fred', ' john']
 351     >>> alias['all'] = ['fred ', 'john', '   mary   ']
 352     >>> alias[os.getenv('USER')] = ['this-is-me@me.com']
 353     >>> series = series.Series()
 354     >>> series.to = ['fred']
 355     >>> series.cc = ['mary']
 356     >>> EmailPatches(series, 'cover', ['p1', 'p2'], True, True, 'cc-fname', \
 357             False, alias)
 358     'git send-email --annotate --to "f.bloggs@napier.co.nz" --cc \
 359 "m.poppins@cloud.net" --cc-cmd "./patman --cc-cmd cc-fname" cover p1 p2'
 360     >>> EmailPatches(series, None, ['p1'], True, True, 'cc-fname', False, \
 361             alias)
 362     'git send-email --annotate --to "f.bloggs@napier.co.nz" --cc \
 363 "m.poppins@cloud.net" --cc-cmd "./patman --cc-cmd cc-fname" p1'
 364     >>> series.cc = ['all']
 365     >>> EmailPatches(series, 'cover', ['p1', 'p2'], True, True, 'cc-fname', \
 366             True, alias)
 367     'git send-email --annotate --to "this-is-me@me.com" --cc-cmd "./patman \
 368 --cc-cmd cc-fname" cover p1 p2'
 369     >>> EmailPatches(series, 'cover', ['p1', 'p2'], True, True, 'cc-fname', \
 370             False, alias)
 371     'git send-email --annotate --to "f.bloggs@napier.co.nz" --cc \
 372 "f.bloggs@napier.co.nz" --cc "j.bloggs@napier.co.nz" --cc \
 373 "m.poppins@cloud.net" --cc-cmd "./patman --cc-cmd cc-fname" cover p1 p2'
 374
 375     # Restore argv[0] since we clobbered it.
 376     >>> sys.argv[0] = _old_argv0
 377     """
 378     to = BuildEmailList(series.get('to'), '--to', alias, raise_on_error)
 379     if not to:
 380         print ("No recipient, please add something like this to a commit\n"
 381             "Series-to: Fred Bloggs <f.blogs@napier.co.nz>")
 382         return
 383     cc = BuildEmailList(series.get('cc'), '--cc', alias, raise_on_error)
 384     if self_only:
 385         to = BuildEmailList([os.getenv('USER')], '--to', alias, raise_on_error)
 386         cc = []
 387     cmd = ['git', 'send-email', '--annotate']
 388     if in_reply_to:
 389         cmd.append('--in-reply-to="%s"' % in_reply_to)
 390
 391     cmd += to
 392     cmd += cc
 393     cmd += ['--cc-cmd', '"%s --cc-cmd %s"' % (sys.argv[0], cc_fname)]
 394     if cover_fname:
 395         cmd.append(cover_fname)
 396     cmd += args
 397     str = ' '.join(cmd)
 398     if not dry_run:
 399         os.system(str)
 400     return str
 401
 402
 403 def LookupEmail(lookup_name, alias=None, raise_on_error=True, level=0):
 404     """If an email address is an alias, look it up and return the full name
 405
 406     TODO: Why not just use git's own alias feature?
 407
 408     Args:
 409         lookup_name: Alias or email address to look up
 410         alias: Dictionary containing aliases (None to use settings default)
 411         raise_on_error: True to raise an error when an alias fails to match,
 412                 False to just print a message.
 413
 414     Returns:
 415         tuple:
 416             list containing a list of email addresses
 417
 418     Raises:
 419         OSError if a recursive alias reference was found
 420         ValueError if an alias was not found
 421
 422     >>> alias = {}
 423     >>> alias['fred'] = ['f.bloggs@napier.co.nz']
 424     >>> alias['john'] = ['j.bloggs@napier.co.nz']
 425     >>> alias['mary'] = ['m.poppins@cloud.net']
 426     >>> alias['boys'] = ['fred', ' john', 'f.bloggs@napier.co.nz']
 427     >>> alias['all'] = ['fred ', 'john', '   mary   ']
 428     >>> alias['loop'] = ['other', 'john', '   mary   ']
 429     >>> alias['other'] = ['loop', 'john', '   mary   ']
 430     >>> LookupEmail('mary', alias)
 431     ['m.poppins@cloud.net']
 432     >>> LookupEmail('arthur.wellesley@howe.ro.uk', alias)
 433     ['arthur.wellesley@howe.ro.uk']
 434     >>> LookupEmail('boys', alias)
 435     ['f.bloggs@napier.co.nz', 'j.bloggs@napier.co.nz']
 436     >>> LookupEmail('all', alias)
 437     ['f.bloggs@napier.co.nz', 'j.bloggs@napier.co.nz', 'm.poppins@cloud.net']
 438     >>> LookupEmail('odd', alias)
 439     Traceback (most recent call last):
 440     ...
 441     ValueError: Alias 'odd' not found
 442     >>> LookupEmail('loop', alias)
 443     Traceback (most recent call last):
 444     ...
 445     OSError: Recursive email alias at 'other'
 446     >>> LookupEmail('odd', alias, raise_on_error=False)
 447     \033[1;31mAlias 'odd' not found\033[0m
 448     []
 449     >>> # In this case the loop part will effectively be ignored.
 450     >>> LookupEmail('loop', alias, raise_on_error=False)
 451     \033[1;31mRecursive email alias at 'other'\033[0m
 452     \033[1;31mRecursive email alias at 'john'\033[0m
 453     \033[1;31mRecursive email alias at 'mary'\033[0m
 454     ['j.bloggs@napier.co.nz', 'm.poppins@cloud.net']
 455     """
 456     if not alias:
 457         alias = settings.alias
 458     lookup_name = lookup_name.strip()
 459     if '@' in lookup_name: # Perhaps a real email address
 460         return [lookup_name]
 461
 462     lookup_name = lookup_name.lower()
 463     col = terminal.Color()
 464
 465     out_list = []
 466     if level > 10:
 467         msg = "Recursive email alias at '%s'" % lookup_name
 468         if raise_on_error:
 469             raise OSError, msg
 470         else:
 471             print col.Color(col.RED, msg)
 472             return out_list
 473
 474     if lookup_name:
 475         if not lookup_name in alias:
 476             msg = "Alias '%s' not found" % lookup_name
 477             if raise_on_error:
 478                 raise ValueError, msg
 479             else:
 480                 print col.Color(col.RED, msg)
 481                 return out_list
 482         for item in alias[lookup_name]:
 483             todo = LookupEmail(item, alias, raise_on_error, level + 1)
 484             for new_item in todo:
 485                 if not new_item in out_list:
 486                     out_list.append(new_item)
 487
 488     #print "No match for alias '%s'" % lookup_name
 489     return out_list
 490
 491 def GetTopLevel():
 492     """Return name of top-level directory for this git repo.
 493
 494     Returns:
 495         Full path to git top-level directory
 496
 497     This test makes sure that we are running tests in the right subdir
 498
 499     >>> os.path.realpath(os.path.dirname(__file__)) == \
 500             os.path.join(GetTopLevel(), 'tools', 'patman')
 501     True
 502     """
 503     return command.OutputOneLine('git', 'rev-parse', '--show-toplevel')
 504
 505 def GetAliasFile():
 506     """Gets the name of the git alias file.
 507
 508     Returns:
 509         Filename of git alias file, or None if none
 510     """
 511     fname = command.OutputOneLine('git', 'config', 'sendemail.aliasesfile',
 512             raise_on_error=False)
 513     if fname:
 514         fname = os.path.join(GetTopLevel(), fname.strip())
 515     return fname
 516
 517 def GetDefaultUserName():
 518     """Gets the user.name from .gitconfig file.
 519
 520     Returns:
 521         User name found in .gitconfig file, or None if none
 522     """
 523     uname = command.OutputOneLine('git', 'config', '--global', 'user.name')
 524     return uname
 525
 526 def GetDefaultUserEmail():
 527     """Gets the user.email from the global .gitconfig file.
 528
 529     Returns:
 530         User's email found in .gitconfig file, or None if none
 531     """
 532     uemail = command.OutputOneLine('git', 'config', '--global', 'user.email')
 533     return uemail
 534
 535 def Setup():
 536     """Set up git utils, by reading the alias files."""
 537     # Check for a git alias file also
 538     alias_fname = GetAliasFile()
 539     if alias_fname:
 540         settings.ReadGitAliases(alias_fname)
 541
 542 def GetHead():
 543     """Get the hash of the current HEAD
 544
 545     Returns:
 546         Hash of HEAD
 547     """
 548     return command.OutputOneLine('git', 'show', '-s', '--pretty=format:%H')
 549
 550 if __name__ == "__main__":
 551     import doctest
 552
 553     doctest.testmod()