cq/scripts/slave/step/landmine_step.py

"""Landmine step for clobbering output directories when needed.

This step checks for a landmine file in continuous-tests/cq/landmines, and
clobbers all the output directories if it finds a new landmine.

For more information on using landmines, checkout out the documentation at
continous-tests/cq/landmines/README.md
"""

from __future__ import absolute_import
import logging
import os
import re

from helpers import git_utils
from slave.step import shell_step

LANDMINE_PATTERN = r'landmine\.(?P<landmine_number>\d+)'


class LandmineStep(shell_step.ShellStep):
  """Landmine step for clobbering output directories when needed."""

  def __init__(self, eureka_root, **kwargs):
    """Creates a LandmineStep instance."""
    shell_step.ShellStep.__init__(self,
                                  name='landmine',
                                  halt_on_failure=True,
                                  **kwargs)
    self._landmine_path = eureka_root / '.landmines'

  def get_landmine_dir(self):
    """Return the directory where new landmine files get checked in."""
    continuous_tests = self.get_project_path('continuous-tests')
    return os.path.join(continuous_tests, 'cq', 'landmines')

  def get_landmine(self):
    """Returns the name of landmine file with the highest number (or None)."""
    landmine = None
    max_landmine_number = float('-inf')

    landmine_dir = self.get_landmine_dir()
    if not os.path.isdir(landmine_dir):
      return landmine

    for filename in os.listdir(landmine_dir):
      match = re.search(LANDMINE_PATTERN, filename)
      if match:
        logging.info('Found landmine file: %s', filename)
        landmine_number = int(match.group('landmine_number'))
        if landmine_number > max_landmine_number:
          landmine, max_landmine_number = filename, landmine_number
    logging.info('Highest landmine: %s', landmine)
    return landmine

  def should_clobber(self, landmine):
    """Returns True if |landmine| is new, indicating a clobber should happen."""
    return not (self._landmine_path / landmine).exists()

  def should_mark_landmine_used(self, landmine):
    """A landmine should only be marked used once it is committed.

    If a landmine is in the patches running through CQ, then do not mark
    the landmine as used, so that if a builder runs another CL between running
    the landmine CL and the landmine actually landing, the landmine can run
    again.
    """
    all_patches = [{'project': self.patch_project}] + self.depends_on_list
    changed_files = git_utils.get_changed_files_in_all_patches(
        self, all_patches, self.get_project_path_lookup_table())
    return all(landmine not in f for f in changed_files)

  def get_commands(self):
    commands = []
    landmine = self.get_landmine()
    if landmine and self.should_clobber(landmine):
      commands.append(self.ShellCommand(['make', 'clean'], cwd=os.getcwd()))
      commands.append(self.ShellCommand(
          ['mkdir', '-p', str(self._landmine_path)],
          cwd=os.getcwd()))

      if self.should_mark_landmine_used(landmine):
        commands.append(self.ShellCommand(
            ['touch', str(self._landmine_path / landmine)],
            cwd=os.getcwd()))

    return commands