bisect_kit/repo_util.py

# -*- coding: utf-8 -*-
# Copyright 2018 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.
"""Repo utility.

This module provides wrapper for "repo" (a Google-built repository management
tool that runs on top of git) and related utility functions.
"""

from __future__ import print_function
import logging
import os
import re
import subprocess
import xml.etree.ElementTree

from six.moves import urllib

from bisect_kit import codechange
from bisect_kit import errors
from bisect_kit import git_util
from bisect_kit import util

logger = logging.getLogger(__name__)


def get_manifest_url(manifest_dir):
  """Get manifest URL of repo project.

  Args:
    manifest_dir: path of manifest directory

  Returns:
    manifest URL.
  """
  url = util.check_output(
      'git', 'config', 'remote.origin.url', cwd=manifest_dir)
  return url


def find_repo_root(path):
  """Find the root path of a repo project

  Args:
    path: path

  Returns:
    project root if path is inside a repo project; otherwise None
  """
  path = os.path.abspath(path)
  while not os.path.exists(os.path.join(path, '.repo')):
    if path == '/':
      return None
    path = os.path.dirname(path)
  return path


def init(repo_dir,
         manifest_url,
         manifest_branch=None,
         manifest_name=None,
         repo_url=None,
         reference=None,
         mirror=False):
  """Repo init.

  Args:
    repo_dir: root directory of repo
    manifest_url: manifest repository location
    manifest_branch: manifest branch or revision
    manifest_name: initial manifest file name
    repo_url: repo repository location
    reference: location of mirror directory
    mirror: indicates repo mirror
  """
  root = find_repo_root(repo_dir)
  if root and root != repo_dir:
    raise errors.ExternalError(
        '%s should not be inside another repo project at %s' % (repo_dir, root))

  cmd = ['repo', 'init', '--manifest-url', manifest_url]
  if manifest_name:
    cmd += ['--manifest-name', manifest_name]
  if manifest_branch:
    cmd += ['--manifest-branch', manifest_branch]
  if repo_url:
    cmd += ['--repo-url', repo_url]
  if reference:
    cmd += ['--reference', reference]
  if mirror:
    cmd.append('--mirror')
  util.check_call(*cmd, cwd=repo_dir)


def cleanup_repo_generated_files(repo_dir, manifest_name='default.xml'):
  """Cleanup files generated by <copyfile> <linkfile> tags.

  Args:
    repo_dir: root directory of repo
    manifest_name: filename of manifest
  """
  manifest_dir = os.path.join(repo_dir, '.repo', 'manifests')
  manifest_path = os.path.join(manifest_dir, manifest_name)
  if os.path.islink(manifest_path):
    manifest_name = os.readlink(manifest_path)
  parser = ManifestParser(manifest_dir)
  manifest = parser.parse_xml_recursive('HEAD', manifest_name)

  for copyfile in manifest.findall('.//copyfile'):
    dest = copyfile.get('dest')
    if not dest:
      continue
    # `dest` is relative to the top of the tree
    dest_path = os.path.join(repo_dir, dest)
    if not os.path.isfile(dest_path):
      continue
    logger.debug('delete file %r', dest_path)
    os.unlink(dest_path)

  for linkfile in manifest.findall('.//linkfile'):
    dest = linkfile.get('dest')
    if not dest:
      continue
    # `dest` is relative to the top of the tree
    dest_path = os.path.join(repo_dir, dest)
    if not os.path.islink(dest_path):
      continue
    logger.debug('delete link %r', dest_path)
    os.unlink(dest_path)


def sync(repo_dir, jobs=16, manifest_name=None, current_branch=False):
  """Repo sync.

  Args:
    repo_dir: root directory of repo
    jobs: projects to fetch simultaneously
    manifest_name: filename of manifest
    current_branch: fetch only current branch
  """
  # Workaround to prevent garbage files left between repo syncs
  # (http://crbug.com/881783).
  cleanup_repo_generated_files(repo_dir)

  cmd = ['repo', 'sync', '-q', '--force-sync']
  if jobs:
    cmd += ['-j', str(jobs)]
  if manifest_name:
    cmd += ['--manifest-name', manifest_name]
  if current_branch:
    cmd += ['--current-branch']
  util.check_call(*cmd, cwd=repo_dir)


