summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorkiorky <kiorky@cryptelium.net>2017-08-31 09:36:57 +0200
committerGitHub <noreply@github.com>2017-08-31 09:36:57 +0200
commite46f26bdb9522193cdfbc7286df51669eb83ec82 (patch)
tree982cda8def08aa3bbf3f5f68f35b8821de33501d
parent1f4ad0b7fb0f0886db245091c4d627a4218d966f (diff)
parent5f98fb2fea1c3b791fcfa29a72ee9b13ea68c58d (diff)
downloadcroniter-e46f26bdb9522193cdfbc7286df51669eb83ec82.tar.gz
Merge pull request #18 from taichino/master
back
-rw-r--r--README.rst5
-rw-r--r--src/croniter/croniter.py226
-rwxr-xr-xsrc/croniter/tests/test_croniter.py16
3 files changed, 142 insertions, 105 deletions
diff --git a/README.rst b/README.rst
index fd04f5a..e91791d 100644
--- a/README.rst
+++ b/README.rst
@@ -82,6 +82,11 @@ 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
+You can validate your crons using ``is_valid`` class method. (>= 0.3.18)::
+
+ >>> croniter.is_valid('0 0 1 * *') # True
+ >>> croniter.is_valid('0 wrong_value 1 * *') # False
+
About DST
=========
Be sure to init your croniter instance with a TZ aware datetime for this to work !::
diff --git a/src/croniter/croniter.py b/src/croniter/croniter.py
index 55bce86..a9f10a3 100644
--- a/src/croniter/croniter.py
+++ b/src/croniter/croniter.py
@@ -14,18 +14,23 @@ search_re = re.compile(r'^([^-]+)-([^-/]+)(/(.*))?$')
only_int_re = re.compile(r'^\d+$')
any_int_re = re.compile(r'^\d+')
star_or_int_re = re.compile(r'^(\d+|\*)$')
+VALID_LEN_EXPRESSION = [5, 6]
-class CroniterBadCronError(ValueError):
- '''.'''
+class CroniterError(ValueError):
+ pass
-class CroniterBadDateError(ValueError):
- '''.'''
+class CroniterBadCronError(CroniterError):
+ pass
-class CroniterNotAlphaError(ValueError):
- '''.'''
+class CroniterBadDateError(CroniterError):
+ pass
+
+
+class CroniterNotAlphaError(CroniterError):
+ pass
class croniter(object):
@@ -79,106 +84,16 @@ class croniter(object):
self.start_time = start_time
self.cur = start_time
- self.exprs = expr_format.split()
-
- if len(self.exprs) != 5 and len(self.exprs) != 6:
- raise CroniterBadCronError(self.bad_length)
-
- expanded = []
- nth_weekday_of_month = {}
-
- for i, expr in enumerate(self.exprs):
- e_list = expr.split(',')
- res = []
-
- while len(e_list) > 0:
- e = e_list.pop()
-
- if i == 4:
- e, sep, nth = str(e).partition('#')
- if nth and not re.match(r'[1-5]', nth):
- raise CroniterBadDateError(
- "[{0}] is not acceptable".format(expr_format))
-
- t = re.sub(r'^\*(\/.+)$', r'%d-%d\1' % (
- self.RANGES[i][0],
- self.RANGES[i][1]),
- str(e))
- m = search_re.search(t)
- if not m:
- t = re.sub(r'^(.+)\/(.+)$', r'\1-%d/\2' % (
- self.RANGES[i][1]),
- str(e))
- m = step_search_re.search(t)
+ self.expanded, self.nth_weekday_of_month = self.expand(expr_format)
- if m:
- (low, high, step) = m.group(1), m.group(2), m.group(4) or 1
-
- if not any_int_re.search(low):
- low = "{0}".format(self._alphaconv(i, low))
-
- if not any_int_re.search(high):
- high = "{0}".format(self._alphaconv(i, high))
-
- if (
- not low or not high or int(low) > int(high)
- or not only_int_re.search(str(step))
- ):
- raise CroniterBadDateError(
- "[{0}] is not acceptable".format(expr_format))
-
- low, high, step = map(int, [low, high, step])
- rng = range(low, high + 1, step)
- e_list += (["{0}#{1}".format(item, nth) for item in rng]
- if i == 4 and nth else rng)
- else:
- if t.startswith('-'):
- raise CroniterBadCronError(
- "[{0}] is not acceptable,\
- negative numbers not allowed".format(
- expr_format))
- if not star_or_int_re.search(t):
- t = self._alphaconv(i, t)
-
- try:
- t = int(t)
- except:
- pass
-
- if t in self.LOWMAP[i]:
- t = self.LOWMAP[i][t]
-
- if (
- t not in ["*", "l"]
- and (int(t) < self.RANGES[i][0] or
- int(t) > self.RANGES[i][1])
- ):
- raise CroniterBadCronError(
- "[{0}] is not acceptable, out of range".format(
- expr_format))
-
- res.append(t)
-
- if i == 4 and nth:
- if t not in nth_weekday_of_month:
- nth_weekday_of_month[t] = set()
- nth_weekday_of_month[t].add(int(nth))
-
- res.sort()
- expanded.append(['*'] if (len(res) == 1
- and res[0] == '*')
- else res)
-
- self.expanded = expanded
- self.nth_weekday_of_month = nth_weekday_of_month
-
- def _alphaconv(self, index, key):
+ @classmethod
+ def _alphaconv(cls, index, key, expressions):
try:
- return self.ALPHACONV[index][key.lower()]
+ return cls.ALPHACONV[index][key.lower()]
except KeyError:
raise CroniterNotAlphaError(
- "[{0}] is not acceptable".format(" ".join(self.exprs)))
+ "[{0}] is not acceptable".format(" ".join(expressions)))
def get_next(self, ret_type=None):
return self._get_next(ret_type or self._ret_type, is_prev=False)
@@ -192,14 +107,15 @@ class croniter(object):
return self._timestamp_to_datetime(self.cur)
return self.cur
- def _datetime_to_timestamp(self, d):
+ @classmethod
+ def _datetime_to_timestamp(cls, d):
"""
Converts a `datetime` object `d` into a UNIX timestamp.
"""
if d.tzinfo is not None:
d = d.replace(tzinfo=None) - d.utcoffset()
- return self._timedelta_to_seconds(d - datetime.datetime(1970, 1, 1))
+ return cls._timedelta_to_seconds(d - datetime.datetime(1970, 1, 1))
def _timestamp_to_datetime(self, timestamp):
"""
@@ -532,3 +448,107 @@ class croniter(object):
return True
else:
return False
+
+ @classmethod
+ def expand(cls, expr_format):
+ expressions = expr_format.split()
+
+ if len(expressions) not in VALID_LEN_EXPRESSION:
+ raise CroniterBadCronError(cls.bad_length)
+
+ expanded = []
+ nth_weekday_of_month = {}
+
+ for i, expr in enumerate(expressions):
+ e_list = expr.split(',')
+ res = []
+
+ while len(e_list) > 0:
+ e = e_list.pop()
+
+ if i == 4:
+ e, sep, nth = str(e).partition('#')
+ if nth and not re.match(r'[1-5]', nth):
+ raise CroniterBadDateError(
+ "[{0}] is not acceptable".format(expr_format))
+
+ t = re.sub(r'^\*(\/.+)$', r'%d-%d\1' % (
+ cls.RANGES[i][0],
+ cls.RANGES[i][1]),
+ str(e))
+ m = search_re.search(t)
+
+ if not m:
+ t = re.sub(r'^(.+)\/(.+)$', r'\1-%d/\2' % (
+ cls.RANGES[i][1]),
+ str(e))
+ m = step_search_re.search(t)
+
+ if m:
+ (low, high, step) = m.group(1), m.group(2), m.group(4) or 1
+
+ if not any_int_re.search(low):
+ low = "{0}".format(cls._alphaconv(i, low, expressions))
+
+ if not any_int_re.search(high):
+ high = "{0}".format(cls._alphaconv(i, high, expressions))
+
+ if (
+ not low or not high or int(low) > int(high)
+ or not only_int_re.search(str(step))
+ ):
+ raise CroniterBadDateError(
+ "[{0}] is not acceptable".format(expr_format))
+
+ low, high, step = map(int, [low, high, step])
+ rng = range(low, high + 1, step)
+ e_list += (["{0}#{1}".format(item, nth) for item in rng]
+ if i == 4 and nth else rng)
+ else:
+ if t.startswith('-'):
+ raise CroniterBadCronError(
+ "[{0}] is not acceptable,\
+ negative numbers not allowed".format(
+ expr_format))
+ if not star_or_int_re.search(t):
+ t = cls._alphaconv(i, t, expressions)
+
+ try:
+ t = int(t)
+ except:
+ pass
+
+ if t in cls.LOWMAP[i]:
+ t = cls.LOWMAP[i][t]
+
+ if (
+ t not in ["*", "l"]
+ and (int(t) < cls.RANGES[i][0] or
+ int(t) > cls.RANGES[i][1])
+ ):
+ raise CroniterBadCronError(
+ "[{0}] is not acceptable, out of range".format(
+ expr_format))
+
+ res.append(t)
+
+ if i == 4 and nth:
+ if t not in nth_weekday_of_month:
+ nth_weekday_of_month[t] = set()
+ nth_weekday_of_month[t].add(int(nth))
+
+ res.sort()
+ expanded.append(['*'] if (len(res) == 1
+ and res[0] == '*')
+ else res)
+
+ return expanded, nth_weekday_of_month
+
+ @classmethod
+ def is_valid(cls, expression):
+ try:
+ cls.expand(expression)
+ except CroniterError:
+ return False
+ else:
+ return True
diff --git a/src/croniter/tests/test_croniter.py b/src/croniter/tests/test_croniter.py
index 9238daa..267791d 100755
--- a/src/croniter/tests/test_croniter.py
+++ b/src/croniter/tests/test_croniter.py
@@ -2,10 +2,10 @@
# -*- coding: utf-8 -*-
import unittest
-from datetime import datetime, timedelta
+from datetime import datetime
from time import sleep
import pytz
-from croniter import croniter, CroniterBadDateError
+from croniter import croniter, CroniterBadDateError, CroniterBadCronError, CroniterNotAlphaError
from croniter.tests import base
@@ -733,5 +733,17 @@ class CroniterTest(base.TestCase):
val = croniter('0 * * * *', local_date).get_next(datetime)
self.assertEqual(val, tz.localize(datetime(2017, 10, 29, 6)))
+ def test_error_alpha_cron(self):
+ self.assertRaises(CroniterNotAlphaError, croniter.expand, '* * * janu-jun *')
+
+ def test_error_bad_cron(self):
+ self.assertRaises(CroniterBadCronError, croniter.expand, '* * * *')
+ self.assertRaises(CroniterBadCronError, croniter.expand, '* * * * * * *')
+
+ def test_is_valid(self):
+ self.assertTrue(croniter.is_valid('0 * * * *'))
+ self.assertFalse(croniter.is_valid('0 * *'))
+ self.assertFalse(croniter.is_valid('* * * janu-jun *'))
+
if __name__ == '__main__':
unittest.main()