recipe_modules/gerrit_auto_submit/test_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.

from recipe_engine import recipe_test_api
from PB.recipe_modules.fuchsia.gerrit_auto_submit import options as gas_options


def _step_name_prefix(host, iteration):
    parts = []
    if iteration is not None:
        parts.append(str(iteration))
    if host:
        parts.append(host)
    return "".join(part + "." for part in parts)


class GerritAutoSubmitTestApi(recipe_test_api.RecipeTestApi):
    def host_config(self, tree_status_host="fuchsia-stem-status.appspot.com", **kwargs):
        return gas_options.HostConfig(tree_status_host=tree_status_host, **kwargs)

    def options(self, *host_configs, max_attempts=4, **kwargs):
        if not host_configs:
            host_configs = [self.host_config()]

        opts = gas_options.Options(
            host_configs=host_configs,
            max_attempts=max_attempts,
            **kwargs,
        )

        return self.m.properties(**{"gerrit_auto_submit_options": opts})

    def cq_success(self, gerrit_host=None, iteration=None):
        return self.step_data(
            "{}cq.3965".format(_step_name_prefix(gerrit_host, iteration)),
            self.m.json.output({}),
            retcode=0,
        )

    def changes_query_test_data(self, gerrit_host=None, iteration=None):
        return self.step_data(
            "{}get eligible.get changes".format(
                _step_name_prefix(gerrit_host, iteration)
            ),
            self.m.json.output(
                [
                    {
                        "id": "myProject~main~I8473b95934b5732ac55d26311a706c9c2bde9939",
                        "project": "myProject",
                        "branch": "main",
                        "change_id": "I8473b95934b5732ac55d26311a706c9c2bde9939",
                        "subject": 'Revert "Implementing Feature X"',
                        "status": "NEW",
                        "created": "2013-02-01 09:59:32.126000000",
                        "updated": "2013-02-21 11:16:36.775000000",
                        "_number": 3965,
                        "owner": {"name": "John Doe"},
                        "current_revision": "27cc4558b5a3d3387dd11ee2df7a117e7e581822",
                    }
                ]
            ),
        )

    def change_detail_messages(self, num, gerrit_host=None, iteration=None):
        # Include a message posted without an author email address because
        # some prod accounts don't use an email address.
        messages = [
            {"author": {"email": "nobody@example.com"}},
            {"author": {}},
        ]

        for _ in range(num):
            messages.append({"author": {"email": "auto-submit@example.com"}})
            messages.append({"tag": "autogenerated:cq:full-run"})

        return self.step_data(
            "{}get eligible.get messages for 3965.get details".format(
                _step_name_prefix(gerrit_host, iteration)
            ),
            self.m.json.output(
                {
                    "id": "myProject~main~I8473b95934b5732ac55d26311a706c9c2bde9939",
                    "status": "NEW",
                    "messages": messages,
                }
            ),
        )

    def changes_submitted_together_test_data(
        self, other_change_count=0, gerrit_host=None, iteration=None
    ):
        return self.step_data(
            ("{}get eligible.get details for 3965." "find dependent changes").format(
                _step_name_prefix(gerrit_host, iteration)
            ),
            self.m.json.output(
                {
                    "changes": [
                        {
                            "id": "myProject~main~I8473b95934b5732ac55d26311a706c9c2bde994%d"
                            % n,
                        }
                        for n in range(other_change_count)
                    ],
                    "non_visible_changes": other_change_count,
                }
            ),
        )

    def mergeable_test_data(
        self, mergeable=True, retcode=0, gerrit_host=None, iteration=None
    ):
        return self.step_data(
            "{}get eligible.get details for 3965.get mergeable".format(
                _step_name_prefix(gerrit_host, iteration)
            ),
            self.m.json.output({"mergeable": mergeable}),
            retcode=retcode,
        )

    def rebase_test_data(self, success=True, gerrit_host=None, iteration=None):
        return self.step_data(
            "{}get eligible.get details for 3965.rebase".format(
                _step_name_prefix(gerrit_host, iteration)
            ),
            retcode=0 if success else 1,
        )