def abandon(repo_dir, branch_name):
  """Repo abandon.

  Args:
    repo_dir: root directory of repo
    branch_name: branch name to abandon
  """
  # Ignore errors if failed, which means the branch didn't exist beforehand.
  util.call('repo', 'abandon', branch_name, cwd=repo_dir)


def info(repo_dir, query):
  """Repo info.

  Args:
    repo_dir: root directory of repo
    query: key to query
  """
  try:
    output = util.check_output('repo', 'info', '.', cwd=repo_dir)
  except subprocess.CalledProcessError as e:
    if 'Manifest branch:' not in e.output:
      raise
    # "repo info" may exit with error while the data we want is already
    # printed. Ignore errors for such case.
    output = e.output
  for line in output.splitlines():
    key, value = map(str.strip, line.split(':'))
    if key == query:
      return value

  return None


def get_current_branch(repo_dir):
  """Get manifest branch of existing repo directory."""
  return info(repo_dir, 'Manifest branch')


def get_manifest_groups(repo_dir):
  """Get manifest group of existing repo directory."""
  return info(repo_dir, 'Manifest groups')


def list_projects(repo_dir):
  """Repo list.

  Args:
    repo_dir: root directory of repo

  Returns:
    list of paths, relative to repo_dir
  """
  result = []
  for line in util.check_output(
      'repo', 'list', '--path-only', cwd=repo_dir).splitlines():
    result.append(line)
  return result


def cleanup_unexpected_files(repo_dir):
  """Clean up unexpected files in repo tree.

  Note this is not fully equivalent to 'repo sync' from scratch because:
    - This only handle git repo folders. In other words, directories under
      repo_dir not inside any git repo will not be touched.
    - It ignores files if matching gitignore pattern.
  So we can keep cache files to speed up incremental build next time.

  If you want truly clean tree, delete entire tree and repo sync directly
  instead.

  Args:
    repo_dir: root directory of repo
  """
  projects = list_projects(repo_dir)

  # When we clean up project X, we don't want to touch files under X's
  # subprojects. Collect the nested project relationship here.
  nested = {}
  # By sorting, parent directory will loop before subdirectories.
  for project_path in sorted(projects):
    components = project_path.split(os.sep)
    for i in range(len(components) - 1, 0, -1):
      head = os.sep.join(components[:i])
      tail = os.sep.join(components[i:])
      if head in nested:
        nested[head].append(tail)
        break
    nested[project_path] = []

  for project_path in projects:
    git_repo = os.path.join(repo_dir, project_path)
    if not os.path.exists(git_repo):
      # It should be harmless to ignore git repo nonexistence because 'repo
      # sync' will restore them.
      logger.warning('git repo not found: %s', git_repo)
      continue
    git_util.distclean(git_repo, nested[project_path])


def _urljoin(base, url):
  # urlparse.urljoin doesn't recognize "persistent-https://" protocol.
  # Following hack replaces "persistent-https" by obsolete protocol "gopher"
  # before urlparse.urljoin and replaces back after urlparse.urljoin calls.
  dummy_scheme = 'gopher://'
  new_scheme = 'persistent-https://'
  assert not base.startswith(dummy_scheme)
  assert not url.startswith(dummy_scheme)
  base = re.sub('^' + new_scheme, dummy_scheme, base)
  url = re.sub('^' + new_scheme, dummy_scheme, url)
  result = urllib.parse.urljoin(base, url)
  result = re.sub('^' + dummy_scheme, new_scheme, result)
  return result


