BOSWatch/BW3-Core
1
0
Fork 0
mirror of https://github.com/BOSWatch/BW3-Core.git synced 2026-04-21 06:03:50 +00:00
Code Issues Projects Releases Packages Wiki Activity

edit pytest suite

Browse source
This commit is contained in:
Bastian Schroll 2019-03-03 11:13:27 +01:00
parent e372fb75b4
commit a35dc15d60
11 changed files with 408 additions and 371 deletions
Download patch file Download diff file Expand all files Collapse all files

269
test/test_ServerClient.py
View file

@ -12,7 +12,7 @@
@file: test_ServerClient.py @file: test_ServerClient.py
@date: 10.12.2017 @date: 10.12.2017
@author: Bastian Schroll @author: Bastian Schroll
@description: Unittests for BOSWatch. File must be _run as "pytest" unittest @description: Unittests for BOSWatch. File have to run as "pytest" unittest
""" """
import logging import logging
import time import time
@ -23,149 +23,156 @@ from boswatch.network.server import TCPServer
from boswatch.network.client import TCPClient from boswatch.network.client import TCPClient
class Test_ServerClient: def setup_method(method):
"""!Unittest for the server/client environment""" logging.debug("[TEST] %s.%s", method.__module__, method.__name__)
def setup_method(self, method):
logging.debug("[TEST] %s.%s", type(self).__name__, method.__name__)
@pytest.fixture(scope="function") @pytest.fixture
def useServer(self): def getClient():
"""!Start and serve the sever for each functions where useServer is given""" return TCPClient()
self.dataQueue = queue.Queue()
self.testServer = TCPServer(self.dataQueue)
logging.debug("start server")
assert self.testServer.start()
time.sleep(0.1) # wait for server
# serv the instances - created in self context
yield 1
try:
logging.debug("stop server")
self.testServer.stop()
except:
logging.warning("server still stopped")
time.sleep(0.1) # wait for server
def test_clientConnectFailed(self): @pytest.fixture
"""!Connect to a non available server""" def getServer():
self.testClient = TCPClient() """!Start and serve the sever for each functions where useServer is given"""
assert not self.testClient.connect() dataQueue = queue.Queue()
testServer = TCPServer(dataQueue)
return testServer
def test_clientDisconnectFailed(self):
"""!Disconnect while no connection is established"""
self.testClient = TCPClient()
assert self.testClient.disconnect()
def test_clientTransmitFailed(self): @pytest.fixture
"""!Transmit while no connection is established""" def getRunningServer(getServer):
self.testClient = TCPClient() logging.debug("start server")
assert not self.testClient.transmit("test") assert getServer.start()
if not getServer.isRunning:
pytest.fail("server not running")
yield getServer
logging.debug("stop server")
assert getServer.stop()
def test_clientReceiveFailed(self):
"""!Receive while no connection is established"""
self.testClient = TCPClient()
assert not self.testClient.receive()
def test_clientConnect(self, useServer): def test_clientConnectFailed(getClient):
"""!Connect to a server""" """!Connect to a non available server"""
self.testClient = TCPClient() assert not getClient.connect()
assert self.testClient.connect()
assert self.testClient.disconnect()
def test_clientReconnect(self, useServer):
"""!Try a reconnect after a established connection"""
self.testClient = TCPClient()
assert self.testClient.connect()
assert self.testClient.disconnect()
assert self.testClient.connect()
assert self.testClient.disconnect()
def test_clientMultiConnect(self, useServer): def test_clientDisconnectFailed(getClient):
"""!Connect with 2 clients to the server""" """!Disconnect while no connection is established"""
self.testClient1 = TCPClient() assert getClient.disconnect()
assert self.testClient1.connect()
self.testClient2 = TCPClient()
assert self.testClient2.connect()
time.sleep(0.1) # wait for all clients connected
# check connected clients
assert self.testServer.countClientsConnected() == 2
# disconnect all
assert self.testClient1.disconnect()
assert self.testClient2.disconnect()
def test_clientCommunicate(self, useServer):
"""!Try to send data to the server and check on '[ack]'"""
self.testClient = TCPClient()
assert self.testClient.connect()
assert self.testClient.transmit("test")
assert self.testClient.receive() == "[ack]"
assert self.testClient.disconnect()
def test_clientMultiCommunicate(self, useServer): def test_clientTransmitFailed(getClient):
"""!Try to send data to the server with 3 clients and check on '[ack]'""" """!Transmit while no connection is established"""
# connect all assert not getClient.transmit("test")
self.testClient1 = TCPClient()
assert self.testClient1.connect()
self.testClient2 = TCPClient()
assert self.testClient2.connect()
self.testClient3 = TCPClient()
assert self.testClient3.connect()
# send all
assert self.testClient1.transmit("test")
assert self.testClient2.transmit("test")
assert self.testClient3.transmit("test")
# recv all
assert self.testClient3.receive() == "[ack]"
assert self.testClient2.receive() == "[ack]"
assert self.testClient1.receive() == "[ack]"
# check server msg queue
assert self.dataQueue.qsize() == 3
# disconnect all
assert self.testClient1.disconnect()
assert self.testClient2.disconnect()
assert self.testClient3.disconnect()
def test_serverRestart(self, useServer):
"""!Test a stop and restart of the server"""
assert self.testServer.stop()
assert self.testServer.start()
assert self.testServer.stop()
def test_serverStopFailed(self, useServer): def test_clientReceiveFailed(getClient):
"""!Test to stop a stopped server""" """!Receive while no connection is established"""
assert self.testServer.stop() assert not getClient.receive()
assert self.testServer.stop()
def test_serverDoubleStart(self):
"""!Test to start the server twice"""
self.dataQueue = queue.Queue()
self.testServer1 = TCPServer(self.dataQueue)
self.testServer2 = TCPServer(self.dataQueue)
assert self.testServer1.start()
assert not self.testServer2.start()
assert self.testServer1.stop()
assert self.testServer2.stop()
def test_serverGetOutput(self, useServer): def test_clientConnect(getClient, getRunningServer):
"""!Send data to server with 2 clients, check '[ack]' and data on server queue""" """!Connect to a server"""
# connect all assert getClient.connect()
self.testClient1 = TCPClient() assert getClient.disconnect()
assert self.testClient1.connect()
self.testClient2 = TCPClient()
assert self.testClient2.connect() def test_clientReconnect(getClient, getRunningServer):
# send all """!Try a reconnect after a established connection"""
assert self.testClient1.transmit("test1") assert getClient.connect()
time.sleep(0.1) # wait for recv to prevent fail of false order assert getClient.disconnect()
assert self.testClient2.transmit("test2") assert getClient.connect()
# recv all assert getClient.disconnect()
assert self.testClient1.receive() == "[ack]"
assert self.testClient2.receive() == "[ack]"
# _check server output data def test_clientMultiConnect(getClient, getRunningServer):
assert self.dataQueue.qsize() == 2 """!Connect with 2 clients to the server"""
assert self.dataQueue.get(True, 1)[1] == "test1" assert getClient.connect()
assert self.dataQueue.get(True, 1)[1] == "test2" testClient2 = TCPClient()
assert self.dataQueue.qsize() is 0 # Last _check must be None assert testClient2.connect()
# disconnect all time.sleep(0.1) # wait for all clients connected
assert self.testClient1.disconnect() # check connected clients
assert self.testClient2.disconnect() assert getRunningServer.countClientsConnected() == 2
# disconnect all
assert getClient.disconnect()
assert testClient2.disconnect()
def test_clientCommunicate(getClient, getRunningServer):
"""!Try to send data to the server and check on '[ack]'"""
assert getClient.connect()
assert getClient.transmit("test")
assert getClient.receive() == "[ack]"
assert getClient.disconnect()
def test_clientMultiCommunicate(getServer):
"""!Try to send data to the server with 3 clients and check on '[ack]'"""
# connect all
testClient1 = TCPClient()
assert testClient1.connect()
testClient2 = TCPClient()
assert testClient2.connect()
testClient3 = TCPClient()
assert testClient3.connect()
# send all
assert testClient1.transmit("test")
assert testClient2.transmit("test")
assert testClient3.transmit("test")
# recv all
assert testClient3.receive() == "[ack]"
assert testClient2.receive() == "[ack]"
assert testClient1.receive() == "[ack]"
# check server msg queue
assert dataQueue.qsize() == 3
# disconnect all
assert testClient1.disconnect()
assert testClient2.disconnect()
assert testClient3.disconnect()
def test_serverRestart(getRunningServer):
"""!Test a stop and restart of the server"""
assert getRunningServer.stop()
assert getRunningServer.start()
assert getRunningServer.stop()
def test_serverStopFailed(getServer):
"""!Test to stop a stopped server"""
assert getServer.stop()
def test_serverDoubleStart():
"""!Test to start the server twice"""
dataQueue = queue.Queue()
testServer1 = TCPServer(dataQueue)
testServer2 = TCPServer(dataQueue)
assert testServer1.start()
assert not testServer2.start()
assert testServer1.stop()
assert testServer2.stop()
def test_serverGetOutput(getRunningServer):
"""!Send data to server with 2 clients, check '[ack]' and data on server queue"""
# connect all
testClient1 = TCPClient()
assert testClient1.connect()
testClient2 = TCPClient()
assert testClient2.connect()
# send all
assert testClient1.transmit("test1")
time.sleep(0.1) # wait for recv to prevent fail of false order
assert testClient2.transmit("test2")
# recv all
assert testClient1.receive() == "[ack]"
assert testClient2.receive() == "[ack]"
# _check server output data
assert dataQueue.qsize() == 2
assert dataQueue.get(True, 1)[1] == "test1"
assert dataQueue.get(True, 1)[1] == "test2"
assert dataQueue.qsize() is 0 # Last _check must be None
# disconnect all
assert testClient1.disconnect()
assert testClient2.disconnect()

