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_DRY_RUN_MESSAGE_TAG,
    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,
    ):
        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_DRY_RUN_MESSAGE_TAG,
                },
            ],
        }
        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_DRY_RUN_MESSAGE_TAG,
                },
                {"message": "Dry run: builds failed", "tag": CQ_DRY_RUN_MESSAGE_TAG,},
                {
                    # 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_DRY_RUN_MESSAGE_TAG,
                },
                {
                    "message": "Dry run: %s" % PASSED_DRY_RUN_MESSAGE,
                    "tag": CQ_DRY_RUN_MESSAGE_TAG,
                },
            ],
        }
        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)