improve timer tests and add some fixes

boswatch/utils/timer.py

@@ -37,6 +37,8 @@ class RepeatedTimer:
         self._kwargs = kwargs
         self._start = 0
         self._overdueCount = 0
+        self._lostEvents = 0
+        self._isRunning = False
         self._event = Event()
         self._thread = None
 
@@ -45,14 +47,18 @@ class RepeatedTimer:
 
         @return True or False"""
         try:
-            self._event.clear()
-            self._thread = Thread(target=self._target)
-            self._thread.name = "RepTim(" + str(self._interval) + ")"
-            self._thread.daemon = True  # start as daemon (thread dies if main program ends)
-            self._thread.start()
-            logging.debug("start repeatedTimer: %s", self._thread.name)
-            return True
-        except:
+            if self._thread is None:
+                self._event.clear()
+                self._thread = Thread(target=self._target)
+                self._thread.name = "RepTim(" + str(self._interval) + ")"
+                self._thread.daemon = True  # start as daemon (thread dies if main program ends)
+                self._thread.start()
+                logging.debug("start repeatedTimer: %s", self._thread.name)
+                return True
+            else:
+                logging.debug("repeatedTimer always started")
+                return True
+        except:  # pragma: no cover
             logging.exception("cannot start timer worker thread")
             return False
 
@@ -64,6 +70,7 @@ class RepeatedTimer:
         if self._thread is not None:
             logging.debug("stop repeatedTimer: %s", self._thread.name)
             self._thread.join()
+            self._thread = None
             return True
         else:
             logging.warning("repeatedTimer always stopped")
@@ -78,14 +85,17 @@ class RepeatedTimer:
 
             try:
                 self._function(*self._args, **self._kwargs)
-            except:
+            except:  # pragma: no cover
                 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:
-                logging.warning("timer overdue! interval: %0.3f sec. - runtime: %0.3f sec.", self._interval, runTime)
+                lostEvents = int(runTime / self._interval)
+                logging.warning("timer overdue! interval: %0.3f sec. - runtime: %0.3f sec. - "
+                                "%d events lost - next call in: %0.3f sec.", self._interval, runTime, lostEvents, self.restTime)
+                self._lostEvents += lostEvents
                 self._overdueCount += 1
         logging.debug("repeatedTimer thread stopped: %s", self._thread.name)
 
@@ -98,3 +108,8 @@ class RepeatedTimer:
     def overdueCount(self):
         """!Property to get a count over all overdues"""
         return self._overdueCount
+
+    @property
+    def lostEvents(self):
+        """!Property to get a count over all los events"""
+        return self._lostEvents

test/test_timer.py

@@ -20,8 +20,6 @@ import pytest
 
 from boswatch.utils.timer import RepeatedTimer
 
-# todo add more tests to overlap all testcases
-
 
 class Test_Timer:
     """!Unittest for the timer class"""
@@ -31,29 +29,59 @@ class Test_Timer:
 
     @staticmethod
     def testTargetFast():
+        """!Fast worker thread"""
         logging.debug("run testTargetFast")
 
     @staticmethod
     def testTargetSlow():
+        """!Slow worker thread"""
         logging.debug("run testTargetSlow start")
-        time.sleep(1)
+        time.sleep(0.51)
         logging.debug("run testTargetSlow end")
 
     @pytest.fixture(scope="function")
-    def useTimer(self):
-        """!Server a RepeatedTimer instance"""
-        self.testTimer = RepeatedTimer(0.5, Test_Timer.testTargetFast)
-        time.sleep(0.1)
+    def useTimerFast(self):
+        """!Server a RepeatedTimer instance with fast worker"""
+        self.testTimer = RepeatedTimer(0.1, Test_Timer.testTargetFast)
         yield 1  # server the timer instance
 
-    def test_timerStartStop(self, useTimer):
+    @pytest.fixture(scope="function")
+    def useTimerSlow(self):
+        """!Server a RepeatedTimer instance slow worker"""
+        self.testTimer = RepeatedTimer(0.1, Test_Timer.testTargetSlow)
+        yield 1  # server the timer instance
+
+    # test cases starts here
+
+    def test_timerStartStop(self, useTimerFast):
         assert self.testTimer.start()
         assert self.testTimer.stop()
 
-    def test_timerStopNotStarted(self, useTimer):
+    def test_timerDoubleSTart(self, useTimerFast):
+        assert self.testTimer.start()
+        assert self.testTimer.start()
+        assert self.testTimer.stop()
+
+    def test_timerStopNotStarted(self, useTimerFast):
         assert not self.testTimer.stop()
 
-    def test_timerRun(self, useTimer):
+    def test_timerRun(self, useTimerFast):
         assert self.testTimer.start()
-        time.sleep(0.6)
+        time.sleep(0.2)
         assert self.testTimer.stop()
+        assert self.testTimer.overdueCount == 0
+        assert self.testTimer.lostEvents == 0
+
+    def test_timerOverdue(self, useTimerSlow):
+        assert self.testTimer.start()
+        time.sleep(0.2)
+        assert self.testTimer.stop()
+        assert self.testTimer.overdueCount == 1
+        assert self.testTimer.lostEvents == 5
+
+    def test_timerOverdueLong(self, useTimerSlow):
+        assert self.testTimer.start()
+        time.sleep(1)
+        assert self.testTimer.stop()
+        assert self.testTimer.overdueCount == 2
+        assert self.testTimer.lostEvents == 10