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


class AutoRollerTestApi(recipe_test_api.RecipeTestApi):

  def success(self):
    """Returns mock data indicating a successful roll."""
    return self.m.json.output({
        'status': 'MERGED',
        'labels': {
            'Commit-Queue': {
                'approved': {}
            }
        }
    })

  def failure(self):
    """Returns mock data indicating a CQ failure."""
    return self.m.json.output({'status': 'NEW', 'labels': {'Commit-Queue': {}}})

  def dry_run(self):
    """Returns mock data indicating a dry run is complete."""
    # Note, this is the same output we expect on failure (CQ just
    # un-sets the CQ label).
    return self.failure()

  def timeout(self):
    """Returns mock data indicating a roller timeout."""
    return self.m.json.output({
        'status': 'NEW',
        'labels': {
            'Commit-Queue': {
                'approved': {}
            }
        }
    })

  def abandoned(self):
    """Returns mock data indicating a successful roll."""
    return self.m.json.output({
        'status': 'ABANDONED',
        'labels': {
            'Commit-Queue': {
                'approved': {}
            }
        }
    })

  def check_done_step_data(self, name, result, iteration=0):
    if not name:
      name = 'check for completion.check if done ({})'
    name = name.format(iteration)

    results_to_outputs = {
        'success': self.success(),
        'failure': self.failure(),
        'dry_run': self.dry_run(),
        'timeout': self.timeout(),
        'abandoned': self.abandoned()
    }
    return self.step_data(name, results_to_outputs[result])

  def success_step_data(self, name=None, iteration=0):
    return self.check_done_step_data(name, 'success', iteration=iteration)

  def failure_step_data(self, name=None, iteration=0):
    return self.check_done_step_data(name, 'failure', iteration=iteration)

  def dry_run_step_data(self, name=None, iteration=0):
    return self.check_done_step_data(name, 'dry_run', iteration=iteration)

  def timeout_step_data(self, name=None, iteration=0):
    return self.check_done_step_data(name, 'timeout', iteration=iteration)

  def abandoned_step_data(self, name=None, iteration=0):
    return self.check_done_step_data(name, 'abandoned', iteration=iteration)