cq/scripts/slave/step/jacoco_coverage_processing_step.py

"""Build step class for processing jacoco exec files."""
import collections
import json
import logging
import os
import pathlib
import zipfile
import xml.etree.ElementTree as ET

from slave.step import cast_shell_step

COVERAGE_SCRIPT_PATH = 'build/android/generate_jacoco_report.py'
SOURCES_JSON_SUFFIX = '__jacoco_sources.json'
MAPPINGS_FILE = 'jcov_file_mappings.json'

class JacocoCoverageProcessingStep(cast_shell_step.NonOtaBuildBaseStep):
  """Build step class to process jacoco code coverage data."""

  def __init__(
      self,
      coverage_dir,
      report_out_dir='report',
      report_format='xml',
      build_args_product=None,
      build_args_official=None,
      **kwargs,
  ):
    """Creates a JacocoCoverageProcessingStep instance.

      Args:
        coverage_dir: Directory under the build directory that contains the
          jacoco exec files.
        report_out_dir: Directory under the gcs directory, to output coverage
          report.
        build_args_product: The name of one of the
          chromecast/internal/build/args/product files to use for the GN args.
        report_format: Output format of the jacoco coverage report
          ('html', 'xml', 'csv').
        build_args_official: Whether the build is an official build. MUST be set
          for any build that will ship to users. Disables debugging; do not
          set for development builds.
        **kwargs: Any additional args to pass to BaseStep.
      """
    super().__init__(
        'process jacoco code coverage data.',
        build_args_product,
        build_args_official,
        **kwargs,
    )
    self._coverage_dir = coverage_dir
    self._report_out_dir = report_out_dir
    self._report_format = report_format

    assert report_format in ['xml', 'html', 'csv']

  def get_output_report_path(self):
    if self._report_format == 'xml':
      logging.info('gcs dir:%s', self.get_gcs_dir())
      path = (
          pathlib.Path(self.get_gcs_dir())
          / self._report_out_dir
          / 'processed_report.xml'
      )
      return path

  def _remap_coverage_report_paths(self, input_xml_report, output_xml_report):
    """Creates a new xml file which has the file path as in the fs.

    Args:
      input_xml_report: path to xml report.
      output_xml_report: path to write the procesed xml report.
    """
    jacoco_xml = ET.parse(input_xml_report)
    mapped_files = set()

    for root, _, files in os.walk(self._outdir, followlinks=True):
      for file in files:
        if file.endswith(SOURCES_JSON_SUFFIX):
          file_path = os.path.join(root, file)
          mapped_files = mapped_files.union(
              self._get_mappings_from_sources_json(file_path, jacoco_xml)
          )

    self._update_coverage_report_xml(
        mapped_files,
        jacoco_xml,
        output_xml_report,
    )

  def _update_coverage_report_xml(
      self,
      source_to_workspace_path_mapping,
      report_contents,
      output_report_path,
  ):
    root = report_contents.getroot()

    new_packages = collections.defaultdict(lambda: ET.Element('package'))

    new_root = ET.Element('report')
    new_tree = ET.ElementTree(new_root)

    for file_mapping in source_to_workspace_path_mapping:
      pkg_name, file_name, real_path = file_mapping
      real_package_path = str(pathlib.Path(real_path).parent)
      file_element = root.find(
          f'.//package[@name="{pkg_name}"]//sourcefile[@name="{file_name}"]'
      )
      if file_element is None:
        continue

      new_packages[real_package_path].set('name', real_package_path)
      new_packages[real_package_path].append(file_element)

    for new_package in new_packages.values():
      new_root.append(new_package)

    new_tree.write(output_report_path, encoding='utf-8')

  def _get_mappings_from_sources_json(self, sources_json_file, jacoco_xml):
    """Maps the files to their real paths.

    Args:
      sources_json_file: path to sources json file.
      jacoco_xml: object for parsing xml report.
    """
    with open(sources_json_file) as file:
      parsed_json = json.load(file)

    mapped_files = set()
    src_dirs_sorted = sorted(parsed_json['source_dirs'], key=len)

    for jar_file in parsed_json['input_path']:
      with zipfile.ZipFile(jar_file, 'r') as zf:
        for java_class_file in zf.infolist():
          package, sourcefile = self._get_filepath_from_classpath(
              java_class_file.filename, jacoco_xml)

          if package and sourcefile:
            real_path = self._get_path_to_file_from_src_dirs(
                src_dirs_sorted, package, sourcefile)

            if real_path:
              mapped_files.add((package, sourcefile, real_path))

    return mapped_files

  def _get_filepath_from_classpath(self, class_file, jacoco_xml):
    """Given the class file name and the jacoco_xml object, this function
       returns the package name and the filename of the class.

    Args:
      class_file: filepath to a class relative to the JAR's root.
      jacoco_xml: the parsed Jacoco XML report.
    """
    class_name = class_file.split('.')[0]
    package = self._get_package_from_class_file(class_file)
    root = jacoco_xml.getroot()

    class_element = root.find(f'.//class[@name="{class_name}"]')

    # For some reason, the bool value of some of the element objects are
    # `False`.
    if class_element is not None:
      filename = class_element.attrib['sourcefilename']

      return package, filename

    return None, None

  def _get_package_from_class_file(self, class_file):
    """Given the class file name, this function returns the package name.

    Args:
      class_file: name of class file.
    """
    return class_file.rsplit('/', 1)[0]

  def _get_path_to_file_from_src_dirs(self, src_dirs, pkg_name, sourcefile):
    """Returns the path to the file, given the `source_dirs` list from the
       sources json file, the name of the package, and the sourcefile name.

    Args:
      src_dirs: `source_dirs` list from sources json file. src_dirs should be
                sorted by length of package to avoid issues with overlapping
                package names.
      pkg_name: name of package.
      sourcefile: name of source file.
    """
    for src_dir in src_dirs:
      if src_dir.endswith(pkg_name):
        path = os.path.join('chromium/src', src_dir, sourcefile)

        # Resolve symlinks.
        path = os.path.join(self._getcwd(), path)
        path = os.path.realpath(path)
        path = os.path.relpath(path, self._getcwd())

        return path

    return None

  def run(self):
    """Runs the generate jacoco report script and generates a coverage report
    from the jacoco `exec` files.

    Returns:
      True iff scripts ran successfuly with no errors.
    """
    coverage_script_path = os.path.join(
        self.get_project_path('chromium/src'), COVERAGE_SCRIPT_PATH)

    vpython3_path = os.path.join(
        self.get_project_path('chromium/tools/depot_tools'),
        'vpython3',
    )

    command = [vpython3_path, coverage_script_path]
    command += ['--format', self._report_format]

    report_path = os.path.join(self.get_gcs_dir(), self._report_out_dir)

    # Make sure directory exists.
    os.makedirs(report_path, exist_ok=True)

    if self._report_format == 'html':
      command += ['--output-dir', report_path]
    else:
      report_file = os.path.join(report_path, f'report.{self._report_format}')

      command += ['--output-file', report_file]

    coverage_dir_path = os.path.join(self._outdir, self._coverage_dir)

    if (not os.path.exists(coverage_dir_path) or
        not os.path.isdir(coverage_dir_path) or
        not os.listdir(coverage_dir_path)):
      logging.info(
          'No files found under coverage directory: %s... Not processing '
          'coverage files',
          coverage_dir_path,
      )
      return True

    command += ['--sources-json-dir', self._outdir]
    command += ['--coverage-dir', coverage_dir_path]
    command += ['--device-or-host', 'host']

    returncode, _, _ = self.exec_subprocess(command)

    # Create the mappings file only for xml reports, as this is the only
    # format that is accepted by the coverage processor at the moment.
    if self._report_format == 'xml':
      output_path = self.get_output_report_path()
      self._remap_coverage_report_paths(report_file, output_path)
      self.set_build_property(
          'coverage_jacoco_files',
          [str(output_path.relative_to(self.get_gcs_dir()))])

    return returncode == 0