recipe_modules/cipd_dependencies/api.py

# Copyright 2019 The Fuchsia Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Provides functions to validate dependencies among packages."""

from recipe_engine import recipe_api


class CipdDependenciesApi(recipe_api.RecipeApi):
  """API for reading and validating CIPD package versions."""

  def get_dependencies(self, name, package, instance_id, versions_file):
    """Reads an embedded json dependencies file.

    Args:
      name: (str) the step name for this operation.
      package: (str) the package name.
      instance_id: (str) the id of the package to download and extract.
      versions_file: (str) the path to the dependencies file within the
        package.
    Returns:
      A dictionary with package/project dependencies as keys and their
      versions as values.
    """
    tmp_dir = self.m.path.mkdtemp('cipd')
    cipd_file = tmp_dir.join('cipd.zip')
    self.m.cipd.pkg_fetch(cipd_file, package, instance_id)
    output_dir = tmp_dir.join('output')
    self.m.archive.extract(
        step_name='extract cipd.zip', archive_file=cipd_file, output=output_dir)
    versions_path = output_dir.join(versions_file)
    return self.m.file.read_json(name, versions_path)

  def validate_against_tree(self,
                            package,
                            dependencies,
                            versions_dict,
                            is_package=True):
    """Validates dependencies are satisfied.

    Args:
      package: (str) name of the current package being rolled.
      dependencies: (list(dict)) list of strings with package dependency
        name and the key used to extract the version for the package from
        versions file.
      versions_json: (dict) dictionary with the content of versions file.
      is_package: (bool) true if we are validating package dependencies,
        False if we are validating project dependencies.

    Returns:
      An empty string if validation succeeded or a string describing the
      packages/projects that failed the validation.
    """
    package_dependencies = {}
    version_key = 'version' if is_package else 'revision'
    dep_name = 'package' if is_package else 'project'
    for dep in dependencies:
      package_name, key = dep.split('=')
      package_dependencies[package_name] = key
    if is_package:
      deps_info = self.m.jiri.package(package_dependencies.keys())
    else:
      deps_info = self.m.jiri.project(package_dependencies.keys())
    deps_to_dict = {d['name']: d[version_key] for d in deps_info.json.output}
    summary = []
    summary_pattern = ('Package {} depends on %s {} with version'
                       ' {} but found {}') % dep_name
    for package_name, key in package_dependencies.items():
      expected_version = ('git_revision:{}'.format(versions_dict[key])
                          if is_package else versions_dict[key])
      if deps_to_dict[package_name] != expected_version:
        summary.append(
            summary_pattern.format(package, package_name, expected_version,
                                   deps_to_dict[package_name]))
    return '\n'.join(summary)

  def validate_against_package(self, package, dependencies, package_versions,
                               dependent_versions):
    """Validates dependencies are satisfied.

    Args:
      package: (str) name of the dependent package.
      dependencies: (list) dependency keys.
      package_versions: (dict) of versions from rolling package.
      dependent_versions: (dict) of versions from dependent definition.

    Returns:
      An empty string if validation succeeded or a string describing the
      packages/projects that failed the validation.
    """
    summary = []
    summary_pattern = (
        'Dependent %s requires %s  with version %s but found version %s')
    for dep in dependencies:
      if package_versions[dep] != dependent_versions[dep]:
        msg = summary_pattern % (package, dep, dependent_versions[dep],
                                 package_versions[dep])
        summary.append(msg)
    return '\n'.join(summary)