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)