~yerinalexey/pcrond

3192416968b175ee3b9500153b2050df513b8fad — Luca Vercelli 2 years ago 1e127b7
threading tests
3 files changed, 25 insertions(+), 11 deletions(-)

M pcrond/job.py
M pcrond/sched.py
M test_scheduler.py
M pcrond/job.py => pcrond/job.py +10 -5
@@ 1,8 1,14 @@

import datetime
from datetime import datetime, timedelta
import logging
logger = logging.getLogger('pcrond')
reboot_time = datetime.datetime.now()

"""
@reboot commands will be executed after this number of seconds
"""
WAIT_BEFORE_REBOOT = 10

reboot_time = datetime.now() + timedelta(seconds=WAIT_BEFORE_REBOOT)

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


@@ 11,7 17,7 @@ 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':    '%d %d %d %d * %d' % (reboot_time.minute,
                                               reboot_time.hour,
                                               reboot_time.day,
                                               reboot_time.month,


@@ 128,8 134,7 @@ class Job(object):
        """
        :return: ``True`` if the job should be run now.
        """
        import datetime
        now = datetime.datetime.now()
        now = datetime.now()
        return self._should_run_at(now)

    def _should_run_at(self, now):

M pcrond/sched.py => pcrond/sched.py +3 -3
@@ 13,13 13,13 @@ def std_launch_func(cmd_splitted, stdin=None):
    """
    if stdin is None:
        def f():
            from subprocess import run
            run(cmd_splitted)
            from subprocess import Popen
            p = Popen(cmd_splitted, stdin=None, stdout=None, stderr=None)
            # not returning anything here
    else:
        def f():
            from subprocess import Popen, PIPE
            p = Popen(cmd_splitted, stdin=PIPE)
            p = Popen(cmd_splitted, stdin=PIPE, stdout=None, stderr=None)
            p.communicate(input=stdin)
            # not returning anything here
    return f

M test_scheduler.py => test_scheduler.py +12 -3
@@ 2,6 2,7 @@
"""Unit tests for pcrond.py"""
import unittest
import logging
import sys
from datetime import datetime as d
from pcrond import scheduler, Job, Parser



@@ 231,7 232,8 @@ 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):
    @unittest.skipIf(sys.platform.startswith("win"), "requires *NIX")
    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


@@ 245,11 247,18 @@ class SchedulerTests(unittest.TestCase):
        start_time = d.now()
        thread = Thread(target=run_in_another_thread)
        thread.start()
        time.sleep(65)
        thread.stop()
        print("Waiting for 15 seconds...")
        time.sleep(15)
        scheduler.ask_for_stop = True
        thread.join()
        assert os.path.isfile(os.path.join("tests", "somefile"))
        assert os.path.getmtime(os.path.join("tests", "somefile")) >= d.utcfromtimestamp(start_time)

    @unittest.skipUnless(sys.platform.startswith("win"), "requires Windows")
    def test_load_crontab_and_main_loop(self):
        # TODO
        pass


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