77
test/test_broadcast.py
View file

@ -12,7 +12,7 @@
@file: test_broadcast.py @file: test_broadcast.py
@date: 25.09.2018 @date: 25.09.2018
@author: Bastian Schroll @author: Bastian Schroll
@description: Unittests for BOSWatch. File must be _run as "pytest" unittest @description: Unittests for BOSWatch. File have to run as "pytest" unittest
""" """
import logging import logging
import pytest import pytest
@ -21,49 +21,50 @@ from boswatch.network.broadcast import BroadcastServer
from boswatch.network.broadcast import BroadcastClient from boswatch.network.broadcast import BroadcastClient
class Test_Broadcast: def setup_method(method):
"""!Unittest for the timer class""" logging.debug("[TEST] %s.%s", method.__module__, method.__name__)
def setup_method(self, method):
logging.debug("[TEST] %s.%s", type(self).__name__, method.__name__)
@pytest.fixture(scope="function") @pytest.fixture()
def useBroadcastServer(self): def broadcastServer():
"""!Server a BroadcastServer instance""" """!Server a BroadcastServer instance"""
self.broadcastServer = BroadcastServer() broadcastServer = BroadcastServer()
yield 1 # server the server instance yield broadcastServer
if self.broadcastServer.isRunning: if broadcastServer.isRunning:
assert self.broadcastServer.stop() assert broadcastServer.stop()
while self.broadcastServer.isRunning: while broadcastServer.isRunning:
pass pass
@pytest.fixture(scope="function")
def useBroadcastClient(self):
"""!Server a BroadcastClient instance"""
self.broadcastClient = BroadcastClient()
yield 1 # server the server instance
# tests start here @pytest.fixture()
def broadcastClient():
"""!Server a BroadcastClient instance"""
return BroadcastClient()
def test_serverStartStop(self, useBroadcastServer):
assert self.broadcastServer.start()
assert self.broadcastServer.isRunning
assert self.broadcastServer.stop()
def test_serverDoubleStart(self, useBroadcastServer): def test_serverStartStop(broadcastServer):
assert self.broadcastServer.start() assert broadcastServer.start()
assert self.broadcastServer.start() assert broadcastServer.isRunning
assert self.broadcastServer.stop() assert broadcastServer.stop()
def test_serverStopNotStarted(self, useBroadcastServer):
assert self.broadcastServer.stop()
def test_clientWithoutServer(self, useBroadcastClient): def test_serverDoubleStart(broadcastServer):
assert not self.broadcastClient.getConnInfo(1) assert broadcastServer.start()
assert broadcastServer.start()
assert broadcastServer.stop()
def test_serverClientFetchConnInfo(self, useBroadcastServer, useBroadcastClient):
assert self.broadcastServer.start() def test_serverStopNotStarted(broadcastServer):
assert self.broadcastClient.getConnInfo() assert broadcastServer.stop()
assert self.broadcastServer.stop()
assert self.broadcastClient.serverIP
assert self.broadcastClient.serverPort def test_clientWithoutServer(broadcastClient):
assert not broadcastClient.getConnInfo(1)
def test_serverClientFetchConnInfo(broadcastClient, broadcastServer):
assert broadcastServer.start()
assert broadcastClient.getConnInfo()
assert broadcastServer.stop()
assert broadcastClient.serverIP
assert broadcastClient.serverPort

