timed_event_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 timed_event unittests."""
# pylint: disable=g-missing-super-call

import datetime
import unittest

import config_reader
import datastore_client
import mock
import task_config_reader
import time_converter
import timed_event

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


class FakeTask(object):

  def __init__(self, name, day=None, hour=None):
    self.name = name
    self.day = day
    self.hour = hour


class TimedEventTestCase(unittest.TestCase):

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

    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.config = config_reader.ConfigReader(None)


class NightlyEventTestCase(TimedEventTestCase):

  _KLASS = timed_event.Nightly
  _DEFAULT_PST_HOUR = _KLASS.DEFAULT_PST_HOUR
  _DEFAULT_UTC_HOUR = time_converter.convert_time_info(
      time_converter.TimeInfo(None, _DEFAULT_PST_HOUR)).hour

  def testCreateNightlyEventWithoutLastExec(self):
    """Create a nightly event without last_exec_utc in datastore."""
    utc_now = datetime.datetime(2017, 8, 1, 4, 15,
                                tzinfo=time_converter.UTC_TZ)
    expected_last_exec_utc = datetime.datetime(2017, 8, 1, 4,
                                               tzinfo=time_converter.UTC_TZ)
    task_config = task_config_reader.TaskConfig(self.config)
    self._mock_utc_now.return_value = utc_now
    last_exec_client = datastore_client.LastExecutionRecordStore()
    event = self._KLASS(
        task_config.get_event_setting(self._KLASS.section_name()),
        last_exec_client.get_last_execute_time(self._KLASS.KEYWORD))

    self.assertEqual(event.target_exec_utc, expected_last_exec_utc)
    self.assertEqual(event.last_exec_utc, expected_last_exec_utc)
    self.assertFalse(event.should_handle)

  def testCreateNightlyEventWithValidLastExecShouldNotHandle(self):
    """Create a nightly event with last_exec_utc but shouldn't be handled."""
    utc_now = datetime.datetime(2017, 8, 1, 4, 15,
                                tzinfo=time_converter.UTC_TZ)
    expected_last_exec_utc = datetime.datetime(2017, 8, 1, 4,
                                               tzinfo=time_converter.UTC_TZ)
    task_config = task_config_reader.TaskConfig(self.config)
    self._mock_utc_now.return_value = utc_now
    last_exec_client = datastore_client.LastExecutionRecordStore()
    last_exec_client.set_last_execute_time(
        self._KLASS.KEYWORD, expected_last_exec_utc)
    event = self._KLASS(
        task_config.get_event_setting(self._KLASS.section_name()),
        last_exec_client.get_last_execute_time(self._KLASS.KEYWORD))

    self.assertEqual(event.target_exec_utc, expected_last_exec_utc)
    self.assertEqual(event.last_exec_utc, expected_last_exec_utc)
    self.assertFalse(event.should_handle)

  def testCreateCantHandleNightlyEventWithValidLastExecWithGAETesting(self):
    """Test a can't-be-handled event can be handled in testing on GAE."""
    utc_now = datetime.datetime(2017, 8, 1, 4, 15,
                                tzinfo=time_converter.UTC_TZ)
    expected_last_exec_utc = datetime.datetime(2017, 8, 1, 4,
                                               tzinfo=time_converter.UTC_TZ)
    task_config = task_config_reader.TaskConfig(self.config)
    self._mock_utc_now.return_value = utc_now
    last_exec_client = datastore_client.LastExecutionRecordStore()
    last_exec_client.set_last_execute_time(
        self._KLASS.KEYWORD, expected_last_exec_utc)

    with mock.patch('global_config.GAE_TESTING', return_value=True):
      event = self._KLASS(
          task_config.get_event_setting(self._KLASS.section_name()),
          last_exec_client.get_last_execute_time(self._KLASS.KEYWORD))
      self.assertEqual(event.target_exec_utc, expected_last_exec_utc)
      self.assertEqual(event.last_exec_utc, expected_last_exec_utc)
      self.assertTrue(event.should_handle)

  def testCreateNightlyEventWithValidLastExecShouldHandle(self):
    """Create a nightly event with last_exec_utc and should be handled."""
    utc_now = datetime.datetime(2017, 8, 1, 5, 15,
                                tzinfo=time_converter.UTC_TZ)
    expected_last_exec_utc = datetime.datetime(2017, 8, 1, 4,
                                               tzinfo=time_converter.UTC_TZ)
    expected_cur_exec_utc = datetime.datetime(2017, 8, 1, 5,
                                              tzinfo=time_converter.UTC_TZ)
    task_config = task_config_reader.TaskConfig(self.config)
    self._mock_utc_now.return_value = utc_now
    last_exec_client = datastore_client.LastExecutionRecordStore()
    last_exec_client.set_last_execute_time(
        self._KLASS.KEYWORD, expected_last_exec_utc)
    event = self._KLASS(
        task_config.get_event_setting(self._KLASS.section_name()),
        last_exec_client.get_last_execute_time(self._KLASS.KEYWORD))

    self.assertEqual(event.target_exec_utc, expected_cur_exec_utc)
    self.assertEqual(event.last_exec_utc, expected_last_exec_utc)
    self.assertTrue(event.should_handle)

  def testCreateNightlyEventWithOutdatedLastExec(self):
    """Create a nightly event with outdated last_exec_utc."""
    utc_now = datetime.datetime(2017, 8, 1, 4, 15,
                                tzinfo=time_converter.UTC_TZ)
    expected_last_exec_utc = datetime.datetime(2017, 8, 1, 4,
                                               tzinfo=time_converter.UTC_TZ)
    task_config = task_config_reader.TaskConfig(self.config)
    self._mock_utc_now.return_value = utc_now
    last_exec_client = datastore_client.LastExecutionRecordStore()
    last_exec_client.set_last_execute_time(
        self._KLASS.KEYWORD,
        expected_last_exec_utc - datetime.timedelta(hours=3))
    event = self._KLASS(
        task_config.get_event_setting(self._KLASS.section_name()),
        last_exec_client.get_last_execute_time(self._KLASS.KEYWORD))

    self.assertEqual(event.target_exec_utc, expected_last_exec_utc)
    self.assertEqual(event.last_exec_utc, expected_last_exec_utc)
    self.assertFalse(event.should_handle)

  def testCreateNightlyEventWithOutdatedLastExecWithGAETesting(self):
    """Test an event with old last exec can be handled in testing on GAE."""
    utc_now = datetime.datetime(2017, 8, 1, 4, 15,
                                tzinfo=time_converter.UTC_TZ)
    expected_last_exec_utc = datetime.datetime(2017, 8, 1, 4,
                                               tzinfo=time_converter.UTC_TZ)
    task_config = task_config_reader.TaskConfig(self.config)
    self._mock_utc_now.return_value = utc_now
    last_exec_client = datastore_client.LastExecutionRecordStore()
    last_exec_client.set_last_execute_time(
        self._KLASS.KEYWORD,
        expected_last_exec_utc - datetime.timedelta(hours=3))

    with mock.patch('global_config.GAE_TESTING', return_value=True):
      event = self._KLASS(
          task_config.get_event_setting(self._KLASS.section_name()),
          last_exec_client.get_last_execute_time(self._KLASS.KEYWORD))
      self.assertEqual(event.target_exec_utc, expected_last_exec_utc)
      self.assertEqual(event.last_exec_utc, expected_last_exec_utc)
      self.assertTrue(event.should_handle)

  def testFilterTasksWithTaskHourForNightlyEvent(self):
    """Test filter_tasks with tasks which contain settings of hour."""
    default_pst_hour = time_converter.convert_time_info(
        time_converter.TimeInfo(None, 0),
        source_tz=time_converter.UTC_TZ,
        target_tz=time_converter.PST_TZ).hour
    utc_now = datetime.datetime(2017, 8, 1, 0, 15,
                                tzinfo=time_converter.UTC_TZ)
    self._mock_utc_now.return_value = utc_now
    event = self._KLASS(
        task_config_reader.EventSettings(True, None),
        datetime.datetime(2017, 7, 31, 23, tzinfo=time_converter.UTC_TZ))

    # It's in PST time.
    task_list = [FakeTask('test1', day=None, hour=default_pst_hour+1),
                 FakeTask('test2', day=None, hour=default_pst_hour),
                 FakeTask('test3', day=None, hour=0),
                 FakeTask('test4', day=None, hour=default_pst_hour)]
    event.set_task_list(task_list)
    self.assertEqual(len(event.task_list), len(task_list))
    event.filter_tasks()
    # Verify only tasks that should be run at 0:00 in UTC are kept,
    # since current time is 0:15.
    self.assertEqual(len(event.task_list), 2)
    self.assertEqual(sorted([t.name for t in event.task_list]),
                     ['test2', 'test4'])

  def testFilterTasksWithDefaultHourForNightlyEvent(self):
    """Test filter_tasks with tasks which don't contain settings of hour."""
    utc_now = datetime.datetime(2017, 8, 1, self._DEFAULT_UTC_HOUR, 15,
                                tzinfo=time_converter.UTC_TZ)
    self._mock_utc_now.return_value = utc_now
    event = self._KLASS(
        task_config_reader.EventSettings(True, None),
        datetime.datetime(2017, 8, 1, 3, tzinfo=time_converter.UTC_TZ))

    # It's in PST time.
    task_list = [FakeTask('test1', day=None, hour=self._DEFAULT_PST_HOUR - 1),
                 FakeTask('test2', day=None, hour=self._DEFAULT_PST_HOUR),
                 FakeTask('test3', day=None, hour=self._DEFAULT_PST_HOUR + 1),
                 FakeTask('test4', day=None, hour=None)]
    event.set_task_list(task_list)
    self.assertEqual(len(event.task_list), len(task_list))
    event.filter_tasks()
    # Verify only tasks that should be run at 21:00 (the default pst hour
    # for nightly events) are kept.
    self.assertEqual(len(event.task_list), 2)
    self.assertEqual(sorted([t.name for t in event.task_list]),
                     ['test2', 'test4'])

  def testFilterTasksWithTaskHourWithGAETesting(self):
    """Verify all tasks are kept when testing on GAE."""
    utc_now = datetime.datetime(2017, 8, 1, 0, 15,
                                tzinfo=time_converter.UTC_TZ)
    self._mock_utc_now.return_value = utc_now

    with mock.patch('global_config.GAE_TESTING', return_value=True):
      event = self._KLASS(
          task_config_reader.EventSettings(True, None),
          datetime.datetime(2017, 7, 31, 23, tzinfo=time_converter.UTC_TZ))

      task_list = [FakeTask('test1', day=None, hour=3),
                   FakeTask('test2', day=None, hour=0),
                   FakeTask('test3', day=None, hour=4),
                   FakeTask('test4', day=None, hour=0)]
      event.set_task_list(task_list)
      self.assertEqual(len(event.task_list), len(task_list))
      event.filter_tasks()
      self.assertEqual(len(event.task_list), 4)
      self.assertEqual(sorted([t.name for t in event.task_list]),
                       ['test1', 'test2', 'test3', 'test4'])

  def testSetTaskListForNightlyEvent(self):
    """Verify tasks' timeinfo are correctly converted to UTC."""
    self._mock_utc_now.return_value = datetime.datetime.now(
        time_converter.UTC_TZ)
    event = self._KLASS(
        task_config_reader.EventSettings(True, None), None)

    # It's in PST time.
    task_list = [FakeTask('test1', day=None, hour=self._DEFAULT_PST_HOUR),
                 FakeTask('test2', day=None, hour=0),
                 FakeTask('test3', day=None, hour=self._DEFAULT_PST_HOUR+1),
                 FakeTask('test4', day=None, hour=None)]
    event.set_task_list(task_list)
    tasks_with_default_utc_hour = [
        t.name for t in event.task_list if t.hour == self._DEFAULT_UTC_HOUR]
    self.assertEqual(sorted(tasks_with_default_utc_hour),
                     ['test1', 'test4'])


