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 recipe_engine import recipe_test_api

from RECIPE_MODULES.fuchsia.auto_roller.api import (
    CQ_MESSAGE_TAGS,
    FAILED_DRY_RUN_MESSAGE,
    PASSED_DRY_RUN_MESSAGE,
)


class AutoRollerTestApi(recipe_test_api.RecipeTestApi):
    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",
                    "tag": CQ_MESSAGE_TAGS[0],
                },
            ],
        }
        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",
                    "tag": CQ_MESSAGE_TAGS[0],
                },
                {"message": FAILED_DRY_RUN_MESSAGE, "tag": CQ_MESSAGE_TAGS[0]},
                {
                    # This is to test that the roller ignores comments that aren't
                    # from CQ.
                    "message": "blah",
                },
            ],
        }
        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",
                    "tag": CQ_MESSAGE_TAGS[0],
                },
                {"message": PASSED_DRY_RUN_MESSAGE, "tag": CQ_MESSAGE_TAGS[0]},
            ],
        }
        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)