timed_event_unittest.py revision 5bde7dc7d8b0608feb43491d5ac648d11c890f54
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.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