class WeeklyEventTestCase(TimedEventTestCase):

  _KLASS = timed_event.Weekly
  _DEFAULT_PST_DAY = _KLASS.DEFAULT_PST_DAY
  _DEFAULT_PST_HOUR = _KLASS.DEFAULT_PST_HOUR
  _DEFAULT_UTC_TIMEINFO = time_converter.convert_time_info(
      time_converter.TimeInfo(_DEFAULT_PST_DAY, _DEFAULT_PST_HOUR))
  _DEFAULT_UTC_DAY = _DEFAULT_UTC_TIMEINFO.weekday
  _DEFAULT_UTC_HOUR = _DEFAULT_UTC_TIMEINFO.hour

  def testCreateWeeklyEventWithoutLastExec(self):
    """Create a weekly event without last_exec_utc in datastore."""
    utc_now = datetime.datetime(2017, 8, 1, 4, 15,
                                tzinfo=time_converter.UTC_TZ)
    expected_last_exec_utc = datetime.datetime(
        2017, 8, 1, 4, tzinfo=time_converter.UTC_TZ)
    task_config = task_config_reader.TaskConfig(self.config)
    self._mock_utc_now.return_value = utc_now
    last_exec_client = datastore_client.LastExecutionRecordStore()
    event = self._KLASS(
        task_config.get_event_setting(self._KLASS.section_name()),
        last_exec_client.get_last_execute_time(self._KLASS.KEYWORD))

    self.assertEqual(event.target_exec_utc, expected_last_exec_utc)
    self.assertEqual(event.last_exec_utc, expected_last_exec_utc)
    self.assertFalse(event.should_handle)

  def testCreateWeeklyEventWithoutLastExecWithGAETesting(self):
    """Test an event without last exec can be handled in testing on GAE."""
    utc_now = datetime.datetime(2017, 8, 1, 4, 15,
                                tzinfo=time_converter.UTC_TZ)
    expected_last_exec_utc = datetime.datetime(
        2017, 8, 1, 4, tzinfo=time_converter.UTC_TZ)
    task_config = task_config_reader.TaskConfig(self.config)
    self._mock_utc_now.return_value = utc_now
    last_exec_client = datastore_client.LastExecutionRecordStore()

    with mock.patch('global_config.GAE_TESTING', return_value=True):
      event = self._KLASS(
          task_config.get_event_setting(self._KLASS.section_name()),
          last_exec_client.get_last_execute_time(self._KLASS.KEYWORD))

      self.assertEqual(event.target_exec_utc, expected_last_exec_utc)
      self.assertEqual(event.last_exec_utc, expected_last_exec_utc)
      self.assertTrue(event.should_handle)

  def testCreateWeeklyEventWithValidLastExecShouldNotHandle(self):
    """Create a weekly event with last_exec_utc but shouldn't be handled."""
    utc_now = datetime.datetime(2017, 8, 1, self._DEFAULT_UTC_HOUR-1, 15,
                                tzinfo=time_converter.UTC_TZ)
    expected_last_exec_utc = datetime.datetime(
        2017, 8, 1, self._DEFAULT_UTC_HOUR-1, tzinfo=time_converter.UTC_TZ)
    task_config = task_config_reader.TaskConfig(self.config)
    self._mock_utc_now.return_value = utc_now
    last_exec_client = datastore_client.LastExecutionRecordStore()
    last_exec_client.set_last_execute_time(
        self._KLASS.KEYWORD, expected_last_exec_utc)

    event = self._KLASS(
        task_config.get_event_setting(self._KLASS.section_name()),
        last_exec_client.get_last_execute_time(self._KLASS.KEYWORD))
    self.assertEqual(event.target_exec_utc, expected_last_exec_utc)
    self.assertEqual(event.last_exec_utc, expected_last_exec_utc)
    self.assertFalse(event.should_handle)

  def testCreateCantHandleWeeklyEventWithValidLastExecWithGAETesting(self):
    """Test a can't-be-handled event can be handled in testing on GAE."""
    utc_now = datetime.datetime(2017, 8, 1, 4, 15,
                                tzinfo=time_converter.UTC_TZ)
    expected_last_exec_utc = datetime.datetime(
        2017, 8, 1, 4, tzinfo=time_converter.UTC_TZ)
    task_config = task_config_reader.TaskConfig(self.config)
    self._mock_utc_now.return_value = utc_now
    last_exec_client = datastore_client.LastExecutionRecordStore()
    last_exec_client.set_last_execute_time(
        self._KLASS.KEYWORD, expected_last_exec_utc)

    with mock.patch('global_config.GAE_TESTING', return_value=True):
      event = self._KLASS(
          task_config.get_event_setting(self._KLASS.section_name()),
          last_exec_client.get_last_execute_time(self._KLASS.KEYWORD))
      self.assertEqual(event.target_exec_utc, expected_last_exec_utc)
      self.assertEqual(event.last_exec_utc, expected_last_exec_utc)
      self.assertTrue(event.should_handle)

  def testCreateWeeklyEventWithValidLastExecShouldHandle(self):
    """Create a nightly event with last_exec_utc and should be handled."""
    utc_now = datetime.datetime(2017, 8, 1, self._DEFAULT_UTC_HOUR, 15,
                                tzinfo=time_converter.UTC_TZ)
    expected_last_exec_utc = datetime.datetime(
        2017, 7, 31, self._DEFAULT_UTC_HOUR, tzinfo=time_converter.UTC_TZ)
    expected_cur_exec_utc = datetime.datetime(
        2017, 8, 1, self._DEFAULT_UTC_HOUR, tzinfo=time_converter.UTC_TZ)
    task_config = task_config_reader.TaskConfig(self.config)
    self._mock_utc_now.return_value = utc_now
    last_exec_client = datastore_client.LastExecutionRecordStore()
    last_exec_client.set_last_execute_time(
        self._KLASS.KEYWORD, expected_last_exec_utc)

    event = self._KLASS(
        task_config.get_event_setting(self._KLASS.section_name()),
        last_exec_client.get_last_execute_time(self._KLASS.KEYWORD))
    self.assertEqual(event.target_exec_utc, expected_cur_exec_utc)
    self.assertEqual(event.last_exec_utc, expected_last_exec_utc)
    self.assertTrue(event.should_handle)

  def testCreateWeeklyEventWithOutdatedLastExec(self):
    """Create a weekly event with outdated last_exec_utc."""
    utc_now = datetime.datetime(2017, 8, 1, 4, 15,
                                tzinfo=time_converter.UTC_TZ)
    expected_last_exec_utc = datetime.datetime(
        2017, 8, 1, 4, tzinfo=time_converter.UTC_TZ)
    task_config = task_config_reader.TaskConfig(self.config)
    self._mock_utc_now.return_value = utc_now
    last_exec_client = datastore_client.LastExecutionRecordStore()
    last_exec_client.set_last_execute_time(
        self._KLASS.KEYWORD,
        expected_last_exec_utc - datetime.timedelta(days=3))

    event = self._KLASS(
        task_config.get_event_setting(self._KLASS.section_name()),
        last_exec_client.get_last_execute_time(self._KLASS.KEYWORD))
    self.assertEqual(event.target_exec_utc, expected_last_exec_utc)
    self.assertEqual(event.last_exec_utc, expected_last_exec_utc)
    self.assertFalse(event.should_handle)

  def testCreateWeeklyEventWithOutdatedLastExecWithGAETesting(self):
    """Test an event with old last exec can be handled in testing on GAE."""
    utc_now = datetime.datetime(2017, 8, 1, 4, 15,
                                tzinfo=time_converter.UTC_TZ)
    expected_last_exec_utc = datetime.datetime(
        2017, 8, 1, 4, tzinfo=time_converter.UTC_TZ)
    task_config = task_config_reader.TaskConfig(self.config)
    self._mock_utc_now.return_value = utc_now
    last_exec_client = datastore_client.LastExecutionRecordStore()
    last_exec_client.set_last_execute_time(
        self._KLASS.KEYWORD,
        expected_last_exec_utc - datetime.timedelta(days=3))

    with mock.patch('global_config.GAE_TESTING', return_value=True):
      event = self._KLASS(
          task_config.get_event_setting(self._KLASS.section_name()),
          last_exec_client.get_last_execute_time(self._KLASS.KEYWORD))
      self.assertEqual(event.target_exec_utc, expected_last_exec_utc)
      self.assertEqual(event.last_exec_utc, expected_last_exec_utc)
      self.assertTrue(event.should_handle)

  def testCreateWeeklyEventWithEventHourSettings(self):
    """Create a weekly event with setting event hour."""
    self.config.add_section(self._KLASS.section_name())
    # Set weekly default kick off time to everyday 14:00 PST (21:00 UTC).
    self.config.set(self._KLASS.section_name(), 'hour', '14')
    expected_utc_hour = time_converter.convert_time_info(
        time_converter.TimeInfo(self._DEFAULT_PST_DAY, 14)).hour
    utc_now = datetime.datetime(2017, 8, 1, expected_utc_hour, 15,
                                tzinfo=time_converter.UTC_TZ)
    expected_last_exec_utc = datetime.datetime(
        2017, 7, 31, expected_utc_hour, tzinfo=time_converter.UTC_TZ)
    expected_cur_exec_utc = datetime.datetime(
        2017, 8, 1, expected_utc_hour, tzinfo=time_converter.UTC_TZ)
    task_config = task_config_reader.TaskConfig(self.config)
    self._mock_utc_now.return_value = utc_now
    last_exec_client = datastore_client.LastExecutionRecordStore()
    last_exec_client.set_last_execute_time(
        self._KLASS.KEYWORD, expected_last_exec_utc)

    event = self._KLASS(
        task_config.get_event_setting(self._KLASS.section_name()),
        last_exec_client.get_last_execute_time(self._KLASS.KEYWORD))
    self.assertEqual(event.target_exec_utc, expected_cur_exec_utc)
    self.assertEqual(event.last_exec_utc, expected_last_exec_utc)
    self.assertTrue(event.should_handle)

  def testFilterTasksWithTaskDayForWeeklyEvent(self):
    """Test filter_tasks with tasks which contain settings of day."""
    utc_now = datetime.datetime(2017, 8, 6, self._DEFAULT_UTC_HOUR, 15,
                                tzinfo=time_converter.UTC_TZ)
    self._mock_utc_now.return_value = utc_now
    event = self._KLASS(
        task_config_reader.EventSettings(True, None),
        datetime.datetime(2017, 8, 5, self._DEFAULT_UTC_HOUR,
                          tzinfo=time_converter.UTC_TZ))

    # It's in PST time.
    task_list = [FakeTask('test1', day=5),
                 FakeTask('test2', day=0),
                 FakeTask('test3', day=3),
                 FakeTask('test4', day=5)]
    event.set_task_list(task_list)
    self.assertEqual(len(event.task_list), len(task_list))
    event.filter_tasks()
    # Verify only tasks that should be run at Saturday 23:00 in PST are kept,
    # since 2017-08-06 6:00 is Sunday in UTC.
    self.assertEqual(len(event.task_list), 2)
    self.assertEqual(sorted([t.name for t in event.task_list]),
                     ['test1', 'test4'])

  def testFilterTasksWithDefaultDayForWeeklyEvent(self):
    """Test filter_tasks with tasks which don't contain settings of day."""
    utc_now = datetime.datetime(2017, 8, 6, self._DEFAULT_UTC_HOUR, 15,
                                tzinfo=time_converter.UTC_TZ)
    self._mock_utc_now.return_value = utc_now
    event = self._KLASS(
        task_config_reader.EventSettings(True, None),
        datetime.datetime(2017, 8, 5, self._DEFAULT_UTC_HOUR,
                          tzinfo=time_converter.UTC_TZ))

    # It's in PST time.
    task_list = [FakeTask('test1', day=5),
                 FakeTask('test2', day=6),
                 FakeTask('test3', day=4),
                 FakeTask('test4', day=None)]
    event.set_task_list(task_list)
    self.assertEqual(len(event.task_list), len(task_list))
    event.filter_tasks()
    # Verify only tasks that should be run at Saturday 23:00 in PST are kept,
    # since 2017-08-06 6:00 is Sunday in UTC.
    # If a task's day is not set, by default it should be run at
    # 6:00 at every Sunday in UTC (equals to 23:00 at every Saturday in PST).
    self.assertEqual(len(event.task_list), 2)
    self.assertEqual(sorted([t.name for t in event.task_list]),
                     ['test1', 'test4'])

  def testFilterTasksWithTaskDayWithGAETesting(self):
    """Verify all tasks are kept in testing on GAE."""
    utc_now = datetime.datetime(2017, 8, 6, self._DEFAULT_UTC_HOUR, 15,
                                tzinfo=time_converter.UTC_TZ)
    self._mock_utc_now.return_value = utc_now

    with mock.patch('global_config.GAE_TESTING', return_value=True):
      event = self._KLASS(
          task_config_reader.EventSettings(True, None),
          datetime.datetime(2017, 8, 5, self._DEFAULT_UTC_HOUR,
                            tzinfo=time_converter.UTC_TZ))
      task_list = [FakeTask('test1', day=6),
                   FakeTask('test2', day=0),
                   FakeTask('test3', day=3),
                   FakeTask('test4', day=6)]
      event.set_task_list(task_list)
      self.assertEqual(len(event.task_list), len(task_list))
      event.filter_tasks()
      self.assertEqual(len(event.task_list), 4)
      self.assertEqual(sorted([t.name for t in event.task_list]),
                       ['test1', 'test2', 'test3', 'test4'])

  def testSetTaskListForWeeklyEventCheckWeekday(self):
    """Verify tasks' timeinfo are correctly converted to UTC."""
    self._mock_utc_now.return_value = datetime.datetime.now(
        time_converter.UTC_TZ)
    event = self._KLASS(
        task_config_reader.EventSettings(True, None), None)

    # It's in PST time.
    task_list = [FakeTask('test1', day=self._DEFAULT_PST_DAY),
                 FakeTask('test2', day=0),
                 FakeTask('test3', day=self._DEFAULT_PST_DAY-1),
                 FakeTask('test4', day=None)]
    event.set_task_list(task_list)
    tasks_with_default_utc_day = [
        t.name for t in event.task_list if t.day == self._DEFAULT_UTC_DAY]
    self.assertEqual(sorted(tasks_with_default_utc_day),
                     ['test1', 'test4'])

  def testSetTaskListForWeeklyEventCheckHour(self):
    """Verify tasks' timeinfo are correctly converted to UTC."""
    self._mock_utc_now.return_value = datetime.datetime.now(
        time_converter.UTC_TZ)
    event = self._KLASS(
        task_config_reader.EventSettings(True, None), None)

    # It's in PST time.
    task_list = [FakeTask('test1', day=self._DEFAULT_PST_DAY),
                 FakeTask('test2', day=self._DEFAULT_PST_DAY, hour=0),
                 FakeTask('test3', day=self._DEFAULT_PST_DAY,
                          hour=self._DEFAULT_PST_HOUR)]
    event.set_task_list(task_list)
    tasks_with_default_utc_hour = [
        t.name for t in event.task_list if t.hour == self._DEFAULT_UTC_HOUR]
    self.assertEqual(sorted(tasks_with_default_utc_hour),
                     ['test1', 'test3'])