trigger_receiver_unittest.py

# Copyright 2017 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""Module for trigger_receiver unittests."""

import datetime
import os
import unittest

import cloud_sql_client
import config_reader
import datastore_client
import file_getter
import mock
import time_converter
import trigger_receiver

from google.appengine.ext import ndb
from google.appengine.ext import testbed

# Ensure that SUITE_SCHEDULER_CONFIG_FILE is read only once.
_SUITE_CONFIG_READER = config_reader.ConfigReader(
    file_getter.TEST_SUITE_SCHEDULER_CONFIG_FILE)


def now_generator(start_time, interval_min=30, last_days=7):
  """A datetime.datetime.now generator.

  The generator will generate 'now' from start_time till start_time+last_days
  for every interval_min.

  Args:
    start_time: A datetime.datetime object representing the initial value of
      the 'now'.
    interval_min: The interval minutes between current 'now' and next 'now.
    last_days: Representing how many days this generator will last.

  Yields:
    a datetime.datetime object to mock the current time.
  """
  cur_time = start_time
  end_time = start_time + datetime.timedelta(days=last_days)
  while cur_time < end_time:
    yield cur_time
    cur_time += datetime.timedelta(minutes=interval_min)


def _get_ground_truth_task_list_from_config():
  """Get the ground truth of to-be-scheduled task list from config file."""
  task_config = config_reader.TaskConfig(_SUITE_CONFIG_READER)
  tasks = {}
  for keyword, klass in config_reader.EVENT_CLASSES.iteritems():
    tasks[keyword] = task_config.get_tasks_by_keyword(klass.KEYWORD)['tasks']

  return tasks


def _should_schedule_nightly_task(last_now, now):
  """Check whether nightly task should be scheduled.

  A nightly task should be schduled when next hour is coming.

  Args:
    last_now: the last time to check if nightly task should be scheduled
      or not.
    now: the current time to check if nightly task should be scheduled.

  Returns:
    a boolean indicating whether there will be nightly tasks scheduled.
  """
  if last_now is not None and last_now.hour != now.hour:
    return True

  return False


def _should_schedule_weekly_task(last_now, now, weekly_time_info):
  """Check whether weekly task should be scheduled.

  A weekly task should be schduled when it comes to the default weekly tasks
  scheduling hour in next day.

  Args:
    last_now: the last time to check if weekly task should be scheduled
      or not.
    now: the current time to check if weekly task should be scheduled.
    weekly_time_info: the default weekly tasks scheduling time info.

  Returns:
    a boolean indicating whether there will be weekly tasks scheduled.
  """
  if (last_now is not None and last_now.hour != now.hour and
      now.hour == weekly_time_info.hour):
    return True

  return False


def _should_schedule_new_build_task(last_now, now):
  """Check whether weekly task should be scheduled.

  A new_build task should be schduled when there're new builds between last
  check and this check.

  Args:
    last_now: the last time to check if new_build task should be scheduled.
    now: the current time to check if new_build task should be scheduled.

  Returns:
    a boolean indicating whether there will be new_build tasks scheduled.
  """
  if last_now is not None and last_now != now:
    return True

  return False


class FakeCIDBClient(object):
  """Mock cloud_sql_client.CIDBClient."""

  def get_passed_builds_since_date(self, since_date):
    """Mock cloud_sql_client.CIDBClient.get_passed_builds_since_date."""
    del since_date  # unused
    return [cloud_sql_client.BuildInfo('link', '62', '9868.0.0',
                                       'link-release')]

  def get_latest_passed_builds(self, build_config):
    """Mock cloud_sql_client.CIDBClient.get_latest_passed_builds."""
    del build_config  # unused
    return cloud_sql_client.BuildInfo('link', '62', '9868.0.0', build_config)


class FakeAndroidBuildRestClient(object):
  """Mock rest_client.AndroidBuildRestClient."""

  def get_latest_build_id(self, branch, target):
    """Mock rest_client.AndroidBuildRestClient.get_latest_build_id."""
    del branch, target  # unused
    return '100'


class FakeLabConfig(object):
  """Mock rest_client.AndroidBuildRestClient."""

  def get_android_board_list(self):
    """Mock config_reader.LabConfig.get_android_board_list."""
    return ('android-angler', 'android-bullhead')

  def get_firmware_ro_build_list(self, release_board):
    """Mock config_reader.LabConfig.get_firmware_ro_build_list."""
    del release_board  # unused
    return 'firmware1,firmware2'


