openstack
/
deb-python-croniter
Code Issues Proposed changes

Implement DST support 

This fixes https://github.com/taichino/croniter/issues/82
This commit is contained in:
Mathieu Le Marec - Pasquet 2017-03-15 23:34:45 +01:00
parent 4916e103b0
commit 1c05bed1c6
3 changed files with 72 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
README.rst
View File

@ -80,6 +80,12 @@ Supported added for ``get_prev`` method. (>= 0.2.0)::
>>> print itr.get_prev(datetime) # 2010-07-01 00:00:00
>>> print itr.get_prev(datetime) # 2010-06-01 00:00:00
About DST
=========
Be sure to init your croniter instance with a TZ aware datetime for this to work !::
>>> local_date = tz.localize(datetime(2017, 3, 26))
>>> val = croniter('0 0 * * *', local_date).get_next(datetime)
Develop this package
====================

27
src/croniter/croniter.py
View File

@ -75,6 +75,7 @@ class croniter(object):
self.tzinfo = start_time.tzinfo
start_time = self._datetime_to_timestamp(start_time)
self.start_time = start_time
self.cur = start_time
self.exprs = expr_format.split()
@ -257,11 +258,31 @@ class croniter(object):
result = t1 if t1 > t2 else t2
else:
result = self._calc(self.cur, expanded, is_prev)
# DST Handling for cron job spanning accross days
dtstarttime = self._timestamp_to_datetime(self.start_time)
dtresult = self._timestamp_to_datetime(result)
dtresult_utcoffset = dtresult.utcoffset() or datetime.timedelta(0)
dtstarttime_utcoffset = (
dtstarttime.utcoffset() or datetime.timedelta(0))
hours_before_midnight = 24 - dtstarttime.hour
lag_hours = (
self._timedelta_to_seconds(dtresult - dtstarttime) / (60*60)
)
if (
lag_hours >= hours_before_midnight and
(dtresult_utcoffset or dtstarttime_utcoffset) and
(dtresult_utcoffset != dtstarttime_utcoffset)
):
lag = self._timedelta_to_seconds(
dtresult_utcoffset - dtstarttime_utcoffset
)
dtresult = dtresult - datetime.timedelta(seconds=lag)
result = self._datetime_to_timestamp(dtresult)
self.cur = result
if issubclass(ret_type, datetime.datetime):
result = self._timestamp_to_datetime(result)
result = dtresult
return result
def _calc(self, now, expanded, is_prev):

43
src/croniter/tests/test_croniter.py
View File

@ -2,7 +2,7 @@
# -*- coding: utf-8 -*-
import unittest
from datetime import datetime
from datetime import datetime, timedelta
from time import sleep
import pytz
from croniter import croniter, CroniterBadDateError
@ -640,6 +640,47 @@ class CroniterTest(base.TestCase):
ct = croniter('*/30 * * * *', tz.localize(start))
self.assertScheduleTimezone(lambda: ct.get_prev(datetime), reversed(expected_schedule))
def test_std_dst(self):
"""
DST tests
This fixes https://github.com/taichino/croniter/issues/82
"""
tz = pytz.timezone('Europe/Warsaw')
# -> 2017-03-26 01:59+1:00 -> 03:00+2:00
local_date = tz.localize(datetime(2017, 3, 26))
val = croniter('0 0 * * *', local_date).get_next(datetime)
self.assertEqual(val, tz.localize(datetime(2017, 3, 27)))
#
local_date = tz.localize(datetime(2017, 3, 26, 1))
cr = croniter('0 * * * *', local_date)
val = cr.get_next(datetime)
self.assertEqual(val, tz.localize(datetime(2017, 3, 26, 3)))
val = cr.get_current(datetime)
self.assertEqual(val, tz.localize(datetime(2017, 3, 26, 3)))
# -> 2017-10-29 02:59+2:00 -> 02:00+1:00
local_date = tz.localize(datetime(2017, 10, 29))
val = croniter('0 0 * * *', local_date).get_next(datetime)
self.assertEqual(val, tz.localize(datetime(2017, 10, 30)))
local_date = tz.localize(datetime(2017, 10, 29, 1, 59))
val = croniter('0 * * * *', local_date).get_next(datetime)
self.assertEqual(
val.replace(tzinfo=None),
tz.localize(datetime(2017, 10, 29, 2)).replace(tzinfo=None))
local_date = tz.localize(datetime(2017, 10, 29, 2))
val = croniter('0 * * * *', local_date).get_next(datetime)
self.assertEqual(val, tz.localize(datetime(2017, 10, 29, 3)))
local_date = tz.localize(datetime(2017, 10, 29, 3))
val = croniter('0 * * * *', local_date).get_next(datetime)
self.assertEqual(val, tz.localize(datetime(2017, 10, 29, 4)))
local_date = tz.localize(datetime(2017, 10, 29, 4))
val = croniter('0 * * * *', local_date).get_next(datetime)
self.assertEqual(val, tz.localize(datetime(2017, 10, 29, 5)))
local_date = tz.localize(datetime(2017, 10, 29, 5))
val = croniter('0 * * * *', local_date).get_next(datetime)
self.assertEqual(val, tz.localize(datetime(2017, 10, 29, 6)))
if __name__ == '__main__':
unittest.main()