cq/scripts/slave/recipe/catatester_binaries_runner.py

"""Recipe for generating and testing test binaries."""
import os
from typing import Dict, Union

from slave import base_recipe
from slave.step import catatester_binaries_step
from slave.step import cipd_upload_step
from slave.step import py_env_setup_step
from slave.step import py_test_step
from slave.test import test_generator
from tools import subprocess42

_CIPD_PACKAGE_ROOT = os.path.join('lab_system', 'cipd_packages')


# (b/149845115) Current lab environment can only support the below libc version
TARGETED_LIBC_VERSION = '2.28'

BUILD_CONFIGS = {
    'catatester_binaries': {
        'py3': False,
        'requirements_file_name': 'requirements.txt',
    },
    'py3_catatester_binaries': {
        'py3': True,
        'requirements_file_name': 'requirements3.txt',
    },
}

Config = Dict[str, Union[bool, str]]


def GetValidBuildNames():
  return list(BUILD_CONFIGS.keys())


def CreateRecipe(build_name: str, eureka_root: str, **kwargs):
  if build_name in BUILD_CONFIGS:
    return CatatesterBinariesRecipe(
        config=BUILD_CONFIGS[build_name], eureka_root=eureka_root, **kwargs)
  raise base_recipe.BuildNameNotSupportedError(build_name)


class CatatesterBinariesRecipe(base_recipe.BaseRecipe):
  """Recipe to generate and test Catatester binaries."""

  def __init__(self, config: Config, eureka_root: str, **kwargs):
    base_recipe.BaseRecipe.__init__(self, eureka_root=eureka_root, **kwargs)
    self._py3 = config.get('py3', False)
    self._requirements_file_name = config.get('requirements_file_name',
                                              'requirements.txt')

  # pylint: disable=method-hidden
  def _supported_libc_found(self):
    """Return True if supported version is found; otherwise, False."""
    short_timeout_secs = 5
    cmd = ['ldd', '--version']
    output, _, returncode, _ = subprocess42.call_with_timeout(
        cmd, timeout=short_timeout_secs, text=True)
    if returncode != 0:
      return False
    return TARGETED_LIBC_VERSION in output

  def get_steps(self):
    requirements_project = 'test'
    requirements_path = os.path.join('lab_system', self._requirements_file_name)
    catatester_python_packages = 'team-catatester/python_packages'

    test_regex = '.*test_binaries.py$'
    test_timeout = 60 * 15  # 15 minutes
    additional_import_paths = [
        self._eureka_root / path for path in (
            'builder/masters',
            'builder/masters/src',
            'chromium/src/third_party/catapult/telemetry',
            'chromium/src/chromecast/internal',
            'chromium/src/chromecast/tools',
            'continuous-tests',
            'continuous-tests/cq/scripts',
            'test',
            'testing/external/autotest',
            'testing/external/chromite',
            'testing/sprockets',
        )
    ]
    steps = [
        py_env_setup_step.PyEnvSetupStep(
            additional_import_paths=additional_import_paths,
            requirements_project=requirements_project,
            requirements_path=requirements_path,
            local_packages_dir=catatester_python_packages,
            py3=self._py3,
            **self._step_kwargs),
        catatester_binaries_step.CatatesterBinariesBuildStep(
            halt_on_failure=True, py3=self._py3, **self._step_kwargs),
        py_test_step.PyTestStep(
            test_file_filter_regex=test_regex,
            timeout=test_timeout,
            project='test',
            halt_on_failure=True,
            **self._step_kwargs),
    ]

    if self._supported_libc_found():
      tool_step = self._GetCIPDUploadStep(
          os.path.join(_CIPD_PACKAGE_ROOT, 'tool_binaries.yaml'),
          'Upload tool binaries')
      test_step = self._GetCIPDUploadStep(
          os.path.join(_CIPD_PACKAGE_ROOT, 'test_binaries.yaml'),
          'Upload test binaries',
          verification_timeout='30m0s')
      steps.extend([tool_step, test_step])

    # Temporary single package upload. See b/118714546.
    standalone_step = self._GetCIPDUploadStep(
        os.path.join(_CIPD_PACKAGE_ROOT, 'standalone_binary.yaml'),
        'Upload standalone binary')
    steps.append(standalone_step)

    return steps

  def _GetCIPDUploadStep(self, pkg_def, step_name, verification_timeout=None):
    """Helper to generate a CIPDUploadStep."""
    return cipd_upload_step.CIPDUploadStep(
        os.path.join('test', pkg_def),
        verification_timeout=verification_timeout,
        step_name=step_name,
        **self._step_kwargs)