~yerinalexey/pcrond

e3907ce836e95a178fb1ec5ffb0c91f4cdda8f5f — Luca Vercelli 2 years ago a000241
param minday
2 files changed, 33 insertions(+), 11 deletions(-)

M pcrond/job.py
M test_scheduler.py
M pcrond/job.py => pcrond/job.py +11 -11
@@ 136,7 136,7 @@ class Job(object):
        # here ranges_with_step == [['2', '5', 3]]
        return (singletons, ranges_no_step + ranges_with_step)

    def _parse_common(self, s, maxval, offsets={}, minval=0, callback=None):
    def _parse_common(self, s, minval, maxval, offsets={}, callback=None):
        """
        Generate a set of integers, corresponding to "allowed values".
        Work for minute, hours, weeks, month, ad days of week, because they


@@ 148,7 148,7 @@ class Job(object):
        :param offsets:
            a dict mapping names (es. "mar") to their offsets (es. 2).
        :param minval:
            es. 1 for days and months
            es. 0 for hours and minutes, 1 for days and months
        :param callback:
            a 2-ary function that pre-elaborates singletons and ranges
        """


@@ 181,10 181,10 @@ class Job(object):
            return [False, set(flatlist)]

    def _parse_min(self, s):
        return self._parse_common(s, 60)
        return self._parse_common(s, 0, 60)

    def _parse_hour(self, s):
        return self._parse_common(s, 24)
        return self._parse_common(s, 0, 24)

    def _parse_day_in_month(self, s):
        def ignore_w(singletons, ranges):


@@ 195,21 195,21 @@ class Job(object):
        def only_w(singletons, ranges):
            return ([x[:-1] for x in singletons if x[-1] == 'w'], [])

        [every, dom] = self._parse_common(s, 31, {'l': '-1'}, 1, ignore_w)
        [every, dom] = self._parse_common(s, 1, 31, {'l': '-1'}, ignore_w)
        if every:
            wdom = None
        else:
            [every, wdom] = self._parse_common(s, 31, {}, 1, only_w)
            [every, wdom] = self._parse_common(s, 1, 31, {}, only_w)
        return [every, dom, wdom]

    def _parse_month(self, s):
        return self._parse_common(s, 12, MONTH_OFFSET, 1)
        return self._parse_common(s, 0, 12, MONTH_OFFSET)

    def _parse_day_in_week(self, s):
        return self._parse_common(s, 7, WEEK_OFFSET)
        return self._parse_common(s, 0, 7, WEEK_OFFSET)

    def _parse_year(self, s):
        return self._parse_common(s, 2099, minval=1970)
        return self._parse_common(s, 1970, 2099)

    def get_last_dom(self, now):
        """ get last day in month determined by given datetime """


@@ 255,10 255,10 @@ class Job(object):
                and (self.allowed_every_hour or now.hour in self.allowed_hours)
                and (self.allowed_every_min or now.minute in self.allowed_min)
                and (self.allowed_every_dow
                     or (now.weekday() in self.allowed_dow
                     or (now.weekday() in self.allowed_dow)
                     or (self.must_consider_wom
                         and now.weekday() in self.allowed_dowl
                         and self.is_last_wom(now))))
                         and self.is_last_wom(now)))
                and (self.allowed_every_dom
                     or now.day in self.allowed_dom
                     or (self.allowed_last_dom and now.day == self.get_last_dom(now)))

M test_scheduler.py => test_scheduler.py +22 -0
@@ 92,12 92,32 @@ class SchedulerTests(unittest.TestCase):

    def test_job_constructor_L_dom(self):
        job = Job("* * L * *")
        assert not job.allowed_every_dom
        assert job.allowed_last_dom
        assert job.allowed_dom == set([-1])
        assert job._should_run_at(d(2019, 3, 31))
        assert not job._should_run_at(d(2019, 3, 28))
        assert job._should_run_at(d(2019, 2, 28))

    def test_job_constructor_W_dom(self):
        job = Job("* * 15w * *")
        assert not job.allowed_every_dom
        assert not job.allowed_last_dom
        assert job.allowed_dom == set()
        assert job.allowed_wdom == set([15])
        assert job._should_run_at(d(2019, 3, 15))       # was fri
        assert not job._should_run_at(d(2018, 12, 15))  # was sat
        assert job._should_run_at(d(2019, 3, 14))       # was fri
        assert not job._should_run_at(d(2018, 7, 15))   # was sun
        assert job._should_run_at(d(2018, 7, 16))       # was mon
        job = Job("* * 1w * *")
        assert not job.allowed_every_dom
        assert not job.allowed_last_dom
        assert job.allowed_dom == set()
        assert job.allowed_wdom == set([1])
        assert job._should_run_at(d(2019, 3, 1))        # was fri
        assert not job._should_run_at(d(2019, 6, 3))    # was mon

    def test_job_constructor_L_dow(self):
        job = Job("* * * * 5l")      # 5=friday, l=only the last one of the month
        assert job.must_consider_wom


@@ 136,6 156,8 @@ class SchedulerTests(unittest.TestCase):
            Job("* @hourly")
        with self.assertRaises(ValueError):
            Job("* L * * *")
        with self.assertRaises(ValueError):
            Job("* * 1w2 * *")
        # currently, hour=25 does not raise errors.

    def test_misconfigured_job_wont_break_scheduler(self):