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