kumi/parse_crontab
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Fixes/adds timezone support to "just work"

Browse source 
Explicitly starts testing against Python 2.6, 2.7, 3.3, 3.4, and 3.5
This commit is contained in:
Josiah Carlson 2016-03-19 22:18:14 -07:00
parent b7c2b5b975
commit fd7a7192f9
5 changed files with 87 additions and 41 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

32
Makefile
View file

@ -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

2
README.rst
View file

@ -70,6 +70,8 @@ Example uses
3600.0
Notes
=====

60
crontab/_crontab.py
View file

@ -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

3
setup.py
View file

@ -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"],
)

31
tests/test_crontab.py
View file

@ -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)