7
test/test_config.py
View file

@ -12,21 +12,22 @@
@file: test_config.py @file: test_config.py
@date: 08.01.2017 @date: 08.01.2017
@author: Bastian Schroll @author: Bastian Schroll
@description: Unittests for BOSWatch. File must be _run as "pytest" unittest @description: Unittests for BOSWatch. File have to run as "pytest" unittest
""" """
import logging import logging
import pytest
from boswatch.utils import paths from boswatch.utils import paths
from boswatch import configYaml from boswatch import configYaml
# FIXME complete tests # FIXME complete tests
@pytest.mark.skip
class Test_Config: class Test_Config:
"""!Unittests for the config""" """!Unittests for the config"""
def setup_method(self, method): def setup_method(self, method):
logging.debug("[TEST] %s.%s", type(self).__name__, method.__name__) logging.debug("[TEST] %s.%s", method.__module__, method.__name__)
def test_loadLocalConfig(self): def test_loadLocalConfig(self):
"""!load a local config file""" """!load a local config file"""

135
test/test_decoder.py
View file

@ -12,84 +12,91 @@
@file: test_Decoder.py @file: test_Decoder.py
@date: 15.12.2017 @date: 15.12.2017
@author: Bastian Schroll @author: Bastian Schroll
@description: Unittests for BOSWatch. File must be _run as "pytest" unittest @description: Unittests for BOSWatch. File have to run as "pytest" unittest
""" """
import logging import logging
from boswatch.decoder.decoder import Decoder from boswatch.decoder.decoder import Decoder
class Test_Decoder: def setup_method(method):
"""!Unittests for the decoder""" logging.debug("[TEST] %s.%s", method.__module__, method.__name__)
def setup_method(self, method):
logging.debug("[TEST] %s.%s", type(self).__name__, method.__name__)
def test_decoderNoData(self): def test_decoderNoData():
"""!Test a empty string""" """!Test a empty string"""
assert Decoder.decode("") is None assert Decoder.decode("") is None
def test_decoderZveiValid(self):
"""!Test valid ZVEI"""
assert not Decoder.decode("ZVEI1: 12345") is None
assert not Decoder.decode("ZVEI1: 12838") is None
assert not Decoder.decode("ZVEI1: 34675") is None
def test_decoderZveiDoubleTone(self): def test_decoderZveiValid():
"""!Test doubleTone included ZVEI""" """!Test valid ZVEI"""
assert not Decoder.decode("ZVEI1: 6E789") is None assert not Decoder.decode("ZVEI1: 12345") is None
assert not Decoder.decode("ZVEI1: 975E7") is None assert not Decoder.decode("ZVEI1: 12838") is None
assert not Decoder.decode("ZVEI1: 2E87E") is None assert not Decoder.decode("ZVEI1: 34675") is None
def test_decoderZveiInvalid(self):
"""Test invalid ZVEI"""
assert Decoder.decode("ZVEI1: 1245A") is None
assert Decoder.decode("ZVEI1: 1245") is None
assert Decoder.decode("ZVEI1: 135") is None
assert Decoder.decode("ZVEI1: 54") is None
assert Decoder.decode("ZVEI1: 54") is None
def test_decoderPocsagValid(self): def test_decoderZveiDoubleTone():
"""!Test valid POCSAG""" """!Test doubleTone included ZVEI"""
assert not Decoder.decode("POCSAG512: Address: 1000000 Function: 0") is None assert not Decoder.decode("ZVEI1: 6E789") is None
assert not Decoder.decode("POCSAG512: Address: 1000001 Function: 1") is None assert not Decoder.decode("ZVEI1: 975E7") is None
assert not Decoder.decode("POCSAG1200: Address: 1000002 Function: 2") is None assert not Decoder.decode("ZVEI1: 2E87E") is None
assert not Decoder.decode("POCSAG2400: Address: 1000003 Function: 3") is None
def test_decoderPocsagText(self):
"""!Test POCSAG with text"""
assert not Decoder.decode("POCSAG512: Address: 1000000 Function: 0 Alpha: test") is None
assert not Decoder.decode("POCSAG512: Address: 1000001 Function: 1 Alpha: test") is None
assert not Decoder.decode("POCSAG1200: Address: 1000002 Function: 2 Alpha: test") is None
assert not Decoder.decode("POCSAG2400: Address: 1000003 Function: 3 Alpha: test") is None
def test_decoderPocsagShortRic(self): def test_decoderZveiInvalid():
"""!Test short POCSAG""" """Test invalid ZVEI"""
assert not Decoder.decode("POCSAG512: Address: 3 Function: 0 Alpha: test") is None assert Decoder.decode("ZVEI1: 1245A") is None
assert not Decoder.decode("POCSAG512: Address: 33 Function: 0 Alpha: test") is None assert Decoder.decode("ZVEI1: 1245") is None
assert not Decoder.decode("POCSAG1200: Address: 333 Function: 0 Alpha: test") is None assert Decoder.decode("ZVEI1: 135") is None
assert not Decoder.decode("POCSAG1200: Address: 3333 Function: 0 Alpha: test") is None assert Decoder.decode("ZVEI1: 54") is None
assert not Decoder.decode("POCSAG2400: Address: 33333 Function: 0 Alpha: test") is None assert Decoder.decode("ZVEI1: 54") is None
assert not Decoder.decode("POCSAG2400: Address: 333333 Function: 0 Alpha: test") is None
assert not Decoder.decode("POCSAG2400: Address: 3333333 Function: 0 Alpha: test") is None
def test_decoderPocsagInvalid(self):
"""!Test invalid POCSAG"""
assert Decoder.decode("POCSAG512: Address: 333333F Function: 0 Alpha: invalid") is None
assert Decoder.decode("POCSAG512: Address: 333333F Function: 1 Alpha: invalid") is None
assert Decoder.decode("POCSAG512: Address: 3333333 Function: 4 Alpha: invalid") is None
def test_decoderFmsValid(self): def test_decoderPocsagValid():
"""!Test valid FMS""" """!Test valid POCSAG"""
assert not Decoder.decode("""FMS: 43f314170000 (9=Rotkreuz 3=Bayern 1 Ort 0x25=037FZG 7141Status 3=Einsatz Ab 0=FZG->LST 2=I (ohneNA,ohneSIGNAL)) CRC correct""") is None assert not Decoder.decode("POCSAG512: Address: 1000000 Function: 0") is None
assert not Decoder.decode("""FMS: 43f314170000 (9=Rotkreuz 3=Bayern 1 Ort 0x25=037FZG 7141Status 3=Einsatz Ab 1=LST->FZG 2=I (ohneNA,ohneSIGNAL)) CRC correct""") is None assert not Decoder.decode("POCSAG512: Address: 1000001 Function: 1") is None
assert not Decoder.decode("""FMS: 43f314170000 (9=Rotkreuz 3=Bayern 1 Ort 0x25=037FZG 7141Status 3=Einsatz Ab 0=FZG->LST 2=II (ohneNA,mit SIGNAL)) CRC correct""") is None assert not Decoder.decode("POCSAG1200: Address: 1000002 Function: 2") is None
assert not Decoder.decode("""FMS: 43f314170000 (9=Rotkreuz 3=Bayern 1 Ort 0x25=037FZG 7141Status 3=Einsatz Ab 1=LST->FZG 2=III(mit NA,ohneSIGNAL)) CRC correct""") is None assert not Decoder.decode("POCSAG2400: Address: 1000003 Function: 3") is None
assert not Decoder.decode("""FMS: 43f314170000 (9=Rotkreuz 3=Bayern 1 Ort 0x25=037FZG 7141Status 3=Einsatz Ab 0=FZG->LST 2=IV (mit NA,mit SIGNAL)) CRC correct""") is None
def test_decoderFmsInvalid(self):
"""!Test invalid FMS""" def test_decoderPocsagText():
assert Decoder.decode("""FMS: 14170000 (9=Rotkreuz 3=Bayern 1 Ort 0x25=037FZG 7141Status 3=Einsatz Ab 1=LST->FZG 2=III(mit NA,ohneSIGNAL)) CRC correct""") is None """!Test POCSAG with text"""
assert Decoder.decode("""FMS: 43f314170000 (9=Rotkreuz 3=Bayern 1 Ort 0x25=037FZG 7141Sta 3=Einsatz Ab 0=FZG->LST 2=IV (mit NA,mit SIGNAL)) CRC correct""") is None assert not Decoder.decode("POCSAG512: Address: 1000000 Function: 0 Alpha: test") is None
assert Decoder.decode("""FMS: 14170000 (9=Rotkreuz 3=Bayern 1 Ort 0x25=037FZG 7141Status 3=Einsatz Ab 1=LST->FZG 2=III(mit NA,ohneSIGNAL)) CRC incorrect""") is None assert not Decoder.decode("POCSAG512: Address: 1000001 Function: 1 Alpha: test") is None
assert Decoder.decode("""FMS: 43f314170000 (9=Rotkreuz 3=Bayern 1 Ort 0x25=037FZG 7141Sta 3=Einsatz Ab 0=FZG->LST 2=IV (mit NA,mit SIGNAL)) CRC incorrect""") is None assert not Decoder.decode("POCSAG1200: Address: 1000002 Function: 2 Alpha: test") is None
assert not Decoder.decode("POCSAG2400: Address: 1000003 Function: 3 Alpha: test") is None
def test_decoderPocsagShortRic():
"""!Test short POCSAG"""
assert not Decoder.decode("POCSAG512: Address: 3 Function: 0 Alpha: test") is None
assert not Decoder.decode("POCSAG512: Address: 33 Function: 0 Alpha: test") is None
assert not Decoder.decode("POCSAG1200: Address: 333 Function: 0 Alpha: test") is None
assert not Decoder.decode("POCSAG1200: Address: 3333 Function: 0 Alpha: test") is None
assert not Decoder.decode("POCSAG2400: Address: 33333 Function: 0 Alpha: test") is None
assert not Decoder.decode("POCSAG2400: Address: 333333 Function: 0 Alpha: test") is None
assert not Decoder.decode("POCSAG2400: Address: 3333333 Function: 0 Alpha: test") is None
def test_decoderPocsagInvalid():
"""!Test invalid POCSAG"""
assert Decoder.decode("POCSAG512: Address: 333333F Function: 0 Alpha: invalid") is None
assert Decoder.decode("POCSAG512: Address: 333333F Function: 1 Alpha: invalid") is None
assert Decoder.decode("POCSAG512: Address: 3333333 Function: 4 Alpha: invalid") is None
def test_decoderFmsValid():
"""!Test valid FMS"""
assert not Decoder.decode("""FMS: 43f314170000 (9=Rotkreuz 3=Bayern 1 Ort 0x25=037FZG 7141Status 3=Einsatz Ab 0=FZG->LST 2=I (ohneNA,ohneSIGNAL)) CRC correct""") is None
assert not Decoder.decode("""FMS: 43f314170000 (9=Rotkreuz 3=Bayern 1 Ort 0x25=037FZG 7141Status 3=Einsatz Ab 1=LST->FZG 2=I (ohneNA,ohneSIGNAL)) CRC correct""") is None
assert not Decoder.decode("""FMS: 43f314170000 (9=Rotkreuz 3=Bayern 1 Ort 0x25=037FZG 7141Status 3=Einsatz Ab 0=FZG->LST 2=II (ohneNA,mit SIGNAL)) CRC correct""") is None
assert not Decoder.decode("""FMS: 43f314170000 (9=Rotkreuz 3=Bayern 1 Ort 0x25=037FZG 7141Status 3=Einsatz Ab 1=LST->FZG 2=III(mit NA,ohneSIGNAL)) CRC correct""") is None
assert not Decoder.decode("""FMS: 43f314170000 (9=Rotkreuz 3=Bayern 1 Ort 0x25=037FZG 7141Status 3=Einsatz Ab 0=FZG->LST 2=IV (mit NA,mit SIGNAL)) CRC correct""") is None
def test_decoderFmsInvalid():
"""!Test invalid FMS"""
assert Decoder.decode("""FMS: 14170000 (9=Rotkreuz 3=Bayern 1 Ort 0x25=037FZG 7141Status 3=Einsatz Ab 1=LST->FZG 2=III(mit NA,ohneSIGNAL)) CRC correct""") is None
assert Decoder.decode("""FMS: 43f314170000 (9=Rotkreuz 3=Bayern 1 Ort 0x25=037FZG 7141Sta 3=Einsatz Ab 0=FZG->LST 2=IV (mit NA,mit SIGNAL)) CRC correct""") is None
assert Decoder.decode("""FMS: 14170000 (9=Rotkreuz 3=Bayern 1 Ort 0x25=037FZG 7141Status 3=Einsatz Ab 1=LST->FZG 2=III(mit NA,ohneSIGNAL)) CRC incorrect""") is None
assert Decoder.decode("""FMS: 43f314170000 (9=Rotkreuz 3=Bayern 1 Ort 0x25=037FZG 7141Sta 3=Einsatz Ab 0=FZG->LST 2=IV (mit NA,mit SIGNAL)) CRC incorrect""") is None

