~yerinalexey/pcrond

746cd13668864cf27419eff6cb3dc5a890655614 — Luca Vercelli 2 years ago e6b9038
more tests
2 files changed, 31 insertions(+), 14 deletions(-)

M pcrond/job.py
M test_scheduler.py
M pcrond/job.py => pcrond/job.py +9 -5
@@ 76,7 76,7 @@ class Job(object):

        # Day of week.
        # 5L = last friday of the month
        # 2#5 = second friday of the month
        # 5#2 = second friday of the month
        [self.allowed_every_dow, self.allowed_dow] = self._parse_day_in_week(crontab_lst[4])

        self.allowed_last_dom = (-1 in self.allowed_dom)


@@ 164,9 164,6 @@ class Job(object):
                    "Wrong format '%s' - expecting an integer after '*/'" % s)
            return [False, set(range(minval, maxval, step))]
        else:                           # at given minutes
            # DEBUG
            # import pdb
            # pdb.set_trace()
            # here s == '1,2-5/3,jul,10-nov'
            (singletons, ranges) = self._split_tokens(s)
            # here singletons == ['1', 'jul'], ranges == [['2', '5', 3], ['10', 'nov', 1]]


@@ 206,9 203,16 @@ class Job(object):
        last_day_of_month = calendar.monthrange(now.year, now.month)[1]
        return last_day_of_month

    def get_num_wom(self, now):
        """
        for a given date, return the number of the week in the month (1..5)
        intended for #
        """
        return ((now.day - 1) // 7) + 1

    def is_last_wom(self, now):
        """ true if given date is in the last week of the month """
        return now.day >= self.get_last_dom(now) - 7
        return now.day > self.get_last_dom(now) - 7

    def __lt__(self, other):
        """

M test_scheduler.py => test_scheduler.py +22 -9
@@ 1,6 1,6 @@
#!/usr/bin/env python
"""Unit tests for pcrond.py"""
import datetime
from datetime import datetime as d
import unittest

from pcrond import scheduler, Job


@@ 41,6 41,19 @@ class SchedulerTests(unittest.TestCase):
        with self.assertRaises(ValueError):
            job._decode_token("goofy", {})

    def test_get_num_wom(self):
        job = Job()
        assert job.get_num_wom(d(2019, 3, 7)) == 1      #first thursday of month
        assert job.get_num_wom(d(2019, 3, 8)) == 2      #second friday of month
        assert job.get_num_wom(d(2019, 3, 31)) == 5     #fifth sunday of month

    def test_is_last_wom(self):
        job = Job()
        assert not job.is_last_wom(d(2019, 3, 1))
        assert not job.is_last_wom(d(2019, 3, 24))
        assert job.is_last_wom(d(2019, 3, 25))
        assert job.is_last_wom(d(2019, 3, 31))

    def test_job_constructor_basic(self):
        job = Job("* 4 * * *")
        ###


@@ 82,9 95,9 @@ class SchedulerTests(unittest.TestCase):
        job = Job("* * L * *")
        assert job.allowed_last_dom
        assert job.allowed_dom == set([-1])
        assert job._should_run_at(datetime.datetime(2019, 3, 31))
        assert not job._should_run_at(datetime.datetime(2019, 3, 28))
        assert job._should_run_at(datetime.datetime(2019, 2, 28))
        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_L_dow(self):
        job = Job("* * * * 5l")      # 5=friday, l=only the last one of the month


@@ 93,9 106,9 @@ class SchedulerTests(unittest.TestCase):
        assert not job.allowed_last_dom
        assert not job.allowed_every_dow
        assert job.allowed_dow == set([5-7])
        assert job._should_run_at(datetime.datetime(2019, 3, 29))       # was fri
        assert not job._should_run_at(datetime.datetime(2019, 3, 28))   # was thu
        assert not job._should_run_at(datetime.datetime(2019, 3, 8))    # was fri
        assert job._should_run_at(d(2019, 3, 29))       # was fri
        assert not job._should_run_at(d(2019, 3, 28))   # was thu
        assert not job._should_run_at(d(2019, 3, 8))    # was fri

    def test_job_constructor_alias(self):
        job = Job("@hourly")


@@ 146,7 159,7 @@ class SchedulerTests(unittest.TestCase):
    def test_add_job_run_pending(self):
        """ schedule a task for this exact minute, then invoke run_pending()"""
        test_obj = {'modified': False}
        now = datetime.datetime.now()
        now = d.now()
        scheduler.cron("%d %d * * *" % (now.minute, now.hour), modify_obj(test_obj))
        assert len(scheduler.jobs) == 1
        scheduler.run_pending()


@@ 155,7 168,7 @@ class SchedulerTests(unittest.TestCase):
    def test_add_job_run_pending_not(self):
        """ schedule a task for this exact minute+5, then invoke run_pending()"""
        test_obj = {'modified': False}
        now = datetime.datetime.now()
        now = d.now()
        scheduler.cron("%d %d * * *" % (now.minute+5, now.hour), modify_obj(test_obj))
        assert len(scheduler.jobs) == 1
        scheduler.run_pending()