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 datetime, logging, mox, unittest

import base_event, deduping_scheduler, 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().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):
        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 tomorrow night
        self.assertFalse(nightly.ShouldHandle())
        self.mox.VerifyAll()

        self.mox.ResetAll()
        fake_now += datetime.timedelta(days=days_to_jump)  # Jump past deadline.
        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']}
        self.mv.ManifestsSinceDays(days, 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.

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

    _HOUR = 20


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


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


    def CreateEvent(self):
        """Return an instance of timed_event.Nightly."""
        return timed_event.Nightly(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.Nightly.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.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._DEFAULT_HOUR),
            timed_event.Nightly.CreateFromConfig(config, self.mv))


    def testDeadlineInPast(self):
        """Ensure we work if the deadline aready passed today."""
        fake_now = self.BaseTime() + datetime.timedelta(hours=1)
        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(hours=1)


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


    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=1)


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


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

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

    _DAY = 5
    _HOUR = 22


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


    def BaseTime(self):
        basetime = datetime.datetime(2012, 1, 1, self._HOUR)
        basetime += datetime.timedelta(self._DAY-basetime.weekday())
        return basetime


    def CreateEvent(self):
        """Return an instance of timed_event.Weekly."""
        return timed_event.Weekly(self.mv, False, self._DAY, 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, 'day', '%d' % self._DAY)
        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 testDeadlineInPast(self):
        """Ensure we work if the deadline already passed this week."""
        fake_now = self.BaseTime() + datetime.timedelta(days=1)
        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.assertFalse(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=1)


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


    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=7)


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


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