Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_days_with_isoweekday() -> None:
t = datetime.datetime(2017, 8, 24, 18, 1, 0)
assert get_next_datetime('0 10 1-7 * *', t) == datetime.datetime(2017, 9, 1, 10, 0)
assert get_next_datetime('0 10 * * mon', t) == datetime.datetime(2017, 8, 28, 10, 0)
assert get_next_datetime('0 10 1-7 * mon', t) == datetime.datetime(2017, 8, 28, 10, 0)
assert get_next_datetime('0 10 1-25 * mon', t) == datetime.datetime(2017, 8, 25, 10, 0)
assert get_next_datetime('0 * 1-25 * mon', t) == datetime.datetime(2017, 8, 24, 19, 0)
assert get_next_datetime('0 * 1-24 * mon', t) == datetime.datetime(2017, 8, 24, 19, 0)
assert get_next_datetime('0 * 1-23 * mon', t) == datetime.datetime(2017, 8, 28, 0, 0)
def test_parser() -> None:
t = datetime.datetime(2017, 6, 15, 10, 16, 50)
assert get_next_datetime('* * * * *', t) == datetime.datetime(2017, 6, 15, 10, 17)
assert get_next_datetime('1 * * * *', t) == datetime.datetime(2017, 6, 15, 11, 1)
assert get_next_datetime('*/2 * * * *', t) == datetime.datetime(2017, 6, 15, 10, 18)
assert get_next_datetime('1/2 * * * *', t) == datetime.datetime(2017, 6, 15, 10, 17)
assert get_next_datetime('1-3/2 * * * *', t) == datetime.datetime(2017, 6, 15, 11, 1)
assert get_next_datetime('0/4 * * * *', t) == datetime.datetime(2017, 6, 15, 10, 20)
assert get_next_datetime('7/8 * * * *', t) == datetime.datetime(2017, 6, 15, 10, 23)
assert get_next_datetime('15 */2 * * *', t) == datetime.datetime(2017, 6, 15, 12, 15)
assert get_next_datetime('15 */2 * * *', t) == datetime.datetime(2017, 6, 15, 12, 15)
assert get_next_datetime('15-30 * * * *', t) == datetime.datetime(2017, 6, 15, 10, 17)
assert get_next_datetime('20-30 * * * *', t) == datetime.datetime(2017, 6, 15, 10, 20)
assert get_next_datetime('5,10,55 * * * *', t) == datetime.datetime(2017, 6, 15, 10, 55)
def test_parser() -> None:
t = datetime.datetime(2017, 6, 15, 10, 16, 50)
assert get_next_datetime('* * * * *', t) == datetime.datetime(2017, 6, 15, 10, 17)
assert get_next_datetime('1 * * * *', t) == datetime.datetime(2017, 6, 15, 11, 1)
assert get_next_datetime('*/2 * * * *', t) == datetime.datetime(2017, 6, 15, 10, 18)
assert get_next_datetime('1/2 * * * *', t) == datetime.datetime(2017, 6, 15, 10, 17)
assert get_next_datetime('1-3/2 * * * *', t) == datetime.datetime(2017, 6, 15, 11, 1)
assert get_next_datetime('0/4 * * * *', t) == datetime.datetime(2017, 6, 15, 10, 20)
assert get_next_datetime('7/8 * * * *', t) == datetime.datetime(2017, 6, 15, 10, 23)
assert get_next_datetime('15 */2 * * *', t) == datetime.datetime(2017, 6, 15, 12, 15)
assert get_next_datetime('15 */2 * * *', t) == datetime.datetime(2017, 6, 15, 12, 15)
assert get_next_datetime('15-30 * * * *', t) == datetime.datetime(2017, 6, 15, 10, 17)
assert get_next_datetime('20-30 * * * *', t) == datetime.datetime(2017, 6, 15, 10, 20)
assert get_next_datetime('5,10,55 * * * *', t) == datetime.datetime(2017, 6, 15, 10, 55)
def test_days_with_isoweekday() -> None:
t = datetime.datetime(2017, 8, 24, 18, 1, 0)
assert get_next_datetime('0 10 1-7 * *', t) == datetime.datetime(2017, 9, 1, 10, 0)
assert get_next_datetime('0 10 * * mon', t) == datetime.datetime(2017, 8, 28, 10, 0)
assert get_next_datetime('0 10 1-7 * mon', t) == datetime.datetime(2017, 8, 28, 10, 0)
assert get_next_datetime('0 10 1-25 * mon', t) == datetime.datetime(2017, 8, 25, 10, 0)
assert get_next_datetime('0 * 1-25 * mon', t) == datetime.datetime(2017, 8, 24, 19, 0)
assert get_next_datetime('0 * 1-24 * mon', t) == datetime.datetime(2017, 8, 24, 19, 0)
assert get_next_datetime('0 * 1-23 * mon', t) == datetime.datetime(2017, 8, 28, 0, 0)
def test_timezones() -> None:
t = datetime.datetime(2017, 6, 15, 10, 16, 50, tzinfo=pytz.UTC)
assert get_next_datetime('* * * * *', t) == datetime.datetime(2017, 6, 15, 10, 17, tzinfo=pytz.UTC)
t = pytz.timezone('Europe/Stockholm').localize(datetime.datetime(2017, 6, 15, 10, 16, 50))
assert get_next_datetime('* * * * *', t) == pytz.timezone('Europe/Stockholm').localize(datetime.datetime(2017, 6, 15, 10, 17))
assert get_next_datetime('15 */2 * * *', t) == pytz.timezone('Europe/Stockholm').localize(datetime.datetime(2017, 6, 15, 12, 15))
def test_isoweekday() -> None:
t = datetime.datetime(2017, 6, 15, 10, 16, 50)
assert get_next_datetime('* * * * 0', t) == datetime.datetime(2017, 6, 18, 0, 0)
assert get_next_datetime('* * * * 1', t) == datetime.datetime(2017, 6, 19, 0, 0)
assert get_next_datetime('* * * * 2', t) == datetime.datetime(2017, 6, 20, 0, 0)
assert get_next_datetime('* * * * 3', t) == datetime.datetime(2017, 6, 21, 0, 0)
assert get_next_datetime('* * * * 4', t) == datetime.datetime(2017, 6, 15, 10, 17)
assert get_next_datetime('* * * * 5', t) == datetime.datetime(2017, 6, 16, 0, 0)
assert get_next_datetime('* * * * 6', t) == datetime.datetime(2017, 6, 17, 0, 0)
assert get_next_datetime('* * * * 7', t) == datetime.datetime(2017, 6, 18, 0, 0)
assert get_next_datetime('* * * * mon-tue', t) == datetime.datetime(2017, 6, 19, 0, 0)
t = datetime.datetime(2017, 6, 19, 10, 16, 50)
assert get_next_datetime('* * * * 5-6', t) == datetime.datetime(2017, 6, 23, 0, 0)
assert get_next_datetime('* * * * 0-4', t) == datetime.datetime(2017, 6, 19, 10, 17)
assert get_next_datetime('* * * * 5-7', t) == datetime.datetime(2017, 6, 23, 0, 0)
assert get_next_datetime('* * * * fri-sat', t) == datetime.datetime(2017, 6, 23, 0, 0)
assert get_next_datetime('* * * * sun-fri', t) == datetime.datetime(2017, 6, 19, 10, 17)
assert get_next_datetime('* * * * fri-sun', t) == datetime.datetime(2017, 6, 23, 0, 0)
assert get_next_datetime('* * * * 0-7', t) == datetime.datetime(2017, 6, 19, 10, 17)
with pytest.raises(Exception):
get_next_datetime('* * * * wed-mon', t)
def test_timezones() -> None:
t = datetime.datetime(2017, 6, 15, 10, 16, 50, tzinfo=pytz.UTC)
assert get_next_datetime('* * * * *', t) == datetime.datetime(2017, 6, 15, 10, 17, tzinfo=pytz.UTC)
t = pytz.timezone('Europe/Stockholm').localize(datetime.datetime(2017, 6, 15, 10, 16, 50))
assert get_next_datetime('* * * * *', t) == pytz.timezone('Europe/Stockholm').localize(datetime.datetime(2017, 6, 15, 10, 17))
assert get_next_datetime('15 */2 * * *', t) == pytz.timezone('Europe/Stockholm').localize(datetime.datetime(2017, 6, 15, 12, 15))
def test_advanced_parsing() -> None:
t = datetime.datetime(2017, 6, 15, 10, 16, 50)
assert get_next_datetime('39 3 L * *', t) == datetime.datetime(2017, 6, 30, 3, 39)
assert get_next_datetime('10-15,30,45 10 * * *', t) == datetime.datetime(2017, 6, 15, 10, 30)
t = datetime.datetime(2017, 6, 15, 10, 12, 50)
assert get_next_datetime('10-15,30,45 10 * * *', t) == datetime.datetime(2017, 6, 15, 10, 13)
t = datetime.datetime(2017, 6, 15, 9, 12, 50)
assert get_next_datetime('10-15,30,45 10 * * *', t) == datetime.datetime(2017, 6, 15, 10, 10)
t = datetime.datetime(2017, 6, 15, 10, 16, 50)
assert get_next_datetime('55 7 * * Lsun', t) == datetime.datetime(2017, 6, 25, 7, 55)
assert get_next_datetime('* * * * Lsun', t) == datetime.datetime(2017, 6, 25)
assert get_next_datetime('* * * * Lwed-fri', t) == datetime.datetime(2017, 6, 28)
assert get_next_datetime('* * 4-15 feb-jun wed-fri', t) == datetime.datetime(2017, 6, 15, 10, 17)
assert get_next_datetime('3-20/2 5 4-15 feb-may *', t) == datetime.datetime(2018, 2, 4, 5, 3)
assert get_next_datetime('3-20/2 5 4-15 feb-may wed', t) == datetime.datetime(2018, 2, 4, 5, 3)
assert get_next_datetime('3-20/2 5 10-15 feb-may wed', t) == datetime.datetime(2018, 2, 7, 5, 3)
assert get_next_datetime('3-20/2 5 * feb-may wed', t) == datetime.datetime(2018, 2, 7, 5, 3)
assert get_next_datetime('3-20/2 5 4-15 feb-may mon,tue-wed', t) == datetime.datetime(2018, 2, 4, 5, 3)
assert get_next_datetime('3-20/2 5 4-15 feb-may wed-fri', t) == datetime.datetime(2018, 2, 1, 5, 3)
assert get_next_datetime('3-20/2 5 7-15 feb-may wed,tue,sat,mon', t) == datetime.datetime(2018, 2, 3, 5, 3)
assert get_next_datetime('* * 29 2 *', t) == datetime.datetime(2020, 2, 29, 0, 0)
assert get_next_datetime('* * 29 2 0', t) == datetime.datetime(2018, 2, 4, 0, 0)
assert get_next_datetime('* * 29 2 0 2020', t) == datetime.datetime(2020, 2, 2, 0, 0)
assert get_next_datetime('30 5 * jan,mar Ltue', t) == datetime.datetime(2018, 1, 30, 5, 30)
datetime_object = datetime.datetime.strptime(timestamp, '%Y-%m-%d %H:%M:%S')
interval = '{} {} {} {} * {}'.format(datetime_object.minute, datetime_object.hour, datetime_object.day, datetime_object.month, datetime_object.year)
second_modifier = 1
if tz.localize(datetime_object) > local_tz.localize(datetime.datetime.fromtimestamp(current_time)):
second_modifier = -60
next_at = get_next_datetime(interval, local_tz.localize(datetime.datetime.fromtimestamp(current_time + second_modifier)).astimezone(tz))
if not next_at:
return int(current_time + 60 * 60 * 24 * 365 * 100)
return int(next_at.timestamp() + datetime_object.second)
except ValueError:
pass
try:
datetime_object = datetime.datetime.strptime(timestamp, '%Y-%m-%d %H:%M')
interval = '{} {} {} {} * {}'.format(datetime_object.minute, datetime_object.hour, datetime_object.day, datetime_object.month, datetime_object.year)
next_at = get_next_datetime(interval, local_tz.localize(datetime.datetime.fromtimestamp(current_time + 1)).astimezone(tz))
if not next_at:
return int(current_time + 60 * 60 * 24 * 365 * 100)
return int(next_at.timestamp())
except ValueError:
pass
try:
datetime_object = datetime.datetime.strptime(timestamp, '%H:%M:%S')
datetime_object = datetime.datetime(datetime.datetime.fromtimestamp(current_time).year, datetime.datetime.fromtimestamp(current_time).month, datetime.datetime.fromtimestamp(current_time).day, datetime_object.hour, datetime_object.minute, datetime_object.second)
interval = '{} {} * * *'.format(datetime_object.minute, datetime_object.hour)
second_modifier = 1
if tz.localize(datetime_object) > local_tz.localize(datetime.datetime.fromtimestamp(current_time)):
second_modifier = -60
next_at = get_next_datetime(interval, local_tz.localize(datetime.datetime.fromtimestamp(current_time + second_modifier)).astimezone(tz))
if not next_at:
return int(current_time + 60 * 60 * 24 * 365 * 100)
if interval.endswith('s') or interval.endswith('seconds'):
try:
interval = int(interval.replace('seconds', '').replace('s', '').replace(' ', ''))
except ValueError:
pass
try:
interval_value = [v for k, v in interval_aliases.items() if interval in k][0] # type: Union[str, int]
except IndexError:
interval_value = interval
if isinstance(interval_value, int):
return int(current_time + interval_value)
try:
next_at = get_next_datetime(interval_value, local_tz.localize(datetime.datetime.fromtimestamp(current_time + 1)).astimezone(tz))
if not next_at:
return int(current_time + 60 * 60 * 24 * 365 * 100)
return int(next_at.timestamp())
except Exception:
raise Exception('Invalid interval')
return int(current_time + 60 * 60 * 24 * 365 * 100)