Close `git-bug` bugs based on entries in git commit messages

close-git-bugs

#!/usr/bin/env python
"""
Close git-bugs based on meta-data from commit messages in a repository.
"""

import argparse
import logging
import os
import subprocess
import sys
import tempfile

import git

logger = logging.getLogger(__name__)


RESOLVED_BUG_MSG = """
This issue was resolved by {} in commit {}.
"""


def get_merge_commits(repo_path, branch='master'):
  """
  Return a generator yielding commits into master.
  """
  repo = git.Repo(repo_path)

  # NOTE(josh): usually we would want --merges, but since skydio has a dumb
  # merge strategy we have to do this nonsense.
  git_proc = subprocess.Popen(
      ['git', 'log', '--no-merges', '--first-parent', '--format=%H', branch],
      cwd=repo_path, stdout=subprocess.PIPE)

  for commit_hash in git_proc.stdout:
    commit = repo.commit(commit_hash.decode("utf-8").strip())

    if commit is not None:
      yield commit

  git_proc.stdout.close()
  git_proc.wait()


def get_merges_that_close_issues(tag_map, repo_path, branch='master'):
  """
  Return a generator yielding pairs (commit, resolutions) where commit is the
  commit object from git python and resolutions is a map from issue key to
  resolution from the commit message (i.e. "resolved", "closed")
  """

  for commit in get_merge_commits(repo_path, branch):
    resolutions = {}
    prev_key = None
    for line in commit.message.splitlines():
      parts = line.strip().split(':', 1)
      if len(parts) == 2:
        key, bugs = parts
        prev_key = key
      elif not prev_key:
        continue
      elif line.startswith(" " * len(prev_key)):
        # This is a continuation of the previous line
        key = prev_key
        bugs = line.strip()
        prev_key = None
      else:
        continue

      resolution = tag_map.get(key, None)
      if resolution is None:
        resolution = tag_map.get(key.lower(), None)
        if resolution is None:
          continue

      if "," in bugs:
        bug_split = bugs.split(",")
      else:
        bug_split = bugs.split()

      for bug in bug_split:
        bug = bug.strip()
        if not bug or bug == 'None':
          continue
        resolutions[bug] = resolution

    if resolutions:
      yield commit, resolutions


def close_git_bugs_with_merged_resolutions(
        code_repo, bug_repo, tag_map=None, branch='master', max_commits=10):
  """
  Walk through a list of commits to a given branch and look for commit
  message tags marking issue resolutions. For each issue that is resolved,
  update it's status in jira.
  """

  if tag_map is None:
    tag_map = {
        "Closes": "close",
        "Resolves": "close"
    }

  tmpdir = tempfile.gettempdir()
  flowtmp = os.path.join(tmpdir, "flow-tools")
  if not os.path.exists(flowtmp):
    os.makedirs(flowtmp)

  tmpargs = {
      "dir": flowtmp,
      "prefix": "git-bug-",
      "suffix": ".log",
      "delete": False,
  }

  iter_count = 0
  for commit, resolutions in (
      get_merges_that_close_issues(tag_map, code_repo, branch)):
    for bugid, resolution in sorted(resolutions.items()):

      try:
        with tempfile.NamedTemporaryFile(**tmpargs) as logfile:
          logpath = logfile.name
          current_status = subprocess.check_output(
              ["git", "bug", "status", bugid], cwd=bug_repo,
              stderr=logfile).decode("utf-8").strip()
        os.unlink(logpath)
      except subprocess.CalledProcessError:
        logger.warning("Failed to query current status of bug %s, see %s",
                       bugid, logpath)
        continue

      if current_status == "closed":
        logger.debug("Skipping: %s", bugid)
        continue

      logger.info("Closing %s", bugid)
      try:
        with tempfile.NamedTemporaryFile(**tmpargs) as logfile:
          logpath = logfile.name
          subprocess.check_call(
              ["git", "bug", "status", resolution, bugid], cwd=bug_repo,
              stdout=logfile, stderr=logfile)
          os.unlink(logpath)
      except subprocess.CalledProcessError:
        logger.warning("Failed to close bug %s, see %s", bugid, logpath)
        continue

      message = RESOLVED_BUG_MSG.format(str(commit.author), commit.hexsha)
      try:
        with tempfile.NamedTemporaryFile(**tmpargs) as logfile:
          logpath = logfile.name
          subprocess.check_call(
              ["git", "bug", "comment", "add", bugid, "--message", message],
              cwd=bug_repo, stdout=logfile, stderr=logfile)
          os.unlink(logpath)
      except subprocess.CalledProcessError:
        logger.warning(
            "Failed to add comment to bug %s, see %s", bugid, logpath)

    iter_count += 1
    if iter_count > max_commits:
      break


def main(argv=None):
  if argv is None:
    argv = sys.argv[1:]

  format_str = '[%(levelname)s] %(message)s'
  logging.basicConfig(level=logging.WARNING, format=format_str)

  parser = argparse.ArgumentParser(description=__doc__)
  parser.add_argument(
      "-l", "--log-level", choices=["debug", "info", "warning", "error"],
      default="info")
  parser.add_argument(
      "-r", "--code-repo", default=os.getcwd(),
      help="Path to the code repository")
  parser.add_argument(
      "-g", "--bug-repo", default=None,
      help="Path to the bug repository (default=code-repo)")
  parser.add_argument(
      "-b", "--branch", default="master")
  parser.add_argument(
      "-m", "--max-commits", default=30,
      help="Don't inspect more than this many commits")

  try:
    import argcomplete
    argcomplete.autocomplete(parser)
  except ImportError:
    pass
  args = parser.parse_args(argv)
  logging.getLogger().setLevel(getattr(logging, args.log_level.upper()))

  if args.bug_repo is None:
    args.bug_repo = args.code_repo

  close_git_bugs_with_merged_resolutions(
      args.code_repo, args.bug_repo, branch=args.branch,
      max_commits=args.max_commits)

  return 0


if __name__ == '__main__':
  sys.exit(main(sys.argv[1:]))