xref: /external/autotest/site_utils/suite_scheduler/timed_event_unittest.py

#!/usr/bin/python
#
# Copyright (c) 2012 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.

"""Unit tests for site_utils/timed_event.py."""

import collections, datetime, mox, unittest

# driver must be imported first due to circular imports in base_event and task
import driver  # pylint: disable-msg=W0611
import base_event, forgiving_config_parser
import manifest_versions, task, timed_event


class TimedEventTestBase(mox.MoxTestBase):
    """Base class for TimedEvent unit test classes."""


    def setUp(self):
        super(TimedEventTestBase, self).setUp()
        self.mox.StubOutWithMock(timed_event.TimedEvent, '_now')
        self.mv = self.mox.CreateMock(manifest_versions.ManifestVersions)


    def BaseTime(self):
        """Return the TimedEvent trigger-time as a datetime instance."""
        raise NotImplementedError()


    def CreateEvent(self):
        """Return an instance of the TimedEvent subclass being tested."""
        raise NotImplementedError()


    def TimeBefore(self, now):
        """Return a datetime that's before |now|."""
        raise NotImplementedError()


    def TimeLaterThan(self, now):
        """Return a datetime that's later than |now|."""
        raise NotImplementedError()


    def doTestDeadlineInFuture(self):
        fake_now = self.TimeBefore(self.BaseTime())
        timed_event.TimedEvent._now().MultipleTimes().AndReturn(fake_now)
        self.mox.ReplayAll()

        t = self.CreateEvent()  # Deadline gets set for a future time.
        self.assertFalse(t.ShouldHandle())
        self.mox.VerifyAll()

        self.mox.ResetAll()
        fake_now = self.TimeLaterThan(fake_now)  # Jump past that future time.
        timed_event.TimedEvent._now().MultipleTimes().AndReturn(fake_now)
        self.mox.ReplayAll()
        self.assertTrue(t.ShouldHandle())


    def doTestDeadlineIsNow(self):
        """We happened to create the trigger at the exact right time."""
        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
        self.mox.ReplayAll()
        to_test = self.CreateEvent()
        self.assertTrue(to_test.ShouldHandle())


    def doTestTOCTOU(self):
        """Even if deadline passes during initialization, trigger must fire."""
        init_now = self.BaseTime() - datetime.timedelta(seconds=1)
        fire_now = self.BaseTime() + datetime.timedelta(seconds=1)
        timed_event.TimedEvent._now().AndReturn(init_now)
        timed_event.TimedEvent._now().MultipleTimes().AndReturn(fire_now)
        self.mox.ReplayAll()

        t = self.CreateEvent()  # Deadline gets set for later tonight...
        # ...but has passed by the time we get around to firing.
        self.assertTrue(t.ShouldHandle())


    def doTestDeadlineUpdate(self, days_to_jump, hours_to_jump=0):
        fake_now = self.TimeBefore(self.BaseTime())
        timed_event.TimedEvent._now().MultipleTimes().AndReturn(fake_now)
        self.mox.ReplayAll()

        nightly = self.CreateEvent()  # Deadline gets set for tonight.
        self.assertFalse(nightly.ShouldHandle())
        self.mox.VerifyAll()

        self.mox.ResetAll()
        fake_now = self.TimeLaterThan(self.BaseTime())  # Jump past deadline.
        timed_event.TimedEvent._now().MultipleTimes().AndReturn(fake_now)
        self.mox.ReplayAll()

        self.assertTrue(nightly.ShouldHandle())
        nightly.UpdateCriteria()  # Deadline moves to an hour later
        self.assertFalse(nightly.ShouldHandle())
        self.mox.VerifyAll()

        self.mox.ResetAll()
        # Jump past deadline.
        fake_now += datetime.timedelta(days=days_to_jump, hours=hours_to_jump)
        timed_event.TimedEvent._now().MultipleTimes().AndReturn(fake_now)
        self.mox.ReplayAll()
        self.assertTrue(nightly.ShouldHandle())


    def doTestGetBranchBuilds(self, days):
        board = 'faux_board'
        branch_manifests = {('factory','16'): ['last16'],
                            ('release','17'): ['first17', 'last17']}
        since_date = self.BaseTime() - datetime.timedelta(days=days)
        self.mv.ManifestsSinceDate(since_date, board).AndReturn(
                branch_manifests)
        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
        self.mox.ReplayAll()
        branch_builds = self.CreateEvent().GetBranchBuildsForBoard(board)
        for (type, milestone), manifests in branch_manifests.iteritems():
            build = None
            if type in task.BARE_BRANCHES:
                self.assertEquals(len(branch_builds[type]), 1)
                build = branch_builds[type][0]
                self.assertTrue(build.startswith('%s-%s' % (board, type)))
            else:
                self.assertEquals(len(branch_builds[milestone]), 1)
                build = branch_builds[milestone][0]
                self.assertTrue(build.startswith('%s-release' % board))
            self.assertTrue('R%s-%s' % (milestone, manifests[-1]) in build)