6
test/test_descriptor.py
View file

@ -12,20 +12,22 @@
@file: test_descriptor.py @file: test_descriptor.py
@date: 07.01.2017 @date: 07.01.2017
@author: Bastian Schroll @author: Bastian Schroll
@description: Unittests for BOSWatch. File must be _run as "pytest" unittest @description: Unittests for BOSWatch. File have to run as "pytest" unittest
""" """
import logging import logging
import pytest
from module.descriptor import Descriptor from module.descriptor import Descriptor
from module.descriptor import DescriptionList from module.descriptor import DescriptionList
from boswatch.packet import Packet from boswatch.packet import Packet
@pytest.mark.skip
class Test_Descriptor: class Test_Descriptor:
"""!Unittests for the descriptor""" """!Unittests for the descriptor"""
def setup_method(self, method): def setup_method(self, method):
logging.debug("[TEST] %s.%s", type(self).__name__, method.__name__) logging.debug("[TEST] %s.%s", method.__module__, method.__name__)
def test_loadCsvNotExist(self): def test_loadCsvNotExist(self):
"""!read CSV file where not exist direct per DescriptionList class""" """!read CSV file where not exist direct per DescriptionList class"""

6
test/test_doubleFilter.py
View file

@ -12,17 +12,19 @@
@file: test_doubleFilter.py @file: test_doubleFilter.py
@date: 15.12.2017 @date: 15.12.2017
@author: Bastian Schroll @author: Bastian Schroll
@description: Unittests for BOSWatch. File must be _run as "pytest" unittest @description: Unittests for BOSWatch. File have to run as "pytest" unittest
""" """
import logging import logging
import pytest
from module.filter.doubleFilter import DoubleFilter from module.filter.doubleFilter import DoubleFilter
@pytest.mark.skip
class Test_DoubleFilter: class Test_DoubleFilter:
def setup_method(self, method): def setup_method(self, method):
logging.debug("[TEST] %s.%s", type(self).__name__, method.__name__) logging.debug("[TEST] %s.%s", method.__module__, method.__name__)
def test_none(self): def test_none(self):
pass pass

