bisect_kit/git_util.py

# -*- coding: utf-8 -*-
# Copyright 2017 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Git utility."""

from __future__ import print_function
import logging
import os
import re
import subprocess

from bisect_kit import cli
from bisect_kit import util

logger = logging.getLogger(__name__)

GIT_FULL_COMMIT_ID_LENGTH = 40

# Minimal acceptable length of git commit id.
#
# For chromium, hash collision rate over number of digits:
#   - 6 digits: 4.85%
#   - 7 digits: 0.32%
#   - 8 digits: 0.01%
# As foolproof check, 7 digits should be enough.
GIT_MIN_COMMIT_ID_LENGTH = 7


def is_git_rev(s):
  """Is a git hash-like version string.

  It accepts shortened hash with at least 7 digits.
  """
  if not GIT_MIN_COMMIT_ID_LENGTH <= len(s) <= GIT_FULL_COMMIT_ID_LENGTH:
    return False
  return bool(re.match(r'^[0-9a-f]+$', s))


def argtype_git_rev(s):
  """Validates git hash."""
  if not is_git_rev(s):
    msg = 'should be git hash, at least %d digits' % GIT_MIN_COMMIT_ID_LENGTH
    raise cli.ArgTypeError(msg, '1a2b3c4d5e')
  return s


def is_git_root(path):
  """Is given path root of git repo."""
  return os.path.exists(os.path.join(path, '.git'))


def clone(git_repo, repo_url, reference=None):
  if not os.path.exists(git_repo):
    os.makedirs(git_repo)
  cmd = ['git', 'clone', repo_url, '.']
  if reference:
    cmd += ['--reference', reference]
  util.check_call(*cmd, cwd=git_repo)


def checkout_version(git_repo, rev):
  """git checkout.

  Args:
    git_repo: path of git repo.
    rev: git commit revision to checkout.
  """
  util.check_call('git', 'checkout', '-q', '-f', rev, cwd=git_repo)


def is_containing_commit(git_repo, rev):
  """Determines given commit exists.

  Args:
    git_repo: path of git repo.
    rev: git commit revision in query.

  Returns:
    True if rev is inside given git repo. If git_repo is not a git folder,
    returns False as well.
  """
  try:
    return util.check_output(
        'git', 'cat-file', '-t', rev, cwd=git_repo) == 'commit\n'
  except subprocess.CalledProcessError:
    return False
  except OSError:
    return False


def is_ancestor_commit(git_repo, old, new):
  """Determines `old` commit is ancestor of `new` commit.

  Args:
    git_repo: path of git repo.
    old: the ancestor commit.
    new: the descendant commit.

  Returns:
    True only if `old` is the ancestor of `new`. One commit is not considered
    as ancestor of itself.
  """
  return util.check_output(
      'git',
      'rev-list',
      '--ancestry-path',
      '-1',
      '%s..%s' % (old, new),
      cwd=git_repo) != ''


def get_revlist(git_repo, old, new):
  """Enumerates git commit between two revisions (inclusive).

  Args:
    git_repo: path of git repo.
    old: git commit revision.
    new: git commit revision.

  Returns:
    list of git revisions. The list contains the input revisions, old and new.
  """
  assert old
  assert new
  cmd = ['git', 'rev-list', '--reverse', '%s^..%s' % (old, new)]
  revlist = util.check_output(*cmd, cwd=git_repo).splitlines()
  return revlist


def get_commit_log(git_repo, rev):
  """Get git commit log.

  Args:
    git_repo: path of git repo.
    rev: git commit revision.

  Returns:
    commit log message
  """
  cmd = ['git', 'log', '-1', '--format=%B', rev]
  msg = util.check_output(*cmd, cwd=git_repo)
  return msg


def get_commit_hash(git_repo, rev):
  """Get git commit hash.

  Args:
    git_repo: path of git repo.
    rev: could be git tag, branch, or (shortened) commit hash

  Returns:
    full git commit hash
  """
  cmd = ['git', 'rev-parse', rev]
  git_rev = util.check_output(*cmd, cwd=git_repo).strip()
  assert git_rev
  return git_rev


def get_commit_time(git_repo, rev, path):
  """Get git commit timestamp.

  Args:
    git_repo: path of git repo
    rev: git commit id, branch name, tag name, or other git object
    path: path, relative to git_repo

  Returns:
    timestamp (int)
  """
  line = util.check_output(
      'git', 'log', '-1', '--format=%ct', rev, '--', path, cwd=git_repo)
  return int(line)


def get_file_from_revision(git_repo, rev, path):
  """Get file content of given revision.

  Args:
    git_repo: path of git repo
    rev: git commit id
    path: file path

  Returns:
    file content (str)
  """
  return util.check_output(
      'git', 'show', '%s:%s' % (rev, path), cwd=git_repo, log_output=False)


def list_dir_from_revision(git_repo, rev, path):
  """Lists entries of directory of given revision.

  Args:
    git_repo: path of git repo
    rev: git commit id
    path: directory path, relative to git root

  Returns:
    list of names

  Raises:
    subprocess.CalledProcessError: if `path` doesn't exists in `rev`
  """
  return util.check_output(
      'git',
      'ls-tree',
      '--name-only',
      '%s:%s' % (rev, path),
      cwd=git_repo,
      log_output=False).splitlines()


def get_rev_by_time(git_repo, timestamp, path=None, branch='HEAD'):
  """Query commit of given time.

  Args:
    git_repo: path of git repo.
    timestamp: timestamp
    path: only query history of path, relative to git_repo
    branch: only the selected subset of history to query. If branch name is
        specified, only parent of the said branch is queried. If omitted, only
        queries the parent of HEAD.

  Returns:
    git commit hash. None if path didn't exist at the given time.
  """

  cmd = [
      'git',
      'rev-list',
      '--first-parent',
      '-1',
      '--before',
      str(timestamp),
      branch,
  ]
  if path:
    cmd += ['--', path]

  result = util.check_output(*cmd, cwd=git_repo).strip()
  return result or None


