mirror of
https://github.com/BOSWatch/BW3-Core.git
synced 2026-01-05 08:09:58 +01:00
improve exc handling and tests
This commit is contained in:
parent
91389a9c5b
commit
11eab418ed
|
|
@ -26,8 +26,7 @@ class TCPClient:
|
|||
def __init__(self, timeout=3):
|
||||
"""!Create a new instance
|
||||
|
||||
Create a new instance of an TCP Client.
|
||||
And set the timeout"""
|
||||
@param timeout: timout for the client in sec. (3)"""
|
||||
try:
|
||||
self._sock = None
|
||||
self._timeout = timeout
|
||||
|
|
|
|||
|
|
@ -75,7 +75,9 @@ class TCPServer(socketserver.ThreadingMixIn, socketserver.TCPServer):
|
|||
"""!TCP server class"""
|
||||
|
||||
def __init__(self, timeout=3):
|
||||
"""!Create a new instance"""
|
||||
"""!Create a new instance
|
||||
|
||||
@param timeout: timeout for the server in sec. (3)"""
|
||||
self._server = None
|
||||
self._server_thread = None
|
||||
self._timeout = timeout
|
||||
|
|
|
|||
|
|
@ -36,17 +36,24 @@ class RepeatedTimer:
|
|||
self._args = args
|
||||
self._kwargs = kwargs
|
||||
self._start = 0
|
||||
self._overdueCount = 0
|
||||
self._event = Event()
|
||||
self._thread = None
|
||||
|
||||
def start(self):
|
||||
"""!Start a new timer worker thread"""
|
||||
self._event.clear()
|
||||
self._thread = Thread(target=self._target)
|
||||
self._thread.name = "RepTim(" + str(self._interval) + ")"
|
||||
self._thread.start()
|
||||
logging.debug("start repeatedTimer: %s", self._thread.name)
|
||||
return True
|
||||
"""!Start a new timer worker thread
|
||||
|
||||
@return True or False"""
|
||||
try:
|
||||
self._event.clear()
|
||||
self._thread = Thread(target=self._target)
|
||||
self._thread.name = "RepTim(" + str(self._interval) + ")"
|
||||
self._thread.start()
|
||||
logging.debug("start repeatedTimer: %s", self._thread.name)
|
||||
return True
|
||||
except:
|
||||
logging.exception("cannot start timer worker thread")
|
||||
return False
|
||||
|
||||
def stop(self):
|
||||
"""!Stop the timer worker thread
|
||||
|
|
@ -78,8 +85,13 @@ class RepeatedTimer:
|
|||
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)
|
||||
|
||||
self._overdueCount += 1
|
||||
@property
|
||||
def restTime(self):
|
||||
"""!Property to get remaining time till next call"""
|
||||
return self._interval - ((time.time() - self._start) % self._interval)
|
||||
|
||||
@property
|
||||
def overdueCount(self):
|
||||
"""!Property to get a count over all iverdues"""
|
||||
return self._overdueCount
|
||||
|
|
|
|||
|
|
@ -20,6 +20,7 @@ import pytest
|
|||
|
||||
from boswatch.utils.timer import RepeatedTimer
|
||||
|
||||
# todo add more tests to overlap all testcases
|
||||
|
||||
class Test_Timer:
|
||||
"""!Unittest for the timer class"""
|
||||
|
|
@ -28,15 +29,21 @@ class Test_Timer:
|
|||
logging.debug("[TEST] %s.%s", type(self).__name__, method.__name__)
|
||||
|
||||
@staticmethod
|
||||
def testTarget():
|
||||
logging.debug("run testTarget")
|
||||
def testTargetFast():
|
||||
logging.debug("run testTargetFast")
|
||||
|
||||
@staticmethod
|
||||
def testTargetSlow():
|
||||
logging.debug("run testTargetSlow start")
|
||||
time.sleep(1)
|
||||
logging.debug("run testTargetSlow end")
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def useTimer(self):
|
||||
"""!Server a RepeatedTimer instance"""
|
||||
self.testTimer = RepeatedTimer(0.5, Test_Timer.testTarget)
|
||||
self.testTimer = RepeatedTimer(0.5, Test_Timer.testTargetFast)
|
||||
time.sleep(0.1)
|
||||
yield 1
|
||||
yield 1 # server the timer instance
|
||||
|
||||
def test_timerStartStop(self, useTimer):
|
||||
assert self.testTimer.start()
|
||||
|
|
|
|||
Loading…
Reference in a new issue