class NightlyTest(TimedEventTestBase):
    """Unit tests for Weekly.
    """


    def setUp(self):
        super(NightlyTest, self).setUp()


    def BaseTime(self):
        return datetime.datetime(2012, 1, 1, 0, 0)


    def CreateEvent(self):
        """Return an instance of timed_event.Nightly."""
        return timed_event.Nightly(self.mv, False)


    def testCreateFromConfig(self):
        """Test that creating from config is equivalent to using constructor."""
        config = forgiving_config_parser.ForgivingConfigParser()
        section = base_event.SectionName(timed_event.Nightly.KEYWORD)
        config.add_section(section)

        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
        self.mox.ReplayAll()

        self.assertEquals(self.CreateEvent(),
                          timed_event.Nightly.CreateFromConfig(config, self.mv))


    def testCreateFromEmptyConfig(self):
        """Test that creating from empty config uses defaults."""
        config = forgiving_config_parser.ForgivingConfigParser()

        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
        self.mox.ReplayAll()

        self.assertEquals(
            timed_event.Nightly(self.mv, False),
            timed_event.Nightly.CreateFromConfig(config, self.mv))


    def testCreateFromAlwaysHandleConfig(self):
        """Test that creating with always_handle works as intended."""
        config = forgiving_config_parser.ForgivingConfigParser()
        section = base_event.SectionName(timed_event.Nightly.KEYWORD)
        config.add_section(section)
        config.set(section, 'always_handle', 'True')

        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
        self.mox.ReplayAll()

        event = timed_event.Nightly.CreateFromConfig(config, self.mv)
        self.assertTrue(event.ShouldHandle())


    def testMerge(self):
        """Test that Merge() works when the deadline time of day changes."""
        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
        self.mox.ReplayAll()

        old = timed_event.Nightly(self.mv, False)
        new = timed_event.Nightly(self.mv, False)
        old.Merge(new)
        self.assertEquals(old._deadline, new._deadline)


    def testSkipMerge(self):
        """Test that deadline is unchanged when time of day is unchanged."""
        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
        self.mox.ReplayAll()

        old = timed_event.Nightly(self.mv, False)
        new = timed_event.Nightly(self.mv, False)
        new._deadline += datetime.timedelta(days=1)
        self.assertNotEquals(old._deadline, new._deadline)
        saved_deadline = old._deadline
        old.Merge(new)
        self.assertEquals(saved_deadline, old._deadline)


    def testDeadlineInPast(self):
        """Ensure we work if the deadline aready passed today."""
        fake_now = self.BaseTime() + datetime.timedelta(hours=0.5)
        timed_event.TimedEvent._now().MultipleTimes().AndReturn(fake_now)
        self.mox.ReplayAll()

        nightly = self.CreateEvent()  # Deadline gets set for tomorrow night.
        self.assertFalse(nightly.ShouldHandle())
        self.mox.VerifyAll()

        self.mox.ResetAll()
        fake_now += datetime.timedelta(days=1)  # Jump to tomorrow night.
        timed_event.TimedEvent._now().MultipleTimes().AndReturn(fake_now)
        self.mox.ReplayAll()
        self.assertTrue(nightly.ShouldHandle())


    def TimeBefore(self, now):
        return now - datetime.timedelta(minutes=1)


    def TimeLaterThan(self, now):
        return now + datetime.timedelta(hours=0.5)


    def testDeadlineInFuture(self):
        """Ensure we work if the deadline is later today."""
        self.doTestDeadlineInFuture()


    def testDeadlineIsNow(self):
        """We happened to create the trigger at the exact right time."""
        self.doTestDeadlineIsNow()


    def testTOCTOU(self):
        """Even if deadline passes during initialization, trigger must fire."""
        self.doTestTOCTOU()


    def testDeadlineUpdate(self):
        """Ensure we update the deadline correctly."""
        self.doTestDeadlineUpdate(days_to_jump=0, hours_to_jump=1)


    #def testGetBranchBuilds(self):
    #    """Ensure Nightly gets most recent builds in last day."""
    #    self.doTestGetBranchBuilds(days=1)


    def testFilterTasks(self):
        """Test FilterTasks function can filter tasks by current hour."""
        Task = collections.namedtuple('Task', 'hour')
        task_1 = Task(hour=0)
        task_2 = Task(hour=10)
        task_3 = Task(hour=11)
        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
        self.mox.ReplayAll()
        event = self.CreateEvent()
        event.tasks = set([task_1, task_2, task_3])
        self.assertEquals([task_1], event.FilterTasks())