21
test/test_header.py
View file

@ -12,23 +12,22 @@
@file: test_header.py @file: test_header.py
@date: 12.12.2017 @date: 12.12.2017
@author: Bastian Schroll @author: Bastian Schroll
@description: Unittests for BOSWatch. File must be _run as "pytest" unittest @description: Unittests for BOSWatch. File have to run as "pytest" unittest
""" """
import logging import logging
from boswatch.utils import header from boswatch.utils import header
class Test_Header: def setup_method(method):
"""!Unittests for the header""" logging.debug("[TEST] %s.%s", method.__module__, method.__name__)
def setup_method(self, method):
logging.debug("[TEST] %s.%s", type(self).__name__, method.__name__)
def test_logoToLog(self): def test_logoToLog():
"""!Test logo to log""" """!Test logo to log"""
assert header.logoToLog() assert header.logoToLog()
def test_infoToLog(self):
"""!Test info to log""" def test_infoToLog():
assert header.infoToLog() """!Test info to log"""
assert header.infoToLog()

59
test/test_packet.py
View file

@ -12,7 +12,7 @@
@file: test_packet.py @file: test_packet.py
@date: 12.12.2017 @date: 12.12.2017
@author: Bastian Schroll @author: Bastian Schroll
@description: Unittests for BOSWatch. File must be _run as "pytest" unittest @description: Unittests for BOSWatch. File have to run as "pytest" unittest
""" """
import logging import logging
import pytest import pytest
@ -20,38 +20,39 @@ import pytest
from boswatch.packet import Packet from boswatch.packet import Packet
class Test_Packet: def setup_method(method):
"""!Unittests for the BOSWatch packet""" logging.debug("[TEST] %s.%s", method.__module__, method.__name__)
def setup_method(self, method):
logging.debug("[TEST] %s.%s", type(self).__name__, method.__name__)
@pytest.fixture(scope="function") @pytest.fixture()
def buildPacket(self): def buildPacket():
"""!Build a BOSWatch packet and serve it to each test""" """!Build a BOSWatch packet and serve it to each test"""
bwPacket = Packet() return Packet()
yield bwPacket
def test_createPacket(self):
"""!Create a packet"""
bwPacket = Packet()
assert bwPacket is not ""
def test_copyPacket(self, buildPacket): def test_createPacket(buildPacket):
"""!Copy a packet to an new instance""" """!Create a packet"""
bwCopyPacket = Packet(buildPacket.__str__()) assert buildPacket is not ""
assert bwCopyPacket is not ""
def test_getPacketString(self, buildPacket):
"""!get the intern packet dict as string"""
assert type(buildPacket.__str__()) is str
assert buildPacket.__str__() is not ""
def test_getNotSetField(self, buildPacket): def test_copyPacket(buildPacket):
"""!try to get a not set field""" """!Copy a packet to an new instance"""
assert not buildPacket.get("testfield") bwCopyPacket = Packet(buildPacket.__str__())
assert bwCopyPacket is not ""
def test_setGetField(self, buildPacket):
"""!set and get a field""" def test_getPacketString(buildPacket):
buildPacket.set("testField", "test") """!get the intern packet dict as string"""
assert buildPacket.get("testField") is "test" assert type(buildPacket.__str__()) is str
assert buildPacket.__str__() is not ""
def test_getNotSetField(buildPacket):
"""!try to get a not set field"""
assert not buildPacket.get("testfield")
def test_setGetField(buildPacket):
"""!set and get a field"""
buildPacket.set("testField", "test")
assert buildPacket.get("testField") is "test"

