mirror of
https://github.com/BOSWatch/BW3-Core.git
synced 2026-02-19 13:54:15 +01:00
edit tests
This commit is contained in:
Bastian Schroll
parent
f14a78984a
commit
0c7fc84089
|
|
@ -8,7 +8,7 @@
|
|||
# by Bastian Schroll
|
||||
|
||||
[pytest]
|
||||
addopts = -v --pep8 --cov --cov-report=term-missing
|
||||
addopts = -v --pep8 --cov --cov-report=term-missing --log-level=CRITICAL
|
||||
|
||||
# classic or progress
|
||||
console_output_style = progress
|
||||
|
|
|
|||
|
|
@ -18,8 +18,8 @@
|
|||
import pytest # import the pytest framework
|
||||
import time
|
||||
|
||||
import boswatch.network.server
|
||||
import boswatch.network.client
|
||||
from boswatch.network.server import TCPServer
|
||||
from boswatch.network.client import TCPClient
|
||||
|
||||
|
||||
class Test_ServerClient:
|
||||
|
|
@ -28,73 +28,73 @@ class Test_ServerClient:
|
|||
@pytest.fixture(scope="function")
|
||||
def useServer(self):
|
||||
"""!Start and serve the sever for each functions where useServer is given"""
|
||||
self.testServer = boswatch.network.server.TCPServer()
|
||||
self.testServer = TCPServer()
|
||||
assert self.testServer.start()
|
||||
time.sleep(0.1) # wait for server
|
||||
yield self.testServer # server to all test where useServer is given
|
||||
assert self.testServer.stop()
|
||||
time.sleep(0.1) # wait for server
|
||||
|
||||
def test_client_failed_connect(self):
|
||||
def test_clientConnectFailed(self):
|
||||
"""!Connect to a non available server"""
|
||||
self.testClient = boswatch.network.client.TCPClient()
|
||||
self.testClient = TCPClient()
|
||||
assert not self.testClient.connect()
|
||||
|
||||
def test_client_failed_disconnect(self):
|
||||
def test_clientDisconnectFailed(self):
|
||||
"""!Disconnect while no connection is established"""
|
||||
self.testClient = boswatch.network.client.TCPClient()
|
||||
self.testClient = TCPClient()
|
||||
assert not self.testClient.disconnect()
|
||||
|
||||
def test_client_failed_transmit(self):
|
||||
def test_clientTransmitFailed(self):
|
||||
"""!Transmit while no connection is established"""
|
||||
self.testClient = boswatch.network.client.TCPClient()
|
||||
self.testClient = TCPClient()
|
||||
assert not self.testClient.transmit("test")
|
||||
|
||||
def test_client_failed_receive(self):
|
||||
def test_clientReceiveFailed(self):
|
||||
"""!Receive while no connection is established"""
|
||||
self.testClient = boswatch.network.client.TCPClient()
|
||||
self.testClient = TCPClient()
|
||||
assert not self.testClient.receive()
|
||||
|
||||
def test_client_success_connect(self, useServer):
|
||||
def test_clientConnect(self, useServer):
|
||||
"""!Connect to a server"""
|
||||
self.testClient = boswatch.network.client.TCPClient()
|
||||
self.testClient = TCPClient()
|
||||
assert self.testClient.connect()
|
||||
assert self.testClient.disconnect()
|
||||
|
||||
def test_client_reconnect(self, useServer):
|
||||
def test_clientReconnect(self, useServer):
|
||||
"""!Try a reconnect after a established connection"""
|
||||
self.testClient = boswatch.network.client.TCPClient()
|
||||
self.testClient = TCPClient()
|
||||
assert self.testClient.connect()
|
||||
assert self.testClient.disconnect()
|
||||
assert self.testClient.connect()
|
||||
assert self.testClient.disconnect()
|
||||
|
||||
def test_multi_connect(self, useServer):
|
||||
def test_clientMultiConnect(self, useServer):
|
||||
"""!Connect with 2 clients to the server"""
|
||||
self.testClient1 = boswatch.network.client.TCPClient()
|
||||
self.testClient1 = TCPClient()
|
||||
assert self.testClient1.connect()
|
||||
self.testClient2 = boswatch.network.client.TCPClient()
|
||||
self.testClient2 = TCPClient()
|
||||
assert self.testClient2.connect()
|
||||
# disconnect all
|
||||
assert self.testClient1.disconnect()
|
||||
assert self.testClient2.disconnect()
|
||||
|
||||
def test_client_communicate(self, useServer):
|
||||
def test_clientCommunicate(self, useServer):
|
||||
"""!Try to send data to the server and check on '[ack]'"""
|
||||
self.testClient = boswatch.network.client.TCPClient()
|
||||
self.testClient = TCPClient()
|
||||
assert self.testClient.connect()
|
||||
assert self.testClient.transmit("test")
|
||||
assert self.testClient.receive() == "[ack]"
|
||||
assert self.testClient.disconnect()
|
||||
|
||||
def test_client_multi_communicate(self, useServer):
|
||||
def test_clientMultiCommunicate(self, useServer):
|
||||
"""!Try to send data to the server with 3 clients and check on '[ack]'"""
|
||||
# connect all
|
||||
self.testClient1 = boswatch.network.client.TCPClient()
|
||||
self.testClient1 = TCPClient()
|
||||
assert self.testClient1.connect()
|
||||
self.testClient2 = boswatch.network.client.TCPClient()
|
||||
self.testClient2 = TCPClient()
|
||||
assert self.testClient2.connect()
|
||||
self.testClient3 = boswatch.network.client.TCPClient()
|
||||
self.testClient3 = TCPClient()
|
||||
assert self.testClient3.connect()
|
||||
# send all
|
||||
assert self.testClient1.transmit("test")
|
||||
|
|
@ -109,34 +109,34 @@ class Test_ServerClient:
|
|||
assert self.testClient2.disconnect()
|
||||
assert self.testClient3.disconnect()
|
||||
|
||||
def test_server_restart(self):
|
||||
def test_serverRestart(self):
|
||||
"""!Test a restart of the server"""
|
||||
self.testServer = boswatch.network.server.TCPServer()
|
||||
self.testServer = TCPServer()
|
||||
assert self.testServer.start()
|
||||
assert self.testServer.stop()
|
||||
assert self.testServer.start()
|
||||
assert self.testServer.stop()
|
||||
|
||||
def test_server_failed_stop(self):
|
||||
def test_serverStopFailed(self):
|
||||
"""!Test to start the server twice"""
|
||||
self.testServer = boswatch.network.server.TCPServer()
|
||||
self.testServer = TCPServer()
|
||||
assert not self.testServer.stop()
|
||||
|
||||
def test_server_doubleStart(self):
|
||||
def test_serverDoubleStart(self):
|
||||
"""!Test to start the server twice"""
|
||||
self.testServer1 = boswatch.network.server.TCPServer()
|
||||
self.testServer2 = boswatch.network.server.TCPServer()
|
||||
self.testServer1 = TCPServer()
|
||||
self.testServer2 = TCPServer()
|
||||
assert self.testServer1.start()
|
||||
assert not self.testServer2.start()
|
||||
assert self.testServer1.stop()
|
||||
assert not self.testServer2.stop()
|
||||
|
||||
def test_server_output(self, useServer):
|
||||
def test_serverGetOutput(self, useServer):
|
||||
"""!Send data to server with 2 clients, check '[ack]' and data on server queue"""
|
||||
# connect all
|
||||
self.testClient1 = boswatch.network.client.TCPClient()
|
||||
self.testClient1 = TCPClient()
|
||||
assert self.testClient1.connect()
|
||||
self.testClient2 = boswatch.network.client.TCPClient()
|
||||
self.testClient2 = TCPClient()
|
||||
assert self.testClient2.connect()
|
||||
# send all
|
||||
useServer.flushData()
|
||||
|
|
|
|||
|
|
@ -34,16 +34,17 @@ class Test_Config:
|
|||
"""!get values from local config file"""
|
||||
bwConfig = Config()
|
||||
bwConfig.loadConfigFile(paths.TEST_PATH + "test.ini")
|
||||
assert bwConfig.get("test", "one") == "1"
|
||||
assert bwConfig.get("test", "two") == "two"
|
||||
assert bwConfig.get("testcase", "test") == "ok"
|
||||
assert bwConfig.getInt("test", "one") == 1
|
||||
assert bwConfig.getBool("test", "one") is True
|
||||
assert bwConfig.getStr("test", "two") == "two"
|
||||
assert bwConfig.getStr("testcase", "test") == "ok"
|
||||
|
||||
def test_getLocalConfigFailed(self):
|
||||
"""!fail while get values from local config file"""
|
||||
bwConfig = Config()
|
||||
bwConfig.loadConfigFile(paths.TEST_PATH + "test.ini")
|
||||
assert bwConfig.get("test", "abc") is None
|
||||
assert bwConfig.get("abc", "test") is None
|
||||
assert bwConfig.getStr("test", "abc") == "None"
|
||||
assert bwConfig.getStr("abc", "test") == "None"
|
||||
|
||||
def test_shareConfig(self):
|
||||
"""!load local config file and share it"""
|
||||
|
|
@ -65,8 +66,8 @@ class Test_Config:
|
|||
"""!try to get values from shared config where not exists"""
|
||||
bwConfig = Config()
|
||||
bwConfig.loadConfigFile(paths.TEST_PATH + "test.ini")
|
||||
assert bwConfig.get("test", "one") == "1"
|
||||
assert bwConfig.get("test", "one", "NotSetSharedConfig") is None
|
||||
assert bwConfig.getInt("test", "one") == 1
|
||||
assert bwConfig.getInt("test", "one", "NotSetSharedConfig") == 0
|
||||
|
||||
def test_getSharedConfig(self):
|
||||
"""!get values from shared config file"""
|
||||
|
|
@ -75,8 +76,8 @@ class Test_Config:
|
|||
assert bwConfig1._sharePoints["test_getSharedConfig"] is not None
|
||||
|
||||
bwConfig2 = Config()
|
||||
assert bwConfig2.get("test", "one") is None
|
||||
assert bwConfig2.get("test", "one", "test_getSharedConfig") == "1"
|
||||
assert bwConfig2.getStr("test", "one") == "None"
|
||||
assert bwConfig2.getInt("test", "one", "test_getSharedConfig") == 1
|
||||
|
||||
def test_getSharedConfigFailed(self):
|
||||
"""!fail while get values from shared config file"""
|
||||
|
|
@ -85,5 +86,5 @@ class Test_Config:
|
|||
assert bwConfig1._sharePoints["test_getSharedConfigFailed"] is not None
|
||||
|
||||
bwConfig2 = Config()
|
||||
assert bwConfig2.get("test", "abc", "test_getSharedConfigFailed") is None
|
||||
assert bwConfig2.get("abc", "test", "test_getSharedConfigFailed") is None
|
||||
assert bwConfig2.getStr("test", "abc", "test_getSharedConfigFailed") == "None"
|
||||
assert bwConfig2.getStr("abc", "test", "test_getSharedConfigFailed") == "None"
|
||||
|
|
|
|||
|
|
@ -21,25 +21,25 @@ from boswatch.decoder import decoder
|
|||
class Test_Decoder:
|
||||
"""!Unittests for the decoder"""
|
||||
|
||||
def test_decoder_no_data(self):
|
||||
def test_decoderNoData(self):
|
||||
"""!Test a empty string"""
|
||||
assert decoder.getDecoder("").decode("") is None
|
||||
|
||||
def test_decoder_zvei_valid(self):
|
||||
def test_decoderZveiValid(self):
|
||||
"""!Test valid ZVEI"""
|
||||
dec = decoder.getDecoder("ZVEI")
|
||||
assert not dec.decode("ZVEI1: 12345") is None
|
||||
assert not dec.decode("ZVEI1: 12838") is None
|
||||
assert not dec.decode("ZVEI1: 34675") is None
|
||||
|
||||
def test_decoder_zvei_doubleTone(self):
|
||||
def test_decoderZveiDoubleTone(self):
|
||||
"""!Test doubleTone included ZVEI"""
|
||||
dec = decoder.getDecoder("ZVEI")
|
||||
assert not dec.decode("ZVEI1: 6E789") is None
|
||||
assert not dec.decode("ZVEI1: 975E7") is None
|
||||
assert not dec.decode("ZVEI1: 2E87E") is None
|
||||
|
||||
def test_decoder_zvei_invalid(self):
|
||||
def test_decoderZveiInvalid(self):
|
||||
"""Test invalid ZVEI"""
|
||||
dec = decoder.getDecoder("ZVEI")
|
||||
assert dec.decode("ZVEI1: 1245A") is None
|
||||
|
|
@ -48,7 +48,7 @@ class Test_Decoder:
|
|||
assert dec.decode("ZVEI1: 54") is None
|
||||
assert dec.decode("ZVEI1: 54") is None
|
||||
|
||||
def test_decoder_pocsag_valid(self):
|
||||
def test_decoderPocsagValid(self):
|
||||
"""!Test valid POCSAG"""
|
||||
dec = decoder.getDecoder("POCSAG")
|
||||
assert not dec.decode("POCSAG512: Address: 1000000 Function: 0") is None
|
||||
|
|
@ -56,7 +56,7 @@ class Test_Decoder:
|
|||
assert not dec.decode("POCSAG1200: Address: 1000002 Function: 2") is None
|
||||
assert not dec.decode("POCSAG2400: Address: 1000003 Function: 3") is None
|
||||
|
||||
def test_decoder_pocsag_text(self):
|
||||
def test_decoderPocsagText(self):
|
||||
"""!Test POCSAG with text"""
|
||||
dec = decoder.getDecoder("POCSAG")
|
||||
assert not dec.decode("POCSAG512: Address: 1000000 Function: 0 Alpha: test") is None
|
||||
|
|
@ -64,7 +64,7 @@ class Test_Decoder:
|
|||
assert not dec.decode("POCSAG1200: Address: 1000002 Function: 2 Alpha: test") is None
|
||||
assert not dec.decode("POCSAG2400: Address: 1000003 Function: 3 Alpha: test") is None
|
||||
|
||||
def test_decoder_pocsag_short(self):
|
||||
def test_decoderPocsagShortRic(self):
|
||||
"""!Test short POCSAG"""
|
||||
dec = decoder.getDecoder("POCSAG")
|
||||
assert not dec.decode("POCSAG512: Address: 3 Function: 0 Alpha: test") is None
|
||||
|
|
@ -75,14 +75,14 @@ class Test_Decoder:
|
|||
assert not dec.decode("POCSAG2400: Address: 333333 Function: 0 Alpha: test") is None
|
||||
assert not dec.decode("POCSAG2400: Address: 3333333 Function: 0 Alpha: test") is None
|
||||
|
||||
def test_decoder_pocsag_invalid(self):
|
||||
def test_decoderPocsagInvalid(self):
|
||||
"""!Test invalid POCSAG"""
|
||||
dec = decoder.getDecoder("POCSAG")
|
||||
assert dec.decode("POCSAG512: Address: 333333F Function: 0 Alpha: invalid") is None
|
||||
assert dec.decode("POCSAG512: Address: 333333F Function: 1 Alpha: invalid") is None
|
||||
assert dec.decode("POCSAG512: Address: 3333333 Function: 4 Alpha: invalid") is None
|
||||
|
||||
def test_decoder_fms_valid(self):
|
||||
def test_decoderFmsValid(self):
|
||||
"""!Test valid FMS"""
|
||||
dec = decoder.getDecoder("FMS")
|
||||
assert not dec.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
|
||||
|
|
@ -91,7 +91,7 @@ class Test_Decoder:
|
|||
assert not dec.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 dec.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_decoder_fms_invalid(self):
|
||||
def test_decoderFmsInvalid(self):
|
||||
"""!Test invalid FMS"""
|
||||
dec = decoder.getDecoder("FMS")
|
||||
assert dec.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
|
||||
|
|
|
|||
|
|
@ -17,38 +17,39 @@
|
|||
|
||||
# import pytest # import the pytest framework
|
||||
|
||||
from boswatch.descriptor import descriptor
|
||||
from boswatch.packet import packet
|
||||
from boswatch.descriptor.descriptor import Descriptor
|
||||
from boswatch.descriptor.descriptor import DescriptionList
|
||||
from boswatch.packet.packet import Packet
|
||||
|
||||
|
||||
class Test_Descriptor:
|
||||
"""!Unittests for the descriptor"""
|
||||
|
||||
def test_loadCSVnotExist(self):
|
||||
def test_loadCsvNotExist(self):
|
||||
"""!read CSV file where not exist direct per DescriptionList class"""
|
||||
descList = descriptor.DescriptionList()
|
||||
descList = DescriptionList()
|
||||
assert not descList.loadCSV("boswatch")
|
||||
|
||||
def test_loadCSV(self):
|
||||
def test_loadCsv(self):
|
||||
"""!read CSV file direct per DescriptionList class"""
|
||||
descList = descriptor.DescriptionList()
|
||||
descList = DescriptionList()
|
||||
assert descList.loadCSV("zvei")
|
||||
|
||||
def test_descriptorLoadFailed(self):
|
||||
"""!read CSV file where not exist"""
|
||||
bwDescriptor = descriptor.Descriptor()
|
||||
bwDescriptor = Descriptor()
|
||||
assert not bwDescriptor.loadDescription("boswatch")
|
||||
|
||||
def test_descriptorLoad(self):
|
||||
"""!read CSV file"""
|
||||
bwDescriptor = descriptor.Descriptor()
|
||||
bwDescriptor = Descriptor()
|
||||
assert bwDescriptor.loadDescription("zvei")
|
||||
|
||||
def test_loadDescriptionsNotSet(self):
|
||||
"""!load descriptions where not set to an bwPacket"""
|
||||
bwDescriptor = descriptor.Descriptor()
|
||||
bwDescriptor = Descriptor()
|
||||
assert bwDescriptor.loadDescription("zvei")
|
||||
bwPacket = packet.Packet()
|
||||
bwPacket = Packet()
|
||||
bwPacket.set("mode", "zvei")
|
||||
bwPacket.set("zvei", "54321")
|
||||
assert bwDescriptor.addDescriptions(bwPacket)
|
||||
|
|
@ -57,9 +58,9 @@ class Test_Descriptor:
|
|||
|
||||
def test_loadDescriptions(self):
|
||||
"""!load descriptions to an bwPacket"""
|
||||
bwDescriptor = descriptor.Descriptor()
|
||||
bwDescriptor = Descriptor()
|
||||
assert bwDescriptor.loadDescription("zvei")
|
||||
bwPacket = packet.Packet()
|
||||
bwPacket = Packet()
|
||||
bwPacket.set("mode", "zvei")
|
||||
bwPacket.set("zvei", "12345")
|
||||
assert bwDescriptor.addDescriptions(bwPacket)
|
||||
|
|
|
|||
|
|
@ -18,7 +18,7 @@
|
|||
|
||||
# import pytest # import the pytest framework
|
||||
|
||||
import boswatch.utils.header
|
||||
from boswatch.utils import header
|
||||
|
||||
|
||||
class Test_Header:
|
||||
|
|
@ -26,12 +26,12 @@ class Test_Header:
|
|||
|
||||
def test_logoToLog(self):
|
||||
"""!Test logo to log"""
|
||||
assert boswatch.utils.header.logoToLog()
|
||||
assert header.logoToLog()
|
||||
|
||||
def test_infoToLog(self):
|
||||
"""!Test info to log"""
|
||||
assert boswatch.utils.header.infoToLog()
|
||||
assert header.infoToLog()
|
||||
|
||||
def test_logoToScreen(self):
|
||||
"""!Test logo to screen"""
|
||||
assert boswatch.utils.header.logoToScreen()
|
||||
assert header.logoToScreen()
|
||||
|
|
|
|||
Loading…
Reference in a new issue