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

from recipe_engine import recipe_api


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

    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 f"gitiles log: {treeish}"
        return self._run(step_name, cmd, **kwargs).json.output

    def fetch(self, host, project, path, ref="main", **kwargs):
        """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.

        Returns:
            Raw file content.
        """
        kwargs.setdefault("step_name", "fetch")

        url = f"https://{host}/{project}/+/{ref}/{path}?format=TEXT"
        step_result = self.m.url.get_text(url=url, **kwargs)
        return base64.b64decode(step_result.output)

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

    def _run(self, step_name, cmd, test_data=None, **kwargs):
        full_cmd = [self._gitiles_path]
        full_cmd.extend(cmd)
        return self.m.step(
            step_name,
            full_cmd,
            step_test_data=test_data,
            infra_step=True,
            timeout=datetime.timedelta(minutes=5),
            **kwargs,
        )