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 collections, datetime, 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, 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().MultipleTimes().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, hours_to_jump=0):
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 an hour later
100        self.assertFalse(nightly.ShouldHandle())
101        self.mox.VerifyAll()
102
103        self.mox.ResetAll()
104        # Jump past deadline.
105        fake_now += datetime.timedelta(days=days_to_jump, hours=hours_to_jump)
106        timed_event.TimedEvent._now().MultipleTimes().AndReturn(fake_now)
107        self.mox.ReplayAll()
108        self.assertTrue(nightly.ShouldHandle())
109
110
111    def doTestGetBranchBuilds(self, days):
112        board = 'faux_board'
113        branch_manifests = {('factory','16'): ['last16'],
114                            ('release','17'): ['first17', 'last17']}
115        since_date = self.BaseTime() - datetime.timedelta(days=days)
116        self.mv.ManifestsSinceDate(since_date, board).AndReturn(
117                branch_manifests)
118        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
119        self.mox.ReplayAll()
120        branch_builds = self.CreateEvent().GetBranchBuildsForBoard(board)
121        for (type, milestone), manifests in branch_manifests.iteritems():
122            build = None
123            if type in task.BARE_BRANCHES:
124                self.assertEquals(len(branch_builds[type]), 1)
125                build = branch_builds[type][0]
126                self.assertTrue(build.startswith('%s-%s' % (board, type)))
127            else:
128                self.assertEquals(len(branch_builds[milestone]), 1)
129                build = branch_builds[milestone][0]
130                self.assertTrue(build.startswith('%s-release' % board))
131            self.assertTrue('R%s-%s' % (milestone, manifests[-1]) in build)
132
133
134class NightlyTest(TimedEventTestBase):
135    """Unit tests for Weekly.
136    """
137
138
139    def setUp(self):
140        super(NightlyTest, self).setUp()
141
142
143    def BaseTime(self):
144        return datetime.datetime(2012, 1, 1, 0, 0)
145
146
147    def CreateEvent(self):
148        """Return an instance of timed_event.Nightly."""
149        return timed_event.Nightly(self.mv, False)
150
151
152    def testCreateFromConfig(self):
153        """Test that creating from config is equivalent to using constructor."""
154        config = forgiving_config_parser.ForgivingConfigParser()
155        section = base_event.SectionName(timed_event.Nightly.KEYWORD)
156        config.add_section(section)
157
158        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
159        self.mox.ReplayAll()
160
161        self.assertEquals(self.CreateEvent(),
162                          timed_event.Nightly.CreateFromConfig(config, self.mv))
163
164
165    def testCreateFromEmptyConfig(self):
166        """Test that creating from empty config uses defaults."""
167        config = forgiving_config_parser.ForgivingConfigParser()
168
169        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
170        self.mox.ReplayAll()
171
172        self.assertEquals(
173            timed_event.Nightly(self.mv, False),
174            timed_event.Nightly.CreateFromConfig(config, self.mv))
175
176
177    def testCreateFromAlwaysHandleConfig(self):
178        """Test that creating with always_handle works as intended."""
179        config = forgiving_config_parser.ForgivingConfigParser()
180        section = base_event.SectionName(timed_event.Nightly.KEYWORD)
181        config.add_section(section)
182        config.set(section, 'always_handle', 'True')
183
184        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
185        self.mox.ReplayAll()
186
187        event = timed_event.Nightly.CreateFromConfig(config, self.mv)
188        self.assertTrue(event.ShouldHandle())
189
190
191    def testMerge(self):
192        """Test that Merge() works when the deadline time of day changes."""
193        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
194        self.mox.ReplayAll()
195
196        old = timed_event.Nightly(self.mv, False)
197        new = timed_event.Nightly(self.mv, False)
198        old.Merge(new)
199        self.assertEquals(old._deadline, new._deadline)
200
201
202    def testSkipMerge(self):
203        """Test that deadline is unchanged when time of day is unchanged."""
204        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
205        self.mox.ReplayAll()
206
207        old = timed_event.Nightly(self.mv, False)
208        new = timed_event.Nightly(self.mv, False)
209        new._deadline += datetime.timedelta(days=1)
210        self.assertNotEquals(old._deadline, new._deadline)
211        saved_deadline = old._deadline
212        old.Merge(new)
213        self.assertEquals(saved_deadline, old._deadline)
214
215
216    def testDeadlineInPast(self):
217        """Ensure we work if the deadline aready passed today."""
218        fake_now = self.BaseTime() + datetime.timedelta(hours=0.5)
219        timed_event.TimedEvent._now().MultipleTimes().AndReturn(fake_now)
220        self.mox.ReplayAll()
221
222        nightly = self.CreateEvent()  # Deadline gets set for tomorrow night.
223        self.assertFalse(nightly.ShouldHandle())
224        self.mox.VerifyAll()
225
226        self.mox.ResetAll()
227        fake_now += datetime.timedelta(days=1)  # Jump to tomorrow night.
228        timed_event.TimedEvent._now().MultipleTimes().AndReturn(fake_now)
229        self.mox.ReplayAll()
230        self.assertTrue(nightly.ShouldHandle())
231
232
233    def TimeBefore(self, now):
234        return now - datetime.timedelta(minutes=1)
235
236
237    def TimeLaterThan(self, now):
238        return now + datetime.timedelta(hours=0.5)
239
240
241    def testDeadlineInFuture(self):
242        """Ensure we work if the deadline is later today."""
243        self.doTestDeadlineInFuture()
244
245
246    def testDeadlineIsNow(self):
247        """We happened to create the trigger at the exact right time."""
248        self.doTestDeadlineIsNow()
249
250
251    def testTOCTOU(self):
252        """Even if deadline passes during initialization, trigger must fire."""
253        self.doTestTOCTOU()
254
255
256    def testDeadlineUpdate(self):
257        """Ensure we update the deadline correctly."""
258        self.doTestDeadlineUpdate(days_to_jump=0, hours_to_jump=1)
259
260
261    def testGetBranchBuilds(self):
262        """Ensure Nightly gets most recent builds in last day."""
263        self.doTestGetBranchBuilds(days=1)
264
265
266    def testFilterTasks(self):
267        """Test FilterTasks function can filter tasks by current hour."""
268        Task = collections.namedtuple('Task', 'hour')
269        task_1 = Task(hour=0)
270        task_2 = Task(hour=10)
271        task_3 = Task(hour=11)
272        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
273        self.mox.ReplayAll()
274        event = self.CreateEvent()
275        event.tasks = set([task_1, task_2, task_3])
276        self.assertEquals([task_1], event.FilterTasks())
277
278
279class WeeklyTest(TimedEventTestBase):
280    """Unit tests for Weekly.
281
282    @var _HOUR: The time of night to use in these unit tests.
283    """
284
285    _HOUR = 22
286
287
288    def setUp(self):
289        super(WeeklyTest, self).setUp()
290
291
292    def BaseTime(self):
293        basetime = datetime.datetime(2012, 1, 1, self._HOUR)
294        return basetime
295
296
297    def CreateEvent(self):
298        """Return an instance of timed_event.Weekly."""
299        return timed_event.Weekly(self.mv, False, self._HOUR)
300
301
302    def testCreateFromConfig(self):
303        """Test that creating from config is equivalent to using constructor."""
304        config = forgiving_config_parser.ForgivingConfigParser()
305        section = base_event.SectionName(timed_event.Weekly.KEYWORD)
306        config.add_section(section)
307        config.set(section, 'hour', '%d' % self._HOUR)
308
309        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
310        self.mox.ReplayAll()
311
312        self.assertEquals(self.CreateEvent(),
313                          timed_event.Weekly.CreateFromConfig(config, self.mv))
314
315
316    def testMergeDueToTimeChange(self):
317        """Test that Merge() works when the deadline time of day changes."""
318        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
319        self.mox.ReplayAll()
320
321        old = timed_event.Weekly(self.mv, False, self._HOUR)
322        new = timed_event.Weekly(self.mv, False, self._HOUR + 1)
323        self.assertNotEquals(old._deadline, new._deadline)
324        old.Merge(new)
325        self.assertEquals(old._deadline, new._deadline)
326
327
328    def testSkipMerge(self):
329        """Test that deadline is unchanged when only the week is changed."""
330        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
331        self.mox.ReplayAll()
332
333        old = timed_event.Weekly(self.mv, False, self._HOUR)
334        new = timed_event.Weekly(self.mv, False, self._HOUR)
335        new._deadline += datetime.timedelta(days=7)
336        self.assertNotEquals(old._deadline, new._deadline)
337        saved_deadline = old._deadline
338        old.Merge(new)
339        self.assertEquals(saved_deadline, old._deadline)
340
341
342    def testDeadlineInPast(self):
343        """Ensure we work if the deadline already passed this week."""
344        fake_now = self.BaseTime() + datetime.timedelta(days=0.5)
345        timed_event.TimedEvent._now().MultipleTimes().AndReturn(fake_now)
346        self.mox.ReplayAll()
347
348        weekly = self.CreateEvent()  # Deadline gets set for next week.
349        self.assertFalse(weekly.ShouldHandle())
350        self.mox.VerifyAll()
351
352        self.mox.ResetAll()
353        fake_now += datetime.timedelta(days=1)  # Jump to tomorrow.
354        timed_event.TimedEvent._now().MultipleTimes().AndReturn(fake_now)
355        self.mox.ReplayAll()
356        self.assertTrue(weekly.ShouldHandle())
357        self.mox.VerifyAll()
358
359        self.mox.ResetAll()
360        fake_now += datetime.timedelta(days=7)  # Jump to next week.
361        timed_event.TimedEvent._now().MultipleTimes().AndReturn(fake_now)
362        self.mox.ReplayAll()
363        self.assertTrue(weekly.ShouldHandle())
364
365
366    def TimeBefore(self, now):
367        return now - datetime.timedelta(days=0.5)
368
369
370    def TimeLaterThan(self, now):
371        return now + datetime.timedelta(days=0.5)
372
373
374    def testDeadlineInFuture(self):
375        """Ensure we work if the deadline is later this week."""
376        self.doTestDeadlineInFuture()
377
378
379    def testDeadlineIsNow(self):
380        """We happened to create the trigger at the exact right time."""
381        self.doTestDeadlineIsNow()
382
383
384    def testTOCTOU(self):
385        """Even if deadline passes during initialization, trigger must fire."""
386        self.doTestTOCTOU()
387
388
389    def testDeadlineUpdate(self):
390        """Ensure we update the deadline correctly."""
391        self.doTestDeadlineUpdate(days_to_jump=1)
392
393
394    def testGetBranchBuilds(self):
395        """Ensure Weekly gets most recent builds in last 7 days."""
396        self.doTestGetBranchBuilds(days=7)
397
398
399    def testFilterTasks(self):
400        """Test FilterTasks function can filter tasks by current day."""
401        Task = collections.namedtuple('Task', 'day')
402        task_1 = Task(day=6)
403        task_2 = Task(day=2)
404        task_3 = Task(day=5)
405        timed_event.TimedEvent._now().MultipleTimes().AndReturn(self.BaseTime())
406        self.mox.ReplayAll()
407        event = self.CreateEvent()
408        event.tasks = set([task_1, task_2, task_3])
409        self.assertEquals([task_1], event.FilterTasks())
410
411
412if __name__ == '__main__':
413  unittest.main()
414