1import pickle 2import random 3import sys 4from datetime import datetime, timedelta, date 5 6import pytest 7import pytz 8 9from apscheduler.triggers.base import BaseTrigger 10from apscheduler.triggers.cron import CronTrigger 11from apscheduler.triggers.date import DateTrigger 12from apscheduler.triggers.interval import IntervalTrigger 13from apscheduler.triggers.combining import AndTrigger, OrTrigger, BaseCombiningTrigger 14 15try: 16 from unittest.mock import Mock 17except ImportError: 18 from mock import Mock 19 20 21class _DummyTriggerWithJitter(BaseTrigger): 22 def __init__(self, dt, jitter): 23 self.dt = dt 24 self.jitter = jitter 25 26 def get_next_fire_time(self, previous_fire_time, now): 27 return self._apply_jitter(self.dt, self.jitter, now) 28 29 30class TestJitter(object): 31 def test_jitter_disabled(self): 32 dt = datetime(2017, 5, 25, 14, 49, 50) 33 trigger = _DummyTriggerWithJitter(dt, None) 34 35 now = datetime(2017, 5, 25, 13, 40, 44) 36 assert trigger.get_next_fire_time(None, now) == dt 37 38 def test_jitter_with_none_next_fire_time(self): 39 trigger = _DummyTriggerWithJitter(None, 5) 40 now = datetime(2017, 5, 25, 13, 40, 44) 41 assert trigger.get_next_fire_time(None, now) is None 42 43 def test_jitter_positive(self, monkeypatch): 44 monkeypatch.setattr(random, 'uniform', lambda a, b: 30.) 45 46 now = datetime(2017, 5, 25, 13, 40, 44) 47 dt = datetime(2017, 5, 25, 14, 49, 50) 48 expected_dt = datetime(2017, 5, 25, 14, 50, 20) 49 50 trigger = _DummyTriggerWithJitter(dt, 60) 51 assert trigger.get_next_fire_time(None, now) == expected_dt 52 53 def test_jitter_in_future_but_initial_date_in_past(self, monkeypatch): 54 monkeypatch.setattr(random, 'uniform', lambda a, b: 30.) 55 56 now = datetime(2017, 5, 25, 13, 40, 44) 57 dt = datetime(2017, 5, 25, 13, 40, 30) 58 expected_dt = datetime(2017, 5, 25, 13, 41, 0) 59 60 trigger = _DummyTriggerWithJitter(dt, 60) 61 assert trigger.get_next_fire_time(None, now) == expected_dt 62 63 def test_jitter_is_now(self, monkeypatch): 64 monkeypatch.setattr(random, 'uniform', lambda a, b: 4.) 65 66 now = datetime(2017, 5, 25, 13, 40, 44) 67 dt = datetime(2017, 5, 25, 13, 40, 40) 68 expected_dt = now 69 70 trigger = _DummyTriggerWithJitter(dt, 60) 71 assert trigger.get_next_fire_time(None, now) == expected_dt 72 73 def test_jitter(self): 74 now = datetime(2017, 5, 25, 13, 36, 44) 75 dt = datetime(2017, 5, 25, 13, 40, 45) 76 min_expected_dt = datetime(2017, 5, 25, 13, 40, 40) 77 max_expected_dt = datetime(2017, 5, 25, 13, 40, 50) 78 79 trigger = _DummyTriggerWithJitter(dt, 5) 80 for _ in range(0, 100): 81 assert min_expected_dt <= trigger.get_next_fire_time(None, now) <= max_expected_dt 82 83 84class TestCronTrigger(object): 85 def test_cron_trigger_1(self, timezone): 86 trigger = CronTrigger(year='2009/2', month='1/3', day='5-13', timezone=timezone) 87 assert repr(trigger) == ("<CronTrigger (year='2009/2', month='1/3', day='5-13', " 88 "timezone='Europe/Berlin')>") 89 assert str(trigger) == "cron[year='2009/2', month='1/3', day='5-13']" 90 start_date = timezone.localize(datetime(2008, 12, 1)) 91 correct_next_date = timezone.localize(datetime(2009, 1, 5)) 92 assert trigger.get_next_fire_time(None, start_date) == correct_next_date 93 94 def test_cron_trigger_2(self, timezone): 95 trigger = CronTrigger(year='2009/2', month='1/3', day='5-13', timezone=timezone) 96 start_date = timezone.localize(datetime(2009, 10, 14)) 97 correct_next_date = timezone.localize(datetime(2011, 1, 5)) 98 assert trigger.get_next_fire_time(None, start_date) == correct_next_date 99 100 def test_cron_trigger_3(self, timezone): 101 trigger = CronTrigger(year='2009', month='feb-dec', hour='8-10', timezone=timezone) 102 assert repr(trigger) == ("<CronTrigger (year='2009', month='feb-dec', hour='8-10', " 103 "timezone='Europe/Berlin')>") 104 start_date = timezone.localize(datetime(2009, 1, 1)) 105 correct_next_date = timezone.localize(datetime(2009, 2, 1, 8)) 106 assert trigger.get_next_fire_time(None, start_date) == correct_next_date 107 108 def test_cron_trigger_4(self, timezone): 109 trigger = CronTrigger(year='2012', month='2', day='last', timezone=timezone) 110 assert repr(trigger) == ("<CronTrigger (year='2012', month='2', day='last', " 111 "timezone='Europe/Berlin')>") 112 start_date = timezone.localize(datetime(2012, 2, 1)) 113 correct_next_date = timezone.localize(datetime(2012, 2, 29)) 114 assert trigger.get_next_fire_time(None, start_date) == correct_next_date 115 116 def test_start_end_times_string(self, timezone, monkeypatch): 117 monkeypatch.setattr('apscheduler.triggers.cron.get_localzone', Mock(return_value=timezone)) 118 trigger = CronTrigger(start_date='2016-11-05 05:06:53', end_date='2017-11-05 05:11:32') 119 assert trigger.start_date == timezone.localize(datetime(2016, 11, 5, 5, 6, 53)) 120 assert trigger.end_date == timezone.localize(datetime(2017, 11, 5, 5, 11, 32)) 121 122 def test_cron_zero_value(self, timezone): 123 trigger = CronTrigger(year=2009, month=2, hour=0, timezone=timezone) 124 assert repr(trigger) == ("<CronTrigger (year='2009', month='2', hour='0', " 125 "timezone='Europe/Berlin')>") 126 127 def test_cron_year_list(self, timezone): 128 trigger = CronTrigger(year='2009,2008', timezone=timezone) 129 assert repr(trigger) == "<CronTrigger (year='2009,2008', timezone='Europe/Berlin')>" 130 assert str(trigger) == "cron[year='2009,2008']" 131 start_date = timezone.localize(datetime(2009, 1, 1)) 132 correct_next_date = timezone.localize(datetime(2009, 1, 1)) 133 assert trigger.get_next_fire_time(None, start_date) == correct_next_date 134 135 def test_cron_start_date(self, timezone): 136 trigger = CronTrigger(year='2009', month='2', hour='8-10', 137 start_date='2009-02-03 11:00:00', timezone=timezone) 138 assert repr(trigger) == ("<CronTrigger (year='2009', month='2', hour='8-10', " 139 "start_date='2009-02-03 11:00:00 CET', " 140 "timezone='Europe/Berlin')>") 141 assert str(trigger) == "cron[year='2009', month='2', hour='8-10']" 142 start_date = timezone.localize(datetime(2009, 1, 1)) 143 correct_next_date = timezone.localize(datetime(2009, 2, 4, 8)) 144 assert trigger.get_next_fire_time(None, start_date) == correct_next_date 145 146 def test_previous_fire_time_1(self, timezone): 147 """Test for previous_fire_time arg in get_next_fire_time()""" 148 trigger = CronTrigger(day="*", timezone=timezone) 149 previous_fire_time = timezone.localize(datetime(2015, 11, 23)) 150 now = timezone.localize(datetime(2015, 11, 26)) 151 correct_next_date = timezone.localize(datetime(2015, 11, 24)) 152 assert trigger.get_next_fire_time(previous_fire_time, now) == correct_next_date 153 154 def test_previous_fire_time_2(self, timezone): 155 trigger = CronTrigger(day="*", timezone=timezone) 156 previous_fire_time = timezone.localize(datetime(2015, 11, 23)) 157 now = timezone.localize(datetime(2015, 11, 22)) 158 correct_next_date = timezone.localize(datetime(2015, 11, 22)) 159 assert trigger.get_next_fire_time(previous_fire_time, now) == correct_next_date 160 161 def test_previous_fire_time_3(self, timezone): 162 trigger = CronTrigger(day="*", timezone=timezone) 163 previous_fire_time = timezone.localize(datetime(2016, 4, 25)) 164 now = timezone.localize(datetime(2016, 4, 25)) 165 correct_next_date = timezone.localize(datetime(2016, 4, 26)) 166 assert trigger.get_next_fire_time(previous_fire_time, now) == correct_next_date 167 168 def test_cron_weekday_overlap(self, timezone): 169 trigger = CronTrigger(year=2009, month=1, day='6-10', day_of_week='2-4', timezone=timezone) 170 assert repr(trigger) == ("<CronTrigger (year='2009', month='1', day='6-10', " 171 "day_of_week='2-4', timezone='Europe/Berlin')>") 172 assert str(trigger) == "cron[year='2009', month='1', day='6-10', day_of_week='2-4']" 173 start_date = timezone.localize(datetime(2009, 1, 1)) 174 correct_next_date = timezone.localize(datetime(2009, 1, 7)) 175 assert trigger.get_next_fire_time(None, start_date) == correct_next_date 176 177 def test_cron_weekday_nomatch(self, timezone): 178 trigger = CronTrigger(year=2009, month=1, day='6-10', day_of_week='0,6', timezone=timezone) 179 assert repr(trigger) == ("<CronTrigger (year='2009', month='1', day='6-10', " 180 "day_of_week='0,6', timezone='Europe/Berlin')>") 181 assert str(trigger) == "cron[year='2009', month='1', day='6-10', day_of_week='0,6']" 182 start_date = timezone.localize(datetime(2009, 1, 1)) 183 correct_next_date = None 184 assert trigger.get_next_fire_time(None, start_date) == correct_next_date 185 186 def test_cron_weekday_positional(self, timezone): 187 trigger = CronTrigger(year=2009, month=1, day='4th wed', timezone=timezone) 188 assert repr(trigger) == ("<CronTrigger (year='2009', month='1', day='4th wed', " 189 "timezone='Europe/Berlin')>") 190 assert str(trigger) == "cron[year='2009', month='1', day='4th wed']" 191 start_date = timezone.localize(datetime(2009, 1, 1)) 192 correct_next_date = timezone.localize(datetime(2009, 1, 28)) 193 assert trigger.get_next_fire_time(None, start_date) == correct_next_date 194 195 def test_week_1(self, timezone): 196 trigger = CronTrigger(year=2009, month=2, week=8, timezone=timezone) 197 assert repr(trigger) == ("<CronTrigger (year='2009', month='2', week='8', " 198 "timezone='Europe/Berlin')>") 199 assert str(trigger) == "cron[year='2009', month='2', week='8']" 200 start_date = timezone.localize(datetime(2009, 1, 1)) 201 correct_next_date = timezone.localize(datetime(2009, 2, 16)) 202 assert trigger.get_next_fire_time(None, start_date) == correct_next_date 203 204 def test_week_2(self, timezone): 205 trigger = CronTrigger(year=2009, week=15, day_of_week=2, timezone=timezone) 206 assert repr(trigger) == ("<CronTrigger (year='2009', week='15', day_of_week='2', " 207 "timezone='Europe/Berlin')>") 208 assert str(trigger) == "cron[year='2009', week='15', day_of_week='2']" 209 start_date = timezone.localize(datetime(2009, 1, 1)) 210 correct_next_date = timezone.localize(datetime(2009, 4, 8)) 211 assert trigger.get_next_fire_time(None, start_date) == correct_next_date 212 213 def test_cron_extra_coverage(self, timezone): 214 # This test has no value other than patching holes in test coverage 215 trigger = CronTrigger(day='6,8', timezone=timezone) 216 assert repr(trigger) == "<CronTrigger (day='6,8', timezone='Europe/Berlin')>" 217 assert str(trigger) == "cron[day='6,8']" 218 start_date = timezone.localize(datetime(2009, 12, 31)) 219 correct_next_date = timezone.localize(datetime(2010, 1, 6)) 220 assert trigger.get_next_fire_time(None, start_date) == correct_next_date 221 222 def test_cron_faulty_expr(self, timezone): 223 pytest.raises(ValueError, CronTrigger, year='2009-fault', timezone=timezone) 224 225 def test_cron_increment_weekday(self, timezone): 226 """ 227 Tests that incrementing the weekday field in the process of calculating the next matching 228 date won't cause problems. 229 230 """ 231 trigger = CronTrigger(hour='5-6', timezone=timezone) 232 assert repr(trigger) == "<CronTrigger (hour='5-6', timezone='Europe/Berlin')>" 233 assert str(trigger) == "cron[hour='5-6']" 234 start_date = timezone.localize(datetime(2009, 9, 25, 7)) 235 correct_next_date = timezone.localize(datetime(2009, 9, 26, 5)) 236 assert trigger.get_next_fire_time(None, start_date) == correct_next_date 237 238 def test_cron_bad_kwarg(self, timezone): 239 pytest.raises(TypeError, CronTrigger, second=0, third=1, timezone=timezone) 240 241 def test_month_rollover(self, timezone): 242 trigger = CronTrigger(timezone=timezone, day=30) 243 now = timezone.localize(datetime(2016, 2, 1)) 244 expected = timezone.localize(datetime(2016, 3, 30)) 245 assert trigger.get_next_fire_time(None, now) == expected 246 247 def test_timezone_from_start_date(self, timezone): 248 """ 249 Tests that the trigger takes the timezone from the start_date parameter if no timezone is 250 supplied. 251 252 """ 253 start_date = timezone.localize(datetime(2014, 4, 13, 5, 30)) 254 trigger = CronTrigger(year=2014, hour=4, start_date=start_date) 255 assert trigger.timezone == start_date.tzinfo 256 257 def test_end_date(self, timezone): 258 end_date = timezone.localize(datetime(2014, 4, 13, 3)) 259 trigger = CronTrigger(year=2014, hour=4, end_date=end_date) 260 261 start_date = timezone.localize(datetime(2014, 4, 13, 2, 30)) 262 assert trigger.get_next_fire_time(None, start_date - timedelta(1)) == \ 263 start_date.replace(day=12, hour=4, minute=0) 264 assert trigger.get_next_fire_time(None, start_date) is None 265 266 def test_different_tz(self, timezone): 267 alter_tz = pytz.FixedOffset(-600) 268 trigger = CronTrigger(year=2009, week=15, day_of_week=2, timezone=timezone) 269 assert repr(trigger) == ("<CronTrigger (year='2009', week='15', day_of_week='2', " 270 "timezone='Europe/Berlin')>") 271 assert str(trigger) == "cron[year='2009', week='15', day_of_week='2']" 272 start_date = alter_tz.localize(datetime(2008, 12, 31, 22)) 273 correct_next_date = timezone.localize(datetime(2009, 4, 8)) 274 assert trigger.get_next_fire_time(None, start_date) == correct_next_date 275 276 @pytest.mark.parametrize('trigger_args, start_date, start_date_dst, correct_next_date', [ 277 ({'hour': 8}, datetime(2013, 3, 9, 12), False, datetime(2013, 3, 10, 8)), 278 ({'hour': 8}, datetime(2013, 11, 2, 12), True, datetime(2013, 11, 3, 8)), 279 ({'minute': '*/30'}, datetime(2013, 3, 10, 1, 35), False, datetime(2013, 3, 10, 3)), 280 ({'minute': '*/30'}, datetime(2013, 11, 3, 1, 35), True, datetime(2013, 11, 3, 1)) 281 ], ids=['absolute_spring', 'absolute_autumn', 'interval_spring', 'interval_autumn']) 282 def test_dst_change(self, trigger_args, start_date, start_date_dst, correct_next_date): 283 """ 284 Making sure that CronTrigger works correctly when crossing the DST switch threshold. 285 Note that you should explicitly compare datetimes as strings to avoid the internal datetime 286 comparison which would test for equality in the UTC timezone. 287 288 """ 289 timezone = pytz.timezone('US/Eastern') 290 trigger = CronTrigger(timezone=timezone, **trigger_args) 291 start_date = timezone.localize(start_date, is_dst=start_date_dst) 292 correct_next_date = timezone.localize(correct_next_date, is_dst=not start_date_dst) 293 assert str(trigger.get_next_fire_time(None, start_date)) == str(correct_next_date) 294 295 def test_timezone_change(self, timezone): 296 """ 297 Ensure that get_next_fire_time method returns datetimes in the timezone of the trigger and 298 not in the timezone of the passed in start_date. 299 300 """ 301 est = pytz.FixedOffset(-300) 302 cst = pytz.FixedOffset(-360) 303 trigger = CronTrigger(hour=11, minute='*/5', timezone=est) 304 start_date = cst.localize(datetime(2009, 9, 26, 10, 16)) 305 correct_next_date = est.localize(datetime(2009, 9, 26, 11, 20)) 306 assert str(trigger.get_next_fire_time(None, start_date)) == str(correct_next_date) 307 308 def test_pickle(self, timezone): 309 """Test that the trigger is pickleable.""" 310 311 trigger = CronTrigger(year=2016, month='5-6', day='20-28', hour=7, minute=25, second='*', 312 timezone=timezone) 313 data = pickle.dumps(trigger, 2) 314 trigger2 = pickle.loads(data) 315 316 for attr in CronTrigger.__slots__: 317 assert getattr(trigger2, attr) == getattr(trigger, attr) 318 319 def test_jitter_produces_differrent_valid_results(self, timezone): 320 trigger = CronTrigger(minute='*', jitter=5) 321 now = timezone.localize(datetime(2017, 11, 12, 6, 55, 30)) 322 323 results = set() 324 for _ in range(0, 100): 325 next_fire_time = trigger.get_next_fire_time(None, now) 326 results.add(next_fire_time) 327 assert timedelta(seconds=25) <= (next_fire_time - now) <= timedelta(seconds=35) 328 329 assert 1 < len(results) 330 331 def test_jitter_with_timezone(self, timezone): 332 est = pytz.FixedOffset(-300) 333 cst = pytz.FixedOffset(-360) 334 trigger = CronTrigger(hour=11, minute='*/5', timezone=est, jitter=5) 335 start_date = cst.localize(datetime(2009, 9, 26, 10, 16)) 336 correct_next_date = est.localize(datetime(2009, 9, 26, 11, 20)) 337 for _ in range(0, 100): 338 assert abs(trigger.get_next_fire_time(None, start_date) - 339 correct_next_date) <= timedelta(seconds=5) 340 341 @pytest.mark.parametrize('trigger_args, start_date, start_date_dst, correct_next_date', [ 342 ({'hour': 8}, datetime(2013, 3, 9, 12), False, datetime(2013, 3, 10, 8)), 343 ({'hour': 8}, datetime(2013, 11, 2, 12), True, datetime(2013, 11, 3, 8)), 344 ({'minute': '*/30'}, datetime(2013, 3, 10, 1, 35), False, datetime(2013, 3, 10, 3)), 345 ({'minute': '*/30'}, datetime(2013, 11, 3, 1, 35), True, datetime(2013, 11, 3, 1)) 346 ], ids=['absolute_spring', 'absolute_autumn', 'interval_spring', 'interval_autumn']) 347 def test_jitter_dst_change(self, trigger_args, start_date, start_date_dst, correct_next_date): 348 timezone = pytz.timezone('US/Eastern') 349 trigger = CronTrigger(timezone=timezone, jitter=5, **trigger_args) 350 start_date = timezone.localize(start_date, is_dst=start_date_dst) 351 correct_next_date = timezone.localize(correct_next_date, is_dst=not start_date_dst) 352 353 for _ in range(0, 100): 354 next_fire_time = trigger.get_next_fire_time(None, start_date) 355 assert abs(next_fire_time - correct_next_date) <= timedelta(seconds=5) 356 357 def test_jitter_with_end_date(self, timezone): 358 now = timezone.localize(datetime(2017, 11, 12, 6, 55, 30)) 359 end_date = timezone.localize(datetime(2017, 11, 12, 6, 56, 0)) 360 trigger = CronTrigger(minute='*', jitter=5, end_date=end_date) 361 362 for _ in range(0, 100): 363 next_fire_time = trigger.get_next_fire_time(None, now) 364 assert next_fire_time is None or next_fire_time <= end_date 365 366 @pytest.mark.parametrize('values, expected', [ 367 (dict(day='*/31'), r"Error validating expression '\*/31': the step value \(31\) is higher " 368 r"than the total range of the expression \(30\)"), 369 (dict(day='4-6/3'), r"Error validating expression '4-6/3': the step value \(3\) is higher " 370 r"than the total range of the expression \(2\)"), 371 (dict(hour='0-24'), r"Error validating expression '0-24': the last value \(24\) is higher " 372 r"than the maximum value \(23\)"), 373 (dict(day='0-3'), r"Error validating expression '0-3': the first value \(0\) is lower " 374 r"than the minimum value \(1\)") 375 ], ids=['too_large_step_all', 'too_large_step_range', 'too_high_last', 'too_low_first']) 376 def test_invalid_ranges(self, values, expected): 377 pytest.raises(ValueError, CronTrigger, **values).match(expected) 378 379 @pytest.mark.parametrize('expr, expected_repr', [ 380 ('* * * * *', 381 "<CronTrigger (month='*', day='*', day_of_week='*', hour='*', minute='*', " 382 "timezone='Europe/Berlin')>"), 383 ('0-14 * 14-28 jul fri', 384 "<CronTrigger (month='jul', day='14-28', day_of_week='fri', hour='*', minute='0-14', " 385 "timezone='Europe/Berlin')>"), 386 (' 0-14 * 14-28 jul fri', 387 "<CronTrigger (month='jul', day='14-28', day_of_week='fri', hour='*', minute='0-14', " 388 "timezone='Europe/Berlin')>") 389 ], ids=['always', 'assorted', 'multiple_spaces_in_format']) 390 def test_from_crontab(self, expr, expected_repr, timezone): 391 trigger = CronTrigger.from_crontab(expr, timezone) 392 assert repr(trigger) == expected_repr 393 394 395class TestDateTrigger(object): 396 @pytest.mark.parametrize('run_date,alter_tz,previous,now,expected', [ 397 (datetime(2009, 7, 6), None, None, datetime(2008, 5, 4), datetime(2009, 7, 6)), 398 (datetime(2009, 7, 6), None, None, datetime(2009, 7, 6), datetime(2009, 7, 6)), 399 (datetime(2009, 7, 6), None, None, datetime(2009, 9, 2), datetime(2009, 7, 6)), 400 ('2009-7-6', None, None, datetime(2009, 9, 2), datetime(2009, 7, 6)), 401 (datetime(2009, 7, 6), None, datetime(2009, 7, 6), datetime(2009, 9, 2), None), 402 (datetime(2009, 7, 5, 22), pytz.FixedOffset(-60), datetime(2009, 7, 6), 403 datetime(2009, 7, 6), None), 404 (None, pytz.FixedOffset(-120), None, datetime(2011, 4, 3, 18, 40), 405 datetime(2011, 4, 3, 18, 40)) 406 ], ids=['earlier', 'exact', 'later', 'as text', 'previously fired', 'alternate timezone', 407 'current_time']) 408 def test_get_next_fire_time(self, run_date, alter_tz, previous, now, expected, timezone, 409 freeze_time): 410 trigger = DateTrigger(run_date, alter_tz or timezone) 411 previous = timezone.localize(previous) if previous else None 412 now = timezone.localize(now) 413 expected = timezone.localize(expected) if expected else None 414 assert trigger.get_next_fire_time(previous, now) == expected 415 416 @pytest.mark.parametrize('is_dst', [True, False], ids=['daylight saving', 'standard time']) 417 def test_dst_change(self, is_dst): 418 """ 419 Test that DateTrigger works during the ambiguous "fall-back" DST period. 420 421 Note that you should explicitly compare datetimes as strings to avoid the internal datetime 422 comparison which would test for equality in the UTC timezone. 423 424 """ 425 eastern = pytz.timezone('US/Eastern') 426 run_date = eastern.localize(datetime(2013, 10, 3, 1, 5), is_dst=is_dst) 427 428 fire_date = eastern.normalize(run_date + timedelta(minutes=55)) 429 trigger = DateTrigger(run_date=fire_date, timezone=eastern) 430 assert str(trigger.get_next_fire_time(None, fire_date)) == str(fire_date) 431 432 def test_repr(self, timezone): 433 trigger = DateTrigger(datetime(2009, 7, 6), timezone) 434 assert repr(trigger) == "<DateTrigger (run_date='2009-07-06 00:00:00 CEST')>" 435 436 def test_str(self, timezone): 437 trigger = DateTrigger(datetime(2009, 7, 6), timezone) 438 assert str(trigger) == "date[2009-07-06 00:00:00 CEST]" 439 440 def test_pickle(self, timezone): 441 """Test that the trigger is pickleable.""" 442 trigger = DateTrigger(date(2016, 4, 3), timezone=timezone) 443 data = pickle.dumps(trigger, 2) 444 trigger2 = pickle.loads(data) 445 assert trigger2.run_date == trigger.run_date 446 447 448class TestIntervalTrigger(object): 449 @pytest.fixture() 450 def trigger(self, timezone): 451 return IntervalTrigger(seconds=1, start_date=datetime(2009, 8, 4, second=2), 452 timezone=timezone) 453 454 def test_invalid_interval(self, timezone): 455 pytest.raises(TypeError, IntervalTrigger, '1-6', timezone=timezone) 456 457 def test_start_end_times_string(self, timezone, monkeypatch): 458 monkeypatch.setattr('apscheduler.triggers.interval.get_localzone', 459 Mock(return_value=timezone)) 460 trigger = IntervalTrigger(start_date='2016-11-05 05:06:53', end_date='2017-11-05 05:11:32') 461 assert trigger.start_date == timezone.localize(datetime(2016, 11, 5, 5, 6, 53)) 462 assert trigger.end_date == timezone.localize(datetime(2017, 11, 5, 5, 11, 32)) 463 464 def test_before(self, trigger, timezone): 465 """Tests that if "start_date" is later than "now", it will return start_date.""" 466 now = trigger.start_date - timedelta(seconds=2) 467 assert trigger.get_next_fire_time(None, now) == trigger.start_date 468 469 def test_within(self, trigger, timezone): 470 """ 471 Tests that if "now" is between "start_date" and the next interval, it will return the next 472 interval. 473 474 """ 475 now = trigger.start_date + timedelta(microseconds=1000) 476 assert trigger.get_next_fire_time(None, now) == trigger.start_date + trigger.interval 477 478 def test_no_start_date(self, timezone): 479 trigger = IntervalTrigger(seconds=2, timezone=timezone) 480 now = datetime.now(timezone) 481 assert (trigger.get_next_fire_time(None, now) - now) <= timedelta(seconds=2) 482 483 def test_different_tz(self, trigger, timezone): 484 alter_tz = pytz.FixedOffset(-60) 485 start_date = alter_tz.localize(datetime(2009, 8, 3, 22, second=2, microsecond=1000)) 486 correct_next_date = timezone.localize(datetime(2009, 8, 4, 1, second=3)) 487 assert trigger.get_next_fire_time(None, start_date) == correct_next_date 488 489 def test_end_date(self, timezone): 490 """Tests that the interval trigger won't return any datetimes past the set end time.""" 491 start_date = timezone.localize(datetime(2014, 5, 26)) 492 trigger = IntervalTrigger(minutes=5, start_date=start_date, 493 end_date=datetime(2014, 5, 26, 0, 7), timezone=timezone) 494 assert trigger.get_next_fire_time(None, start_date + timedelta(minutes=2)) == \ 495 start_date.replace(minute=5) 496 assert trigger.get_next_fire_time(None, start_date + timedelta(minutes=6)) is None 497 498 def test_dst_change(self): 499 """ 500 Making sure that IntervalTrigger works during the ambiguous "fall-back" DST period. 501 Note that you should explicitly compare datetimes as strings to avoid the internal datetime 502 comparison which would test for equality in the UTC timezone. 503 504 """ 505 eastern = pytz.timezone('US/Eastern') 506 start_date = datetime(2013, 3, 1) # Start within EDT 507 trigger = IntervalTrigger(hours=1, start_date=start_date, timezone=eastern) 508 509 datetime_edt = eastern.localize(datetime(2013, 3, 10, 1, 5), is_dst=False) 510 correct_next_date = eastern.localize(datetime(2013, 3, 10, 3), is_dst=True) 511 assert str(trigger.get_next_fire_time(None, datetime_edt)) == str(correct_next_date) 512 513 datetime_est = eastern.localize(datetime(2013, 11, 3, 1, 5), is_dst=True) 514 correct_next_date = eastern.localize(datetime(2013, 11, 3, 1), is_dst=False) 515 assert str(trigger.get_next_fire_time(None, datetime_est)) == str(correct_next_date) 516 517 def test_space_in_expr(self, timezone): 518 trigger = CronTrigger(day='1-2, 4-7', timezone=timezone) 519 assert repr(trigger) == "<CronTrigger (day='1-2,4-7', timezone='Europe/Berlin')>" 520 521 def test_repr(self, trigger): 522 if sys.version_info[:2] < (3, 7): 523 timedelta_args = '0, 1' 524 else: 525 timedelta_args = 'seconds=1' 526 527 assert repr(trigger) == ("<IntervalTrigger (interval=datetime.timedelta({}), " 528 "start_date='2009-08-04 00:00:02 CEST', " 529 "timezone='Europe/Berlin')>".format(timedelta_args)) 530 531 def test_str(self, trigger): 532 assert str(trigger) == "interval[0:00:01]" 533 534 def test_pickle(self, timezone): 535 """Test that the trigger is pickleable.""" 536 537 trigger = IntervalTrigger(weeks=2, days=6, minutes=13, seconds=2, 538 start_date=date(2016, 4, 3), timezone=timezone, 539 jitter=12) 540 data = pickle.dumps(trigger, 2) 541 trigger2 = pickle.loads(data) 542 543 for attr in IntervalTrigger.__slots__: 544 assert getattr(trigger2, attr) == getattr(trigger, attr) 545 546 def test_jitter_produces_different_valid_results(self, timezone): 547 trigger = IntervalTrigger(seconds=5, timezone=timezone, jitter=3) 548 now = datetime.now(timezone) 549 550 results = set() 551 for _ in range(0, 100): 552 next_fire_time = trigger.get_next_fire_time(None, now) 553 results.add(next_fire_time) 554 assert timedelta(seconds=2) <= (next_fire_time - now) <= timedelta(seconds=8) 555 assert 1 < len(results) 556 557 @pytest.mark.parametrize('trigger_args, start_date, start_date_dst, correct_next_date', [ 558 ({'hours': 1}, datetime(2013, 3, 10, 1, 35), False, datetime(2013, 3, 10, 3, 35)), 559 ({'hours': 1}, datetime(2013, 11, 3, 1, 35), True, datetime(2013, 11, 3, 1, 35)) 560 ], ids=['interval_spring', 'interval_autumn']) 561 def test_jitter_dst_change(self, trigger_args, start_date, start_date_dst, correct_next_date): 562 timezone = pytz.timezone('US/Eastern') 563 epsilon = timedelta(seconds=1) 564 start_date = timezone.localize(start_date, is_dst=start_date_dst) 565 trigger = IntervalTrigger(timezone=timezone, start_date=start_date, jitter=5, 566 **trigger_args) 567 correct_next_date = timezone.localize(correct_next_date, is_dst=not start_date_dst) 568 569 for _ in range(0, 100): 570 next_fire_time = trigger.get_next_fire_time(None, start_date + epsilon) 571 assert abs(next_fire_time - correct_next_date) <= timedelta(seconds=5) 572 573 def test_jitter_with_end_date(self, timezone): 574 now = timezone.localize(datetime(2017, 11, 12, 6, 55, 58)) 575 end_date = timezone.localize(datetime(2017, 11, 12, 6, 56, 0)) 576 trigger = IntervalTrigger(seconds=5, jitter=5, end_date=end_date) 577 578 for _ in range(0, 100): 579 next_fire_time = trigger.get_next_fire_time(None, now) 580 assert next_fire_time is None or next_fire_time <= end_date 581 582 583class TestAndTrigger(object): 584 @pytest.fixture 585 def trigger(self, timezone): 586 return AndTrigger([ 587 CronTrigger(month='5-8', day='6-15', 588 end_date=timezone.localize(datetime(2017, 8, 10))), 589 CronTrigger(month='6-9', day='*/3', end_date=timezone.localize(datetime(2017, 9, 7))) 590 ]) 591 592 @pytest.mark.parametrize('start_time, expected', [ 593 (datetime(2017, 8, 6), datetime(2017, 8, 7)), 594 (datetime(2017, 8, 10, 1), None) 595 ], ids=['firstmatch', 'end']) 596 def test_next_fire_time(self, trigger, timezone, start_time, expected): 597 expected = timezone.localize(expected) if expected else None 598 assert trigger.get_next_fire_time(None, timezone.localize(start_time)) == expected 599 600 def test_jitter(self, trigger, timezone): 601 trigger.jitter = 5 602 start_time = timezone.localize(datetime(2017, 8, 6)) 603 expected = timezone.localize(datetime(2017, 8, 7)) 604 for _ in range(100): 605 next_fire_time = trigger.get_next_fire_time(None, start_time) 606 assert abs(expected - next_fire_time) <= timedelta(seconds=5) 607 608 @pytest.mark.parametrize('jitter', [None, 5], ids=['nojitter', 'jitter']) 609 def test_repr(self, trigger, jitter): 610 trigger.jitter = jitter 611 jitter_part = ', jitter={}'.format(jitter) if jitter else '' 612 assert repr(trigger) == ( 613 "<AndTrigger([<CronTrigger (month='5-8', day='6-15', " 614 "end_date='2017-08-10 00:00:00 CEST', timezone='Europe/Berlin')>, <CronTrigger " 615 "(month='6-9', day='*/3', end_date='2017-09-07 00:00:00 CEST', " 616 "timezone='Europe/Berlin')>]{})>".format(jitter_part)) 617 618 def test_str(self, trigger): 619 assert str(trigger) == "and[cron[month='5-8', day='6-15'], cron[month='6-9', day='*/3']]" 620 621 @pytest.mark.parametrize('jitter', [None, 5], ids=['nojitter', 'jitter']) 622 def test_pickle(self, trigger, jitter): 623 """Test that the trigger is pickleable.""" 624 trigger.jitter = jitter 625 data = pickle.dumps(trigger, 2) 626 trigger2 = pickle.loads(data) 627 628 for attr in BaseCombiningTrigger.__slots__: 629 assert repr(getattr(trigger2, attr)) == repr(getattr(trigger, attr)) 630 631 632class TestOrTrigger(object): 633 @pytest.fixture 634 def trigger(self, timezone): 635 return OrTrigger([ 636 CronTrigger(month='5-8', day='6-15', 637 end_date=timezone.localize(datetime(2017, 8, 10))), 638 CronTrigger(month='6-9', day='*/3', end_date=timezone.localize(datetime(2017, 9, 7))) 639 ]) 640 641 @pytest.mark.parametrize('start_time, expected', [ 642 (datetime(2017, 8, 6), datetime(2017, 8, 6)), 643 (datetime(2017, 9, 7, 1), None) 644 ], ids=['earliest', 'end']) 645 def test_next_fire_time(self, trigger, timezone, start_time, expected): 646 expected = timezone.localize(expected) if expected else None 647 assert trigger.get_next_fire_time(None, timezone.localize(start_time)) == expected 648 649 def test_jitter(self, trigger, timezone): 650 trigger.jitter = 5 651 start_time = expected = timezone.localize(datetime(2017, 8, 6)) 652 for _ in range(100): 653 next_fire_time = trigger.get_next_fire_time(None, start_time) 654 assert abs(expected - next_fire_time) <= timedelta(seconds=5) 655 656 @pytest.mark.parametrize('jitter', [None, 5], ids=['nojitter', 'jitter']) 657 def test_repr(self, trigger, jitter): 658 trigger.jitter = jitter 659 jitter_part = ', jitter={}'.format(jitter) if jitter else '' 660 assert repr(trigger) == ( 661 "<OrTrigger([<CronTrigger (month='5-8', day='6-15', " 662 "end_date='2017-08-10 00:00:00 CEST', timezone='Europe/Berlin')>, <CronTrigger " 663 "(month='6-9', day='*/3', end_date='2017-09-07 00:00:00 CEST', " 664 "timezone='Europe/Berlin')>]{})>".format(jitter_part)) 665 666 def test_str(self, trigger): 667 assert str(trigger) == "or[cron[month='5-8', day='6-15'], cron[month='6-9', day='*/3']]" 668 669 @pytest.mark.parametrize('jitter', [None, 5], ids=['nojitter', 'jitter']) 670 def test_pickle(self, trigger, jitter): 671 """Test that the trigger is pickleable.""" 672 trigger.jitter = jitter 673 data = pickle.dumps(trigger, 2) 674 trigger2 = pickle.loads(data) 675 676 for attr in BaseCombiningTrigger.__slots__: 677 assert repr(getattr(trigger2, attr)) == repr(getattr(trigger, attr)) 678