cq/scripts/slave/step/precommit_step.py

"""A step to run pre-commit hooks.

See documentation on the precommit recipe - this module
contains the single step in that recipe."""
import logging
import os
import pprint

from helpers import git_utils
from helpers import os_utils
from slave import base_step


class PreCommitStep(base_step.BaseStep):
  """Step for running pre-commit hook checks."""

  def __init__(self, config_file, repo_root_directory, show_output, patch_project: str, commit_counts: int, **kwargs):
    """
    Args:
      config_file: The pre-commit config file to use.
      repo_root_directory: The root directory to the repo checkout.
      show_output: True to make the output a comment on the CL.
      patch_project: The project to check the precommit.
      commit_counts: Number of commits made to the current project.
      **kwargs: Additional args passed to BaseStep.
    """
    base_step.BaseStep.__init__(self, name=f"{patch_project}_precommit", **kwargs)
    self._config_file = config_file
    self._repo_root_directory = repo_root_directory
    self._show_output = show_output
    self._patch_project = patch_project
    self._commit_counts = commit_counts

  def run(self) -> bool:
    """Run the step.

    Returns:
      True if either:
        1. There is a pre-commit config and there were no pre-commit failures.
        2. There was no pre-commit config.
    """
    try:
        directory = self.get_project_path(self._patch_project)
    except KeyError:
        logging.info("Unable to get directory of project %s. Skipping pre-commit.", self._patch_project)
        return True

    pre_commit_config = os.path.join(directory, self._config_file)
    if not os.path.exists(pre_commit_config):
      logging.info("No pre-commit config at %s (%s), skipping",
          pre_commit_config, os.path.abspath(pre_commit_config))
      return True

    results = [self._run_commit_message_hooks(directory, i) for i in range(self._commit_counts)]
    results += [
      self._run_pre_commit_hooks(directory),
      self._run_push_hooks(directory),
    ]
    returncode = max(returncode for returncode, _, _ in results)
    stdout = "\n".join(stdout for _, stdout, _ in results)
    stderr = "\n".join(stderr for _, _, stderr in results)
    if self._show_output:
      try:
        self.add_review({
          "comments": {},
          "message": "stdout: {}\n\nstderr: {}".format(stdout, stderr),
        })
      except UnicodeDecodeError:
        self.add_review({
          "comments": {},
          "message": ("Failed to decode precommit output as unicode, "
            "please check the builder log directly."),
        })
    else:
      self.add_review({
        "comments": {},
        "message": ("This builder is non-blocking and its output can be "
                    "ignored."),
      })
    if returncode:
      return False
    return True

  def _get_env(self):
    env = os.environ.copy()
    env["EUREKAROOT"] = self._repo_root_directory
    logging.info("Using PATH: %s",
      pprint.pformat(env.get("PATH", "").split(":")))
    try:
      chromium_src = self.get_project_path("chromium/src")
      buildtools_path = os.path.join(
        self._repo_root_directory,
        chromium_src,
        "buildtools")
      logging.info("Setting CHROMIUM_BUILDTOOLS_PATH to %s", buildtools_path)
      env["CHROMIUM_BUILDTOOLS_PATH"] = buildtools_path
    except KeyError:
      logging.info("Unable to set CHROMIUM_BUILDTOOLS_PATH, "
        "no chromium/src found.")
    return env

  def _precommit_binary(self) -> str:
    """Get the path to pre-commit in our recipe's virtualenv."""
    return "/workspace/recipe_python_virtual_env/bin/pre-commit"

  def _run_commit_message_hooks(self, directory: str, skips: int) -> tuple[int, str, str]:
    """Run all the hooks for commit message checks.

    Args:
      directory: The directory to execute the command.
      skips: Number of git commit messages to skip.

    Returns:
      A tuple of returncode, stdout, stderr from pre-commit
    """
    env = self._get_env()
    with os_utils.change_cwd(directory):
      commit_message = git_utils.commit_message(self, skip_counts=skips)
      commit_message_path = self.write_temp_file(commit_message)
      logging.debug("Using commit message tempfile %s", commit_message_path)
      cmd = [
        self._precommit_binary(),
        "run",
        "--config", self._config_file,
        "--hook-stage", "commit-msg",
        "--commit-msg-filename", commit_message_path,
      ]
      return self.exec_subprocess(cmd, env=env)

  def _run_pre_commit_hooks(self, directory: str) -> tuple[int, str, str]:
    """Run all the hooks of the 'pre-commit' type.

    This is an unfortunate naming problem - 'pre-commit' the framework
    shares a name with the 'pre-commit' git hook type, which are hooks
    to run before committing. This function runs just those hooks, as
    opposted to pre-push, commit-message, or other hook types.

    Args:
      directory: The directory to execute the command.

    Returns:
      A tuple of returncode, stdout, stderr from pre-commit
    """
    env = self._get_env()
    with os_utils.change_cwd(directory):
      cmd = [
        self._precommit_binary(),
        "run",
        "--config", self._config_file,
        "--source", "HEAD^",
        "--origin", "HEAD"
      ]
      return self.exec_subprocess(cmd, env=env)

  def _run_push_hooks(self, directory: str) -> tuple[int, str, str]:
    """Run all the hooks of the 'push' stage.

    Args:
      directory: The directory to execute the command.

    Returns:
      A tuple of returncode, stdout, stderr from pre-commit
    """
    env = self._get_env()
    with os_utils.change_cwd(directory):
      cmd = [
        self._precommit_binary(),
        "run",
        "--hook-stage", "push",
        "--config", self._config_file,
        "--source", "HEAD^",
        "--origin", "HEAD"
      ]
      return self.exec_subprocess(cmd, env=env)