timed_event_unittest.py revision c7bcf8bf70d6e40aa4d37528f75b5a98b0f7a00e 
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
11# driver must be imported first due to circular imports in base_event and task
12import driver  # pylint: disable-msg=W0611
13import base_event, deduping_scheduler, forgiving_config_parser
14import manifest_versions, task, timed_event
15
16
17class TimedEventTestBase(mox.MoxTestBase):
18    """Base class for TimedEvent unit test classes."""
19
20
21    def setUp(self):
22        super(TimedEventTestBase, self).setUp()
23        self.mox.StubOutWithMock(timed_event.TimedEvent, '_now')
24        self.mv = self.mox.CreateMock(manifest_versions.ManifestVersions)
25
26
27    def BaseTime(self):
28        """Return the TimedEvent trigger-time as a datetime instance."""
29        raise NotImplementedError()
30
31
32    def CreateEvent(self):
33        """Return an instance of the TimedEvent subclass being tested."""
34        raise NotImplementedError()
35
36
37    def TimeBefore(self, now):
38        """Return a datetime that's before |now|."""
39        raise NotImplementedError()
40
41
42    def TimeLaterThan(self, now):
43        """Return a datetime that's later than |now|."""
44        raise NotImplementedError()
45
46
47    def doTestDeadlineInFuture(self):
48        fake_now = self.TimeBefore(self.BaseTime())
49        timed_event.TimedEvent._now().MultipleTimes().AndReturn(fake_now)
50        self.mox.ReplayAll()
51
52        t = self.CreateEvent()  # Deadline gets set for a future time.
53        self.assertFalse(t.ShouldHandle())
54        self.mox.VerifyAll()
55
56        self.mox.ResetAll()
57        fake_now = self.TimeLaterThan(fake_now)  # Jump past that future time.
58        timed_event.TimedEvent._now().MultipleTimes().AndReturn(fake_now)
59        self.mox.ReplayAll()
60        self.assertTrue(t.ShouldHandle())
61
62
63    def doTestDeadlineIsNow(self):
64        """We happened to create the trigger at the exact right time."""
65        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
66        self.mox.ReplayAll()
67        to_test = self.CreateEvent()
68        self.assertTrue(to_test.ShouldHandle())
69
70
71    def doTestTOCTOU(self):
72        """Even if deadline passes during initialization, trigger must fire."""
73        init_now = self.BaseTime() - datetime.timedelta(seconds=1)
74        fire_now = self.BaseTime() + datetime.timedelta(seconds=1)
75        timed_event.TimedEvent._now().AndReturn(init_now)
76        timed_event.TimedEvent._now().AndReturn(fire_now)
77        self.mox.ReplayAll()
78
79        t = self.CreateEvent()  # Deadline gets set for later tonight...
80        # ...but has passed by the time we get around to firing.
81        self.assertTrue(t.ShouldHandle())
82
83
84    def doTestDeadlineUpdate(self, days_to_jump):
85        fake_now = self.TimeBefore(self.BaseTime())
86        timed_event.TimedEvent._now().MultipleTimes().AndReturn(fake_now)
87        self.mox.ReplayAll()
88
89        nightly = self.CreateEvent()  # Deadline gets set for tonight.
90        self.assertFalse(nightly.ShouldHandle())
91        self.mox.VerifyAll()
92
93        self.mox.ResetAll()
94        fake_now = self.TimeLaterThan(self.BaseTime())  # Jump past deadline.
95        timed_event.TimedEvent._now().MultipleTimes().AndReturn(fake_now)
96        self.mox.ReplayAll()
97
98        self.assertTrue(nightly.ShouldHandle())
99        nightly.UpdateCriteria()  # Deadline moves to tomorrow night
100        self.assertFalse(nightly.ShouldHandle())
101        self.mox.VerifyAll()
102
103        self.mox.ResetAll()
104        fake_now += datetime.timedelta(days=days_to_jump)  # Jump past deadline.
105        timed_event.TimedEvent._now().MultipleTimes().AndReturn(fake_now)
106        self.mox.ReplayAll()
107        self.assertTrue(nightly.ShouldHandle())
108
109
110    def doTestGetBranchBuilds(self, days):
111        board = 'faux_board'
112        branch_manifests = {('factory','16'): ['last16'],
113                            ('release','17'): ['first17', 'last17']}
114        self.mv.ManifestsSinceDays(days, board).AndReturn(branch_manifests)
115        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
116        self.mox.ReplayAll()
117
118        branch_builds = self.CreateEvent().GetBranchBuildsForBoard(board)
119        for (type, milestone), manifests in branch_manifests.iteritems():
120            build = None
121            if type in task.BARE_BRANCHES:
122                self.assertEquals(len(branch_builds[type]), 1)
123                build = branch_builds[type][0]
124                self.assertTrue(build.startswith('%s-%s' % (board, type)))
125            else:
126                self.assertEquals(len(branch_builds[milestone]), 1)
127                build = branch_builds[milestone][0]
128                self.assertTrue(build.startswith('%s-release' % board))
129            self.assertTrue('R%s-%s' % (milestone, manifests[-1]) in build)
130
131
132class NightlyTest(TimedEventTestBase):
133    """Unit tests for Weekly.
134
135    @var _HOUR: The time of night to use in these unit tests.
136    """
137
138    _HOUR = 20
139
140
141    def setUp(self):
142        super(NightlyTest, self).setUp()
143
144
145    def BaseTime(self):
146        return datetime.datetime(2012, 1, 1, self._HOUR)
147
148
149    def CreateEvent(self):
150        """Return an instance of timed_event.Nightly."""
151        return timed_event.Nightly(self.mv, False, self._HOUR)
152
153
154    def testCreateFromConfig(self):
155        """Test that creating from config is equivalent to using constructor."""
156        config = forgiving_config_parser.ForgivingConfigParser()
157        section = base_event.SectionName(timed_event.Nightly.KEYWORD)
158        config.add_section(section)
159        config.set(section, 'hour', '%d' % self._HOUR)
160
161        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
162        self.mox.ReplayAll()
163
164        self.assertEquals(self.CreateEvent(),
165                          timed_event.Nightly.CreateFromConfig(config, self.mv))
166
167
168    def testCreateFromEmptyConfig(self):
169        """Test that creating from empty config uses defaults."""
170        config = forgiving_config_parser.ForgivingConfigParser()
171
172        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
173        self.mox.ReplayAll()
174
175        self.assertEquals(
176            timed_event.Nightly(self.mv, False,
177                                timed_event.Nightly._DEFAULT_HOUR),
178            timed_event.Nightly.CreateFromConfig(config, self.mv))
179
180
181    def testCreateFromAlwaysHandleConfig(self):
182        """Test that creating with always_handle works as intended."""
183        config = forgiving_config_parser.ForgivingConfigParser()
184        section = base_event.SectionName(timed_event.Nightly.KEYWORD)
185        config.add_section(section)
186        config.set(section, 'hour', '%d' % (self._HOUR + 1))
187        config.set(section, 'always_handle', 'True')
188
189        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
190        self.mox.ReplayAll()
191
192        event = timed_event.Nightly.CreateFromConfig(config, self.mv)
193        self.assertTrue(event.ShouldHandle())
194
195
196    def testMerge(self):
197        """Test that Merge() works when the deadline time of day changes."""
198        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
199        self.mox.ReplayAll()
200
201        old = timed_event.Nightly(self.mv, False, self._HOUR)
202        new = timed_event.Nightly(self.mv, False, (self._HOUR + 23) % 24)
203        self.assertNotEquals(old._deadline, new._deadline)
204        old.Merge(new)
205        self.assertEquals(old._deadline, new._deadline)
206
207
208    def testSkipMerge(self):
209        """Test that deadline is unchanged when time of day is unchanged."""
210        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
211        self.mox.ReplayAll()
212
213        old = timed_event.Nightly(self.mv, False, self._HOUR)
214        new = timed_event.Nightly(self.mv, False, self._HOUR)
215        new._deadline += datetime.timedelta(days=1)
216        self.assertNotEquals(old._deadline, new._deadline)
217        saved_deadline = old._deadline
218        old.Merge(new)
219        self.assertEquals(saved_deadline, old._deadline)
220
221
222    def testDeadlineInPast(self):
223        """Ensure we work if the deadline aready passed today."""
224        fake_now = self.BaseTime() + datetime.timedelta(hours=1)
225        timed_event.TimedEvent._now().MultipleTimes().AndReturn(fake_now)
226        self.mox.ReplayAll()
227
228        nightly = self.CreateEvent()  # Deadline gets set for tomorrow night.
229        self.assertFalse(nightly.ShouldHandle())
230        self.mox.VerifyAll()
231
232        self.mox.ResetAll()
233        fake_now += datetime.timedelta(days=1)  # Jump to tomorrow night.
234        timed_event.TimedEvent._now().MultipleTimes().AndReturn(fake_now)
235        self.mox.ReplayAll()
236        self.assertTrue(nightly.ShouldHandle())
237
238
239    def TimeBefore(self, now):
240        return now - datetime.timedelta(hours=1)
241
242
243    def TimeLaterThan(self, now):
244        return now + datetime.timedelta(hours=2)
245
246
247    def testDeadlineInFuture(self):
248        """Ensure we work if the deadline is later today."""
249        self.doTestDeadlineInFuture()
250
251
252    def testDeadlineIsNow(self):
253        """We happened to create the trigger at the exact right time."""
254        self.doTestDeadlineIsNow()
255
256
257    def testTOCTOU(self):
258        """Even if deadline passes during initialization, trigger must fire."""
259        self.doTestTOCTOU()
260
261
262    def testDeadlineUpdate(self):
263        """Ensure we update the deadline correctly."""
264        self.doTestDeadlineUpdate(days_to_jump=1)
265
266
267    def testGetBranchBuilds(self):
268        """Ensure Nightly gets most recent builds in last day."""
269        self.doTestGetBranchBuilds(days=1)
270
271
272class WeeklyTest(TimedEventTestBase):
273    """Unit tests for Weekly.
274
275    @var _DAY: The day of the week to use in these unit tests.
276    @var _HOUR: The time of night to use in these unit tests.
277    """
278
279    _DAY = 5
280    _HOUR = 22
281
282
283    def setUp(self):
284        super(WeeklyTest, self).setUp()
285
286
287    def BaseTime(self):
288        basetime = datetime.datetime(2012, 1, 1, self._HOUR)
289        basetime += datetime.timedelta(self._DAY-basetime.weekday())
290        return basetime
291
292
293    def CreateEvent(self):
294        """Return an instance of timed_event.Weekly."""
295        return timed_event.Weekly(self.mv, False, self._DAY, self._HOUR)
296
297
298    def testCreateFromConfig(self):
299        """Test that creating from config is equivalent to using constructor."""
300        config = forgiving_config_parser.ForgivingConfigParser()
301        section = base_event.SectionName(timed_event.Weekly.KEYWORD)
302        config.add_section(section)
303        config.set(section, 'day', '%d' % self._DAY)
304        config.set(section, 'hour', '%d' % self._HOUR)
305
306        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
307        self.mox.ReplayAll()
308
309        self.assertEquals(self.CreateEvent(),
310                          timed_event.Weekly.CreateFromConfig(config, self.mv))
311
312
313    def testMergeDueToTimeChange(self):
314        """Test that Merge() works when the deadline time of day changes."""
315        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
316        self.mox.ReplayAll()
317
318        old = timed_event.Weekly(self.mv, False, self._DAY, self._HOUR)
319        new = timed_event.Weekly(self.mv, False, self._DAY, self._HOUR + 1)
320        self.assertNotEquals(old._deadline, new._deadline)
321        old.Merge(new)
322        self.assertEquals(old._deadline, new._deadline)
323
324
325    def testMergeDueToDayChange(self):
326        """Test that Merge() works when the deadline day of week changes."""
327        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
328        self.mox.ReplayAll()
329
330        old = timed_event.Weekly(self.mv, False, self._DAY, self._HOUR)
331        new = timed_event.Weekly(self.mv, False, self._DAY, self._HOUR)
332        new._deadline += datetime.timedelta(days=1)
333        self.assertNotEquals(old._deadline, new._deadline)
334        old.Merge(new)
335        self.assertEquals(old._deadline, new._deadline)
336
337
338    def testSkipMerge(self):
339        """Test that deadline is unchanged when only the week is changed."""
340        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
341        self.mox.ReplayAll()
342
343        old = timed_event.Weekly(self.mv, False, self._DAY, self._HOUR)
344        new = timed_event.Weekly(self.mv, False, self._DAY, self._HOUR)
345        new._deadline += datetime.timedelta(days=7)
346        self.assertNotEquals(old._deadline, new._deadline)
347        saved_deadline = old._deadline
348        old.Merge(new)
349        self.assertEquals(saved_deadline, old._deadline)
350
351
352    def testDeadlineInPast(self):
353        """Ensure we work if the deadline already passed this week."""
354        fake_now = self.BaseTime() + datetime.timedelta(days=1)
355        timed_event.TimedEvent._now().MultipleTimes().AndReturn(fake_now)
356        self.mox.ReplayAll()
357
358        weekly = self.CreateEvent()  # Deadline gets set for next week.
359        self.assertFalse(weekly.ShouldHandle())
360        self.mox.VerifyAll()
361
362        self.mox.ResetAll()
363        fake_now += datetime.timedelta(days=1)  # Jump to tomorrow.
364        timed_event.TimedEvent._now().MultipleTimes().AndReturn(fake_now)
365        self.mox.ReplayAll()
366        self.assertFalse(weekly.ShouldHandle())
367        self.mox.VerifyAll()
368
369        self.mox.ResetAll()
370        fake_now += datetime.timedelta(days=7)  # Jump to next week.
371        timed_event.TimedEvent._now().MultipleTimes().AndReturn(fake_now)
372        self.mox.ReplayAll()
373        self.assertTrue(weekly.ShouldHandle())
374
375
376    def TimeBefore(self, now):
377        return now - datetime.timedelta(days=1)
378
379
380    def TimeLaterThan(self, now):
381        return now + datetime.timedelta(days=2)
382
383
384    def testDeadlineInFuture(self):
385        """Ensure we work if the deadline is later this week."""
386        self.doTestDeadlineInFuture()
387
388
389    def testDeadlineIsNow(self):
390        """We happened to create the trigger at the exact right time."""
391        self.doTestDeadlineIsNow()
392
393
394    def testTOCTOU(self):
395        """Even if deadline passes during initialization, trigger must fire."""
396        self.doTestTOCTOU()
397
398
399    def testDeadlineUpdate(self):
400        """Ensure we update the deadline correctly."""
401        self.doTestDeadlineUpdate(days_to_jump=7)
402
403
404    def testGetBranchBuilds(self):
405        """Ensure Weekly gets most recent builds in last 7 days."""
406        self.doTestGetBranchBuilds(days=7)
407
408
409if __name__ == '__main__':
410  unittest.main()
411