class TriggerReceiverTestCase(unittest.TestCase):

  def setUp(self):
    self.testbed = testbed.Testbed()
    self.testbed.activate()
    self.addCleanup(self.testbed.deactivate)

    self.testbed.init_datastore_v3_stub()
    self.testbed.init_memcache_stub()
    ndb.get_context().clear_cache()
    self.testbed.init_taskqueue_stub(
        root_path=os.path.join(os.path.dirname(__file__)))
    self.taskqueue_stub = self.testbed.get_stub(
        testbed.TASKQUEUE_SERVICE_NAME)

    mock_cidb_client = mock.patch('cloud_sql_client.CIDBClient')
    self._mock_cidb_client = mock_cidb_client.start()
    self.addCleanup(mock_cidb_client.stop)

    mock_android_client = mock.patch('rest_client.AndroidBuildRestClient')
    self._mock_android_client = mock_android_client.start()
    self.addCleanup(mock_android_client.stop)

    mock_config_reader = mock.patch('config_reader.ConfigReader')
    self._mock_config_reader = mock_config_reader.start()
    self.addCleanup(mock_config_reader.stop)

    mock_lab_config = mock.patch('config_reader.LabConfig')
    self._mock_lab_config = mock_lab_config.start()
    self.addCleanup(mock_lab_config.stop)

    mock_utc_now = mock.patch('time_converter.utc_now')
    self._mock_utc_now = mock_utc_now.start()
    self.addCleanup(mock_utc_now.stop)

    self._mock_cidb_client.return_value = FakeCIDBClient()
    self._mock_android_client.return_value = FakeAndroidBuildRestClient()
    self._mock_config_reader.return_value = _SUITE_CONFIG_READER
    self._mock_lab_config.return_value = FakeLabConfig()

  def testCronWithoutLastExec(self):
    """Test the first round of cron can be successfully executed."""
    self._mock_utc_now.return_value = datetime.datetime.now(
        time_converter.UTC_TZ)
    suite_trigger = trigger_receiver.TriggerReceiver()
    suite_trigger.cron()
    self.assertFalse(suite_trigger.events['nightly'].should_handle)
    self.assertFalse(suite_trigger.events['weekly'].should_handle)
    self.assertFalse(suite_trigger.events['new_build'].should_handle)

    self.assertEqual(suite_trigger.event_results, {})

  def testCronTriggerNightly(self):
    """Test nightly event is read with available nightly last_exec_time."""
    utc_now = datetime.datetime.now(time_converter.UTC_TZ)
    last_exec_client = datastore_client.LastExecutionRecordStore()
    last_exec_client.set_last_execute_time(
        'nightly', utc_now - datetime.timedelta(hours=1))
    self._mock_utc_now.return_value = utc_now
    suite_trigger = trigger_receiver.TriggerReceiver()
    self.assertTrue(suite_trigger.events['nightly'].should_handle)
    self.assertFalse(suite_trigger.events['weekly'].should_handle)
    self.assertFalse(suite_trigger.events['new_build'].should_handle)

  def testCronTriggerNightlyOutdated(self):
    """Test nightly event is read with available nightly last_exec_time."""
    utc_now = datetime.datetime.now(time_converter.UTC_TZ)
    last_exec_client = datastore_client.LastExecutionRecordStore()
    last_exec_client.set_last_execute_time(
        'nightly', utc_now - datetime.timedelta(days=3))
    self._mock_utc_now.return_value = utc_now
    suite_trigger = trigger_receiver.TriggerReceiver()
    self.assertFalse(suite_trigger.events['nightly'].should_handle)

  def testCronTriggerWeeklyOutdated(self):
    """Test weekly event is read with available weekly last_exec_time."""
    utc_now = datetime.datetime.now(time_converter.UTC_TZ)
    last_exec_client = datastore_client.LastExecutionRecordStore()
    last_exec_client.set_last_execute_time(
        'weekly', utc_now - datetime.timedelta(days=8))
    self._mock_utc_now.return_value = utc_now
    suite_trigger = trigger_receiver.TriggerReceiver()
    self.assertFalse(suite_trigger.events['weekly'].should_handle)

  def testCronForWeeks(self):
    """Ensure cron job can be successfully scheduled for several weeks."""
    all_tasks = _get_ground_truth_task_list_from_config()
    nightly_time_info = time_converter.convert_time_info_to_utc(
        time_converter.TimeInfo(
            config_reader.EVENT_CLASSES['nightly'].DEFAULT_PST_DAY,
            config_reader.EVENT_CLASSES['nightly'].DEFAULT_PST_HOUR))
    weekly_time_info = time_converter.convert_time_info_to_utc(
        time_converter.TimeInfo(
            config_reader.EVENT_CLASSES['weekly'].DEFAULT_PST_DAY,
            config_reader.EVENT_CLASSES['weekly'].DEFAULT_PST_HOUR))
    last_now = None

    for now in now_generator(datetime.datetime.now(time_converter.UTC_TZ)):
      self._mock_utc_now.return_value = now
      suite_trigger = trigger_receiver.TriggerReceiver()
      suite_trigger.cron()

      # Verify nightly tasks
      should_scheduled_nightly_tasks = [
          t.name for t in all_tasks['nightly']
          if (t.hour is not None and t.hour == now.hour) or
          (t.hour is None and now.hour == nightly_time_info.hour)]
      if (_should_schedule_nightly_task(last_now, now) and
          should_scheduled_nightly_tasks):
        self.assertEqual(suite_trigger.event_results['nightly'],
                         should_scheduled_nightly_tasks)
      else:
        self.assertNotIn('nightly', suite_trigger.event_results.keys())

      # Verify weekly tasks
      should_scheduled_weekly_tasks = [
          t.name for t in all_tasks['weekly']
          if (t.day is not None and now.weekday() == t.day) or
          (t.day is None and now.weekday() == weekly_time_info.weekday)]
      if (_should_schedule_weekly_task(last_now, now, weekly_time_info) and
          should_scheduled_weekly_tasks):
        self.assertEqual(suite_trigger.event_results['weekly'],
                         should_scheduled_weekly_tasks)
      else:
        self.assertNotIn('weekly', suite_trigger.event_results.keys())

      # Verify new_build tasks
      should_scheduled_new_build_tasks = [
          t.name for t in all_tasks['new_build']]
      if (_should_schedule_new_build_task(last_now, now) and
          should_scheduled_new_build_tasks):
        self.assertEqual(suite_trigger.event_results['new_build'],
                         should_scheduled_new_build_tasks)
      else:
        self.assertNotIn('new_build', suite_trigger.event_results.keys())

      last_now = now


if __name__ == '__main__':
  unittest.main()