cq/scripts/slave/step/lucicfg_step.py

"""Steps for executing lucicfg commands."""

from __future__ import absolute_import
import logging
import os

from slave import base_step
from helpers import os_utils

_CONFIG_PATHS_KEY = 'lucicfg_paths'


class FindConfigsStep(base_step.BaseStep):
  """Step for finding all `main.star` luci configs."""

  def __init__(self, **kwargs):
    super(FindConfigsStep, self).__init__(name='Find luci configs', **kwargs)

  def _get_config_paths(self):
    return os_utils.find_files(self.directory, 'main.star')

  def run(self):
    config_paths = self._get_config_paths()
    logging.info(config_paths)
    self.add_step_data(_CONFIG_PATHS_KEY, config_paths)
    return True


class LucicfgValidateStep(base_step.BaseStep):
  """Step for running `lucicfg validate` on `main.star` files."""

  def __init__(self, **kwargs):
    super(LucicfgValidateStep, self).__init__(name='Validate configs', **kwargs)

  def _validate_config(self, config_path):
    """Runs `lucicfg validate` against a single config."""
    cmd = ['lucicfg', 'validate', config_path]

    # Validate with GCE service account, if specified.
    if self.get_property('gce_service_account'):
      cmd.extend(['-service-account-json',
                  self.get_property('gce_service_account')])

    retcode, _, stderr = self.exec_subprocess(cmd)

    if retcode == 0:
      logging.info('%s is valid!', config_path)
      return True

    self.add_review({'message': stderr})
    logging.error('%s is NOT valid!', config_path)
    return False

  def run(self):
    """Runs the lucicfg validate step.

    Attempts to validate every luci config found in the directory.
    Expects depot_tools to be on $PATH.
    """
    config_paths = self.get_step_data(_CONFIG_PATHS_KEY)
    results = []
    for config_path in config_paths:
      result = self._validate_config(config_path)
      results.append(result)

    return all(results)