cq/scripts/slave/test/test_generator.py

"""Tools for generating custom recipe tests."""

from __future__ import absolute_import
import copy
import re
import sys

import six

DEFAULT_TEST_PROPERTIES = {
    'revision_ref': 'refs/changes/45/12345/1',
    'patch_branch': 'master',
    'patch_project': 'fake/unittest',
    'gerrit': 'eureka-internal',
    'issue': 12345,
    'patchset': 3,
    'manifest_url': 'https://eureka-internal.googlesource.com/eureka/manifest',
    'shallow_checkout': False,
}

# Mock manifest used by default in recipes_test.
# Defined here to be used by custom test_generators as well.
DEFAULT_MANIFEST_STRING = """
<?xml version="1.0" encoding="UTF-8"?>
<manifest>
  <remote fetch="https://chromium.googlesource.com/" name="chromium" review="https://chromium-review.googlesource.com/"/>
  <remote fetch="https://eureka-internal.googlesource.com/" name="eureka" review="sso://eureka-internal"/>
  <remote fetch="https://github.com/" name="github"/>

  <default revision="master"/>

  <project name="chromium/src" remote="eureka" revision="unfork_m43"/>
  <project name="chromium/tools/depot_tools" remote="chromium"/>
  <project name="chromiumos/chromite" path="chromite/chromite" remote="chromium"/>
  <project name="chromiumos/third_party/autotest" path="autotest" remote="chromium"/>
  <project name="continuous-tests" remote="eureka"/>
  <project name="fake/unittest" remote="eureka"/>
  <project name="fragglerock" remote="eureka"/>
  <project name="prebuilt/toolchain" path="toolchain" remote="eureka"/>
  <project name="google/sprockets" path="test/sprockets" remote="github"/>
  <project name="test" remote="eureka"/>
  <project name="chromecast/internal" path="chromium/src/chromecast/internal" remote="eureka"/>
  <project name="chromecast/internal/voice_ui" path="chromium/src/chromecast/internal/chirp" remote="eureka"/>
  <project name="vendor/amlogic" remote="eureka"/>
  <project name="vendor/marvell" remote="eureka"/>
</manifest>
"""


def repo_version_matcher():
  return match('repo', 'version')


def gclient_sync_matcher():
  return match(
      'luci-auth', 'context -disable-git-auth -service-account-json :gce -- ' +
      'gclient sync --force --with_branch_heads ' +
      '--delete_unversioned_trees --reset --jobs=[0-9]+')


def match(tool, optional_arg_regex=''):
  return '{} {}'.format(any_path_to(tool), optional_arg_regex)


def any_path_to(tool):
  return '(\/?[A-z]\/?)*\/?' + tool


