import sys import time from contextlib import contextmanager from datetime import datetime, timedelta, timezone from pickle import dumps, loads from unittest import TestCase from unittest.mock import Mock import pytest from celery.schedules import ParseException, crontab, crontab_parser, schedule, solar if sys.version_info >= (3, 9): from zoneinfo import ZoneInfo else: from backports.zoneinfo import ZoneInfo assertions = TestCase('__init__') @contextmanager def patch_crontab_nowfun(cls, retval): prev_nowfun = cls.nowfun cls.nowfun = lambda: retval try: yield finally: cls.nowfun = prev_nowfun class test_solar: def setup_method(self): pytest.importorskip('ephem') self.s = solar('sunrise', 60, 30, app=self.app) def test_reduce(self): fun, args = self.s.__reduce__() assert fun(*args) == self.s def test_eq(self): assert self.s == solar('sunrise', 60, 30, app=self.app) assert self.s != solar('sunset', 60, 30, app=self.app) assert self.s != schedule(10) def test_repr(self): assert repr(self.s) def test_is_due(self): self.s.remaining_estimate = Mock(name='rem') self.s.remaining_estimate.return_value = timedelta(seconds=0) assert self.s.is_due(datetime.now(timezone.utc)).is_due def test_is_due__not_due(self): self.s.remaining_estimate = Mock(name='rem') self.s.remaining_estimate.return_value = timedelta(hours=10) assert not self.s.is_due(datetime.now(timezone.utc)).is_due def test_remaining_estimate(self): self.s.cal = Mock(name='cal') self.s.cal.next_rising().datetime.return_value = datetime.now(timezone.utc) self.s.remaining_estimate(datetime.now(timezone.utc)) def test_coordinates(self): with pytest.raises(ValueError): solar('sunrise', -120, 60, app=self.app) with pytest.raises(ValueError): solar('sunrise', 120, 60, app=self.app) with pytest.raises(ValueError): solar('sunrise', 60, -200, app=self.app) with pytest.raises(ValueError): solar('sunrise', 60, 200, app=self.app) def test_invalid_event(self): with pytest.raises(ValueError): solar('asdqwewqew', 60, 60, app=self.app) def test_event_uses_center(self): s = solar('solar_noon', 60, 60, app=self.app) for ev, is_center in s._use_center_l.items(): s.method = s._methods[ev] s.is_center = s._use_center_l[ev] try: s.remaining_estimate(datetime.now(timezone.utc)) except TypeError: pytest.fail( f"{s.method} was called with 'use_center' which is not a " "valid keyword for the function.") class test_schedule: def test_ne(self): s1 = schedule(10, app=self.app) s2 = schedule(12, app=self.app) s3 = schedule(10, app=self.app) assert s1 == s3 assert s1 != s2 def test_pickle(self): s1 = schedule(10, app=self.app) fun, args = s1.__reduce__() s2 = fun(*args) assert s1 == s2 # This is needed for test_crontab_parser because datetime.utcnow doesn't pickle # in python 2 def utcnow(): return datetime.now(timezone.utc) class test_crontab_parser: def crontab(self, *args, **kwargs): return crontab(*args, **dict(kwargs, app=self.app)) def test_crontab_reduce(self): c = self.crontab('*') assert c == loads(dumps(c)) c = self.crontab( minute='1', hour='2', day_of_week='3', day_of_month='4', month_of_year='5', nowfun=utcnow) assert c == loads(dumps(c)) def test_range_steps_not_enough(self): with pytest.raises(crontab_parser.ParseException): crontab_parser(24)._range_steps([1]) def test_parse_star(self): assert crontab_parser(24).parse('*') == set(range(24)) assert crontab_parser(60).parse('*') == set(range(60)) assert crontab_parser(7).parse('*') == set(range(7)) assert crontab_parser(31, 1).parse('*') == set(range(1, 31 + 1)) assert crontab_parser(12, 1).parse('*') == set(range(1, 12 + 1)) def test_parse_range(self): assert crontab_parser(60).parse('1-10') == set(range(1, 10 + 1)) assert crontab_parser(24).parse('0-20') == set(range(0, 20 + 1)) assert crontab_parser().parse('2-10') == set(range(2, 10 + 1)) assert crontab_parser(60, 1).parse('1-10') == set(range(1, 10 + 1)) def test_parse_range_wraps(self): assert crontab_parser(12).parse('11-1') == {11, 0, 1} assert crontab_parser(60, 1).parse('2-1') == set(range(1, 60 + 1)) def test_parse_groups(self): assert crontab_parser().parse('1,2,3,4') == {1, 2, 3, 4} assert crontab_parser().parse('0,15,30,45') == {0, 15, 30, 45} assert crontab_parser(min_=1).parse('1,2,3,4') == {1, 2, 3, 4} def test_parse_steps(self): assert crontab_parser(8).parse('*/2') == {0, 2, 4, 6} assert crontab_parser().parse('*/2') == {i * 2 for i in range(30)} assert crontab_parser().parse('*/3') == {i * 3 for i in range(20)} assert crontab_parser(8, 1).parse('*/2') == {1, 3, 5, 7} assert crontab_parser(min_=1).parse('*/2') == { i * 2 + 1 for i in range(30) } assert crontab_parser(min_=1).parse('*/3') == { i * 3 + 1 for i in range(20) } def test_parse_composite(self): assert crontab_parser(8).parse('*/2') == {0, 2, 4, 6} assert crontab_parser().parse('2-9/5') == {2, 7} assert crontab_parser().parse('2-10/5') == {2, 7} assert crontab_parser(min_=1).parse('55-5/3') == {55, 58, 1, 4} assert crontab_parser().parse('2-11/5,3') == {2, 3, 7} assert crontab_parser().parse('2-4/3,*/5,0-21/4') == { 0, 2, 4, 5, 8, 10, 12, 15, 16, 20, 25, 30, 35, 40, 45, 50, 55, } assert crontab_parser().parse('1-9/2') == {1, 3, 5, 7, 9} assert crontab_parser(8, 1).parse('*/2') == {1, 3, 5, 7} assert crontab_parser(min_=1).parse('2-9/5') == {2, 7} assert crontab_parser(min_=1).parse('2-10/5') == {2, 7} assert crontab_parser(min_=1).parse('2-11/5,3') == {2, 3, 7} assert crontab_parser(min_=1).parse('2-4/3,*/5,1-21/4') == { 1, 2, 5, 6, 9, 11, 13, 16, 17, 21, 26, 31, 36, 41, 46, 51, 56, } assert crontab_parser(min_=1).parse('1-9/2') == {1, 3, 5, 7, 9} def test_parse_errors_on_empty_string(self): with pytest.raises(ParseException): crontab_parser(60).parse('') def test_parse_errors_on_empty_group(self): with pytest.raises(ParseException): crontab_parser(60).parse('1,,2') def test_parse_errors_on_empty_steps(self): with pytest.raises(ParseException): crontab_parser(60).parse('*/') def test_parse_errors_on_negative_number(self): with pytest.raises(ParseException): crontab_parser(60).parse('-20') def test_parse_errors_on_lt_min(self): crontab_parser(min_=1).parse('1') with pytest.raises(ValueError): crontab_parser(12, 1).parse('0') with pytest.raises(ValueError): crontab_parser(24, 1).parse('12-0') def test_parse_errors_on_gt_max(self): crontab_parser(1).parse('0') with pytest.raises(ValueError): crontab_parser(1).parse('1') with pytest.raises(ValueError): crontab_parser(60).parse('61-0') def test_expand_cronspec_eats_iterables(self): assert crontab._expand_cronspec(iter([1, 2, 3]), 100) == {1, 2, 3} assert crontab._expand_cronspec(iter([1, 2, 3]), 100, 1) == {1, 2, 3} def test_expand_cronspec_invalid_type(self): with pytest.raises(TypeError): crontab._expand_cronspec(object(), 100) def test_repr(self): assert '*' in repr(self.crontab('*')) def test_eq(self): assert (self.crontab(day_of_week='1, 2') == self.crontab(day_of_week='1-2')) assert (self.crontab(day_of_month='1, 16, 31') == self.crontab(day_of_month='*/15')) assert ( self.crontab( minute='1', hour='2', day_of_week='5', day_of_month='10', month_of_year='5') == self.crontab( minute='1', hour='2', day_of_week='5', day_of_month='10', month_of_year='5')) assert crontab(minute='1') != crontab(minute='2') assert (self.crontab(month_of_year='1') != self.crontab(month_of_year='2')) assert object() != self.crontab(minute='1') assert self.crontab(minute='1') != object() assert crontab(month_of_year='1') != schedule(10) class test_crontab_from_string: def test_every_minute(self): assert crontab.from_string('* * * * *') == crontab() def test_every_minute_on_sunday(self): assert crontab.from_string('* * * * SUN') == crontab(day_of_week='SUN') def test_once_per_month(self): assert crontab.from_string('0 8 5 * *') == crontab(minute=0, hour=8, day_of_month=5) def test_invalid_crontab_string(self): with pytest.raises(ValueError): crontab.from_string('*') class test_crontab_remaining_estimate: def crontab(self, *args, **kwargs): return crontab(*args, **dict(kwargs, app=self.app)) def next_occurrence(self, crontab, now): crontab.nowfun = lambda: now return now + crontab.remaining_estimate(now) def test_next_minute(self): next = self.next_occurrence( self.crontab(), datetime(2010, 9, 11, 14, 30, 15), ) assert next == datetime(2010, 9, 11, 14, 31) def test_not_next_minute(self): next = self.next_occurrence( self.crontab(), datetime(2010, 9, 11, 14, 59, 15), ) assert next == datetime(2010, 9, 11, 15, 0) def test_this_hour(self): next = self.next_occurrence( self.crontab(minute=[5, 42]), datetime(2010, 9, 11, 14, 30, 15), ) assert next == datetime(2010, 9, 11, 14, 42) def test_not_this_hour(self): next = self.next_occurrence( self.crontab(minute=[5, 10, 15]), datetime(2010, 9, 11, 14, 30, 15), ) assert next == datetime(2010, 9, 11, 15, 5) def test_today(self): next = self.next_occurrence( self.crontab(minute=[5, 42], hour=[12, 17]), datetime(2010, 9, 11, 14, 30, 15), ) assert next == datetime(2010, 9, 11, 17, 5) def test_not_today(self): next = self.next_occurrence( self.crontab(minute=[5, 42], hour=[12]), datetime(2010, 9, 11, 14, 30, 15), ) assert next == datetime(2010, 9, 12, 12, 5) def test_weekday(self): next = self.next_occurrence( self.crontab(minute=30, hour=14, day_of_week='sat'), datetime(2010, 9, 11, 14, 30, 15), ) assert next == datetime(2010, 9, 18, 14, 30) def test_not_weekday(self): next = self.next_occurrence( self.crontab(minute=[5, 42], day_of_week='mon-fri'), datetime(2010, 9, 11, 14, 30, 15), ) assert next == datetime(2010, 9, 13, 0, 5) def test_monthyear(self): next = self.next_occurrence( self.crontab(minute=30, hour=14, month_of_year='oct', day_of_month=18), datetime(2010, 9, 11, 14, 30, 15), ) assert next == datetime(2010, 10, 18, 14, 30) def test_not_monthyear(self): next = self.next_occurrence( self.crontab(minute=[5, 42], month_of_year='nov-dec', day_of_month=13), datetime(2010, 9, 11, 14, 30, 15), ) assert next == datetime(2010, 11, 13, 0, 5) def test_monthday(self): next = self.next_occurrence( self.crontab(minute=30, hour=14, day_of_month=18), datetime(2010, 9, 11, 14, 30, 15), ) assert next == datetime(2010, 9, 18, 14, 30) def test_not_monthday(self): next = self.next_occurrence( self.crontab(minute=[5, 42], day_of_month=29), datetime(2010, 1, 22, 14, 30, 15), ) assert next == datetime(2010, 1, 29, 0, 5) def test_weekday_monthday(self): next = self.next_occurrence( self.crontab(minute=30, hour=14, day_of_week='mon', day_of_month=18), datetime(2010, 1, 18, 14, 30, 15), ) assert next == datetime(2010, 10, 18, 14, 30) def test_monthday_not_weekday(self): next = self.next_occurrence( self.crontab(minute=[5, 42], day_of_week='sat', day_of_month=29), datetime(2010, 1, 29, 0, 5, 15), ) assert next == datetime(2010, 5, 29, 0, 5) def test_weekday_not_monthday(self): next = self.next_occurrence( self.crontab(minute=[5, 42], day_of_week='mon', day_of_month=18), datetime(2010, 1, 11, 0, 5, 15), ) assert next == datetime(2010, 1, 18, 0, 5) def test_not_weekday_not_monthday(self): next = self.next_occurrence( self.crontab(minute=[5, 42], day_of_week='mon', day_of_month=18), datetime(2010, 1, 10, 0, 5, 15), ) assert next == datetime(2010, 1, 18, 0, 5) def test_leapday(self): next = self.next_occurrence( self.crontab(minute=30, hour=14, day_of_month=29), datetime(2012, 1, 29, 14, 30, 15), ) assert next == datetime(2012, 2, 29, 14, 30) def test_not_leapday(self): next = self.next_occurrence( self.crontab(minute=30, hour=14, day_of_month=29), datetime(2010, 1, 29, 14, 30, 15), ) assert next == datetime(2010, 3, 29, 14, 30) def test_weekmonthdayyear(self): next = self.next_occurrence( self.crontab(minute=30, hour=14, day_of_week='fri', day_of_month=29, month_of_year=1), datetime(2010, 1, 22, 14, 30, 15), ) assert next == datetime(2010, 1, 29, 14, 30) def test_monthdayyear_not_week(self): next = self.next_occurrence( self.crontab(minute=[5, 42], day_of_week='wed,thu', day_of_month=29, month_of_year='1,4,7'), datetime(2010, 1, 29, 14, 30, 15), ) assert next == datetime(2010, 4, 29, 0, 5) def test_weekdaymonthyear_not_monthday(self): next = self.next_occurrence( self.crontab(minute=30, hour=14, day_of_week='fri', day_of_month=29, month_of_year='1-10'), datetime(2010, 1, 29, 14, 30, 15), ) assert next == datetime(2010, 10, 29, 14, 30) def test_weekmonthday_not_monthyear(self): next = self.next_occurrence( self.crontab(minute=[5, 42], day_of_week='fri', day_of_month=29, month_of_year='2-10'), datetime(2010, 1, 29, 14, 30, 15), ) assert next == datetime(2010, 10, 29, 0, 5) def test_weekday_not_monthdayyear(self): next = self.next_occurrence( self.crontab(minute=[5, 42], day_of_week='mon', day_of_month=18, month_of_year='2-10'), datetime(2010, 1, 11, 0, 5, 15), ) assert next == datetime(2010, 10, 18, 0, 5) def test_monthday_not_weekdaymonthyear(self): next = self.next_occurrence( self.crontab(minute=[5, 42], day_of_week='mon', day_of_month=29, month_of_year='2-4'), datetime(2010, 1, 29, 0, 5, 15), ) assert next == datetime(2010, 3, 29, 0, 5) def test_monthyear_not_weekmonthday(self): next = self.next_occurrence( self.crontab(minute=[5, 42], day_of_week='mon', day_of_month=29, month_of_year='2-4'), datetime(2010, 2, 28, 0, 5, 15), ) assert next == datetime(2010, 3, 29, 0, 5) def test_not_weekmonthdayyear(self): next = self.next_occurrence( self.crontab(minute=[5, 42], day_of_week='fri,sat', day_of_month=29, month_of_year='2-10'), datetime(2010, 1, 28, 14, 30, 15), ) assert next == datetime(2010, 5, 29, 0, 5) def test_invalid_specification(self): # *** WARNING *** # This test triggers an infinite loop in case of a regression with pytest.raises(RuntimeError): self.next_occurrence( self.crontab(day_of_month=31, month_of_year=4), datetime(2010, 1, 28, 14, 30, 15), ) def test_leapyear(self): next = self.next_occurrence( self.crontab(minute=30, hour=14, day_of_month=29, month_of_year=2), datetime(2012, 2, 29, 14, 30), ) assert next == datetime(2016, 2, 29, 14, 30) def test_day_after_dst_end(self): # Test for #1604 issue with region configuration using DST tzname = "Europe/Paris" self.app.timezone = tzname tz = ZoneInfo(tzname) crontab = self.crontab(minute=0, hour=9) # Set last_run_at Before DST end last_run_at = datetime(2017, 10, 28, 9, 0, tzinfo=tz) # Set now after DST end now = datetime(2017, 10, 29, 7, 0, tzinfo=tz) crontab.nowfun = lambda: now next = now + crontab.remaining_estimate(last_run_at) assert next.utcoffset().seconds == 3600 assert next == datetime(2017, 10, 29, 9, 0, tzinfo=tz) def test_day_after_dst_start(self): # Test for #1604 issue with region configuration using DST tzname = "Europe/Paris" self.app.timezone = tzname tz = ZoneInfo(tzname) crontab = self.crontab(minute=0, hour=9) # Set last_run_at Before DST start last_run_at = datetime(2017, 3, 25, 9, 0, tzinfo=tz) # Set now after DST start now = datetime(2017, 3, 26, 7, 0, tzinfo=tz) crontab.nowfun = lambda: now next = now + crontab.remaining_estimate(last_run_at) assert next.utcoffset().seconds == 7200 assert next == datetime(2017, 3, 26, 9, 0, tzinfo=tz) def test_negative_utc_timezone_with_day_of_month(self): # UTC-8 tzname = "America/Los_Angeles" self.app.timezone = tzname tz = ZoneInfo(tzname) # set day_of_month to test on _delta_to_next crontab = self.crontab(minute=0, day_of_month='27-31') # last_run_at: '2023/01/28T23:00:00-08:00' last_run_at = datetime(2023, 1, 28, 23, 0, tzinfo=tz) # now: '2023/01/29T00:00:00-08:00' now = datetime(2023, 1, 29, 0, 0, tzinfo=tz) crontab.nowfun = lambda: now next = now + crontab.remaining_estimate(last_run_at) assert next == datetime(2023, 1, 29, 0, 0, tzinfo=tz) class test_crontab_is_due: def setup_method(self): self.now = self.app.now() self.next_minute = 60 - self.now.second - 1e-6 * self.now.microsecond self.every_minute = self.crontab() self.quarterly = self.crontab(minute='*/15') self.hourly = self.crontab(minute=30) self.daily = self.crontab(hour=7, minute=30) self.weekly = self.crontab(hour=7, minute=30, day_of_week='thursday') self.monthly = self.crontab( hour=7, minute=30, day_of_week='thursday', day_of_month='8-14', ) self.monthly_moy = self.crontab( hour=22, day_of_week='*', month_of_year='2', day_of_month='26,27,28', ) self.yearly = self.crontab( hour=7, minute=30, day_of_week='thursday', day_of_month='8-14', month_of_year=3, ) def crontab(self, *args, **kwargs): return crontab(*args, app=self.app, **kwargs) def test_default_crontab_spec(self): c = self.crontab() assert c.minute == set(range(60)) assert c.hour == set(range(24)) assert c.day_of_week == set(range(7)) assert c.day_of_month == set(range(1, 32)) assert c.month_of_year == set(range(1, 13)) def test_simple_crontab_spec(self): c = self.crontab(minute=30) assert c.minute == {30} assert c.hour == set(range(24)) assert c.day_of_week == set(range(7)) assert c.day_of_month == set(range(1, 32)) assert c.month_of_year == set(range(1, 13)) @pytest.mark.parametrize('minute,expected', [ (30, {30}), ('30', {30}), ((30, 40, 50), {30, 40, 50}), ((30, 40, 50, 51), {30, 40, 50, 51}) ]) def test_crontab_spec_minute_formats(self, minute, expected): c = self.crontab(minute=minute) assert c.minute == expected @pytest.mark.parametrize('minute', [60, '0-100']) def test_crontab_spec_invalid_minute(self, minute): with pytest.raises(ValueError): self.crontab(minute=minute) @pytest.mark.parametrize('hour,expected', [ (6, {6}), ('5', {5}), ((4, 8, 12), {4, 8, 12}), ]) def test_crontab_spec_hour_formats(self, hour, expected): c = self.crontab(hour=hour) assert c.hour == expected @pytest.mark.parametrize('hour', [24, '0-30']) def test_crontab_spec_invalid_hour(self, hour): with pytest.raises(ValueError): self.crontab(hour=hour) @pytest.mark.parametrize('day_of_week,expected', [ (5, {5}), ('5', {5}), ('fri', {5}), ('tuesday,sunday,fri', {0, 2, 5}), ('mon-fri', {1, 2, 3, 4, 5}), ('*/2', {0, 2, 4, 6}), ]) def test_crontab_spec_dow_formats(self, day_of_week, expected): c = self.crontab(day_of_week=day_of_week) assert c.day_of_week == expected @pytest.mark.parametrize('day_of_week', [ 'fooday-barday', '1,4,foo', '7', '12', ]) def test_crontab_spec_invalid_dow(self, day_of_week): with pytest.raises(ValueError): self.crontab(day_of_week=day_of_week) @pytest.mark.parametrize('day_of_month,expected', [ (5, {5}), ('5', {5}), ('2,4,6', {2, 4, 6}), ('*/5', {1, 6, 11, 16, 21, 26, 31}), ]) def test_crontab_spec_dom_formats(self, day_of_month, expected): c = self.crontab(day_of_month=day_of_month) assert c.day_of_month == expected @pytest.mark.parametrize('day_of_month', [0, '0-10', 32, '31,32']) def test_crontab_spec_invalid_dom(self, day_of_month): with pytest.raises(ValueError): self.crontab(day_of_month=day_of_month) @pytest.mark.parametrize('month_of_year,expected', [ (1, {1}), ('1', {1}), ('feb', {2}), ('Mar', {3}), ('april', {4}), ('may,jun,jul', {5, 6, 7}), ('aug-oct', {8, 9, 10}), ('2,4,6', {2, 4, 6}), ('*/2', {1, 3, 5, 7, 9, 11}), ('2-12/2', {2, 4, 6, 8, 10, 12}), ]) def test_crontab_spec_moy_formats(self, month_of_year, expected): c = self.crontab(month_of_year=month_of_year) assert c.month_of_year == expected @pytest.mark.parametrize('month_of_year', [0, '0-5', 13, '12,13', 'jaan', 'sebtember']) def test_crontab_spec_invalid_moy(self, month_of_year): with pytest.raises(ValueError): self.crontab(month_of_year=month_of_year) def seconds_almost_equal(self, a, b, precision): for index, skew in enumerate((+1, -1, 0)): try: assertions.assertAlmostEqual(a, b + skew, precision) except Exception as exc: # AssertionError != builtins.AssertionError in pytest if 'AssertionError' in str(exc): if index + 1 >= 3: raise else: break def test_every_minute_execution_is_due(self): last_ran = self.now - timedelta(seconds=61) due, remaining = self.every_minute.is_due(last_ran) self.assert_relativedelta(self.every_minute, last_ran) assert due self.seconds_almost_equal(remaining, self.next_minute, 1) def assert_relativedelta(self, due, last_ran): try: from dateutil.relativedelta import relativedelta except ImportError: return l1, d1, n1 = due.remaining_delta(last_ran) l2, d2, n2 = due.remaining_delta(last_ran, ffwd=relativedelta) if not isinstance(d1, relativedelta): assert l1 == l2 for field, value in d1._fields().items(): assert getattr(d1, field) == value assert not d2.years assert not d2.months assert not d2.days assert not d2.leapdays assert not d2.hours assert not d2.minutes assert not d2.seconds assert not d2.microseconds def test_every_minute_execution_is_not_due(self): last_ran = self.now - timedelta(seconds=self.now.second) due, remaining = self.every_minute.is_due(last_ran) assert not due self.seconds_almost_equal(remaining, self.next_minute, 1) def test_execution_is_due_on_saturday(self): # 29th of May 2010 is a saturday with patch_crontab_nowfun(self.hourly, datetime(2010, 5, 29, 10, 30)): last_ran = self.now - timedelta(seconds=61) due, remaining = self.every_minute.is_due(last_ran) assert due self.seconds_almost_equal(remaining, self.next_minute, 1) def test_execution_is_due_on_sunday(self): # 30th of May 2010 is a sunday with patch_crontab_nowfun(self.hourly, datetime(2010, 5, 30, 10, 30)): last_ran = self.now - timedelta(seconds=61) due, remaining = self.every_minute.is_due(last_ran) assert due self.seconds_almost_equal(remaining, self.next_minute, 1) def test_execution_is_due_on_monday(self): # 31st of May 2010 is a monday with patch_crontab_nowfun(self.hourly, datetime(2010, 5, 31, 10, 30)): last_ran = self.now - timedelta(seconds=61) due, remaining = self.every_minute.is_due(last_ran) assert due self.seconds_almost_equal(remaining, self.next_minute, 1) def test_every_hour_execution_is_due(self): with patch_crontab_nowfun(self.hourly, datetime(2010, 5, 10, 10, 30)): due, remaining = self.hourly.is_due(datetime(2010, 5, 10, 6, 30)) assert due assert remaining == 60 * 60 def test_every_hour_execution_is_not_due(self): with patch_crontab_nowfun(self.hourly, datetime(2010, 5, 10, 10, 29)): due, remaining = self.hourly.is_due(datetime(2010, 5, 10, 9, 30)) assert not due assert remaining == 60 def test_first_quarter_execution_is_due(self): with patch_crontab_nowfun( self.quarterly, datetime(2010, 5, 10, 10, 15)): due, remaining = self.quarterly.is_due( datetime(2010, 5, 10, 6, 30), ) assert due assert remaining == 15 * 60 def test_second_quarter_execution_is_due(self): with patch_crontab_nowfun( self.quarterly, datetime(2010, 5, 10, 10, 30)): due, remaining = self.quarterly.is_due( datetime(2010, 5, 10, 6, 30), ) assert due assert remaining == 15 * 60 def test_first_quarter_execution_is_not_due(self): with patch_crontab_nowfun( self.quarterly, datetime(2010, 5, 10, 10, 14)): due, remaining = self.quarterly.is_due( datetime(2010, 5, 10, 10, 0), ) assert not due assert remaining == 60 def test_second_quarter_execution_is_not_due(self): with patch_crontab_nowfun( self.quarterly, datetime(2010, 5, 10, 10, 29)): due, remaining = self.quarterly.is_due( datetime(2010, 5, 10, 10, 15), ) assert not due assert remaining == 60 def test_daily_execution_is_due(self): with patch_crontab_nowfun(self.daily, datetime(2010, 5, 10, 7, 30)): due, remaining = self.daily.is_due(datetime(2010, 5, 9, 7, 30)) assert due assert remaining == 24 * 60 * 60 def test_daily_execution_is_not_due(self): with patch_crontab_nowfun(self.daily, datetime(2010, 5, 10, 10, 30)): due, remaining = self.daily.is_due(datetime(2010, 5, 10, 7, 30)) assert not due assert remaining == 21 * 60 * 60 def test_weekly_execution_is_due(self): with patch_crontab_nowfun(self.weekly, datetime(2010, 5, 6, 7, 30)): due, remaining = self.weekly.is_due(datetime(2010, 4, 30, 7, 30)) assert due assert remaining == 7 * 24 * 60 * 60 def test_weekly_execution_is_not_due(self): with patch_crontab_nowfun(self.weekly, datetime(2010, 5, 7, 10, 30)): due, remaining = self.weekly.is_due(datetime(2010, 5, 6, 7, 30)) assert not due assert remaining == 6 * 24 * 60 * 60 - 3 * 60 * 60 def test_monthly_execution_is_due(self): with patch_crontab_nowfun(self.monthly, datetime(2010, 5, 13, 7, 30)): due, remaining = self.monthly.is_due(datetime(2010, 4, 8, 7, 30)) assert due assert remaining == 28 * 24 * 60 * 60 def test_monthly_execution_is_not_due(self): with patch_crontab_nowfun(self.monthly, datetime(2010, 5, 9, 10, 30)): due, remaining = self.monthly.is_due(datetime(2010, 4, 8, 7, 30)) assert not due assert remaining == 4 * 24 * 60 * 60 - 3 * 60 * 60 def test_monthly_moy_execution_is_due(self): with patch_crontab_nowfun( self.monthly_moy, datetime(2014, 2, 26, 22, 0)): due, remaining = self.monthly_moy.is_due( datetime(2013, 7, 4, 10, 0), ) assert due assert remaining == 60.0 @pytest.mark.skip('TODO: unstable test') def test_monthly_moy_execution_is_not_due(self): with patch_crontab_nowfun( self.monthly_moy, datetime(2013, 6, 28, 14, 30)): due, remaining = self.monthly_moy.is_due( datetime(2013, 6, 28, 22, 14), ) assert not due attempt = ( time.mktime(datetime(2014, 2, 26, 22, 0).timetuple()) - time.mktime(datetime(2013, 6, 28, 14, 30).timetuple()) - 60 * 60 ) assert remaining == attempt def test_monthly_moy_execution_is_due2(self): with patch_crontab_nowfun( self.monthly_moy, datetime(2014, 2, 26, 22, 0)): due, remaining = self.monthly_moy.is_due( datetime(2013, 2, 28, 10, 0), ) assert due assert remaining == 60.0 def test_monthly_moy_execution_is_not_due2(self): with patch_crontab_nowfun( self.monthly_moy, datetime(2014, 2, 26, 21, 0)): due, remaining = self.monthly_moy.is_due( datetime(2013, 6, 28, 22, 14), ) assert not due attempt = 60 * 60 assert remaining == attempt def test_yearly_execution_is_due(self): with patch_crontab_nowfun(self.yearly, datetime(2010, 3, 11, 7, 30)): due, remaining = self.yearly.is_due(datetime(2009, 3, 12, 7, 30)) assert due assert remaining == 364 * 24 * 60 * 60 def test_yearly_execution_is_not_due(self): with patch_crontab_nowfun(self.yearly, datetime(2010, 3, 7, 10, 30)): due, remaining = self.yearly.is_due(datetime(2009, 3, 12, 7, 30)) assert not due assert remaining == 4 * 24 * 60 * 60 - 3 * 60 * 60 def test_execution_not_due_if_task_not_run_at_last_feasible_time_outside_deadline( self): """If the crontab schedule was added after the task was due, don't immediately fire the task again""" # could have feasibly been run on 12/5 at 7:30, but wasn't. self.app.conf.beat_cron_starting_deadline = 3600 last_run = datetime(2022, 12, 4, 10, 30) now = datetime(2022, 12, 5, 10, 30) expected_next_execution_time = datetime(2022, 12, 6, 7, 30) expected_remaining = ( expected_next_execution_time - now).total_seconds() # Run the daily (7:30) crontab with the current date with patch_crontab_nowfun(self.daily, now): due, remaining = self.daily.is_due(last_run) assert remaining == expected_remaining assert not due def test_execution_not_due_if_task_not_run_at_last_feasible_time_no_deadline_set( self): """Same as above test except there's no deadline set, so it should be due""" last_run = datetime(2022, 12, 4, 10, 30) now = datetime(2022, 12, 5, 10, 30) expected_next_execution_time = datetime(2022, 12, 6, 7, 30) expected_remaining = ( expected_next_execution_time - now).total_seconds() # Run the daily (7:30) crontab with the current date with patch_crontab_nowfun(self.daily, now): due, remaining = self.daily.is_due(last_run) assert remaining == expected_remaining assert due def test_execution_due_if_task_not_run_at_last_feasible_time_within_deadline( self): # Could have feasibly been run on 12/5 at 7:30, but wasn't. We are # still within a 1 hour deadline from the # last feasible run, so the task should still be due. self.app.conf.beat_cron_starting_deadline = 3600 last_run = datetime(2022, 12, 4, 10, 30) now = datetime(2022, 12, 5, 8, 0) expected_next_execution_time = datetime(2022, 12, 6, 7, 30) expected_remaining = ( expected_next_execution_time - now).total_seconds() # run the daily (7:30) crontab with the current date with patch_crontab_nowfun(self.daily, now): due, remaining = self.daily.is_due(last_run) assert remaining == expected_remaining assert due def test_execution_due_if_task_not_run_at_any_feasible_time_within_deadline( self): # Could have feasibly been run on 12/4 at 7:30, or 12/5 at 7:30, # but wasn't. We are still within a 1 hour # deadline from the last feasible run (12/5), so the task should # still be due. self.app.conf.beat_cron_starting_deadline = 3600 last_run = datetime(2022, 12, 3, 10, 30) now = datetime(2022, 12, 5, 8, 0) expected_next_execution_time = datetime(2022, 12, 6, 7, 30) expected_remaining = ( expected_next_execution_time - now).total_seconds() # Run the daily (7:30) crontab with the current date with patch_crontab_nowfun(self.daily, now): due, remaining = self.daily.is_due(last_run) assert remaining == expected_remaining assert due def test_execution_not_due_if_task_not_run_at_any_feasible_time_outside_deadline( self): """Verifies that remaining is still the time to the next feasible run date even though the original feasible date was passed over in favor of a newer one.""" # Could have feasibly been run on 12/4 or 12/5 at 7:30, # but wasn't. self.app.conf.beat_cron_starting_deadline = 3600 last_run = datetime(2022, 12, 3, 10, 30) now = datetime(2022, 12, 5, 11, 0) expected_next_execution_time = datetime(2022, 12, 6, 7, 30) expected_remaining = ( expected_next_execution_time - now).total_seconds() # run the daily (7:30) crontab with the current date with patch_crontab_nowfun(self.daily, now): due, remaining = self.daily.is_due(last_run) assert remaining == expected_remaining assert not due def test_execution_not_due_if_last_run_in_future(self): # Should not run if the last_run hasn't happened yet. last_run = datetime(2022, 12, 6, 7, 30) now = datetime(2022, 12, 5, 10, 30) expected_next_execution_time = datetime(2022, 12, 7, 7, 30) expected_remaining = ( expected_next_execution_time - now).total_seconds() # Run the daily (7:30) crontab with the current date with patch_crontab_nowfun(self.daily, now): due, remaining = self.daily.is_due(last_run) assert not due assert remaining == expected_remaining def test_execution_not_due_if_last_run_at_last_feasible_time(self): # Last feasible time is 12/5 at 7:30 last_run = datetime(2022, 12, 5, 7, 30) now = datetime(2022, 12, 5, 10, 30) expected_next_execution_time = datetime(2022, 12, 6, 7, 30) expected_remaining = ( expected_next_execution_time - now).total_seconds() # Run the daily (7:30) crontab with the current date with patch_crontab_nowfun(self.daily, now): due, remaining = self.daily.is_due(last_run) assert remaining == expected_remaining assert not due def test_execution_not_due_if_last_run_past_last_feasible_time(self): # Last feasible time is 12/5 at 7:30 last_run = datetime(2022, 12, 5, 8, 30) now = datetime(2022, 12, 5, 10, 30) expected_next_execution_time = datetime(2022, 12, 6, 7, 30) expected_remaining = ( expected_next_execution_time - now).total_seconds() # Run the daily (7:30) crontab with the current date with patch_crontab_nowfun(self.daily, now): due, remaining = self.daily.is_due(last_run) assert remaining == expected_remaining assert not due def test_execution_due_for_negative_utc_timezone_with_day_of_month(self): # UTC-8 tzname = "America/Los_Angeles" self.app.timezone = tzname tz = ZoneInfo(tzname) # set day_of_month to test on _delta_to_next crontab = self.crontab(minute=0, day_of_month='27-31') # last_run_at: '2023/01/28T23:00:00-08:00' last_run_at = datetime(2023, 1, 28, 23, 0, tzinfo=tz) # now: '2023/01/29T00:00:00-08:00' now = datetime(2023, 1, 29, 0, 0, tzinfo=tz) with patch_crontab_nowfun(crontab, now): due, remaining = crontab.is_due(last_run_at) assert (due, remaining) == (True, 3600)