cq/scripts/slave/recipe/catabuilder_e2e_mock.py

"""Recipe for Catabuilder E2E tests."""
import enum
import json
import logging
import os
from typing import Dict, List, Union

from slave import base_recipe
from slave import base_step


@enum.unique
class Status(enum.Enum):
  PASS = 'PASSES'
  FAIL = 'FAILS'


_RESULTS_FILE = 'results.json'


def GetValidBuildNames() -> List[str]:
  return [
      'catabuilder-e2e-mock-1',
      'catabuilder-e2e-mock-2',
      'catabuilder-e2e-mock-3',
  ]


def CreateRecipe(build_name: str, **kwargs) -> 'CatabuilderE2EMockRecipe':
  """Defines how this recipe gets built."""
  return CatabuilderE2EMockRecipe(build_name, **kwargs)


class CatabuilderE2EMockRecipe(base_recipe.BaseRecipe):
  """Recipe to execute e2e fake build."""

  def __init__(self, builder_name, **kwargs):
    base_recipe.BaseRecipe.__init__(self, **kwargs)
    self._builder_name = builder_name

  def get_steps(self) -> List['CatabuilderE2EMockStep']:
    return [
        CatabuilderE2EMockStep(
            builder_name=self._builder_name, **self._step_kwargs),
    ]


class CatabuilderE2EMockStep(base_step.BaseStep):
  """Step that does the verification of the expected CB behavior."""

  def __init__(self, builder_name, **kwargs):
    base_step.BaseStep.__init__(self, name='e2e_mock', **kwargs)
    self._builder_name = builder_name

  def run(self) -> bool:
    results = self._process_instructions()
    self._save_results_to_gcs_dir(results)
    outcome = self.get_property('outcome')
    self.set_build_property('e2e_tests', 'yes')
    if outcome == Status.PASS.value:
      return True
    if outcome == Status.FAIL.value:
      return False
    raise ValueError('Status must be set in the "outcome" property.')

  def _process_instructions(self) -> Dict[str, Union[str, int, bool]]:
    instruction_project = self.get_property('project')
    project_path = self.get_project_path(instruction_project)
    instruction_filename = self.get_property('instructions_file')
    instruction_path = os.path.join(project_path, instruction_filename)
    logging.info('Opening instruction file: %s', instruction_path)
    with open(instruction_path, 'r') as instruction_file:
      instructions = json.load(instruction_file)
      assert isinstance(instructions, dict)
      logging.info('Instruction file content: %s', instructions)
      return instructions

  def _save_results_to_gcs_dir(self, results) -> None:
    gcs_dir = self._executor.get_gcs_dir()
    results_path = os.path.join(gcs_dir, _RESULTS_FILE)
    with open(results_path, 'w') as results_file:
      logging.info('Writing results to: %s', results_path)
      json.dump(results, results_file, indent=2, sort_keys=True)