class ManifestParser(object):
  """Enumerates historical manifest files and parses them."""

  def __init__(self, manifest_dir):
    self.manifest_dir = manifest_dir
    self.manifest_url = get_manifest_url(self.manifest_dir)

  def parse_single_xml(self, content, allow_include=False):
    root = xml.etree.ElementTree.fromstring(content)
    if not allow_include and root.find('include') is not None:
      raise errors.InternalError(
          'Expects self-contained manifest. <include> is not allowed')
    return root

  def parse_xml_recursive(self, git_rev, path):
    content = git_util.get_file_from_revision(self.manifest_dir, git_rev, path)
    root = self.parse_single_xml(content, allow_include=True)

    result = xml.etree.ElementTree.Element('manifest')
    for node in root:
      if node.tag == 'include':
        for subnode in self.parse_xml_recursive(git_rev, node.get('name')):
          result.append(subnode)
      else:
        result.append(node)
    return result

  def process_parsed_result(self, root):
    result = {}
    default = root.find('default')
    if default is None:
      default = {}

    remote_fetch_map = {}
    for remote in root.findall('.//remote'):
      name = remote.get('name')
      fetch_url = _urljoin(self.manifest_url, remote.get('fetch'))
      if urllib.parse.urlparse(fetch_url).path not in ('', '/'):
        # TODO(kcwu): support remote url with sub folders
        raise errors.InternalError(
            'only support git repo at root path of remote server: %s' %
            fetch_url)
      remote_fetch_map[name] = fetch_url

    assert root.find('include') is None

    for project in root.findall('.//project'):
      if 'notdefault' in project.get('groups', ''):
        continue
      for subproject in project.findall('.//project'):
        logger.warning('nested project %s.%s is not supported and ignored',
                       project.get('name'), subproject.get('name'))

      # default path is its name
      path = project.get('path', project.get('name'))
      revision = project.get('revision', default.get('revision'))

      remote_name = project.get('remote', default.get('remote'))
      if remote_name not in remote_fetch_map:
        raise errors.InternalError('unknown remote name=%s' % remote_name)
      fetch_url = remote_fetch_map.get(remote_name)
      repo_url = _urljoin(fetch_url, project.get('name'))

      result[path] = codechange.PathSpec(path, repo_url, revision)
    return result

  def enumerate_manifest_commits(self, start_time, end_time, path):

    def parse_dependencies(path, content):
      try:
        root = self.parse_single_xml(content, allow_include=True)
      except xml.etree.ElementTree.ParseError:
        logger.warning('%s syntax error, skip', path)
        return None

      result = []
      for include in root.findall('.//include'):
        result.append(include.get('name'))
      return result

    return git_util.get_history_recursively(self.manifest_dir, path, start_time,
                                            end_time, parse_dependencies)


class RepoMirror(codechange.CodeStorage):
  """Repo git mirror."""

  def __init__(self, mirror_dir):
    self.mirror_dir = mirror_dir

  def _url_to_cache_dir(self, url):
    # Here we assume remote fetch url is always at root of server url, so we can
    # simply treat whole path as repo project name.
    path = urllib.parse.urlparse(url).path
    assert path[0] == '/'
    return '%s.git' % path[1:]

  def cached_git_root(self, repo_url):
    cache_path = self._url_to_cache_dir(repo_url)

    # The location of chromeos manifest-internal repo mirror is irregular
    # (http://crbug.com/895957). This is a workaround.
    if cache_path == 'chromeos/manifest-internal.git':
      cache_path = 'manifest-internal.git'

    return os.path.join(self.mirror_dir, cache_path)

  def _load_project_list(self, project_root):
    repo_project_list = os.path.join(project_root, '.repo', 'project.list')
    with open(repo_project_list) as f:
      return f.readlines()

  def _save_project_list(self, project_root, lines):
    repo_project_list = os.path.join(project_root, '.repo', 'project.list')
    with open(repo_project_list, 'w') as f:
      f.write(''.join(sorted(lines)))

  def add_to_project_list(self, project_root, path, repo_url):
    lines = self._load_project_list(project_root)

    line = path + '\n'
    if line not in lines:
      lines.append(line)

    self._save_project_list(project_root, lines)

  def remove_from_project_list(self, project_root, path):
    lines = self._load_project_list(project_root)

    line = path + '\n'
    if line in lines:
      lines.remove(line)

    self._save_project_list(project_root, lines)