recipe_modules/auto_roller/test_api.py

# Copyright 2018 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 google.protobuf import json_format as jsonpb

from recipe_engine import recipe_test_api

from PB.recipe_modules.fuchsia.auto_roller.options import Options

from RECIPE_MODULES.fuchsia.auto_roller.api import (
    CQ_SERVICE_ACCOUNT_DOMAIN,
    FAILED_DRY_RUN_MESSAGES,
    PASSED_DRY_RUN_MESSAGES,
)


class AutoRollerTestApi(recipe_test_api.RecipeTestApi):
    def Options(self, **kwargs):
        return jsonpb.ParseDict(kwargs, Options())

    def _check_done_step_data(
        self,
        output,
        name="check for completion.check if done ({})",
        iteration=0,
    ):
        output.setdefault("current_revision", "abc123")
        name = name.format(iteration)
        return self.step_data(name, self.m.json.output(output))

    def success(self, *args, **kwargs):
        """Returns mock data indicating a successful roll."""
        output = {
            "status": "MERGED",
            "labels": kwargs.pop("labels", {"Commit-Queue": {"approved": {}}}),
        }
        return self._check_done_step_data(output, *args, **kwargs)

    def failure(self, *args, **kwargs):
        """Returns mock data indicating a CQ failure."""
        output = {
            "status": "NEW",
            "labels": kwargs.pop("labels", {"Commit-Queue": {}}),
        }
        return self._check_done_step_data(output, *args, **kwargs)

    def dry_run_incomplete(self, *args, **kwargs):
        """Returns mock data indicating a dry run is in progress."""
        labels = kwargs.pop(
            "labels",
            {
                "Commit-Queue": {
                    "recommended": {"email": "roller@fuchsia-service-account.com"}
                },
            },
        )
        output = {
            "status": "NEW",
            "labels": labels,
            "messages": [
                {
                    "message": "Dry run: CQ is trying the patch",
                    "real_author": {
                        "email": f"foo@{CQ_SERVICE_ACCOUNT_DOMAIN}",
                    },
                },
            ],
        }
        return self._check_done_step_data(output, *args, **kwargs)

    def dry_run_failure(self, *args, **kwargs):
        """Returns mock data indicating a dry run is complete."""
        output = {
            "status": "NEW",
            "labels": kwargs.pop("labels", {"Commit-Queue": {}}),
            "messages": [
                {
                    "message": "Dry run: CQ is trying the patch",
                    "real_author": {
                        "email": f"foo@{CQ_SERVICE_ACCOUNT_DOMAIN}",
                    },
                },
                {
                    "message": FAILED_DRY_RUN_MESSAGES[0],
                    "real_author": {
                        "email": f"foo@{CQ_SERVICE_ACCOUNT_DOMAIN}",
                    },
                },
                {
                    # This is to test that the roller ignores comments that aren't
                    # from CQ.
                    "message": "blah",
                    "real_author": {
                        "email": "foo@google.com",
                    },
                },
            ],
        }
        return self._check_done_step_data(output, *args, **kwargs)

    def dry_run_success(self, *args, **kwargs):
        """Returns mock data indicating a dry run is complete."""
        output = {
            "status": "NEW",
            "labels": kwargs.pop("labels", {"Commit-Queue": {}}),
            "messages": [
                {
                    "message": "Dry run: CQ is trying the patch",
                    "real_author": {
                        "email": f"foo@{CQ_SERVICE_ACCOUNT_DOMAIN}",
                    },
                },
                {
                    "message": PASSED_DRY_RUN_MESSAGES[0],
                    "real_author": {
                        "email": f"foo@{CQ_SERVICE_ACCOUNT_DOMAIN}",
                    },
                },
            ],
        }
        return self._check_done_step_data(output, *args, **kwargs)

    def timeout(self, *args, **kwargs):
        """Returns mock data indicating a roller timeout."""
        output = {
            "status": "NEW",
            "labels": kwargs.pop("labels", {"Commit-Queue": {"approved": {}}}),
        }
        return self._check_done_step_data(output, *args, **kwargs)

    def abandoned(self, *args, **kwargs):
        """Returns mock data indicating a successful roll."""
        output = {
            "status": "ABANDONED",
            "labels": kwargs.pop("labels", {"Commit-Queue": {"approved": {}}}),
        }
        return self._check_done_step_data(output, *args, **kwargs)