class TestGenerator(object):
  """The state needed to run a single recipe test.

  An instance of this class should represent a unique test for a build recipe,
  allowing the recipe to provide fake data for one or more steps in order to
  trigger a certain result.
  """

  def __init__(self, test_name, buildername, branch='master', properties=None):
    """Instantiates a TestGenerator instance.

    Args:
      test_name: name for this test case. Unique within a recipe.
      buildername: the name of the build used for the test. Must be valid within
        the recipe this instance is used with.
      branch: The manifest branch used to generate the manifest. Defaults to a
        reasonable value.
      properties: If set, these properties will be added to the dict returned by
        get_properties, overriding the DEFAULT_TEST_PROPERTIES when duplicate
        keys exist.
    """
    assert buildername
    assert test_name
    self._test_name = test_name
    self._buildername = buildername
    self._repo_branch = branch
    self._custom_properties = properties

    self._mock_files = {}
    self._mock_executions = []

  def add_mock_execution(self,
                         execution_regex,
                         retcode=0,
                         stdout=None,
                         stderr=None,
                         reusable=True):
    """Adds a mock execution result.

    Args:
      execution_regex: regex used to match an executed command.
      retcode: return code to mock if the regex matches.
      stdout: stdout to mock if the regex matches.
      stderr: stderr to mock if the regex matches.
    """
    assert isinstance(execution_regex, six.string_types)
    assert isinstance(retcode, int)
    self._mock_executions.append({
        'regex': execution_regex,
        'retcode': retcode,
        'used': False,
        'stdout': stdout,
        'stderr': stderr,
        'reusable': reusable,
    })

  def exec_command(self, command):
    """Mocks a command execution and returns exit values as appropriate.

    Args:
      command: the command to run, as a single string.

    Returns:
      The tuple (return code, stdout, stderr). If values are explicitly mocked,
      those values will be inserted; otherwise, the default is (0, '', '').
    """
    assert isinstance(command, six.string_types)
    for execution in self._mock_executions:
      if execution['used'] and not execution['reusable']:
        continue
      if re.search(execution['regex'], command):
        execution['used'] = True
        stdout = execution['stdout'] or ''
        stderr = execution['stderr'] or ''
        return (execution['retcode'], stdout, stderr)
    return (0, '', '')

  def exec_non_blocking_command(self, command):
    """Mocks a non blocking command execution and returns a mock subprocess.

    Args:
      command: the command to run, as a single string.

    Returns:
      A mock subprocess instance.
    """
    assert isinstance(command, six.string_types)

    log_file_descriptor = 9999
    log_file_path = '/tmpFolder/tmpPath'

    class MockSubprocess(object):
      returncode = None
      pid = 123

      def terminate(self, *args, **kwargs):  # pylint: disable=unused-argument,invalid-name
        return 0

      def poll(self, *args, **kwargs):  # pylint: disable=unused-argument,invalid-name
        return 0

      def kill(self):  # pylint: disable-unused-argument,invalid-name
        return 0

    return MockSubprocess(), log_file_descriptor, log_file_path

  def add_mock_file(self, filename, file_contents):
    """Adds a mock file result by filename.

    A mock result will override file contents read using
    BaseStep.exec_read_file.

    Args:
      filename: the path, relative or absolute, of the file to mock.
      file_contents: the contents of the file to return when read.

    Raises:
      KeyError: if the filename was already mocked in this TestGenerator.
    """
    assert isinstance(filename, six.string_types)
    assert isinstance(file_contents, six.string_types)
    if filename in self._mock_files:
      raise KeyError(filename)
    self._mock_files[filename] = {'contents': file_contents, 'used': False}

  def all_mocks_used(self):
    """Returns whether or not all mocks were exercised for this test."""
    for filename in self._mock_files:
      mock = self._mock_files[filename]
      if not mock['used']:
        return False

    for mock in self._mock_executions:
      if not mock['used']:
        return False

    return True

  def get_unused_mocks_message(self):
    """Returns the list of mocks that were unused for this test."""
    results = []
    for filename in self._mock_files:
      mock = self._mock_files[filename]
      if not mock['used']:
        results.append(str(mock))

    for mock in self._mock_executions:
      if not mock['used']:
        results.append(str(mock))

    if results:
      return 'Unused mocks:\n{}'.format('\n'.join(results))
    return None

  def read_file(self, filename):
    """Mocks reading file contents within a test context.

    Args:
      filename: the path, relative or absolute, of the file to read.

    Returns:
      The mocked file contents, or an empty string if not explicitly mocked.
    """
    if filename in self._mock_files:
      self._mock_files[filename]['used'] = True
      return self._mock_files[filename]['contents']
    return ''

  def get_test_name(self):
    """Returns the test name represented by this instance."""
    return self._test_name

  def get_properties(self):
    """Returns a set of build properties used for running this test."""
    properties = copy.deepcopy(DEFAULT_TEST_PROPERTIES)
    properties['manifest_branch'] = self._repo_branch
    properties['buildername'] = self._buildername

    if self._custom_properties:
      properties.update(self._custom_properties)

    return properties