recipe_modules/repo/api.py

# 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 datetime

from recipe_engine import recipe_api


class RepoApi(recipe_api.RecipeApi):
    """APIs for invoking the repo tool."""

    def checkout_from_build_input(
        self, remote, path, fallback_ref=None, timeout=datetime.timedelta(minutes=15)
    ):
        """Checkout a meta-repository from the build input.

        If the build input has a GitilesCommit which matches the specified
        remote, checkout GitilesCommit.Id. Otherwise, checkout HEAD. Do not
        attempt to patch anything even if the build input has GerritChanges.

        Args:
            remote (str): Remote URL of the meta-repository to checkout.
            path (Path): Directory to checkout in.
            fallback_ref (str): Fallback ref to checkout if build input does
              not match the specified remote.
            timeout (timedelta): Timeout of `sync` command.
        """
        with self.m.step.nest("checkout"), self.m.context(infra_steps=True):
            self.m.file.ensure_directory("ensure directory", path)
            build_input = self.m.buildbucket.build.input
            args = [
                "init",
                "--manifest-url",
                remote,
            ]
            ref = fallback_ref or "HEAD"
            gitiles_commit = build_input.gitiles_commit
            if (
                gitiles_commit
                and f"https://{gitiles_commit.host}/{gitiles_commit.project}"
                == self.m.sso.sso_to_https(remote)
            ):
                ref = gitiles_commit.id
            args += [
                # This flag is misleadingly named. It accepts a revision or a
                # floating ref.
                "--manifest-branch",
                ref,
            ]
            with self.m.context(cwd=path):
                self._run(f"repo init {remote}", args)
                self._run("repo sync", ["sync"], timeout=timeout)

    def snapshot(self, step_name, path):
        """Get a snapshot of project revisions in a repo meta-repository.

        This is comparable to a Jiri snapshot, albeit much more lightweight. It
        may be useful for tracking exact pins when the repo meta-repository only
        tracks floating refs.

        Args:
            step_name (str): Name of the step.
            path (Path): Directory to checkout in.

        Returns:
            dict: Map of project_name: git sha1.
        """
        with self.m.step.nest(step_name), self.m.context(cwd=path):
            step = self._run("repo list", ["list"], stdout=self.m.raw_io.output_text())
            projects = [l.split(" : ") for l in step.stdout.strip().split("\n")]
            snapshot = {}
            for project_path, project in projects:
                with self.m.context(cwd=path.join(project_path)):
                    snapshot[project] = self.m.git.rev_parse(
                        step_name=f"get {project} HEAD",
                        ref="HEAD",
                    )
            return snapshot

    @property
    def _repo_tool(self):
        return self.m.ensure_tool("repo", self.resource("tool_manifest.json"))

    def _run(self, step_name, args, **kwargs):
        return self.m.step(
            step_name, [self._repo_tool] + args, infra_step=True, **kwargs
        )