From fd7a7192f9eb1f38fcdd06ff608ce5bf95c6587a Mon Sep 17 00:00:00 2001 From: Josiah Carlson Date: Sat, 19 Mar 2016 22:18:14 -0700 Subject: [PATCH] Fixes/adds timezone support to "just work" Explicitly starts testing against Python 2.6, 2.7, 3.3, 3.4, and 3.5 --- Makefile | 32 +++++++++++++---------- README.rst | 2 ++ crontab/_crontab.py | 60 +++++++++++++++++++++++++++++++------------ setup.py | 3 ++- tests/test_crontab.py | 31 ++++++++++++++-------- 5 files changed, 87 insertions(+), 41 deletions(-) diff --git a/Makefile b/Makefile index 5ca2e71..f60b7f2 100644 --- a/Makefile +++ b/Makefile @@ -1,14 +1,18 @@ -SHELL=/bin/bash - -clean: - -rm -f *.pyc crontab/*.pyc README.html MANIFEST - -rm -rf build dist - -install: - python setup.py install - -test: - python -m tests.test_crontab - -upload: - python setup.py sdist upload +SHELL=/bin/bash + +clean: + -rm -f *.pyc crontab/*.pyc README.html MANIFEST + -rm -rf build dist + +install: + python setup.py install + +test: + python2.6 -m tests.test_crontab + python2.7 -m tests.test_crontab + python3.3 -m tests.test_crontab + python3.4 -m tests.test_crontab + python3.5 -m tests.test_crontab + +upload: + python setup.py sdist upload diff --git a/README.rst b/README.rst index 56f5e17..79db752 100644 --- a/README.rst +++ b/README.rst @@ -70,6 +70,8 @@ Example uses 3600.0 + + Notes ===== diff --git a/crontab/_crontab.py b/crontab/_crontab.py index b300cd4..500ddeb 100644 --- a/crontab/_crontab.py +++ b/crontab/_crontab.py @@ -13,8 +13,9 @@ Other licenses may be available upon request. ''' from collections import namedtuple -import datetime +from datetime import datetime, timedelta import sys +import warnings _ranges = [ (0, 59), @@ -46,18 +47,31 @@ _aliases = { '@hourly': '0 * * * *', } +WARNING_CHANGE_MESSAGE = '''\ +Version 0.22.0+ of crontab will use datetime.utcnow() and +datetime.utcfromtimestamp() instead of datetime.now() and +datetime.fromtimestamp() as was previous. This had been a bug, which will be +remedied. If you would like to keep the *old* behavior: +`ct.next(..., default_utc=False)` . If you want to use the new behavior *now*: +`ct.next(..., default_utc=True)`. If you pass a datetime object with a tzinfo +attribute that is not None, timezones will *just work* to the best of their +ability. There are tests...''' + + if sys.version_info >= (3, 0): _number_types = (int, float) xrange = range else: _number_types = (int, long, float) -MINUTE = datetime.timedelta(minutes=1) -HOUR = datetime.timedelta(hours=1) -DAY = datetime.timedelta(days=1) -WEEK = datetime.timedelta(days=7) -MONTH = datetime.timedelta(days=28) -YEAR = datetime.timedelta(days=365) +MINUTE = timedelta(minutes=1) +HOUR = timedelta(hours=1) +DAY = timedelta(days=1) +WEEK = timedelta(days=7) +MONTH = timedelta(days=28) +YEAR = timedelta(days=365) + +WARN_CHANGE = object() # find the next scheduled time def _end_of_month(dt): @@ -305,6 +319,7 @@ class _Matcher(object): return good, _end + class CronTab(object): __slots__ = 'matchers', def __init__(self, crontab): @@ -343,15 +358,22 @@ class CronTab(object): attr = attr() % 7 return self.matchers[index](attr, dt) - def next(self, now=None, increments=_increments, delta=True): + def next(self, now=None, increments=_increments, delta=True, default_utc=WARN_CHANGE): ''' How long to wait in seconds before this crontab entry can next be executed. ''' - now = now or datetime.datetime.now() + if default_utc is WARN_CHANGE and (isinstance(now, _number_types) or (now and not now.tzinfo)): + warnings.warn(WARNING_CHANGE_MESSAGE, FutureWarning, 2) + default_utc = False + + now = now or (datetime.utcnow() if default_utc else datetime.now()) if isinstance(now, _number_types): - now = datetime.datetime.fromtimestamp(now) - # get a reasonable future/past start time + now = datetime.utcfromtimestamp(now) if default_utc else datetime.fromtimestamp(now) + + # handle timezones if the datetime object has a timezone and get a + # reasonable future/past start time + onow, now = now, now.replace(tzinfo=None) future = now.replace(second=0, microsecond=0) + increments[0]() if future < now: # we are going backwards... @@ -387,17 +409,23 @@ class CronTab(object): "author with the following information:\n" \ "crontab: %r\n" \ "now: %r", ' '.join(m.input for m in self.matchers), now) - delay = future - now + if onow.tzinfo: + future = onow.tzinfo.localize(future) + + delay = future - onow if not delta: - delay = future - datetime.datetime(1970, 1, 1) + begin = datetime(1970, 1, 1) + if onow.tzinfo: + begin = onow.tzinfo.localize(begin) + delay = future - begin return delay.days * 86400 + delay.seconds + delay.microseconds / 1000000. - def previous(self, now=None, delta=True): - return self.next(now, _decrements, delta) + def previous(self, now=None, delta=True, default_utc=WARN_CHANGE): + return self.next(now, _decrements, delta, default_utc) def test(self, entry): if isinstance(entry, _number_types): - entry = datetime.datetime.utcfromtimestamp(entry) + entry = datetime.utcfromtimestamp(entry) for index in xrange(6): if not self._test_match(index, entry): return False diff --git a/setup.py b/setup.py index 3ce4ba8..e3bdf47 100644 --- a/setup.py +++ b/setup.py @@ -10,7 +10,7 @@ except: setup( name='crontab', - version='0.20.5', + version='0.21.0', description='Parse and use crontab schedules in Python', author='Josiah Carlson', author_email='josiah.carlson@gmail.com', @@ -31,4 +31,5 @@ setup( ], license='GNU LGPL v2.1', long_description=long_description, + install_requires=["pytz"], ) diff --git a/tests/test_crontab.py b/tests/test_crontab.py index f8cccc8..8ad1f73 100644 --- a/tests/test_crontab.py +++ b/tests/test_crontab.py @@ -1,20 +1,23 @@ + import datetime import unittest +import pytz + from crontab import CronTab class TestCrontab(unittest.TestCase): def _run_test(self, crontab, max_delay, now=None, min_delay=None): ct = CronTab(crontab) - now = now or datetime.datetime.now() - delay = ct.next(now) + now = now or datetime.datetime.utcnow() + delay = ct.next(now, default_utc=True) assert delay is not None dd = (crontab, delay, max_delay, now, now+datetime.timedelta(seconds=delay)) assert delay <= max_delay, dd if min_delay is not None: assert delay >= min_delay, dd if not crontab.endswith(' 2099'): - delay2 = ct.previous(now + datetime.timedelta(seconds=delay)) + delay2 = ct.previous(now + datetime.timedelta(seconds=delay), default_utc=True) dd = (crontab, delay, max_delay, now, now+datetime.timedelta(seconds=delay)) assert abs(delay2) >= delay, (delay, delay2) pt = now + datetime.timedelta(seconds=delay) + datetime.timedelta(seconds=delay2) @@ -22,7 +25,7 @@ class TestCrontab(unittest.TestCase): def _run_impossible(self, crontab, now): ct = CronTab(crontab) - delay = ct.next(now) + delay = ct.next(now, default_utc=True) assert delay is None, (crontab, delay, now, now+datetime.timedelta(seconds=delay)) def test_closest(self): @@ -35,14 +38,14 @@ class TestCrontab(unittest.TestCase): assert not ce.test(s1245) assert ce.test(t1245) - n = datetime.datetime.utcfromtimestamp(ce.next(t945, delta=False)) + n = datetime.datetime.utcfromtimestamp(ce.next(t945, delta=False, default_utc=True)) assert n == datetime.datetime(2013, 1, 1, 10, 0), n - p = datetime.datetime.utcfromtimestamp(ce.previous(t945, delta=False)) + p = datetime.datetime.utcfromtimestamp(ce.previous(t945, delta=False, default_utc=True)) assert p == datetime.datetime(2012, 12, 31, 15, 45), p - n = datetime.datetime.utcfromtimestamp(ce.next(s1245, delta=False)) + n = datetime.datetime.utcfromtimestamp(ce.next(s1245, delta=False, default_utc=True)) assert n == datetime.datetime(2013, 1, 7, 10, 0), n - p = datetime.datetime.utcfromtimestamp(ce.previous(s1245, delta=False)) + p = datetime.datetime.utcfromtimestamp(ce.previous(s1245, delta=False, default_utc=True)) assert p == datetime.datetime(2013, 1, 4, 15, 45), p def test_normal(self): @@ -154,12 +157,20 @@ class TestCrontab(unittest.TestCase): schedule = CronTab('0 * * * *') ts = datetime.datetime(2014, 6, 6, 9, 0, 0) for i in range(70): - next = schedule.next(ts) + next = schedule.next(ts, default_utc=True) self.assertTrue(0 <= next <= 3600, next) - previous = schedule.previous(ts) + previous = schedule.previous(ts, default_utc=True) self.assertTrue(-3600 <= previous <= 0, previous) ts += datetime.timedelta(seconds=1) + def test_timezones(self): + s = CronTab('0 9 13 3 * 2016') + + self.assertEquals(s.next(datetime.datetime(2016, 3, 13), default_utc=True), 32400) + self.assertEquals(s.next(pytz.utc.localize(datetime.datetime(2016, 3, 13)), default_utc=True), 32400) + + self.assertEquals(s.next(pytz.timezone('US/Eastern').localize(datetime.datetime(2016, 3, 13))), 28800) + def test(): suite = unittest.TestLoader().loadTestsFromTestCase(TestCrontab)