| # Copyright 2021 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. |
| |
| import re |
| |
| from recipe_engine import recipe_api |
| |
| |
| class CIPDUtilApi(recipe_api.RecipeApi): |
| """Utilities for interacting with CIPD.""" |
| |
| OS_RE = re.compile(r"\${os=?([^}]*)}") |
| ARCH_RE = re.compile(r"\${arch=?(.*)}") |
| PLATFORM_RE = re.compile(r"\${platform}") |
| |
| @property |
| def platform_name(self): |
| """Returns CIPD's name for the current platform. |
| |
| This is the value that the CIPD CLI will substitute for ${platform} |
| in any package names that it sees. |
| """ |
| os = self.m.platform.name.replace("win", "windows") |
| arch = {"intel": {32: "386", 64: "amd64"}, "arm": {32: "armv6", 64: "arm64"}}[ |
| self.m.platform.arch |
| ][self.m.platform.bits] |
| return "%s-%s" % (os, arch) |
| |
| def get_platforms(self, step_name, ensure_file): |
| """Get platforms specified in a CIPD ensure file. |
| |
| Args: |
| step_name (str): Name of the step. |
| ensure_file (Path): Path to ensure file. |
| |
| Returns: |
| list(str): List of platforms. |
| """ |
| with self.m.step.nest(step_name): |
| lines = self.m.file.read_text("read ensure file", ensure_file).split("\n") |
| for line in lines: |
| # Get platforms from "$VerifiedPlatform <platform>...". |
| if line.startswith("$VerifiedPlatform"): |
| return line.split(" ")[1:] |
| # If a $VerifiedPlatform directive was not found, return an empty |
| # list as the directive is not required assuming there are no |
| # platform-specific packages in the ensure file. |
| return [] |
| |
| def expand_packages_by_platforms(self, packages, platforms): |
| """Expand packages by platforms. |
| |
| e.g. for platforms ["linux-amd64", "linux-arm64", "mac-amd64"], |
| |
| ["pkg/${platform}"] -> ["pkg/linux-amd64", "pkg/linux-arm64", "pkg/mac-amd64"] |
| ["pkg/${os}-{arch}"] -> ["pkg/linux-amd64", "pkg/linux-arm64", "pkg/mac-amd64"] |
| ["pkg/${os=linux}-{arch}"] -> ["pkg/linux-amd64", "pkg/linux-arm64"] |
| ["pkg/${os=mac}-{arch=amd64}"] -> ["pkg/mac-amd64"] |
| ["pkg/${os=linux,mac}"] -> ["pkg/linux", "pkg/mac"] |
| |
| Args: |
| packages (list(str)): Package names which may contain CIPD magic |
| strings. |
| platforms (list(str)): List of platform names. |
| |
| Returns: |
| list(str): Expanded list of platform-specific packages. |
| """ |
| if not platforms: |
| return packages |
| |
| def parse_allowlist(match): |
| if not match: |
| return [] |
| allowed = match.group(1) |
| return allowed.split(",") if allowed else [] |
| |
| expanded_packages = set() |
| |
| for package in packages: |
| match_platform = self.PLATFORM_RE.search(package) |
| match_os = self.OS_RE.search(package) |
| match_arch = self.ARCH_RE.search(package) |
| os_allowlist = parse_allowlist(match_os) or [ |
| p.split("-")[0] for p in platforms |
| ] |
| arch_allowlist = parse_allowlist(match_arch) or [ |
| p.split("-")[1] for p in platforms |
| ] |
| |
| for platform in platforms: |
| os, arch = platform.split("-") |
| # Replace all instances ${platform}, ${os}, and ${arch}. Apply |
| # allowlists if specified by the package. If an allowlist is |
| # empty, all values are valid. |
| if match_platform: |
| expanded_packages.add( |
| package.replace(match_platform.group(0), platform) |
| ) |
| elif match_os and match_arch: |
| if os in os_allowlist and arch in arch_allowlist: |
| expanded_packages.add( |
| package.replace(match_os.group(0), os).replace( |
| match_arch.group(0), arch |
| ) |
| ) |
| elif match_os: |
| if os in os_allowlist: |
| expanded_packages.add(package.replace(match_os.group(0), os)) |
| elif match_arch: |
| if arch in arch_allowlist: |
| expanded_packages.add( |
| package.replace(match_arch.group(0), arch) |
| ) |
| else: |
| expanded_packages.add(package) |
| |
| return sorted(list(expanded_packages)) |
| |
| def update_packages( |
| self, |
| step_name, |
| ensure_file, |
| packages, |
| version, |
| ): |
| """Updates an ensure file's packages to the target version. |
| |
| NOTE: This is a custom parser. This should be relatively safe because |
| the ensure file format is very stable, but we should eventually make |
| parsing natively supported by the CIPD client for robustness' sake. See |
| go/rolling-cipd-ensure-files. |
| |
| Args: |
| step_name (str): Name of the step. |
| ensure_file (Path): Path to ensure file. |
| packages (seq(str)): Packages to update. |
| version (str): Target version to update to. |
| """ |
| with self.m.step.nest(step_name): |
| lines = self.m.file.read_text("read ensure file", ensure_file).split("\n") |
| modified_lines = [] |
| |
| for line in lines: |
| for package in packages: |
| # Lines which do not start with the name of a package are |
| # comments, settings, or directives, which should remain |
| # untouched. |
| # |
| # Look for the terminating space character to ensure that we |
| # don't accidentally touch a package whose name is a prefix |
| # of another package. |
| if line.startswith("%s " % package): |
| line = "%s %s" % (package, version) |
| break |
| modified_lines.append(line) |
| |
| self.m.file.write_text( |
| "write ensure file", ensure_file, "\n".join(modified_lines) |
| ) |
| |
| def upload_package( |
| self, |
| pkg_name, |
| pkg_root, |
| search_tag, |
| pkg_paths=None, |
| repository=None, |
| install_mode="copy", |
| refs=("latest",), |
| metadata=None, |
| add_version_file=True, |
| name=None, |
| step_name="cipd", |
| ): |
| """Creates and uploads a CIPD package containing the tool at pkg_dir. |
| |
| The tool is published to CIPD under the path pkg_name. |
| |
| Args: |
| pkg_name (basestr): The CIPD package to publish to. |
| pkg_root (Path): The absolute path to the parent directory of the |
| package. |
| search_tag (dict): The tag to search for the CIPD pin with. This |
| should contain one element and be either `git_revision` or |
| `version`. |
| pkg_paths (list(Path)): A list of Path objects which specify the |
| paths to directories or files to upload. Defaults to [pkg_root]. |
| repository (str or None): The git repository where code for the |
| package lives. |
| install_mode (str or None): The install mode for the package. |
| refs (str): Refs to set on the package. |
| metadata (list of pair/Metadata or None): Metadata to add to the |
| package. |
| add_version_file (bool): Include a .cipd_version file in the |
| package. |
| name (str or None): Logical name of the package. Defaults to |
| second-to-last part of pkg_name. |
| step_name (str): Name of the step. |
| |
| Returns: |
| The CIPDApi.Pin instance_id. |
| """ |
| pkg_paths = pkg_paths or [pkg_root] |
| |
| with self.m.step.nest(step_name) as presentation: |
| dry_run = self.m.buildbucket_util.is_tryjob or not pkg_name |
| if dry_run: |
| presentation.step_text = "dry run" |
| pkg_name = pkg_name or "experimental/fuchsia/cipd-util-test/${platform}" |
| |
| pkg_def = self.m.cipd.PackageDefinition( |
| package_name=pkg_name, package_root=pkg_root, install_mode=install_mode |
| ) |
| |
| # Mock the existence of the package root directory so `isdir()` will |
| # return True. |
| self.m.path.mock_add_directory(pkg_root) |
| for path in pkg_paths: |
| if self.m.path.isdir(path): |
| pkg_def.add_dir(path) |
| else: |
| pkg_def.add_file(path) |
| |
| if not name: |
| # E.g., "fuchsia/go/linux-amd64" -> "go". |
| name = str(pkg_name.split("/")[-2]) |
| if add_version_file: |
| pkg_def.add_version_file(".versions/%s.cipd_version" % name) |
| |
| pkg_path = self.m.path.mkdtemp("cipd-util-build").join("%s.pkg" % name) |
| cipd_pin = self.m.cipd.build_from_pkg(pkg_def, pkg_path) |
| |
| if dry_run: |
| return cipd_pin.instance_id |
| |
| assert ( |
| len(search_tag) == 1 |
| ), "search_tag must contain one (key: value) pair to search for." |
| search_tag_key = list(search_tag.keys())[0] |
| search_tag_value = search_tag[search_tag_key] |
| assert search_tag_value, "value for search tag %r is unset: %r" % ( |
| search_tag_key, |
| search_tag_value, |
| ) |
| cipd_pins = self.m.cipd.search( |
| pkg_name, |
| "%s:%s" % (search_tag_key, search_tag_value), |
| test_instances=[], |
| ) |
| if cipd_pins: |
| self.m.step.empty("package is up-to-date") |
| assert len(cipd_pins) == 1, "%s has too many pins" % pkg_name |
| return cipd_pins[0].instance_id |
| |
| tags = {} |
| tags.update(search_tag) |
| |
| final_metadata = [ |
| self.m.cipd.Metadata("bbid", self.m.buildbucket_util.id), |
| ] |
| if repository: |
| final_metadata.append( |
| self.m.cipd.Metadata("git_repository", repository) |
| ) |
| |
| if metadata: |
| for metadatum in metadata: |
| if isinstance(metadatum, self.m.cipd.Metadata): |
| final_metadata.append(metadatum) |
| else: |
| key, value = metadatum |
| final_metadata.append(self.m.cipd.Metadata(key, value)) |
| # TODO(fxbug.dev/85982) Remove metadata from tags. |
| tags[key] = value |
| |
| cipd_pin = self.m.cipd.register( |
| package_name=pkg_name, |
| package_path=pkg_path, |
| refs=refs, |
| tags=tags, |
| metadata=final_metadata, |
| ) |
| presentation.properties.update(cipd_pin._asdict()) |
| if search_tag: |
| # Return search_tag to output properties so it can be |
| # used by builders like goma_toolchain. |
| presentation.properties.update(search_tag) |
| |
| return cipd_pin.instance_id |
