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 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", test_data=None, **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.
            test_data (bytes): Test file contents.

        Returns:
            Raw file content.
        """
        cmd = [
            "download-file",
            f"https://{host}/{project}",
            ref,
            path,
        ]

        return self._run(
            kwargs.pop("step_name", "fetch"),
            cmd,
            test_data=test_data,
            stdout=self.m.raw_io.output(),
        ).stdout

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

    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=lambda: self.m.raw_io.test_api.output(test_data),
            infra_step=True,
            timeout=datetime.timedelta(minutes=5),
            **kwargs,
        )