~yerinalexey/pcrond

7d24b53930870e1f0954b1469a89be6545d8256b — Luca Vercelli 2 years ago 19d99b9
more tests
6 files changed, 70 insertions(+), 23 deletions(-)

M .gitignore
M pcrond/job.py
M pcrond/sched.py
M test_scheduler.py
M tests/crontab.txt
A tests/crontab2.txt
M .gitignore => .gitignore +3 -0
@@ 102,3 102,6 @@ venv.bak/

# mypy
.mypy_cache/

# this is specific for our tests
tests/somefile

M pcrond/job.py => pcrond/job.py +8 -6
@@ 1,7 1,8 @@


import datetime
import logging
logger = logging.getLogger('schedule')
logger = logging.getLogger('pcrond')
reboot_time = datetime.datetime.now()

ALIASES = {'@yearly':    '0 0 1 1 *',
           '@annually':  '0 0 1 1 *',


@@ 10,6 11,11 @@ ALIASES = {'@yearly':    '0 0 1 1 *',
           '@daily':     '0 0 * * *',
           '@midnight':  '0 0 * * *',
           '@hourly':    '0 * * * *',
           '@reboot':    '%d %d %d %d * %d' % (reboot_time.minute + 1,
                                               reboot_time.hour,
                                               reboot_time.day,
                                               reboot_time.month,
                                               reboot_time.year)
           }




@@ 46,10 52,6 @@ class Job(object):

        if crontab in ALIASES.keys():
            crontab = ALIASES[crontab]
        elif crontab == "@reboot":
            import datetime
            now = datetime.datetime.now()
            crontab = "%d %d * %d * %d" % (now.minute + 1, now.hour, now.month, now.year)

        crontab_lst = crontab.split()


M pcrond/sched.py => pcrond/sched.py +24 -15
@@ 4,17 4,24 @@ from .job import Job, ALIASES
import logging
import time

logger = logging.getLogger('schedule')
logger = logging.getLogger('pcrond')


def std_launch_func(cmd_splitted):
def std_launch_func(cmd_splitted, stdin=None):
    """
    Default way of executing commands is to invoke subprocess.run()
    """
    def f():
        import subprocess
        subprocess.run(cmd_splitted)
        # not returning anything here
    if stdin is None:
        def f():
            from subprocess import run
            run(cmd_splitted)
            # not returning anything here
    else:
        def f():
            from subprocess import Popen, PIPE
            p = Popen(cmd_splitted, stdin=PIPE)
            p.communicate(input=stdin)
            # not returning anything here
    return f




@@ 27,6 34,7 @@ class Scheduler(object):
    def __init__(self):
        self.delay = 60         # in seconds
        self.jobs = []
        self.ask_for_stop = False

    def run_pending(self):
        """


@@ 88,7 96,7 @@ class Scheduler(object):
        self.jobs.append(job)
        return job

    def _load_crontab_line(self, rownum, crontab_line, job_func_func=std_launch_func):
    def _load_crontab_line(self, rownum, crontab_line, job_func_func=std_launch_func, stdin=None):
        """
        create a Job from a single crontab entry, and add it to this Scheduler
        :param crontab_line:


@@ 105,10 113,10 @@ class Scheduler(object):
                # CASE 1 - pattern using alias
                job = self.cron(pieces[0], job_func_func(pieces[1:]))
                return job
            except ValueError:
            except ValueError as e:
                # shouldn't happen
                logger.error("Error at line %d, cannot parse pattern, the line will be ignored" % rownum)
                logger.exception('Caused by:')
                logger.error(("Error at line %d, cannot parse pattern, the line will be ignored.\r\n" +
                              "Inner Exception: %s") % (rownum, str(e)))
                return None
        if len(pieces) < 6:
            logger.error("Error at line %d, expected at least 6 tokens" % rownum)


@@ 124,9 132,9 @@ class Scheduler(object):
            # CASE 3 - pattern not including  year
            job = self.cron(" ".join(pieces[0:5]), job_func_func(pieces[5:]))
            return job
        except ValueError:
            logger.error("Error at line %d, cannot parse pattern, the line will be ignored" % rownum)
            logger.exception('Caused by:')
        except ValueError as e:
            logger.error(("Error at line %d, cannot parse pattern, the line will be ignored.\r\n" +
                          "Inner Exception: %s") % (rownum, str(e)))
            return None

    def _split_input_line(self, s):


@@ 162,7 170,8 @@ class Scheduler(object):
                    if line != "" and line[0] != "#":
                        # skip empty lines and comments
                        pieces = self._split_input_line(line)
                        self._load_crontab_line(rownum, pieces[0], job_func_func)
                        stdin = pieces[1] if len(pieces) > 1 else None
                        self._load_crontab_line(rownum, pieces[0], job_func_func, stdin)
                        # TODO support % sign inside command, should consider pieces[1] if any
        logger.info(str(len(self.jobs)) + " jobs loaded from configuration file")



@@ 171,6 180,6 @@ class Scheduler(object):
        Perform main run-and-wait loop.
        """
        import time
        while True:
        while not self.ask_for_stop:
            self.run_pending()
            time.sleep(self.delay)

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

import logging
from datetime import datetime as d
from pcrond import scheduler, Job, Parser

# when tests with a logger fail, you can set this to True
SHOW_LOGGING = False

logger = logging.getLogger()
if SHOW_LOGGING:
    logging.basicConfig()
else:
    logger.addHandler(logging.NullHandler())  # do not show logs.


def do_nothing():
    pass


@@ 222,6 231,25 @@ class SchedulerTests(unittest.TestCase):
        scheduler.load_crontab_file(os.path.join("tests", "crontab.txt"))
        assert len(scheduler.jobs) == 4

    def _test_load_crontab_and_main_loop(self):
        # FIXME not working
        # and even if it worked, this is a long test, and will run on *nix only
        import os
        import time
        from threading import Thread

        def run_in_another_thread():
            scheduler.load_crontab_file(os.path.join("tests", "crontab2.txt"))
            scheduler.main_loop()

        start_time = d.now()
        thread = Thread(target=run_in_another_thread)
        thread.start()
        time.sleep(65)
        thread.stop()
        assert os.path.isfile(os.path.join("tests", "somefile"))
        assert os.path.getmtime(path) >= d.utcfromtimestamp(start_time)


if __name__ == '__main__':
    unittest.main()

M tests/crontab.txt => tests/crontab.txt +2 -0
@@ 1,5 1,7 @@
# crontab entries used during tests

30 4 * * mon echo goofy
30 5-6 * * mon echo donald duck
30 * */3 * mon 2020 echo mickey mouse
here is some very bad pattern that must not break other tasks to run
@daily echo it's midnight

A tests/crontab2.txt => tests/crontab2.txt +3 -0
@@ 0,0 1,3 @@
# crontab entries used during tests

@reboot touch tests/somefile