From 40c07dc48eadc189fa6ee9e34269f17a6052151e Mon Sep 17 00:00:00 2001 From: otherpirate Date: Thu, 6 Jul 2017 19:01:36 -0300 Subject: Creating base croniter error --- src/croniter/croniter.py | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/src/croniter/croniter.py b/src/croniter/croniter.py index 55bce86..3e837f4 100644 --- a/src/croniter/croniter.py +++ b/src/croniter/croniter.py @@ -16,16 +16,20 @@ any_int_re = re.compile(r'^\d+') star_or_int_re = re.compile(r'^(\d+|\*)$') -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): -- cgit v1.2.1 From 30caae2ae7717726a4736ae0df3e101addc452bb Mon Sep 17 00:00:00 2001 From: otherpirate Date: Thu, 6 Jul 2017 19:32:14 -0300 Subject: Adding class method is_valid to validate cron syntax --- src/croniter/croniter.py | 210 +++++++++++++++++++----------------- src/croniter/tests/test_croniter.py | 16 ++- 2 files changed, 127 insertions(+), 99 deletions(-) diff --git a/src/croniter/croniter.py b/src/croniter/croniter.py index 3e837f4..a9f10a3 100644 --- a/src/croniter/croniter.py +++ b/src/croniter/croniter.py @@ -14,6 +14,7 @@ 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 CroniterError(ValueError): @@ -83,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) + self.expanded, self.nth_weekday_of_month = self.expand(expr_format) - 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) - - 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) @@ -196,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): """ @@ -536,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() -- cgit v1.2.1 From 66cdb9e755127053af25082817f3d8318136a41f Mon Sep 17 00:00:00 2001 From: otherpirate Date: Thu, 6 Jul 2017 19:38:06 -0300 Subject: Adding is_valid class method to readme --- README.rst | 5 +++++ 1 file changed, 5 insertions(+) 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 !:: -- cgit v1.2.1