~yerinalexey/pcrond

61b6791c81c1e3e3a6a2c93e7a40626a686d178d — Luca Vercelli 2 years ago aa84c38
run test
2 files changed, 14 insertions(+), 4 deletions(-)

M pcrond/sched.py
M test_scheduler.py
M pcrond/sched.py => pcrond/sched.py +7 -0
@@ 13,11 13,13 @@ def std_launch_func(cmd_splitted, stdin=None):
    """
    if stdin is None:
        def f():
            logger.info("Now running: " + str(cmd_splitted))
            from subprocess import Popen
            Popen(cmd_splitted, stdin=None, stdout=None, stderr=None)
            # not returning anything here
    else:
        def f():
            logger.info("Now running: " + str(cmd_splitted))
            from subprocess import Popen, PIPE
            p = Popen(cmd_splitted, stdin=PIPE, stdout=None, stderr=None)
            p.communicate(input=stdin)


@@ 45,7 47,9 @@ class Scheduler(object):
        in one hour increments then your job won't be run 60 times in
        between but only once.
        """
        logger.debug("available jobs: " + str(self.jobs))
        runnable_jobs = (job for job in self.jobs if job.should_run())
        logger.debug("runnable jobs: " + str(self.jobs))
        for job in runnable_jobs:
            job.run()



@@ 183,3 187,6 @@ class Scheduler(object):
        while not self.ask_for_stop:
            self.run_pending()
            time.sleep(self.delay)
            # FIXME this will look at self.ask_for_stop only every self.delay seconds
            # see https://stackoverflow.com/questions/5114292/break-interrupt-a-time-sleep-in-python


M test_scheduler.py => test_scheduler.py +7 -4
@@ 11,7 11,7 @@ SHOW_LOGGING = False

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



@@ 234,8 234,7 @@ class SchedulerTests(unittest.TestCase):

    @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
        # FIXME this is a slow test, currently 1 minute
        import os
        import time
        from threading import Thread


@@ 250,9 249,13 @@ class SchedulerTests(unittest.TestCase):
        print("Waiting for 15 seconds...")
        time.sleep(15)
        scheduler.ask_for_stop = True
        print("Waiting other thread to stop...")
        thread.join()
        print("Other thread stopped.")
        assert os.path.isfile(os.path.join("tests", "somefile"))
        assert os.path.getmtime(os.path.join("tests", "somefile")) >= d.utcfromtimestamp(start_time)
        #FIXME system may be utc or not...
        #assert d.utcfromtimestamp(os.path.getmtime(os.path.join("tests", "somefile"))) >= start_time
        assert d.fromtimestamp(os.path.getmtime(os.path.join("tests", "somefile"))) >= start_time

    @unittest.skipUnless(sys.platform.startswith("win"), "requires Windows")
    def test_load_crontab_and_main_loop_win(self):