mirror of
https://github.com/BOSWatch/BW3-Core.git
synced 2026-04-04 13:57:44 +00:00
move test files - fix config tests
This commit is contained in:
parent
82e3eb719d
commit
2ca62d90f2
17 changed files with 97 additions and 110 deletions
1
test/boswatch/__init__.py
Normal file
1
test/boswatch/__init__.py
Normal file
|
|
@ -0,0 +1 @@
|
|||
# coding=utf-8
|
||||
178
test/boswatch/test_ServerClient.py
Normal file
178
test/boswatch/test_ServerClient.py
Normal file
|
|
@ -0,0 +1,178 @@
|
|||
#!/usr/bin/python
|
||||
# -*- coding: utf-8 -*-
|
||||
"""!
|
||||
____ ____ ______ __ __ __ _____
|
||||
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
|
||||
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
|
||||
/ /_/ / /_/ /___/ /| |/ |/ / /_/ / /_/ /__/ / / / ___/ /
|
||||
/_____/\____//____/ |__/|__/\__,_/\__/\___/_/ /_/ /____/
|
||||
German BOS Information Script
|
||||
by Bastian Schroll
|
||||
|
||||
@file: test_ServerClient.py
|
||||
@date: 10.12.2017
|
||||
@author: Bastian Schroll
|
||||
@description: Unittests for BOSWatch. File have to run as "pytest" unittest
|
||||
"""
|
||||
import logging
|
||||
import time
|
||||
import queue
|
||||
import pytest
|
||||
|
||||
from boswatch.network.server import TCPServer
|
||||
from boswatch.network.client import TCPClient
|
||||
|
||||
|
||||
def setup_method(method):
|
||||
logging.debug("[TEST] %s.%s", method.__module__, method.__name__)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def getClient():
|
||||
return TCPClient()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def getServer():
|
||||
"""!Start and serve the sever for each functions where useServer is given"""
|
||||
dataQueue = queue.Queue()
|
||||
testServer = TCPServer(dataQueue)
|
||||
return testServer
|
||||
|
||||
|
||||
@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_clientConnectFailed(getClient):
|
||||
"""!Connect to a non available server"""
|
||||
assert not getClient.connect()
|
||||
|
||||
|
||||
def test_clientDisconnectFailed(getClient):
|
||||
"""!Disconnect while no connection is established"""
|
||||
assert getClient.disconnect()
|
||||
|
||||
|
||||
def test_clientTransmitFailed(getClient):
|
||||
"""!Transmit while no connection is established"""
|
||||
assert not getClient.transmit("test")
|
||||
|
||||
|
||||
def test_clientReceiveFailed(getClient):
|
||||
"""!Receive while no connection is established"""
|
||||
assert not getClient.receive()
|
||||
|
||||
|
||||
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()
|
||||
70
test/boswatch/test_broadcast.py
Normal file
70
test/boswatch/test_broadcast.py
Normal file
|
|
@ -0,0 +1,70 @@
|
|||
#!/usr/bin/python
|
||||
# -*- coding: utf-8 -*-
|
||||
"""!
|
||||
____ ____ ______ __ __ __ _____
|
||||
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
|
||||
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
|
||||
/ /_/ / /_/ /___/ /| |/ |/ / /_/ / /_/ /__/ / / / ___/ /
|
||||
/_____/\____//____/ |__/|__/\__,_/\__/\___/_/ /_/ /____/
|
||||
German BOS Information Script
|
||||
by Bastian Schroll
|
||||
|
||||
@file: test_broadcast.py
|
||||
@date: 25.09.2018
|
||||
@author: Bastian Schroll
|
||||
@description: Unittests for BOSWatch. File have to run as "pytest" unittest
|
||||
"""
|
||||
import logging
|
||||
import pytest
|
||||
|
||||
from boswatch.network.broadcast import BroadcastServer
|
||||
from boswatch.network.broadcast import BroadcastClient
|
||||
|
||||
|
||||
def setup_method(method):
|
||||
logging.debug("[TEST] %s.%s", method.__module__, method.__name__)
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def broadcastServer():
|
||||
"""!Server a BroadcastServer instance"""
|
||||
broadcastServer = BroadcastServer()
|
||||
yield broadcastServer
|
||||
if broadcastServer.isRunning:
|
||||
assert broadcastServer.stop()
|
||||
while broadcastServer.isRunning:
|
||||
pass
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def broadcastClient():
|
||||
"""!Server a BroadcastClient instance"""
|
||||
return BroadcastClient()
|
||||
|
||||
|
||||
def test_serverStartStop(broadcastServer):
|
||||
assert broadcastServer.start()
|
||||
assert broadcastServer.isRunning
|
||||
assert broadcastServer.stop()
|
||||
|
||||
|
||||
def test_serverDoubleStart(broadcastServer):
|
||||
assert broadcastServer.start()
|
||||
assert broadcastServer.start()
|
||||
assert broadcastServer.stop()
|
||||
|
||||
|
||||
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
|
||||
75
test/boswatch/test_config.py
Normal file
75
test/boswatch/test_config.py
Normal file
|
|
@ -0,0 +1,75 @@
|
|||
#!/usr/bin/python
|
||||
# -*- coding: utf-8 -*-
|
||||
"""!
|
||||
____ ____ ______ __ __ __ _____
|
||||
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
|
||||
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
|
||||
/ /_/ / /_/ /___/ /| |/ |/ / /_/ / /_/ /__/ / / / ___/ /
|
||||
/_____/\____//____/ |__/|__/\__,_/\__/\___/_/ /_/ /____/
|
||||
German BOS Information Script
|
||||
by Bastian Schroll
|
||||
|
||||
@file: test_config.py
|
||||
@date: 08.01.2017
|
||||
@author: Bastian Schroll
|
||||
@description: Unittests for BOSWatch. File have to run as "pytest" unittest
|
||||
"""
|
||||
import logging
|
||||
from typing import Dict
|
||||
|
||||
import pytest
|
||||
|
||||
from boswatch.configYaml import ConfigYAML
|
||||
|
||||
|
||||
def setup_method(method):
|
||||
logging.debug("[TEST] %s.%s", method.__module__, method.__name__)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def getConfig():
|
||||
return ConfigYAML()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def getFilledConfig():
|
||||
filledConfig = ConfigYAML()
|
||||
filledConfig.loadConfigFile("test_config.yaml")
|
||||
return filledConfig
|
||||
|
||||
|
||||
def test_loadConfigFile(getConfig):
|
||||
"""!load a local config file"""
|
||||
assert getConfig.loadConfigFile("test_config.yaml") is True
|
||||
|
||||
|
||||
def test_loadConfigFileFailed(getConfig):
|
||||
"""!load a local config file with syntax error"""
|
||||
assert getConfig.loadConfigFile("test_configFailed.yaml") is False
|
||||
|
||||
|
||||
def test_loadConfigFileNotFound(getConfig):
|
||||
"""!load a local config file where is not available"""
|
||||
assert getConfig.loadConfigFile("test_configNotFound.yaml") is False
|
||||
|
||||
|
||||
def test_getTypes(getFilledConfig):
|
||||
assert type(getFilledConfig.get("types")) is ConfigYAML
|
||||
assert type(getFilledConfig.get("types", "string")) is str
|
||||
assert type(getFilledConfig.get("types", "bool")) is bool
|
||||
assert type(getFilledConfig.get("types", "integer")) is int
|
||||
assert type(getFilledConfig.get("types", "float")) is float
|
||||
|
||||
|
||||
def test_getNestedConfig(getFilledConfig):
|
||||
nestedConfig = getFilledConfig.get("types")
|
||||
assert type(nestedConfig) is ConfigYAML
|
||||
assert nestedConfig.get("string") == "Hello World"
|
||||
|
||||
|
||||
def test_configIterationList(getFilledConfig):
|
||||
counter = 0
|
||||
for item in getFilledConfig.get("list"):
|
||||
assert type(item) is str
|
||||
counter += 1
|
||||
assert counter == 3
|
||||
10
test/boswatch/test_config.yaml
Normal file
10
test/boswatch/test_config.yaml
Normal file
|
|
@ -0,0 +1,10 @@
|
|||
types:
|
||||
string: Hello World
|
||||
bool: true
|
||||
integer: 11
|
||||
float: 3.14
|
||||
|
||||
list:
|
||||
- one
|
||||
- two
|
||||
- three
|
||||
5
test/boswatch/test_configFailed.yaml
Normal file
5
test/boswatch/test_configFailed.yaml
Normal file
|
|
@ -0,0 +1,5 @@
|
|||
types:
|
||||
string: Hello World
|
||||
boolean: true
|
||||
integer: 11
|
||||
float: 3.14
|
||||
102
test/boswatch/test_decoder.py
Normal file
102
test/boswatch/test_decoder.py
Normal file
|
|
@ -0,0 +1,102 @@
|
|||
#!/usr/bin/python
|
||||
# -*- coding: utf-8 -*-
|
||||
"""!
|
||||
____ ____ ______ __ __ __ _____
|
||||
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
|
||||
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
|
||||
/ /_/ / /_/ /___/ /| |/ |/ / /_/ / /_/ /__/ / / / ___/ /
|
||||
/_____/\____//____/ |__/|__/\__,_/\__/\___/_/ /_/ /____/
|
||||
German BOS Information Script
|
||||
by Bastian Schroll
|
||||
|
||||
@file: test_Decoder.py
|
||||
@date: 15.12.2017
|
||||
@author: Bastian Schroll
|
||||
@description: Unittests for BOSWatch. File have to run as "pytest" unittest
|
||||
"""
|
||||
import logging
|
||||
|
||||
from boswatch.decoder.decoder import Decoder
|
||||
|
||||
|
||||
def setup_method(method):
|
||||
logging.debug("[TEST] %s.%s", method.__module__, method.__name__)
|
||||
|
||||
|
||||
def test_decoderNoData():
|
||||
"""!Test a empty string"""
|
||||
assert Decoder.decode("") 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_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_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_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_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
|
||||
33
test/boswatch/test_header.py
Normal file
33
test/boswatch/test_header.py
Normal file
|
|
@ -0,0 +1,33 @@
|
|||
#!/usr/bin/python
|
||||
# -*- coding: utf-8 -*-
|
||||
"""!
|
||||
____ ____ ______ __ __ __ _____
|
||||
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
|
||||
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
|
||||
/ /_/ / /_/ /___/ /| |/ |/ / /_/ / /_/ /__/ / / / ___/ /
|
||||
/_____/\____//____/ |__/|__/\__,_/\__/\___/_/ /_/ /____/
|
||||
German BOS Information Script
|
||||
by Bastian Schroll
|
||||
|
||||
@file: test_header.py
|
||||
@date: 12.12.2017
|
||||
@author: Bastian Schroll
|
||||
@description: Unittests for BOSWatch. File have to run as "pytest" unittest
|
||||
"""
|
||||
import logging
|
||||
|
||||
from boswatch.utils import header
|
||||
|
||||
|
||||
def setup_method(method):
|
||||
logging.debug("[TEST] %s.%s", method.__module__, method.__name__)
|
||||
|
||||
|
||||
def test_logoToLog():
|
||||
"""!Test logo to log"""
|
||||
assert header.logoToLog()
|
||||
|
||||
|
||||
def test_infoToLog():
|
||||
"""!Test info to log"""
|
||||
assert header.infoToLog()
|
||||
58
test/boswatch/test_packet.py
Normal file
58
test/boswatch/test_packet.py
Normal file
|
|
@ -0,0 +1,58 @@
|
|||
#!/usr/bin/python
|
||||
# -*- coding: utf-8 -*-
|
||||
"""!
|
||||
____ ____ ______ __ __ __ _____
|
||||
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
|
||||
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
|
||||
/ /_/ / /_/ /___/ /| |/ |/ / /_/ / /_/ /__/ / / / ___/ /
|
||||
/_____/\____//____/ |__/|__/\__,_/\__/\___/_/ /_/ /____/
|
||||
German BOS Information Script
|
||||
by Bastian Schroll
|
||||
|
||||
@file: test_packet.py
|
||||
@date: 12.12.2017
|
||||
@author: Bastian Schroll
|
||||
@description: Unittests for BOSWatch. File have to run as "pytest" unittest
|
||||
"""
|
||||
import logging
|
||||
import pytest
|
||||
|
||||
from boswatch.packet import Packet
|
||||
|
||||
|
||||
def setup_method(method):
|
||||
logging.debug("[TEST] %s.%s", method.__module__, method.__name__)
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def buildPacket():
|
||||
"""!Build a BOSWatch packet and serve it to each test"""
|
||||
return Packet()
|
||||
|
||||
|
||||
def test_createPacket(buildPacket):
|
||||
"""!Create a packet"""
|
||||
assert buildPacket is not ""
|
||||
|
||||
|
||||
def test_copyPacket(buildPacket):
|
||||
"""!Copy a packet to an new instance"""
|
||||
bwCopyPacket = Packet(buildPacket.__str__())
|
||||
assert bwCopyPacket is not ""
|
||||
|
||||
|
||||
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"
|
||||
47
test/boswatch/test_paths.py
Normal file
47
test/boswatch/test_paths.py
Normal file
|
|
@ -0,0 +1,47 @@
|
|||
#!/usr/bin/python
|
||||
# -*- coding: utf-8 -*-
|
||||
"""!
|
||||
____ ____ ______ __ __ __ _____
|
||||
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
|
||||
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
|
||||
/ /_/ / /_/ /___/ /| |/ |/ / /_/ / /_/ /__/ / / / ___/ /
|
||||
/_____/\____//____/ |__/|__/\__,_/\__/\___/_/ /_/ /____/
|
||||
German BOS Information Script
|
||||
by Bastian Schroll
|
||||
|
||||
@file: test_paths.py
|
||||
@date: 22.02.2017
|
||||
@author: Bastian Schroll
|
||||
@description: Unittests for BOSWatch. File have to run as "pytest" unittest
|
||||
"""
|
||||
import logging
|
||||
import os
|
||||
|
||||
from boswatch.utils import paths
|
||||
|
||||
|
||||
def setup_method(method):
|
||||
logging.debug("[TEST] %s.%s", method.__module__, method.__name__)
|
||||
|
||||
|
||||
def test_fileExists():
|
||||
"""!load a local config file"""
|
||||
assert paths.fileExist("README.md")
|
||||
|
||||
|
||||
def test_fileNotExists():
|
||||
"""!load a local config file"""
|
||||
assert not paths.fileExist("notFound.txt")
|
||||
|
||||
|
||||
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")
|
||||
100
test/boswatch/test_timer.py
Normal file
100
test/boswatch/test_timer.py
Normal file
|
|
@ -0,0 +1,100 @@
|
|||
#!/usr/bin/python
|
||||
# -*- coding: utf-8 -*-
|
||||
"""!
|
||||
____ ____ ______ __ __ __ _____
|
||||
/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ /
|
||||
/ __ / / / /\__ \| | /| / / __ `/ __/ ___/ __ \ /_ <
|
||||
/ /_/ / /_/ /___/ /| |/ |/ / /_/ / /_/ /__/ / / / ___/ /
|
||||
/_____/\____//____/ |__/|__/\__,_/\__/\___/_/ /_/ /____/
|
||||
German BOS Information Script
|
||||
by Bastian Schroll
|
||||
|
||||
@file: test_timer.py
|
||||
@date: 21.09.2018
|
||||
@author: Bastian Schroll
|
||||
@description: Unittests for BOSWatch. File have to run as "pytest" unittest
|
||||
"""
|
||||
import logging
|
||||
import time
|
||||
import pytest
|
||||
|
||||
from boswatch.utils.timer import RepeatedTimer
|
||||
|
||||
|
||||
def setup_method(method):
|
||||
logging.debug("[TEST] %s.%s", method.__module__, method.__name__)
|
||||
|
||||
|
||||
def testTargetFast():
|
||||
"""!Fast worker thread"""
|
||||
logging.debug("run testTargetFast")
|
||||
|
||||
|
||||
def testTargetSlow():
|
||||
"""!Slow worker thread"""
|
||||
logging.debug("run testTargetSlow start")
|
||||
time.sleep(0.51)
|
||||
logging.debug("run testTargetSlow end")
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def useTimerFast():
|
||||
"""!Server a RepeatedTimer instance with fast worker"""
|
||||
testTimer = RepeatedTimer(0.1, testTargetFast)
|
||||
yield testTimer
|
||||
if testTimer.isRunning:
|
||||
assert 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_timerStartStop(useTimerFast):
|
||||
assert useTimerFast.start()
|
||||
assert useTimerFast.stop()
|
||||
|
||||
|
||||
def test_timerDoubleStart(useTimerFast):
|
||||
assert useTimerFast.start()
|
||||
assert useTimerFast.start()
|
||||
assert useTimerFast.stop()
|
||||
|
||||
|
||||
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
|
||||
Loading…
Add table
Add a link
Reference in a new issue