recipe_modules/gerrit/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 functools
from urllib.parse import urlparse

from recipe_engine import recipe_api


class GerritApi(recipe_api.RecipeApi):
    """Module for querying a Gerrit host through the Gerrit API."""

    def normalize_host(self, host):
        """Remove most elements if present, then add them back."""
        host_basename = (
            host.rstrip("/")
            .removeprefix("https://")
            .removesuffix(".googlesource.com")
            .removesuffix("-review")
        )
        return f"{host_basename}-review.googlesource.com"

    def host_from_remote_url(self, remote):
        """Given a git remote URL, returns the Gerrit host."""
        # urlparse doesn't work as expected on scheme-less URLs.
        if "://" not in remote:
            remote = f"https://{remote}"
        parsed_remote = urlparse(self.m.sso.sso_to_https(remote))
        host = parsed_remote.netloc
        host_parts = host.split(".")
        host_parts[0] += "-review"
        return ".".join(host_parts)

    def project_from_remote_url(self, remote):
        """Given a git remote URL, returns the Gerrit project."""
        # urlparse doesn't work as expected on scheme-less URLs.
        if "://" not in remote:
            remote = f"https://{remote}"
        parsed_remote = urlparse(self.m.sso.sso_to_https(remote))
        return parsed_remote.path.strip("/")

    def resolve_change(self, change):
        """Resolve a GerritChange to a URL and ref."""
        url = f"https://{change.host.replace('-review', '')}/{change.project}"
        ref = "refs/changes/%02d/%d/%d" % (
            change.change % 100,
            change.change,
            change.patchset,
        )
        return url, ref

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

    @functools.cached_property
    def host(self):
        bb_input = self.m.buildbucket.build.input
        if bb_input.gerrit_changes:
            return bb_input.gerrit_changes[0].host
        elif bb_input.gitiles_commit.host:
            gitiles_host = bb_input.gitiles_commit.host
            return gitiles_host.replace(".", "-review.", 1)
        return None  # pragma: no cover

    def abandon(self, name, change_id, message=None, notify=None, **kwargs):
        """Abandons a change.

        Returns the details of the change, after attempting to abandon.

        Args:
          name (str): The name of the step.
          change_id (str): A change ID that uniquely defines a change on the
            host.
          message (str): A message explaining the reason for abandoning the
            change.
          notify (str): One of "NONE", "OWNER", "OWNER_REVIEWERS", "ALL"
            specifying whom to send notifications to.
          **kwargs: Passed through to __call__().
        """
        input_json = {"change_id": str(change_id)}
        if message:
            input_json["input"] = {"message": message}
        if notify:
            assert notify in ("NONE", "OWNER", "OWNER_REVIEWERS", "ALL")
            input_json["input"]["notify"] = notify

        return self._run(
            name=name, subcmd="change-abandon", input_json=input_json, **kwargs
        )

    def create_change(self, name, project, subject, branch, topic=None, **kwargs):
        """Creates a new change for a given project on the gerrit host.

        Returns the details of the newly-created change.

        Args:
          name (str): The name of the step.
          project (str): The name of the project on the host to create a change
            for.
          subject (str): The subject of the new change.
          branch (str): The branch onto which the change will be made.
          topic (str): A gerrit topic that can be used to atomically land the
            change with other changes in the same topic.
          **kwargs: Passed through to __call__().
        """
        input_json = {
            "input": {"project": project, "subject": subject, "branch": branch}
        }
        if topic:
            input_json["input"]["topic"] = topic
        return self._run(
            name=name, subcmd="change-create", input_json=input_json, **kwargs
        )

    def set_review(
        self,
        name,
        change_id,
        labels=None,
        message=None,
        reviewers=None,
        ccs=None,
        revision="current",
        notify=None,
        tag=None,
        patchset_level_comment=None,
        robot_comments=None,
        ignore_automatic_attention_set_rules=None,
        **kwargs,
    ):
        """Sets a change at a revision for review. Can optionally set labels, reviewers,
        CCs, and tag.

        Returns updated labels, reviewers, and whether the change is ready for
        review as a JSON dict.

        Args:
          name (str): The name of the step.
          change_id (str): A change ID that uniquely defines a change on the
            host.
          labels (dict): A map of labels (with names as strings, e.g.
            'Code-Review') to the integral values you wish to set them to.
          message (str): A message to be added as a review comment.
          reviewers (list): A list of strings containing reviewer IDs (e.g.
            email addresses).
          ccs (list): A list of strings containing reviewer IDs (e.g. email
            addresses).
          revision (str): A revision ID that identifies a revision for the
            change (default is 'current').
          notify (str): One of "NONE", "OWNER", "OWNER_REVIEWERS", "ALL"
            specifying whom to send notifications to.
          tag (str): A tag to attach to the message, votes and comments.
          patchset_level_comment (dict): A dict in this format
            https://gerrit-review.googlesource.com/Documentation/rest-api-changes.html#comment-input
            to add as a comment to the entire patchset.
          robot_comments (dict): Robot comments to send. Should be a mapping
            from file path to list of dicts of this format:
            https://gerrit-review.googlesource.com/Documentation/rest-api-changes.html#robot-comment-input
          ignore_automatic_attention_set_rules (bool): Don't automatically add
            anybody to the attention set.
          **kwargs: Passed through to __call__().
        """
        input_json = {
            "change_id": str(change_id),
            "revision_id": str(revision),
            "input": {},
        }
        if labels:
            input_json["input"]["labels"] = labels
        if reviewers or ccs:
            input_json["input"]["reviewers"] = []
        if reviewers:
            input_json["input"]["reviewers"] += [{"reviewer": i} for i in reviewers]
        if ccs:
            input_json["input"]["reviewers"] += [
                {"reviewer": i, "state": "CC"} for i in ccs
            ]
        if notify:
            assert notify in ("NONE", "OWNER", "OWNER_REVIEWERS", "ALL")
            input_json["input"]["notify"] = notify

        if not tag:
            parts = [
                "autogenerated",
                self.m.buildbucket.build.builder.project,
                self.m.buildbucket.build.builder.builder,
                f"bbid={self.m.buildbucket.build.id}",
            ]
            tag = ":".join(parts)
        input_json["input"]["tag"] = tag

        # TODO(akbiggs): Consider merging the logic that
        # sets "message" and "comments" into just setting
        # "comments", since a patchset-level resolved
        # comment is functionally similar to a message.
        if message:
            input_json["input"]["message"] = message
        if patchset_level_comment:
            input_json["input"]["comments"] = {
                "/PATCHSET_LEVEL": [patchset_level_comment]
            }
        if robot_comments:
            input_json["input"]["robot_comments"] = robot_comments

        if ignore_automatic_attention_set_rules:
            input_json["input"][
                "ignore_automatic_attention_set_rules"
            ] = ignore_automatic_attention_set_rules

        return self._run(
            name=name, subcmd="set-review", input_json=input_json, **kwargs
        )

    def submit(self, name, change_id, **kwargs):
        """Submit a change a repository.

        Returns a JSON dict of details regarding the change.

        Args:
          name (str): The name of the step.
          change_id (str): A change ID that uniquely defines a change on the
            host.
          **kwargs: Passed through to __call__().
        """
        return self._run(
            name=name,
            subcmd="submit",
            input_json={"change_id": str(change_id)},
            **kwargs,
        )

    def rebase(self, name, change_id, on_behalf_of_uploader=False, **kwargs):
        """Rebase a change.

        Returns a JSON dict of details regarding the change.

        Args:
            name (str): The name of the step.
            change_id (str): A change ID that uniquely defines a change on the
                host.
            on_behalf_of_uploader (bool): Rebase on behalf of the user that
                uploaded the change instead of as the caller.
            **kwargs: Passed through to __call__().
        """
        input_json = {"change_id": str(change_id), "input": {}}
        if on_behalf_of_uploader:
            input_json["input"]["on_behalf_of_uploader"] = True

        return self._run(
            name=name,
            subcmd="rebase",
            input_json=input_json,
            **kwargs,
        )

    def change_details(self, name, change_id, query_params=None, **kwargs):
        """Returns a JSON dict of details regarding a specific change.

        Args:
          name (str): The name of the step.
          change_id (str): A change ID that uniquely defines a change on the
            host.
          query_params (list): A list of Gerrit REST query parameters (strings).
            Documented at
            https://gerrit-review.googlesource.com/Documentation/rest-api-changes.html#query-options
          **kwargs: Passed through to __call__().
        """
        query_params = query_params or []
        input_json = {"change_id": str(change_id)}
        if query_params:
            input_json["params"] = {"o": query_params}
        return self._run(
            name=name, subcmd="change-detail", input_json=input_json, **kwargs
        )

    def restore_change(self, name, change_id, message=None, **kwargs):
        """Restores a change.

        Returns the details of the change, after attempting to restore.

        Args:
          name (str): The name of the step.
          change_id (str): A change ID that uniquely defines a change on the
            host.
          message (str): An optional message explaining the reason for restoring
            the change.
          **kwargs: Passed through to __call__().
        """
        input_json = {"change_id": str(change_id)}
        if message:
            input_json["input"] = {"message": message}
        return self._run(name=name, subcmd="restore", input_json=input_json, **kwargs)

    def change_query(self, name, query_string, query_params=None, **kwargs):
        """Searches for changes.

        Returns a list of matched changes (or None if no changes were matched).
        See the Gerrit documentation for supported query params:

        Args:
          name (str): The name of the step.
          query_string (str): The parameters to use for the search, e.g.
            'status:open+-status:wip. See the Gerrit docs for supported parameters:
            https://gerrit-review.googlesource.com/Documentation/user-search.html#_search_operators
          query_params (list): A list of Gerrit REST query parameters (strings).
            Documented at
            https://gerrit-review.googlesource.com/Documentation/rest-api-changes.html#query-options
          **kwargs: Passed through to __call__().

        Returns:
          A `StepData` with the search results in its `json.output`.
        """
        input_json = {"params": {"q": query_string}}
        if query_params:
            input_json["params"]["o"] = query_params
        return self._run(
            name=name, subcmd="change-query", input_json=input_json, **kwargs
        )

    def get_mergeable(self, name, change_id, revision_id="current", **kwargs):
        """Query whether a change is submittable.

        Returns a JSON dict of details regarding the change.

        Args:
          name (str): The name of the step.
          change_id (str): A change ID that uniquely defines a change on the
            host.
          revision_id (str): An ID that uniquely identifies the patchset in
            the change.
          **kwargs: Passed through to __call__().
        """
        return self._run(
            name=name,
            subcmd="get-mergeable",
            input_json={"change_id": str(change_id), "revision_id": revision_id},
            **kwargs,
        )

    def changes_submitted_together(self, name, change_id, query_params=None, **kwargs):
        """Returns a JSON dict of changes that would be submitted with the
        current change.

        Args:
          name (str): The name of the step.
          change_id (str): A change ID that uniquely defines a change on the
            host.
          query_params (list): A list of Gerrit REST query parameters (strings).
            Documented at
            https://gerrit-review.googlesource.com/Documentation/rest-api-changes.html#query-options
          **kwargs: Passed through to __call__().
        """
        input_json = {"change_id": str(change_id)}
        if query_params:
            input_json["params"] = {"o": query_params}
        return self._run(
            name=name,
            subcmd="changes-submitted-together",
            input_json=input_json,
            **kwargs,
        )

    def list_robot_comments(self, name, change_id, revision_id="", **kwargs):
        """List comments on a change.

        Returns a list of comments on the given change.
        See the Gerrit documentation for supported query params:

        Args:
          name (str): The name of the step.
          change_id (str): A change ID that uniquely defines a change on the
            host.
          revision_id (str): An ID that uniquely identifies the patchset in the
          change. If omitted, comments from all patchsets will be returned.
          **kwargs: Passed through to __call__().

        Returns:
          A `StepData` with the results in its `json.output`.
        """
        input_json = {
            "change_id": str(change_id),
            "revision_id": str(revision_id),
        }
        return self._run(
            name=name, subcmd="list-robot-comments", input_json=input_json, **kwargs
        )

    def list_change_comments(self, name, change_id, **kwargs):
        """List comments on a change.

        Returns a list of comments on the given change.
        See the Gerrit documentation for supported query params:

        Args:
          name (str): The name of the step.
          change_id (str): A change ID that uniquely defines a change on the
            host.
          **kwargs: Passed through to __call__().

        Returns:
          A `StepData` with the results in its `json.output`.
        """
        input_json = {"change_id": str(change_id)}
        return self._run(
            name=name, subcmd="list-change-comments", input_json=input_json, **kwargs
        )

    def create_branch(self, name, project, ref, revision, **kwargs):
        """Create a branch.

        Args:
          name (str): The name of the step.
          project (str): The name of the project.
          ref (str): The name of the branch to create.
          revision (str): The revision that the branch should point to.
        """
        input_json = {
            "project_id": project,
            "input": {"ref": ref, "revision": revision},
        }
        return self._run(
            name=name, subcmd="create-branch", input_json=input_json, **kwargs
        )

    def account_query(self, name, query_string, query_params=None, **kwargs):
        """Search for accounts.

        Returns a list of matching accounts.
        See the Gerrit documentation for supported query params:

        Args:
          name (str): The name of the step.
          query_string (str): The parameters to use for the search, e.g.
            'email:nobody@example.com'. See the Gerrit docs for supported parameters:
            https://gerrit-review.googlesource.com/Documentation/user-search-accounts.html
          query_params (list): A list of Gerrit REST query parameters (strings).
            Documented at
            https://gerrit-review.googlesource.com/Documentation/rest-api-accounts.html
          **kwargs: Passed through to __call__().

        Returns:
          A `StepData` with the results in its `json.output`.
        """
        input_json = {"params": {"q": query_string}}
        if query_params:
            input_json["params"]["o"] = query_params
        return self._run(
            name=name, subcmd="account-query", input_json=input_json, **kwargs
        )

    def _run(
        self,
        name,
        subcmd,
        input_json,
        host=None,
        test_data=None,
        max_attempts=1,
        infra_step=True,
        **kwargs,
    ):
        """Invoke the gerrit command-line interface.

        Args:
          name (str): The name of the step.
          subcmd (str): The subcommand to invoke.
          input_json (json): Input to the subcommand.
          host (str): The Gerrit host to make the query against.
            If not set, inferred from Buildbucket input.
          test_data (recipe_test_api.StepTestData): Test JSON output data for
            this step.
          max_attempts (int): The maximum number of times to attempt this
            command.
          infra_step (bool): Whether this should be considered an infra step.
          **kwargs (dict): Passed through to api.step().
        """
        if not host:
            assert self.host
            host = self.host
        host = self.normalize_host(host)

        cmd = [
            self._gerrit_path,
            subcmd,
            "-host",
            "https://" + host,
            "-input",
            self.m.json.input(input_json),
            "-output",
            self.m.json.output(),
        ]

        # If there's test data, create a factory for it.
        step_test_data = None
        if test_data is not None:
            step_test_data = lambda: test_data

        kwargs.setdefault("timeout", 10 * 60)

        sleep = 5
        for i in range(max_attempts):
            try:
                # Run the gerrit client command.
                return self.m.step(
                    name,
                    cmd,
                    step_test_data=step_test_data,
                    infra_step=infra_step,
                    **kwargs,
                )
            except self.m.step.StepFailure:
                # Raise exception if not a transient failure or on the last attempt
                step = self.m.step.active_result
                # Most gerrit command failures result in an exit code of 1, but
                # timeouts have an exit code of -15 because the recipe engine
                # kills them with SIGTERM which is 15.
                if step.retcode not in (1, -15) or i == max_attempts - 1:
                    raise
                self.m.time.sleep(sleep)
                sleep *= 2
            finally:
                try:
                    step = self.m.step.active_result
                except ValueError:  # pragma: no cover
                    # Trying to access `api.step.active_result` may raise a
                    # ValueError if the expected step did not get run, e.g.
                    # because another exception occurred first.
                    pass
                else:
                    step.presentation.logs["json.input"] = self.m.json.dumps(
                        input_json, indent=2
                    ).splitlines()

                    # If this call operates on a single change id, link to it.
                    if "change_id" in input_json:
                        step.presentation.links["gerrit link"] = (
                            f"https://{host}/q/{input_json['change_id']}"
                        )