41
test/test_paths.py
View file

@ -12,7 +12,7 @@
@file: test_paths.py @file: test_paths.py
@date: 22.02.2017 @date: 22.02.2017
@author: Bastian Schroll @author: Bastian Schroll
@description: Unittests for BOSWatch. File must be _run as "pytest" unittest @description: Unittests for BOSWatch. File have to run as "pytest" unittest
""" """
import logging import logging
import os import os
@ -20,27 +20,28 @@ import os
from boswatch.utils import paths from boswatch.utils import paths
class Test_Config: def setup_method(method):
"""!Unittests for the paths""" logging.debug("[TEST] %s.%s", method.__module__, method.__name__)
def setup_method(self, method):
logging.debug("[TEST] %s.%s", type(self).__name__, method.__name__)
def test_fileExists(self): def test_fileExists():
"""!load a local config file""" """!load a local config file"""
assert paths.fileExist("README.md") assert paths.fileExist("README.md")
def test_fileNotExists(self):
"""!load a local config file"""
assert not paths.fileExist("notFound.txt")
def test_makeDirNotExisting(self): def test_fileNotExists():
"""!load a local config file""" """!load a local config file"""
assert paths.makeDirIfNotExist("UnItTeSt") assert not paths.fileExist("notFound.txt")
os.removedirs("UnItTeSt")
def test_makeDirExisting(self):
"""!load a local config file""" def test_makeDirNotExisting():
paths.makeDirIfNotExist("UnItTeSt") """!load a local config file"""
assert paths.makeDirIfNotExist("UnItTeSt") assert paths.makeDirIfNotExist("UnItTeSt")
os.removedirs("UnItTeSt") os.removedirs("UnItTeSt")
def test_makeDirExisting():
"""!load a local config file"""
paths.makeDirIfNotExist("UnItTeSt")
assert paths.makeDirIfNotExist("UnItTeSt")
os.removedirs("UnItTeSt")

