timed_event_unittest.py revision 05b1944ebcb8a62ee0cc38a7f90b4131ededab8f 
1#!/usr/bin/python
2#
3# Copyright (c) 2012 The Chromium OS Authors. All rights reserved.
4# Use of this source code is governed by a BSD-style license that can be
5# found in the LICENSE file.
6
7"""Unit tests for site_utils/timed_event.py."""
8
9import datetime, logging, mox,  unittest
10
11import deduping_scheduler, forgiving_config_parser, task, timed_event
12import manifest_versions
13
14
15class TimedEventTestBase(mox.MoxTestBase):
16    """Base class for TimedEvent unit test classes."""
17
18
19    def setUp(self):
20        super(TimedEventTestBase, self).setUp()
21        self.mox.StubOutWithMock(timed_event.TimedEvent, '_now')
22        self.mv = self.mox.CreateMock(manifest_versions.ManifestVersions)
23
24
25    def BaseTime(self):
26        """Return the TimedEvent trigger-time as a datetime instance."""
27        raise NotImplementedError()
28
29
30    def CreateEvent(self):
31        """Return an instance of the TimedEvent subclass being tested."""
32        raise NotImplementedError()
33
34
35    def TimeBefore(self, now):
36        """Return a datetime that's before |now|."""
37        raise NotImplementedError()
38
39
40    def TimeLaterThan(self, now):
41        """Return a datetime that's later than |now|."""
42        raise NotImplementedError()
43
44
45    def doTestDeadlineInFuture(self):
46        fake_now = self.TimeBefore(self.BaseTime())
47        timed_event.TimedEvent._now().MultipleTimes().AndReturn(fake_now)
48        self.mox.ReplayAll()
49
50        t = self.CreateEvent()  # Deadline gets set for a future time.
51        self.assertFalse(t.ShouldHandle())
52        self.mox.VerifyAll()
53
54        self.mox.ResetAll()
55        fake_now = self.TimeLaterThan(fake_now)  # Jump past that future time.
56        timed_event.TimedEvent._now().MultipleTimes().AndReturn(fake_now)
57        self.mox.ReplayAll()
58        self.assertTrue(t.ShouldHandle())
59
60
61    def doTestDeadlineIsNow(self):
62        """We happened to create the trigger at the exact right time."""
63        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
64        self.mox.ReplayAll()
65        to_test = self.CreateEvent()
66        self.assertTrue(to_test.ShouldHandle())
67
68
69    def doTestTOCTOU(self):
70        """Even if deadline passes during initialization, trigger must fire."""
71        init_now = self.BaseTime() - datetime.timedelta(seconds=1)
72        fire_now = self.BaseTime() + datetime.timedelta(seconds=1)
73        timed_event.TimedEvent._now().AndReturn(init_now)
74        timed_event.TimedEvent._now().AndReturn(fire_now)
75        self.mox.ReplayAll()
76
77        t = self.CreateEvent()  # Deadline gets set for later tonight...
78        # ...but has passed by the time we get around to firing.
79        self.assertTrue(t.ShouldHandle())
80
81
82    def doTestGetBranchBuilds(self, days):
83        board = 'faux_board'
84        branch_manifests = {('factory','16'): ['last16'],
85                            ('release','17'): ['first17', 'last17']}
86        self.mv.ManifestsSince(days, board).AndReturn(branch_manifests)
87        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
88        self.mox.ReplayAll()
89
90        branch_builds = self.CreateEvent().GetBranchBuildsForBoard(board,
91                                                                   self.mv)
92        for (type, milestone), manifests in branch_manifests.iteritems():
93            build = None
94            if type in task.BARE_BRANCHES:
95                build = branch_builds[type]
96                self.assertTrue(build.startswith('%s-%s' % (board, type)))
97            else:
98                build = branch_builds[milestone]
99                self.assertTrue(build.startswith('%s-release' % board))
100            self.assertTrue('R%s-%s' % (milestone, manifests[-1:]) in build)
101
102
103class NightlyTest(TimedEventTestBase):
104    """Unit tests for Weekly.
105
106    @var _HOUR: The time of night to use in these unit tests.
107    """
108
109    _HOUR = 20
110
111
112    def setUp(self):
113        super(NightlyTest, self).setUp()
114
115
116    def BaseTime(self):
117        return datetime.datetime(2012, 1, 1, self._HOUR)
118
119
120    def CreateEvent(self):
121        """Return an instance of timed_event.Nightly."""
122        return timed_event.Nightly(self._HOUR)
123
124
125    def testCreateFromConfig(self):
126        """Test that creating from config is equivalent to using constructor."""
127        config = forgiving_config_parser.ForgivingConfigParser()
128        section = timed_event.TimedEvent.section_name(
129            timed_event.Nightly.KEYWORD)
130        config.add_section(section)
131        config.set(section, 'hour', '%d' % self._HOUR)
132
133        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
134        self.mox.ReplayAll()
135
136        self.assertEquals(timed_event.Nightly(self._HOUR),
137                          timed_event.Nightly.CreateFromConfig(config))
138
139
140    def testCreateFromEmptyConfig(self):
141        """Test that creating from empty config uses defaults."""
142        config = forgiving_config_parser.ForgivingConfigParser()
143
144        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
145        self.mox.ReplayAll()
146
147        self.assertEquals(
148            timed_event.Nightly(timed_event.Nightly._DEFAULT_HOUR),
149            timed_event.Nightly.CreateFromConfig(config))
150
151
152    def testDeadlineInPast(self):
153        """Ensure we work if the deadline aready passed today."""
154        fake_now = self.BaseTime() + datetime.timedelta(hours=1)
155        timed_event.TimedEvent._now().MultipleTimes().AndReturn(fake_now)
156        self.mox.ReplayAll()
157
158        nightly = self.CreateEvent()  # Deadline gets set for tomorrow night.
159        self.assertFalse(nightly.ShouldHandle())
160        self.mox.VerifyAll()
161
162        self.mox.ResetAll()
163        fake_now += datetime.timedelta(days=1)  # Jump to tomorrow night.
164        timed_event.TimedEvent._now().MultipleTimes().AndReturn(fake_now)
165        self.mox.ReplayAll()
166        self.assertTrue(nightly.ShouldHandle())
167
168
169    def TimeBefore(self, now):
170        return now - datetime.timedelta(hours=1)
171
172
173    def TimeLaterThan(self, now):
174        return now + datetime.timedelta(hours=2)
175
176
177    def testDeadlineInFuture(self):
178        """Ensure we work if the deadline is later today."""
179        self.doTestDeadlineInFuture()
180
181
182    def testDeadlineIsNow(self):
183        """We happened to create the trigger at the exact right time."""
184        self.doTestDeadlineIsNow()
185
186
187    def testTOCTOU(self):
188        """Even if deadline passes during initialization, trigger must fire."""
189        self.doTestTOCTOU()
190
191
192    def testGetBranchBuilds(self):
193        """Ensure Nightly gets most recent builds in last day."""
194        self.doTestGetBranchBuilds(days=1)
195
196
197class WeeklyTest(TimedEventTestBase):
198    """Unit tests for Weekly.
199
200    @var _DAY: The day of the week to use in these unit tests.
201    @var _HOUR: The time of night to use in these unit tests.
202    """
203
204    _DAY = 5
205    _HOUR = 22
206
207
208    def setUp(self):
209        super(WeeklyTest, self).setUp()
210
211
212    def BaseTime(self):
213        basetime = datetime.datetime(2012, 1, 1, self._HOUR)
214        basetime += datetime.timedelta(self._DAY-basetime.weekday())
215        return basetime
216
217
218    def CreateEvent(self):
219        """Return an instance of timed_event.Weekly."""
220        return timed_event.Weekly(self._DAY, self._HOUR)
221
222
223    def testCreateFromConfig(self):
224        """Test that creating from config is equivalent to using constructor."""
225        config = forgiving_config_parser.ForgivingConfigParser()
226        section = timed_event.TimedEvent.section_name(
227            timed_event.Weekly.KEYWORD)
228        config.add_section(section)
229        config.set(section, 'day', '%d' % self._DAY)
230        config.set(section, 'hour', '%d' % self._HOUR)
231
232        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
233        self.mox.ReplayAll()
234
235        self.assertEquals(timed_event.Weekly(self._DAY, self._HOUR),
236                          timed_event.Weekly.CreateFromConfig(config))
237
238
239    def testDeadlineInPast(self):
240        """Ensure we work if the deadline already passed this week."""
241        fake_now = self.BaseTime() + datetime.timedelta(days=1)
242        timed_event.TimedEvent._now().MultipleTimes().AndReturn(fake_now)
243        self.mox.ReplayAll()
244
245        weekly = self.CreateEvent()  # Deadline gets set for next week.
246        self.assertFalse(weekly.ShouldHandle())
247        self.mox.VerifyAll()
248
249        self.mox.ResetAll()
250        fake_now += datetime.timedelta(days=1)  # Jump to tomorrow.
251        timed_event.TimedEvent._now().MultipleTimes().AndReturn(fake_now)
252        self.mox.ReplayAll()
253        self.assertFalse(weekly.ShouldHandle())
254        self.mox.VerifyAll()
255
256        self.mox.ResetAll()
257        fake_now += datetime.timedelta(days=7)  # Jump to next week.
258        timed_event.TimedEvent._now().MultipleTimes().AndReturn(fake_now)
259        self.mox.ReplayAll()
260        self.assertTrue(weekly.ShouldHandle())
261
262
263    def TimeBefore(self, now):
264        return now - datetime.timedelta(days=1)
265
266
267    def TimeLaterThan(self, now):
268        return now + datetime.timedelta(days=2)
269
270
271    def testDeadlineInFuture(self):
272        """Ensure we work if the deadline is later this week."""
273        self.doTestDeadlineInFuture()
274
275
276    def testDeadlineIsNow(self):
277        """We happened to create the trigger at the exact right time."""
278        self.doTestDeadlineIsNow()
279
280
281    def testTOCTOU(self):
282        """Even if deadline passes during initialization, trigger must fire."""
283        self.doTestTOCTOU()
284
285
286    def testGetBranchBuilds(self):
287        """Ensure Weekly gets most recent builds in last 7 days."""
288        self.doTestGetBranchBuilds(days=7)
289
290
291if __name__ == '__main__':
292  unittest.main()
293