From a35dc15d60e264fd37d5553aba209e8e546a4c87 Mon Sep 17 00:00:00 2001 From: Bastian Schroll Date: Sun, 3 Mar 2019 11:13:27 +0100 Subject: [PATCH] edit pytest suite --- test/test_ServerClient.py | 269 ++++++++++---------- test/test_broadcast.py | 77 +++--- test/test_config.py | 7 +- test/test_decoder.py | 135 +++++----- test/test_descriptor.py | 6 +- test/test_doubleFilter.py | 6 +- test/test_header.py | 21 +- test/test_packet.py | 59 ++--- test/test_paths.py | 41 +-- test/{test_watchdog.py => test_template.py} | 32 ++- test/test_timer.py | 126 ++++----- 11 files changed, 408 insertions(+), 371 deletions(-) rename test/{test_watchdog.py => test_template.py} (51%) diff --git a/test/test_ServerClient.py b/test/test_ServerClient.py index 0e2ddff..aa89706 100644 --- a/test/test_ServerClient.py +++ b/test/test_ServerClient.py @@ -12,7 +12,7 @@ @file: test_ServerClient.py @date: 10.12.2017 @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 time @@ -23,149 +23,156 @@ from boswatch.network.server import TCPServer from boswatch.network.client import TCPClient -class Test_ServerClient: - """!Unittest for the server/client environment""" +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__) - @pytest.fixture(scope="function") - def useServer(self): - """!Start and serve the sever for each functions where useServer is given""" - 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") +@pytest.fixture +def getClient(): + return TCPClient() - time.sleep(0.1) # wait for server - def test_clientConnectFailed(self): - """!Connect to a non available server""" - self.testClient = TCPClient() - assert not self.testClient.connect() +@pytest.fixture +def getServer(): + """!Start and serve the sever for each functions where useServer is given""" + 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): - """!Transmit while no connection is established""" - self.testClient = TCPClient() - assert not self.testClient.transmit("test") +@pytest.fixture +def getRunningServer(getServer): + logging.debug("start server") + 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): - """!Connect to a server""" - self.testClient = TCPClient() - assert self.testClient.connect() - assert self.testClient.disconnect() +def test_clientConnectFailed(getClient): + """!Connect to a non available server""" + assert not getClient.connect() - 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): - """!Connect with 2 clients to the server""" - self.testClient1 = TCPClient() - 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_clientDisconnectFailed(getClient): + """!Disconnect while no connection is established""" + assert getClient.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): - """!Try to send data to the server with 3 clients and check on '[ack]'""" - # connect all - 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_clientTransmitFailed(getClient): + """!Transmit while no connection is established""" + assert not getClient.transmit("test") - 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): - """!Test to stop a stopped server""" - assert self.testServer.stop() - assert self.testServer.stop() +def test_clientReceiveFailed(getClient): + """!Receive while no connection is established""" + assert not getClient.receive() - 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): - """!Send data to server with 2 clients, check '[ack]' and data on server queue""" - # connect all - self.testClient1 = TCPClient() - assert self.testClient1.connect() - self.testClient2 = TCPClient() - assert self.testClient2.connect() - # send all - assert self.testClient1.transmit("test1") - time.sleep(0.1) # wait for recv to prevent fail of false order - assert self.testClient2.transmit("test2") - # recv all - assert self.testClient1.receive() == "[ack]" - assert self.testClient2.receive() == "[ack]" - # _check server output data - assert self.dataQueue.qsize() == 2 - assert self.dataQueue.get(True, 1)[1] == "test1" - assert self.dataQueue.get(True, 1)[1] == "test2" - assert self.dataQueue.qsize() is 0 # Last _check must be None - # disconnect all - assert self.testClient1.disconnect() - assert self.testClient2.disconnect() +def test_clientConnect(getClient, getRunningServer): + """!Connect to a server""" + assert getClient.connect() + assert getClient.disconnect() + + +def test_clientReconnect(getClient, getRunningServer): + """!Try a reconnect after a established connection""" + assert getClient.connect() + assert getClient.disconnect() + assert getClient.connect() + assert getClient.disconnect() + + +def test_clientMultiConnect(getClient, getRunningServer): + """!Connect with 2 clients to the server""" + assert getClient.connect() + testClient2 = TCPClient() + assert testClient2.connect() + time.sleep(0.1) # wait for all clients connected + # check connected clients + 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() diff --git a/test/test_broadcast.py b/test/test_broadcast.py index f6a97cc..e0a37a6 100644 --- a/test/test_broadcast.py +++ b/test/test_broadcast.py @@ -12,7 +12,7 @@ @file: test_broadcast.py @date: 25.09.2018 @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 pytest @@ -21,49 +21,50 @@ from boswatch.network.broadcast import BroadcastServer from boswatch.network.broadcast import BroadcastClient -class Test_Broadcast: - """!Unittest for the timer class""" +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__) - @pytest.fixture(scope="function") - def useBroadcastServer(self): - """!Server a BroadcastServer instance""" - self.broadcastServer = BroadcastServer() - yield 1 # server the server instance - if self.broadcastServer.isRunning: - assert self.broadcastServer.stop() - while self.broadcastServer.isRunning: - pass +@pytest.fixture() +def broadcastServer(): + """!Server a BroadcastServer instance""" + broadcastServer = BroadcastServer() + yield broadcastServer + if broadcastServer.isRunning: + assert broadcastServer.stop() + while broadcastServer.isRunning: + 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): - assert self.broadcastServer.start() - assert self.broadcastServer.start() - assert self.broadcastServer.stop() +def test_serverStartStop(broadcastServer): + assert broadcastServer.start() + assert broadcastServer.isRunning + assert broadcastServer.stop() - def test_serverStopNotStarted(self, useBroadcastServer): - assert self.broadcastServer.stop() - def test_clientWithoutServer(self, useBroadcastClient): - assert not self.broadcastClient.getConnInfo(1) +def test_serverDoubleStart(broadcastServer): + assert broadcastServer.start() + assert broadcastServer.start() + assert broadcastServer.stop() - def test_serverClientFetchConnInfo(self, useBroadcastServer, useBroadcastClient): - assert self.broadcastServer.start() - assert self.broadcastClient.getConnInfo() - assert self.broadcastServer.stop() - assert self.broadcastClient.serverIP - assert self.broadcastClient.serverPort + +def test_serverStopNotStarted(broadcastServer): + assert broadcastServer.stop() + + +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 diff --git a/test/test_config.py b/test/test_config.py index 03cda7e..f8dbc68 100644 --- a/test/test_config.py +++ b/test/test_config.py @@ -12,21 +12,22 @@ @file: test_config.py @date: 08.01.2017 @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 pytest from boswatch.utils import paths from boswatch import configYaml # FIXME complete tests - +@pytest.mark.skip class Test_Config: """!Unittests for the config""" 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): """!load a local config file""" diff --git a/test/test_decoder.py b/test/test_decoder.py index 11f0ba6..63e74ee 100644 --- a/test/test_decoder.py +++ b/test/test_decoder.py @@ -12,84 +12,91 @@ @file: test_Decoder.py @date: 15.12.2017 @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 from boswatch.decoder.decoder import Decoder -class Test_Decoder: - """!Unittests for the decoder""" +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_decoderNoData(self): - """!Test a empty string""" - assert Decoder.decode("") is None +def test_decoderNoData(): + """!Test a empty string""" + 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): - """!Test doubleTone included ZVEI""" - assert not Decoder.decode("ZVEI1: 6E789") is None - assert not Decoder.decode("ZVEI1: 975E7") is None - assert not Decoder.decode("ZVEI1: 2E87E") is None +def test_decoderZveiValid(): + """!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_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): - """!Test valid POCSAG""" - assert not Decoder.decode("POCSAG512: Address: 1000000 Function: 0") is None - assert not Decoder.decode("POCSAG512: Address: 1000001 Function: 1") is None - assert not Decoder.decode("POCSAG1200: Address: 1000002 Function: 2") is None - assert not Decoder.decode("POCSAG2400: Address: 1000003 Function: 3") is None +def test_decoderZveiDoubleTone(): + """!Test doubleTone included ZVEI""" + assert not Decoder.decode("ZVEI1: 6E789") is None + assert not Decoder.decode("ZVEI1: 975E7") is None + assert not Decoder.decode("ZVEI1: 2E87E") 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): - """!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_decoderZveiInvalid(): + """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_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): - """!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_decoderPocsagValid(): + """!Test valid POCSAG""" + assert not Decoder.decode("POCSAG512: Address: 1000000 Function: 0") is None + assert not Decoder.decode("POCSAG512: Address: 1000001 Function: 1") is None + assert not Decoder.decode("POCSAG1200: Address: 1000002 Function: 2") is None + assert not Decoder.decode("POCSAG2400: Address: 1000003 Function: 3") is None - def test_decoderFmsInvalid(self): - """!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 + +def test_decoderPocsagText(): + """!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(): + """!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 diff --git a/test/test_descriptor.py b/test/test_descriptor.py index d5ae3b4..ca6f53e 100644 --- a/test/test_descriptor.py +++ b/test/test_descriptor.py @@ -12,20 +12,22 @@ @file: test_descriptor.py @date: 07.01.2017 @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 pytest from module.descriptor import Descriptor from module.descriptor import DescriptionList from boswatch.packet import Packet +@pytest.mark.skip class Test_Descriptor: """!Unittests for the descriptor""" 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): """!read CSV file where not exist direct per DescriptionList class""" diff --git a/test/test_doubleFilter.py b/test/test_doubleFilter.py index 0123be1..ab92c0f 100644 --- a/test/test_doubleFilter.py +++ b/test/test_doubleFilter.py @@ -12,17 +12,19 @@ @file: test_doubleFilter.py @date: 15.12.2017 @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 pytest from module.filter.doubleFilter import DoubleFilter +@pytest.mark.skip class Test_DoubleFilter: 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): pass diff --git a/test/test_header.py b/test/test_header.py index 1732543..49270af 100644 --- a/test/test_header.py +++ b/test/test_header.py @@ -12,23 +12,22 @@ @file: test_header.py @date: 12.12.2017 @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 from boswatch.utils import header -class Test_Header: - """!Unittests for the header""" +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_logoToLog(self): - """!Test logo to log""" - assert header.logoToLog() +def test_logoToLog(): + """!Test logo to log""" + assert header.logoToLog() - def test_infoToLog(self): - """!Test info to log""" - assert header.infoToLog() + +def test_infoToLog(): + """!Test info to log""" + assert header.infoToLog() diff --git a/test/test_packet.py b/test/test_packet.py index 7cdf10d..31a738c 100644 --- a/test/test_packet.py +++ b/test/test_packet.py @@ -12,7 +12,7 @@ @file: test_packet.py @date: 12.12.2017 @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 pytest @@ -20,38 +20,39 @@ import pytest from boswatch.packet import Packet -class Test_Packet: - """!Unittests for the BOSWatch packet""" +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__) - @pytest.fixture(scope="function") - def buildPacket(self): - """!Build a BOSWatch packet and serve it to each test""" - bwPacket = Packet() - yield bwPacket +@pytest.fixture() +def buildPacket(): + """!Build a BOSWatch packet and serve it to each test""" + return Packet() - def test_createPacket(self): - """!Create a packet""" - bwPacket = Packet() - assert bwPacket is not "" - def test_copyPacket(self, buildPacket): - """!Copy a packet to an new instance""" - bwCopyPacket = Packet(buildPacket.__str__()) - assert bwCopyPacket is not "" +def test_createPacket(buildPacket): + """!Create a packet""" + assert buildPacket 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): - """!try to get a not set field""" - assert not buildPacket.get("testfield") +def test_copyPacket(buildPacket): + """!Copy a packet to an new instance""" + bwCopyPacket = Packet(buildPacket.__str__()) + assert bwCopyPacket is not "" - def test_setGetField(self, buildPacket): - """!set and get a field""" - buildPacket.set("testField", "test") - assert buildPacket.get("testField") is "test" + +def test_getPacketString(buildPacket): + """!get the intern packet dict as string""" + 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" diff --git a/test/test_paths.py b/test/test_paths.py index d8891a8..d1f04b4 100644 --- a/test/test_paths.py +++ b/test/test_paths.py @@ -12,7 +12,7 @@ @file: test_paths.py @date: 22.02.2017 @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 os @@ -20,27 +20,28 @@ import os from boswatch.utils import paths -class Test_Config: - """!Unittests for the paths""" +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_fileExists(self): - """!load a local config file""" - assert paths.fileExist("README.md") +def test_fileExists(): + """!load a local config file""" + assert paths.fileExist("README.md") - def test_fileNotExists(self): - """!load a local config file""" - assert not paths.fileExist("notFound.txt") - def test_makeDirNotExisting(self): - """!load a local config file""" - assert paths.makeDirIfNotExist("UnItTeSt") - os.removedirs("UnItTeSt") +def test_fileNotExists(): + """!load a local config file""" + assert not paths.fileExist("notFound.txt") - def test_makeDirExisting(self): - """!load a local config file""" - paths.makeDirIfNotExist("UnItTeSt") - assert paths.makeDirIfNotExist("UnItTeSt") - os.removedirs("UnItTeSt") + +def test_makeDirNotExisting(): + """!load a local config file""" + assert paths.makeDirIfNotExist("UnItTeSt") + os.removedirs("UnItTeSt") + + +def test_makeDirExisting(): + """!load a local config file""" + paths.makeDirIfNotExist("UnItTeSt") + assert paths.makeDirIfNotExist("UnItTeSt") + os.removedirs("UnItTeSt") diff --git a/test/test_watchdog.py b/test/test_template.py similarity index 51% rename from test/test_watchdog.py rename to test/test_template.py index 5a1496b..ccb8c24 100644 --- a/test/test_watchdog.py +++ b/test/test_template.py @@ -9,20 +9,32 @@ German BOS Information Script by Bastian Schroll -@file: test_watchdog.py -@date: 15.12.2017 +@file: test_template.py +@date: 03.03.2019 @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 boswatch.watchdog.watchdog +import pytest -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): - pass +@pytest.fixture +def fixtureTemplate(): + return None + + +@pytest.mark.skip +def test_skippedTest(): + pass + + +def test_testName(): + pass + + +def test_withFixture(fixtureTemplate): + assert fixtureTemplate is None diff --git a/test/test_timer.py b/test/test_timer.py index 49ff4a6..a0df0e5 100644 --- a/test/test_timer.py +++ b/test/test_timer.py @@ -12,7 +12,7 @@ @file: test_timer.py @date: 21.09.2018 @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 time @@ -21,76 +21,80 @@ import pytest from boswatch.utils.timer import RepeatedTimer -class Test_Timer: - """!Unittest for the timer class""" +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__) - @staticmethod - def testTargetFast(): - """!Fast worker thread""" - logging.debug("run testTargetFast") +def testTargetFast(): + """!Fast worker thread""" + 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 useTimerFast(self): - """!Server a RepeatedTimer instance with fast worker""" - self.testTimer = RepeatedTimer(0.1, Test_Timer.testTargetFast) - yield 1 # server the timer instance - if self.testTimer.isRunning: - assert self.testTimer.stop() +def testTargetSlow(): + """!Slow worker thread""" + logging.debug("run testTargetSlow start") + time.sleep(0.51) + logging.debug("run testTargetSlow end") - @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): - assert self.testTimer.start() - assert self.testTimer.start() - assert self.testTimer.stop() +@pytest.fixture() +def useTimerSlow(): + """!Server a RepeatedTimer instance slow worker""" + 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): - assert self.testTimer.start() - assert self.testTimer.isRunning - assert self.testTimer.stop() +def test_timerStartStop(useTimerFast): + assert useTimerFast.start() + assert useTimerFast.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): - assert self.testTimer.start() - time.sleep(0.2) - assert self.testTimer.stop() - assert self.testTimer.overdueCount == 1 - assert self.testTimer.lostEvents == 5 +def test_timerDoubleStart(useTimerFast): + assert useTimerFast.start() + assert useTimerFast.start() + assert useTimerFast.stop() - 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 + +def test_timerStopNotStarted(useTimerFast): + assert useTimerFast.stop() + + +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