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.

import datetime
from typing import Optional

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):
    DEFAULT_UPDATED_DATETIME = datetime.datetime(2024, 1, 1, 13, 14, 15)
    FORMAT = "%Y-%m-%d %H:%M:%S.%f000"

    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(
            f"{_step_name_prefix(gerrit_host, iteration)}cq.3965.set CQ+2",
            self.m.json.output({}),
            retcode=0,
        )

    def changes_query_test_data(self, gerrit_host=None, iteration=None):
        return self.step_data(
            f"{_step_name_prefix(gerrit_host, iteration)}get eligible.get changes",
            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: int,
        gerrit_host: Optional[str] = None,
        iteration: Optional[int] = None,
        time_gap: datetime.timedelta = datetime.timedelta(hours=1),
    ):
        updated_date = self.DEFAULT_UPDATED_DATETIME
        message_date = updated_date - time_gap
        updated_formatted = updated_date.strftime(self.FORMAT)
        message_formatted = message_date.strftime(self.FORMAT)

        # 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"}, "date": message_formatted},
            {"author": {}, "date": message_formatted},
        ]

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

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

    def changes_submitted_together_test_data(
        self, other_change_count=0, gerrit_host=None, iteration=None
    ):
        return self.step_data(
            f"{_step_name_prefix(gerrit_host, iteration)}get eligible.get details for 3965.find dependent changes",
            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(
            f"{_step_name_prefix(gerrit_host, iteration)}cq.3965.get mergeable",
            self.m.json.output({"mergeable": mergeable}),
            retcode=retcode,
        )

    def rebase_test_data(self, success=True, gerrit_host=None, iteration=None):
        return self.step_data(
            f"{_step_name_prefix(gerrit_host, iteration)}cq.3965.rebase",
            retcode=0 if success else 1,
        )