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.
        """
        pkg_dir = self.m.path.mkdtemp("cipd")
        pkgs = self.m.cipd.EnsureFile().add_package(package, instance_id)
        self.m.cipd.ensure(pkg_dir, pkgs, f"download {package}")
        return self.m.file.read_json(name, pkg_dir.join(versions_file))

    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 = []
        for package_name, key in sorted(package_dependencies.items()):
            expected_version = (
                f"git_revision:{versions_dict[key]}"
                if is_package
                else versions_dict[key]
            )
            if deps_to_dict[package_name] != expected_version:
                summary.append(
                    f"package {package} depends on {dep_name} {package_name} "
                    f"with version {expected_version} but found "
                    f"{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 = []
        for dep in dependencies:
            if package_versions[dep] != dependent_versions[dep]:
                summary.append(
                    f"Dependent {package} requires {dep} with version "
                    f"{dependent_versions[dep]} but found version "
                    f"{package_versions[dep]}"
                )
        return "\n".join(summary)