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 9 import collections, datetime, mox, unittest 10 11 # driver must be imported first due to circular imports in base_event and task 12 import driver # pylint: disable-msg=W0611 13 import base_event, forgiving_config_parser 14 import manifest_versions, task, timed_event 15 16 17 class 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 134 class 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 279 class 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 412 if __name__ == '__main__': 413 unittest.main() 414