def get_history(git_repo, path, after=None, before=None, padding=False):
  """Get commit history of given path.

  `after` and `before` could be outside of lifetime of `path`. `padding` is
  used to control what to return for such cases.

  Args:
    git_repo: path of git repo.
    path: path to query, relative to git_repo
    after: limit history after given time (inclusive)
    before: limit history before given time (inclusive)
    padding: If True, pads returned result with dummy record at exact 'after'
        and 'before' time, if 'path' existed at that time. Otherwise, only
        returns real commits.

  Returns:
    List of (timestamp, git hash); They are all events when `path` was added,
    removed, modified, and start and end time if `padding` is true.

    For each pair, at `timestamp`, the repo state is `git hash`. In other
    words, `timestamp` is not necessary the commit time of `git hash` for the
    padded entries.
  """
  cmd = ['git', 'log', '--reverse', '--first-parent', '--format=%ct %H']
  if after:
    cmd += ['--after', str(after)]
  if before:
    cmd += ['--before', str(before)]
  # '--' is necessary otherwise if `path` is removed in current revision, git
  # will complain it's an ambiguous argument which may be path or something
  # else (like git branch name, tag name, etc.)
  cmd += ['--', path]

  result = []
  for line in util.check_output(*cmd, cwd=git_repo).splitlines():
    commit_time, git_rev = line.split()
    result.append((int(commit_time), git_rev))

  if padding:
    assert before or after, "padding=True make no sense if they are both None"
    if before is not None and get_rev_by_time(git_repo, before, path=path):
      before = int(before)
      if not result or result[-1][0] != before:
        git_rev = get_rev_by_time(git_repo, before)
        assert git_rev
        result.append((before, git_rev))
    if after is not None and get_rev_by_time(git_repo, after, path=path):
      after = int(after)
      if not result or result[0][0] != after:
        git_rev = get_rev_by_time(git_repo, after)
        assert git_rev
        result.insert(0, (after, git_rev))

  return result


def get_history_recursively(git_repo, path, after, before, parser_callback):
  """Get commit history of given path and its dependencies.

  In comparison to get_history(), get_history_recursively also takes
  dependencies into consideration. For example, if file A referenced file B,
  get_history_recursively(A) will return commits of B in addition to A. This
  applies recursively, so commits of C will be included if file B referenced
  file C, and so on.

  This function is file type neutral. `parser_callback(filename, content)` will
  be invoked to parse file content and should return list of filename of
  dependencies.

  Args:
    git_repo: path of git repo
    path: path to query, relative to git_repo
    after: limit history after given time (inclusive)
    before: limit history before given time (inclusive)
    parser_callback: callback to parse file content. See above comment.

  Returns:
    list of (commit timestamp, git hash)
  """
  history = get_history(
      git_repo, path, after=after, before=before, padding=True)

  # Collect include information of each commit.
  includes = {}
  for commit_time, git_rev in history:
    content = get_file_from_revision(git_repo, git_rev, path)
    for include_name in parser_callback(path, content):
      if include_name not in includes:
        includes[include_name] = set()
      includes[include_name].add(git_rev)

  # Analyze the start time and end time of each include.
  dependencies = []
  for include in includes:
    appeared = None
    for commit_time, git_rev in history:
      if git_rev in includes[include]:
        if not appeared:
          appeared = commit_time
      else:
        if appeared:
          dependencies.append((include, appeared, commit_time))
          appeared = None

    if appeared is not None:
      dependencies.append((include, appeared, before))

  # Recursion and merge.
  result = list(history)
  for include, appeared, disappeared in dependencies:
    result += get_history_recursively(git_repo, include, appeared, disappeared,
                                      parser_callback)

  # Sort and dedup.
  result2 = []
  for x in sorted(result):
    if result2 and result2[-1] == x:
      continue
    result2.append(x)

  return result2


def list_commits_between_commits(git_repo, old, new):
  """Get all commits between (old, new].

  Args:
    git_repo: path of git repo.
    old: old commit hash (exclusive)
    new: new commit hash (inclusive)

  Returns:
    list of (timestamp, rev)
  """
  assert old and new
  assert old == new or is_ancestor_commit(git_repo, old, new)
  commits = []
  # --first-parent is necessary for Android, see following link for more
  # discussion.
  # https://docs.google.com/document/d/1c8qiq14_ObRRjLT62sk9r5V5cyCGHX66dLYab4MVnks/edit#heading=h.n3i6mt2n6xuu
  for line in util.check_output(
      'git',
      'rev-list',
      '--timestamp',
      '--reverse',
      '--first-parent',
      '%s..%s' % (old, new),
      cwd=git_repo).splitlines():
    timestamp, git_rev = line.split()
    commits.append([int(timestamp), git_rev])

  # bisect-kit has a fundamental assumption that commit timestamps are
  # increasing because we sort and bisect the commits by timestamp across git
  # repos. If not increasing, we have to adjust the timestamp as workaround.
  # This might lead to bad bisect result, however the bad probability is low in
  # practice since most machines' clocks are good enough.
  if commits != sorted(commits, key=lambda x: x[0]):
    logger.warning('Commit timestamps are not increasing')
    last_timestamp = -1
    adjusted = 0
    for commit in commits:
      if commit[0] < last_timestamp:
        commit[0] = last_timestamp
        adjusted += 1

      last_timestamp = commit[0]
    logger.warning('%d timestamps adjusted', adjusted)

  return commits