mirror of
https://github.com/BOSWatch/BW3-Core.git
synced 2025-12-06 07:12:04 +01:00
edit some tests
This commit is contained in:
parent
f14a78984a
commit
405c632ceb
|
|
@ -72,9 +72,9 @@ class Config:
|
||||||
except KeyError:
|
except KeyError:
|
||||||
logging.error("no sharePoint named: %s", sharePoint)
|
logging.error("no sharePoint named: %s", sharePoint)
|
||||||
except configparser.NoSectionError:
|
except configparser.NoSectionError:
|
||||||
logging.error("no shared config section: %s", section)
|
logging.warning("no shared config section: %s", section)
|
||||||
except configparser.NoOptionError:
|
except configparser.NoOptionError:
|
||||||
logging.error("no shared config option: %s", key)
|
logging.warning("no shared config option: %s", key)
|
||||||
except: # pragma: no cover
|
except: # pragma: no cover
|
||||||
logging.exception("error while reading shared config")
|
logging.exception("error while reading shared config")
|
||||||
return None
|
return None
|
||||||
|
|
@ -83,9 +83,15 @@ class Config:
|
||||||
try:
|
try:
|
||||||
return self._config.get(section, key)
|
return self._config.get(section, key)
|
||||||
except configparser.NoSectionError:
|
except configparser.NoSectionError:
|
||||||
logging.error("no local config section: %s", section)
|
logging.warning("no local config section: %s", section)
|
||||||
except configparser.NoOptionError:
|
except configparser.NoOptionError:
|
||||||
logging.error("no local config option: %s", key)
|
logging.warning("no local config option: %s", key)
|
||||||
except: # pragma: no cover
|
except: # pragma: no cover
|
||||||
logging.exception("error while reading local config")
|
logging.exception("error while reading local config")
|
||||||
return None
|
return None
|
||||||
|
|
||||||
|
def getAllSharepoints(self):
|
||||||
|
"""!Return a python dict of all set sharepoint's
|
||||||
|
|
||||||
|
@return Sharepoint dict"""
|
||||||
|
return self._sharePoints
|
||||||
|
|
|
||||||
|
|
@ -18,8 +18,8 @@
|
||||||
import pytest # import the pytest framework
|
import pytest # import the pytest framework
|
||||||
import time
|
import time
|
||||||
|
|
||||||
import boswatch.network.server
|
from boswatch.network.server import TCPServer
|
||||||
import boswatch.network.client
|
from boswatch.network.client import TCPClient
|
||||||
|
|
||||||
|
|
||||||
class Test_ServerClient:
|
class Test_ServerClient:
|
||||||
|
|
@ -28,7 +28,7 @@ class Test_ServerClient:
|
||||||
@pytest.fixture(scope="function")
|
@pytest.fixture(scope="function")
|
||||||
def useServer(self):
|
def useServer(self):
|
||||||
"""!Start and serve the sever for each functions where useServer is given"""
|
"""!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()
|
assert self.testServer.start()
|
||||||
time.sleep(0.1) # wait for server
|
time.sleep(0.1) # wait for server
|
||||||
yield self.testServer # server to all test where useServer is given
|
yield self.testServer # server to all test where useServer is given
|
||||||
|
|
@ -37,33 +37,33 @@ class Test_ServerClient:
|
||||||
|
|
||||||
def test_client_failed_connect(self):
|
def test_client_failed_connect(self):
|
||||||
"""!Connect to a non available server"""
|
"""!Connect to a non available server"""
|
||||||
self.testClient = boswatch.network.client.TCPClient()
|
self.testClient = TCPClient()
|
||||||
assert not self.testClient.connect()
|
assert not self.testClient.connect()
|
||||||
|
|
||||||
def test_client_failed_disconnect(self):
|
def test_client_failed_disconnect(self):
|
||||||
"""!Disconnect while no connection is established"""
|
"""!Disconnect while no connection is established"""
|
||||||
self.testClient = boswatch.network.client.TCPClient()
|
self.testClient = TCPClient()
|
||||||
assert not self.testClient.disconnect()
|
assert not self.testClient.disconnect()
|
||||||
|
|
||||||
def test_client_failed_transmit(self):
|
def test_client_failed_transmit(self):
|
||||||
"""!Transmit while no connection is established"""
|
"""!Transmit while no connection is established"""
|
||||||
self.testClient = boswatch.network.client.TCPClient()
|
self.testClient = TCPClient()
|
||||||
assert not self.testClient.transmit("test")
|
assert not self.testClient.transmit("test")
|
||||||
|
|
||||||
def test_client_failed_receive(self):
|
def test_client_failed_receive(self):
|
||||||
"""!Receive while no connection is established"""
|
"""!Receive while no connection is established"""
|
||||||
self.testClient = boswatch.network.client.TCPClient()
|
self.testClient = TCPClient()
|
||||||
assert not self.testClient.receive()
|
assert not self.testClient.receive()
|
||||||
|
|
||||||
def test_client_success_connect(self, useServer):
|
def test_client_success_connect(self, useServer):
|
||||||
"""!Connect to a server"""
|
"""!Connect to a server"""
|
||||||
self.testClient = boswatch.network.client.TCPClient()
|
self.testClient = TCPClient()
|
||||||
assert self.testClient.connect()
|
assert self.testClient.connect()
|
||||||
assert self.testClient.disconnect()
|
assert self.testClient.disconnect()
|
||||||
|
|
||||||
def test_client_reconnect(self, useServer):
|
def test_client_reconnect(self, useServer):
|
||||||
"""!Try a reconnect after a established connection"""
|
"""!Try a reconnect after a established connection"""
|
||||||
self.testClient = boswatch.network.client.TCPClient()
|
self.testClient = TCPClient()
|
||||||
assert self.testClient.connect()
|
assert self.testClient.connect()
|
||||||
assert self.testClient.disconnect()
|
assert self.testClient.disconnect()
|
||||||
assert self.testClient.connect()
|
assert self.testClient.connect()
|
||||||
|
|
@ -71,9 +71,9 @@ class Test_ServerClient:
|
||||||
|
|
||||||
def test_multi_connect(self, useServer):
|
def test_multi_connect(self, useServer):
|
||||||
"""!Connect with 2 clients to the server"""
|
"""!Connect with 2 clients to the server"""
|
||||||
self.testClient1 = boswatch.network.client.TCPClient()
|
self.testClient1 = TCPClient()
|
||||||
assert self.testClient1.connect()
|
assert self.testClient1.connect()
|
||||||
self.testClient2 = boswatch.network.client.TCPClient()
|
self.testClient2 = TCPClient()
|
||||||
assert self.testClient2.connect()
|
assert self.testClient2.connect()
|
||||||
# disconnect all
|
# disconnect all
|
||||||
assert self.testClient1.disconnect()
|
assert self.testClient1.disconnect()
|
||||||
|
|
@ -81,7 +81,7 @@ class Test_ServerClient:
|
||||||
|
|
||||||
def test_client_communicate(self, useServer):
|
def test_client_communicate(self, useServer):
|
||||||
"""!Try to send data to the server and check on '[ack]'"""
|
"""!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.connect()
|
||||||
assert self.testClient.transmit("test")
|
assert self.testClient.transmit("test")
|
||||||
assert self.testClient.receive() == "[ack]"
|
assert self.testClient.receive() == "[ack]"
|
||||||
|
|
@ -90,11 +90,11 @@ class Test_ServerClient:
|
||||||
def test_client_multi_communicate(self, useServer):
|
def test_client_multi_communicate(self, useServer):
|
||||||
"""!Try to send data to the server with 3 clients and check on '[ack]'"""
|
"""!Try to send data to the server with 3 clients and check on '[ack]'"""
|
||||||
# connect all
|
# connect all
|
||||||
self.testClient1 = boswatch.network.client.TCPClient()
|
self.testClient1 = TCPClient()
|
||||||
assert self.testClient1.connect()
|
assert self.testClient1.connect()
|
||||||
self.testClient2 = boswatch.network.client.TCPClient()
|
self.testClient2 = TCPClient()
|
||||||
assert self.testClient2.connect()
|
assert self.testClient2.connect()
|
||||||
self.testClient3 = boswatch.network.client.TCPClient()
|
self.testClient3 = TCPClient()
|
||||||
assert self.testClient3.connect()
|
assert self.testClient3.connect()
|
||||||
# send all
|
# send all
|
||||||
assert self.testClient1.transmit("test")
|
assert self.testClient1.transmit("test")
|
||||||
|
|
@ -111,7 +111,7 @@ class Test_ServerClient:
|
||||||
|
|
||||||
def test_server_restart(self):
|
def test_server_restart(self):
|
||||||
"""!Test a restart of the server"""
|
"""!Test a restart of the server"""
|
||||||
self.testServer = boswatch.network.server.TCPServer()
|
self.testServer = TCPServer()
|
||||||
assert self.testServer.start()
|
assert self.testServer.start()
|
||||||
assert self.testServer.stop()
|
assert self.testServer.stop()
|
||||||
assert self.testServer.start()
|
assert self.testServer.start()
|
||||||
|
|
@ -119,13 +119,13 @@ class Test_ServerClient:
|
||||||
|
|
||||||
def test_server_failed_stop(self):
|
def test_server_failed_stop(self):
|
||||||
"""!Test to start the server twice"""
|
"""!Test to start the server twice"""
|
||||||
self.testServer = boswatch.network.server.TCPServer()
|
self.testServer = TCPServer()
|
||||||
assert not self.testServer.stop()
|
assert not self.testServer.stop()
|
||||||
|
|
||||||
def test_server_doubleStart(self):
|
def test_server_doubleStart(self):
|
||||||
"""!Test to start the server twice"""
|
"""!Test to start the server twice"""
|
||||||
self.testServer1 = boswatch.network.server.TCPServer()
|
self.testServer1 = TCPServer()
|
||||||
self.testServer2 = boswatch.network.server.TCPServer()
|
self.testServer2 = TCPServer()
|
||||||
assert self.testServer1.start()
|
assert self.testServer1.start()
|
||||||
assert not self.testServer2.start()
|
assert not self.testServer2.start()
|
||||||
assert self.testServer1.stop()
|
assert self.testServer1.stop()
|
||||||
|
|
@ -134,9 +134,9 @@ class Test_ServerClient:
|
||||||
def test_server_output(self, useServer):
|
def test_server_output(self, useServer):
|
||||||
"""!Send data to server with 2 clients, check '[ack]' and data on server queue"""
|
"""!Send data to server with 2 clients, check '[ack]' and data on server queue"""
|
||||||
# connect all
|
# connect all
|
||||||
self.testClient1 = boswatch.network.client.TCPClient()
|
self.testClient1 = TCPClient()
|
||||||
assert self.testClient1.connect()
|
assert self.testClient1.connect()
|
||||||
self.testClient2 = boswatch.network.client.TCPClient()
|
self.testClient2 = TCPClient()
|
||||||
assert self.testClient2.connect()
|
assert self.testClient2.connect()
|
||||||
# send all
|
# send all
|
||||||
useServer.flushData()
|
useServer.flushData()
|
||||||
|
|
|
||||||
|
|
@ -17,8 +17,9 @@
|
||||||
|
|
||||||
# import pytest # import the pytest framework
|
# import pytest # import the pytest framework
|
||||||
|
|
||||||
from boswatch.descriptor import descriptor
|
from boswatch.descriptor.descriptor import Descriptor
|
||||||
from boswatch.packet import packet
|
from boswatch.descriptor.descriptor import DescriptionList
|
||||||
|
from boswatch.packet.packet import Packet
|
||||||
|
|
||||||
|
|
||||||
class Test_Descriptor:
|
class Test_Descriptor:
|
||||||
|
|
@ -26,29 +27,29 @@ class Test_Descriptor:
|
||||||
|
|
||||||
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"""
|
||||||
descList = descriptor.DescriptionList()
|
descList = DescriptionList()
|
||||||
assert not descList.loadCSV("boswatch")
|
assert not descList.loadCSV("boswatch")
|
||||||
|
|
||||||
def test_loadCSV(self):
|
def test_loadCSV(self):
|
||||||
"""!read CSV file direct per DescriptionList class"""
|
"""!read CSV file direct per DescriptionList class"""
|
||||||
descList = descriptor.DescriptionList()
|
descList = DescriptionList()
|
||||||
assert descList.loadCSV("zvei")
|
assert descList.loadCSV("zvei")
|
||||||
|
|
||||||
def test_descriptorLoadFailed(self):
|
def test_descriptorLoadFailed(self):
|
||||||
"""!read CSV file where not exist"""
|
"""!read CSV file where not exist"""
|
||||||
bwDescriptor = descriptor.Descriptor()
|
bwDescriptor = Descriptor()
|
||||||
assert not bwDescriptor.loadDescription("boswatch")
|
assert not bwDescriptor.loadDescription("boswatch")
|
||||||
|
|
||||||
def test_descriptorLoad(self):
|
def test_descriptorLoad(self):
|
||||||
"""!read CSV file"""
|
"""!read CSV file"""
|
||||||
bwDescriptor = descriptor.Descriptor()
|
bwDescriptor = Descriptor()
|
||||||
assert bwDescriptor.loadDescription("zvei")
|
assert bwDescriptor.loadDescription("zvei")
|
||||||
|
|
||||||
def test_loadDescriptionsNotSet(self):
|
def test_loadDescriptionsNotSet(self):
|
||||||
"""!load descriptions where not set to an bwPacket"""
|
"""!load descriptions where not set to an bwPacket"""
|
||||||
bwDescriptor = descriptor.Descriptor()
|
bwDescriptor = Descriptor()
|
||||||
assert bwDescriptor.loadDescription("zvei")
|
assert bwDescriptor.loadDescription("zvei")
|
||||||
bwPacket = packet.Packet()
|
bwPacket = Packet()
|
||||||
bwPacket.set("mode", "zvei")
|
bwPacket.set("mode", "zvei")
|
||||||
bwPacket.set("zvei", "54321")
|
bwPacket.set("zvei", "54321")
|
||||||
assert bwDescriptor.addDescriptions(bwPacket)
|
assert bwDescriptor.addDescriptions(bwPacket)
|
||||||
|
|
@ -57,9 +58,9 @@ class Test_Descriptor:
|
||||||
|
|
||||||
def test_loadDescriptions(self):
|
def test_loadDescriptions(self):
|
||||||
"""!load descriptions to an bwPacket"""
|
"""!load descriptions to an bwPacket"""
|
||||||
bwDescriptor = descriptor.Descriptor()
|
bwDescriptor = Descriptor()
|
||||||
assert bwDescriptor.loadDescription("zvei")
|
assert bwDescriptor.loadDescription("zvei")
|
||||||
bwPacket = packet.Packet()
|
bwPacket = Packet()
|
||||||
bwPacket.set("mode", "zvei")
|
bwPacket.set("mode", "zvei")
|
||||||
bwPacket.set("zvei", "12345")
|
bwPacket.set("zvei", "12345")
|
||||||
assert bwDescriptor.addDescriptions(bwPacket)
|
assert bwDescriptor.addDescriptions(bwPacket)
|
||||||
|
|
|
||||||
|
|
@ -18,7 +18,7 @@
|
||||||
|
|
||||||
# import pytest # import the pytest framework
|
# import pytest # import the pytest framework
|
||||||
|
|
||||||
import boswatch.utils.header
|
from boswatch.utils import header
|
||||||
|
|
||||||
|
|
||||||
class Test_Header:
|
class Test_Header:
|
||||||
|
|
@ -26,12 +26,12 @@ class Test_Header:
|
||||||
|
|
||||||
def test_logoToLog(self):
|
def test_logoToLog(self):
|
||||||
"""!Test logo to log"""
|
"""!Test logo to log"""
|
||||||
assert boswatch.utils.header.logoToLog()
|
assert header.logoToLog()
|
||||||
|
|
||||||
def test_infoToLog(self):
|
def test_infoToLog(self):
|
||||||
"""!Test info to log"""
|
"""!Test info to log"""
|
||||||
assert boswatch.utils.header.infoToLog()
|
assert header.infoToLog()
|
||||||
|
|
||||||
def test_logoToScreen(self):
|
def test_logoToScreen(self):
|
||||||
"""!Test logo to screen"""
|
"""!Test logo to screen"""
|
||||||
assert boswatch.utils.header.logoToScreen()
|
assert header.logoToScreen()
|
||||||
|
|
|
||||||
65
test/test_packet.py
Normal file
65
test/test_packet.py
Normal file
|
|
@ -0,0 +1,65 @@
|
||||||
|
#!/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 must be _run as "pytest" unittest
|
||||||
|
"""
|
||||||
|
|
||||||
|
|
||||||
|
import pytest # import the pytest framework
|
||||||
|
|
||||||
|
from boswatch.packet.packet import Packet
|
||||||
|
|
||||||
|
|
||||||
|
class Test_Packet:
|
||||||
|
"""!Unittests for the BOSWatch packet"""
|
||||||
|
|
||||||
|
@pytest.fixture(scope="function")
|
||||||
|
def buildPacket(self):
|
||||||
|
"""!Build a BOSWatch packet and serve it to each test"""
|
||||||
|
bwPacket = Packet()
|
||||||
|
yield bwPacket
|
||||||
|
|
||||||
|
def test_crearePacket(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_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_setGetField(self, buildPacket):
|
||||||
|
"""!set and get a field"""
|
||||||
|
buildPacket.set("testField", "test")
|
||||||
|
assert buildPacket.get("testField") is "test"
|
||||||
|
|
||||||
|
def test_addClientData(self, buildPacket):
|
||||||
|
"""!add client data to packet"""
|
||||||
|
buildPacket.addClientData()
|
||||||
|
assert buildPacket.get("clientVersion")
|
||||||
|
|
||||||
|
def test_addServerData(self, buildPacket):
|
||||||
|
"""!add server data to packet"""
|
||||||
|
buildPacket.addServerData()
|
||||||
|
assert buildPacket.get("serverVersion")
|
||||||
Loading…
Reference in a new issue