cq/scripts/slave/step/x86_wifi_step.py

from __future__ import absolute_import
import os
import pathlib
import re
import collections

from slave import base_step
from slave.step import code_coverage_step

BUILD_VARIANT = 'eng'
TARGET_PRODUCT = 'sirocco_x86_64'
BUILD_ITEM = 'sirocco_virtual_device_deps'
UNIT_TESTS = [
  'ap-wireless_unittest',
  'ap-infra_unittests',
  'ap-daemons_unittest',
]

_MAKE_ERROR_PATTERN = (
  r'^(?P<filename>[^:]+):(?P<linenum>\d+):(?P<column>\d+): '
  'error: (?P<err_msg>.*)$',
  r'^(?P<filename>[^\-]+)-(?P<linenum>\d+): error: (?P<err_msg>.*)$',
)

_GTEST_ERROR_TEST_NAME_PATTERN = re.compile(r'^\[. FAILED. \] (\S+\.\S+)$',
                                            re.MULTILINE)

def make_error_to_comments(stderr, directory):
  """
  parse stderr with regex error pattern and return comments dict
  Args:
    stderr (str): logs from stderr
  Return:
    (dict) comments: {
      "filename":
        { "linenum": <int>,
          "message": <str> error message,
        }
  }
  """
  comments = collections.defaultdict(list)

  for line in stderr.splitlines():
    match = [m for m in [
        re.match(err_pat, line) for err_pat in _MAKE_ERROR_PATTERN
        ] if m is not None]
    if match:
      match = match[0]
      filename = match.group('filename')
      if directory and filename.startswith(directory):
        filename = filename[len(directory) + len(os.path.sep):]
      linenum = int(match.group('linenum'))
      comments[filename].append({'line': linenum, 'message': line})
  return comments


class MakeCleanSim(base_step.BaseStep):

  def __init__(self, **kwargs):
    base_step.BaseStep.__init__(self,
                                name='make clean sirocco_x86_64',
                                halt_on_failure=True,
                                **kwargs)

  def run(self):

    command = [
        'make',
        'TARGET_PRODUCT={}'.format(TARGET_PRODUCT),
        'TARGET_BUILD_VARIANT={}'.format(BUILD_VARIANT),
        'clean'
    ]
    returncode, _, stderr = self.exec_subprocess(command)
    if returncode != 0:
      comment = make_error_to_comments(stderr, '')
      self.add_review({
          'comments': comment
      })
      return False

    return True


class BuildSimulatorStep(base_step.BaseStep):

  def __init__(self, code_coverage=False, **kwargs):
    self._code_coverage = code_coverage
    base_step.BaseStep.__init__(self,
                                name='build x86_64 sirocco_x86_64',
                                halt_on_failure=True,
                                **kwargs)

  def run(self):
    command = [
        'make',
        '-j{}'.format(self.get_num_jobs()),
        'TARGET_PRODUCT={}'.format(TARGET_PRODUCT),
        'TARGET_BUILD_VARIANT={}'.format(BUILD_VARIANT)
    ]

    command += self.build_accelerator.make_flags
    command += [BUILD_ITEM]
    returncode, _, stderr = self.exec_subprocess(command)
    if returncode != 0:
      comment = make_error_to_comments(stderr, '')
      self.add_review({
          'comments': comment
      })
      return False

    return True

class BuildUnitTestsStep(base_step.BaseStep):

  def __init__(self, code_coverage=False, **kwargs):
    self._code_coverage = code_coverage
    super().__init__(name='build unittests sirocco_x86_64',
                     halt_on_failure=True,
                     **kwargs)

  def run(self):
    command = [
        'make',
        '-j{}'.format(self.get_num_jobs()),
        'TARGET_PRODUCT={}'.format(TARGET_PRODUCT),
        'TARGET_BUILD_VARIANT={}'.format(BUILD_VARIANT),
    ]

    if self._code_coverage:
      command.append('COMMAND_LINE_GN_ARGS=\''
        'use_clang_coverage=true is_component_build=false\''
      )

    if self.build_accelerator.make_flags:
      command.extend(self.build_accelerator.make_flags)
    command.append('sirocco_virtual_unittests')
    returncode, _, stderr = self.exec_subprocess(command)
    if returncode:
      comment = make_error_to_comments(stderr, '')
      self.add_review({
          'comments': comment
      })
      return False
    test_list = UNIT_TESTS.copy()
    self.add_step_data(code_coverage_step.STEP_DATA_TEST_LIST, test_list)
    return True

class RunUnitTestsStep(base_step.BaseStep):

  def __init__(self, test_env=None, build_out_dir=None, **kwargs):
    self._test_env = test_env
    self._build_out_dir = build_out_dir
    super().__init__(name='run unittests sirocco_x86_64',
                     halt_on_failure=True,
                     **kwargs)

  def run(self):
    # Use to run all tests, then if any failed we fail this step
    errors = []
    for test in UNIT_TESTS:
      returncode, stdout, _ = self.exec_subprocess(
        [f'{self._build_out_dir}/{test}'],
        cwd='chromium/src',
        env=self._test_env
      )
      if returncode:
        gtest_failures = re.findall(_GTEST_ERROR_TEST_NAME_PATTERN, stdout)
        if gtest_failures:
          errors.extend(
            [f'{test_case} from {test}' for test_case in gtest_failures]
          )
        else:
          errors.append(f'Test execution error for {test}')
    if errors:
      error_msg = '\n'.join(errors)
      self.add_review({
        'message': f'{len(errors)} failure(s) from unit tests:\n{error_msg}',
      })
      return False
    return True

class CreateSimTarStep(base_step.BaseStep):

  def __init__(self, **kwargs):
    self._eurekaroot = kwargs['eureka_root']
    base_step.BaseStep.__init__(self,
                                name='tar-ing sim artifacts',
                                halt_on_failure=True,
                                **kwargs)


  def run(self):
    dockerignore = os.path.join(self._eurekaroot,
        'vendor/x86_64_simulator/sirocco_x86_64/products',
        'Dockerfile.dockerignore')

    include_files = []
    exclude_files = []
    try:
      with open(dockerignore) as f:
        # Skip * on first line that excludes everything
        for line in f.readlines()[1:]:
          line = line.replace('\\', '') # Strip escaped + characters
          if line.startswith('!'):
            paths = self._eurekaroot.glob(line[1:].rstrip())
            include_files.extend(
                [str(path.relative_to(self._eurekaroot)) for path in paths])
          else:
            path = line.rstrip()
            exclude_files.append(path)
    except OSError:
      # Ignore missing file here for tests
      # tar would fail because the archive is empty
      pass

    tar_filename = '{}.tar.gz'.format(TARGET_PRODUCT)
    command = ['tar']
    command += ['-C', str(self._eurekaroot)]
    command += [f'--exclude={file}' for file in exclude_files]
    command += [
      '-czf',
      os.path.join(self.get_gcs_dir(), tar_filename),
    ]
    command += include_files

    returncode, _, stderr = self.exec_subprocess(command)
    if returncode != 0:
      comment = f'tar-ing failed: {stderr}'
      self.add_review({
        'message': comment
      })
      return False

    self.set_build_property('ota_files', [tar_filename])
    return True