diff --git a/boswatch/utils/timer.py b/boswatch/utils/timer.py index 49fb809..64cf379 100644 --- a/boswatch/utils/timer.py +++ b/boswatch/utils/timer.py @@ -44,25 +44,36 @@ class RepeatedTimer: self._event.clear() self._thread = Thread(target=self._target) self._thread.name = "RepTim(" + str(self._interval) + ")" - self._start = time.time() self._thread.start() logging.debug("start repeatedTimer: %s", self._thread.name) + return True def stop(self): - """!Stop the timer worker thread""" + """!Stop the timer worker thread + + @return True or False""" self._event.set() - self._thread.join() - logging.debug("stop repeatedTimer: %s", self._thread.name) + if self._thread is not None: + logging.debug("stop repeatedTimer: %s", self._thread.name) + self._thread.join() + return True + else: + logging.warning("repeatedTimer always stopped") + return False def _target(self): """!Runs the target function with his arguments""" + self._start = time.time() while not self._event.wait(self.restTime): logging.debug("work") startTime = time.time() - self._function(*self._args, **self._kwargs) - time.sleep(1.5) - runTime = time.time() - startTime + try: + self._function(*self._args, **self._kwargs) + except: + logging.exception("target throws an exception") + + runTime = time.time() - startTime if runTime < self._interval: logging.debug("ready after: %0.3f sec. - next call in: %0.3f sec.", runTime, self.restTime) else: diff --git a/test/test_timer.py b/test/test_timer.py new file mode 100644 index 0000000..29161ed --- /dev/null +++ b/test/test_timer.py @@ -0,0 +1,51 @@ +#!/usr/bin/python +# -*- coding: utf-8 -*- +"""! + ____ ____ ______ __ __ __ _____ + / __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ / + / __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ < + / /_/ / /_/ /___/ /| |/ |/ / /_/ / /_/ /__/ / / / ___/ / +/_____/\____//____/ |__/|__/\__,_/\__/\___/_/ /_/ /____/ + German BOS Information Script + by Bastian Schroll + +@file: test_timer.py +@date: 21.09.2019 +@author: Bastian Schroll +@description: Unittests for BOSWatch. File must be _run as "pytest" unittest +""" +import logging +import time +import pytest + +from boswatch.utils.timer import RepeatedTimer + + +class Test_Timer: + """!Unittest for the timer class""" + + def setup_method(self, method): + logging.debug("[TEST] %s.%s", type(self).__name__, method.__name__) + + @staticmethod + def testTarget(): + logging.debug("run testTarget") + + @pytest.fixture(scope="function") + def useTimer(self): + """!Server a RepeatedTimer instance""" + self.testTimer = RepeatedTimer(0.5, Test_Timer.testTarget) + time.sleep(0.1) + yield 1 + + def test_timerStartStop(self, useTimer): + assert self.testTimer.start() + assert self.testTimer.stop() + + def test_timerStopNotStarted(self, useTimer): + assert not self.testTimer.stop() + + def test_timerRun(self, useTimer): + assert self.testTimer.start() + time.sleep(0.6) + assert self.testTimer.stop()