class WeeklyTest(TimedEventTestBase):
    """Unit tests for Weekly.

    @var _HOUR: The time of night to use in these unit tests.
    """

    _HOUR = 22


    def setUp(self):
        super(WeeklyTest, self).setUp()


    def BaseTime(self):
        basetime = datetime.datetime(2012, 1, 1, self._HOUR)
        return basetime


    def CreateEvent(self):
        """Return an instance of timed_event.Weekly."""
        return timed_event.Weekly(self.mv, False, self._HOUR)


    def testCreateFromConfig(self):
        """Test that creating from config is equivalent to using constructor."""
        config = forgiving_config_parser.ForgivingConfigParser()
        section = base_event.SectionName(timed_event.Weekly.KEYWORD)
        config.add_section(section)
        config.set(section, 'hour', '%d' % self._HOUR)

        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
        self.mox.ReplayAll()

        self.assertEquals(self.CreateEvent(),
                          timed_event.Weekly.CreateFromConfig(config, self.mv))


    def testMergeDueToTimeChange(self):
        """Test that Merge() works when the deadline time of day changes."""
        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
        self.mox.ReplayAll()

        old = timed_event.Weekly(self.mv, False, self._HOUR)
        new = timed_event.Weekly(self.mv, False, self._HOUR + 1)
        self.assertNotEquals(old._deadline, new._deadline)
        old.Merge(new)
        self.assertEquals(old._deadline, new._deadline)


    def testSkipMerge(self):
        """Test that deadline is unchanged when only the week is changed."""
        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
        self.mox.ReplayAll()

        old = timed_event.Weekly(self.mv, False, self._HOUR)
        new = timed_event.Weekly(self.mv, False, self._HOUR)
        new._deadline += datetime.timedelta(days=7)
        self.assertNotEquals(old._deadline, new._deadline)
        saved_deadline = old._deadline
        old.Merge(new)
        self.assertEquals(saved_deadline, old._deadline)


    def testDeadlineInPast(self):
        """Ensure we work if the deadline already passed this week."""
        fake_now = self.BaseTime() + datetime.timedelta(days=0.5)
        timed_event.TimedEvent._now().MultipleTimes().AndReturn(fake_now)
        self.mox.ReplayAll()

        weekly = self.CreateEvent()  # Deadline gets set for next week.
        self.assertFalse(weekly.ShouldHandle())
        self.mox.VerifyAll()

        self.mox.ResetAll()
        fake_now += datetime.timedelta(days=1)  # Jump to tomorrow.
        timed_event.TimedEvent._now().MultipleTimes().AndReturn(fake_now)
        self.mox.ReplayAll()
        self.assertTrue(weekly.ShouldHandle())
        self.mox.VerifyAll()

        self.mox.ResetAll()
        fake_now += datetime.timedelta(days=7)  # Jump to next week.
        timed_event.TimedEvent._now().MultipleTimes().AndReturn(fake_now)
        self.mox.ReplayAll()
        self.assertTrue(weekly.ShouldHandle())


    def TimeBefore(self, now):
        return now - datetime.timedelta(days=0.5)


    def TimeLaterThan(self, now):
        return now + datetime.timedelta(days=0.5)


    def testDeadlineInFuture(self):
        """Ensure we work if the deadline is later this week."""
        self.doTestDeadlineInFuture()


    def testDeadlineIsNow(self):
        """We happened to create the trigger at the exact right time."""
        self.doTestDeadlineIsNow()


    def testTOCTOU(self):
        """Even if deadline passes during initialization, trigger must fire."""
        self.doTestTOCTOU()


    def testDeadlineUpdate(self):
        """Ensure we update the deadline correctly."""
        self.doTestDeadlineUpdate(days_to_jump=1)


    #def testGetBranchBuilds(self):
    #    """Ensure Weekly gets most recent builds in last 7 days."""
    #    self.doTestGetBranchBuilds(days=7)


    def testFilterTasks(self):
        """Test FilterTasks function can filter tasks by current day."""
        Task = collections.namedtuple('Task', 'day')
        task_1 = Task(day=6)
        task_2 = Task(day=2)
        task_3 = Task(day=5)
        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
        self.mox.ReplayAll()
        event = self.CreateEvent()
        event.tasks = set([task_1, task_2, task_3])
        self.assertEquals([task_1], event.FilterTasks())


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