32
test/test_watchdog.py → test/test_template.py
View file

@ -9,20 +9,32 @@
German BOS Information Script German BOS Information Script
by Bastian Schroll by Bastian Schroll
@file: test_watchdog.py @file: test_template.py
@date: 15.12.2017 @date: 03.03.2019
@author: Bastian Schroll @author: Bastian Schroll
@description: Unittests for BOSWatch. File must be _run as "pytest" unittest @description: Unittests for BOSWatch. File have to run as "pytest" unittest
""" """
import logging import logging
import pytest
# import boswatch.watchdog.watchdog
class Test_Watchdog: def setup_method(method):
logging.debug("[TEST] %s.%s", method.__module__, method.__name__)
def setup_method(self, method):
logging.debug("[TEST] %s.%s", type(self).__name__, method.__name__)
def test_none(self): @pytest.fixture
pass def fixtureTemplate():
return None
@pytest.mark.skip
def test_skippedTest():
pass
def test_testName():
pass
def test_withFixture(fixtureTemplate):
assert fixtureTemplate is None

126
test/test_timer.py
View file

@ -12,7 +12,7 @@
@file: test_timer.py @file: test_timer.py
@date: 21.09.2018 @date: 21.09.2018
@author: Bastian Schroll @author: Bastian Schroll
@description: Unittests for BOSWatch. File must be _run as "pytest" unittest @description: Unittests for BOSWatch. File have to run as "pytest" unittest
""" """
import logging import logging
import time import time
@ -21,76 +21,80 @@ import pytest
from boswatch.utils.timer import RepeatedTimer from boswatch.utils.timer import RepeatedTimer
class Test_Timer: def setup_method(method):
"""!Unittest for the timer class""" logging.debug("[TEST] %s.%s", method.__module__, method.__name__)
def setup_method(self, method):
logging.debug("[TEST] %s.%s", type(self).__name__, method.__name__)
@staticmethod def testTargetFast():
def testTargetFast(): """!Fast worker thread"""
"""!Fast worker thread""" logging.debug("run testTargetFast")
logging.debug("run testTargetFast")
@staticmethod
def testTargetSlow():
"""!Slow worker thread"""
logging.debug("run testTargetSlow start")
time.sleep(0.51)
logging.debug("run testTargetSlow end")
@pytest.fixture(scope="function") def testTargetSlow():
def useTimerFast(self): """!Slow worker thread"""
"""!Server a RepeatedTimer instance with fast worker""" logging.debug("run testTargetSlow start")
self.testTimer = RepeatedTimer(0.1, Test_Timer.testTargetFast) time.sleep(0.51)
yield 1 # server the timer instance logging.debug("run testTargetSlow end")
if self.testTimer.isRunning:
assert self.testTimer.stop()
@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
if self.testTimer.isRunning:
assert self.testTimer.stop()
# test cases starts here @pytest.fixture()
def useTimerFast():
"""!Server a RepeatedTimer instance with fast worker"""
testTimer = RepeatedTimer(0.1, testTargetFast)
yield testTimer
if testTimer.isRunning:
assert testTimer.stop()
def test_timerStartStop(self, useTimerFast):
assert self.testTimer.start()
assert self.testTimer.stop()
def test_timerDoubleStart(self, useTimerFast): @pytest.fixture()
assert self.testTimer.start() def useTimerSlow():
assert self.testTimer.start() """!Server a RepeatedTimer instance slow worker"""
assert self.testTimer.stop() testTimer = RepeatedTimer(0.1, testTargetSlow)
yield testTimer
if testTimer.isRunning:
assert testTimer.stop()
def test_timerStopNotStarted(self, useTimerFast):
assert self.testTimer.stop()
def test_timerIsRunning(self, useTimerFast): def test_timerStartStop(useTimerFast):
assert self.testTimer.start() assert useTimerFast.start()
assert self.testTimer.isRunning assert useTimerFast.stop()
assert self.testTimer.stop()
def test_timerRun(self, useTimerFast):
assert self.testTimer.start()
time.sleep(0.2)
assert self.testTimer.stop()
assert self.testTimer.overdueCount == 0
assert self.testTimer.lostEvents == 0
def test_timerOverdue(self, useTimerSlow): def test_timerDoubleStart(useTimerFast):
assert self.testTimer.start() assert useTimerFast.start()
time.sleep(0.2) assert useTimerFast.start()
assert self.testTimer.stop() assert useTimerFast.stop()
assert self.testTimer.overdueCount == 1
assert self.testTimer.lostEvents == 5
def test_timerOverdueLong(self, useTimerSlow):
assert self.testTimer.start() def test_timerStopNotStarted(useTimerFast):
time.sleep(1) assert useTimerFast.stop()
assert self.testTimer.stop()
assert self.testTimer.overdueCount == 2
assert self.testTimer.lostEvents == 10 def test_timerIsRunning(useTimerFast):
assert useTimerFast.start()
assert useTimerFast.isRunning
assert useTimerFast.stop()
def test_timerRun(useTimerFast):
assert useTimerFast.start()
time.sleep(0.2)
assert useTimerFast.stop()
assert useTimerFast.overdueCount == 0
assert useTimerFast.lostEvents == 0
def test_timerOverdue(useTimerSlow):
assert useTimerSlow.start()
time.sleep(0.2)
assert useTimerSlow.stop()
assert useTimerSlow.overdueCount == 1
assert useTimerSlow.lostEvents == 5
def test_timerOverdueLong(useTimerSlow):
assert useTimerSlow.start()
time.sleep(1)
assert useTimerSlow.stop()
assert useTimerSlow.overdueCount == 2
assert useTimerSlow.lostEvents == 10