recipe_modules/gitiles/api.py

# Copyright 2017 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 base64

from recipe_engine import recipe_api


class GitilesApi(recipe_api.RecipeApi):
    """Module for polling a Git repository using the Gitiles web interface."""

    def __call__(self, step_name, cmd, test_data=None, max_attempts=1, **kwargs):
        full_cmd = [self._gitiles_path]
        full_cmd.extend(cmd)

        def run():
            return self.m.step(step_name, full_cmd, step_test_data=test_data, **kwargs)

        return self.m.utils.retry(run, max_attempts=max_attempts)

    @property
    def _gitiles_path(self):
        return self.m.cipd.ensure_tool("infra/tools/luci/gitiles/${platform}", "latest")

    def log(self, url, treeish, limit=0, step_name=None, **kwargs):
        """Returns the most recent commits for treeish object.

        Args:
          url (str): base URL of the remote repository.
          treeish (str): tree object identifier.
          limit (int): number of commits to limit the fetching to.
          step_name (str): custom name for this step (optional).
        """
        cmd = [
            "log",
            "-json-output",
            self.m.json.output(),
        ]
        if limit:
            cmd.extend(["-limit", limit])
        cmd.extend([url, treeish])
        step_name = step_name or "gitiles log: %s" % treeish
        return self(step_name, cmd, **kwargs).json.output

    def fetch(
        self,
        host,
        project,
        path,
        ref="main",
        step_name=None,
        timeout_secs=None,
    ):
        """Fetches raw file content from a Gitiles repository.

        Args:
          host (str): The Gerrit hostname.
          project (str): The project name.
          path (str): The path to the file relative to the project root.
          ref (str): A ref to fetch from.
          step_name (str): (optional) A custom step name.
          timeout_secs (int): (optional) The number of seconds to wait before
            failing.

        Returns:
          Raw file content.
        """
        url = "https://{host}/{project}/+/{ref}/{path}?format=TEXT".format(
            host=host, project=project, ref=ref, path=path
        )

        step_name = step_name or "fetch"
        step_result = self.m.url.get_text(
            url=url,
            step_name=step_name,
            timeout=timeout_secs,
        )
        return base64.b64decode(step_result.output)