diff --git a/OBSW_TmTcClient.py b/OBSW_TmTcClient.py
index b416a4cadc70b24978ab6dbf56f52d0dc2ec6350..69305d4d979cb613512a8d6596a90aa72b7138d1 100644
--- a/OBSW_TmTcClient.py
+++ b/OBSW_TmTcClient.py
@@ -58,7 +58,7 @@ import signal
import queue
import unittest
-from test import OBSW_UnitTest
+from test import OBSW_PusServiceTest
from config import OBSW_Config as g
from config.OBSW_Config import setGlobals
from tc.OBSW_TcPacker import PUSTelecommand, createTotalTcQueue, serviceTestSelect
@@ -107,7 +107,7 @@ def main():
# Set up test suite and run it with runner
# Verbosity specifies detail level
# noinspection PyTypeChecker
- suite = unittest.TestLoader().loadTestsFromModule(OBSW_UnitTest)
+ suite = unittest.TestLoader().loadTestsFromModule(OBSW_PusServiceTest)
unittest.TextTestRunner(verbosity=2).run(suite)
else:
diff --git a/test/OBSW_PusServiceTest.py b/test/OBSW_PusServiceTest.py
new file mode 100644
index 0000000000000000000000000000000000000000..8b057941d24b004440ee405bb88a036b4474997e
--- /dev/null
+++ b/test/OBSW_PusServiceTest.py
@@ -0,0 +1,119 @@
+import unittest
+from test.OBSW_UnitTest import TestService
+from tc.OBSW_TcPacker import packService17TestInto, packService5TestInto, packService2TestInto
+import config.OBSW_Config as g
+
+
+class TestService2(TestService):
+ @classmethod
+ def setUpClass(cls):
+ super().setUpClass()
+ print("Testing Service 2")
+ # all commands must be sent sequentially, not as a burst
+ cls.waitIntervals = [1, 2, 3, 4]
+ cls.waitTime = [2, 2, 2, 2]
+ packService2TestInto(cls.testQueue)
+
+ def test_Service2(self):
+ assertionDict = self.performTestingAndGenerateAssertionDict()
+ self.eventExpected = 1
+ self.miscExpected = 4
+ super().performGenericAssertionTest(assertionDict)
+
+ def analyseTmInfo(self, tmInfoQueue, assertionDict):
+ while not tmInfoQueue.empty():
+ currentTmInfo = tmInfoQueue.get()
+ # Tc verification scanning is generic and has been moved to the superclass
+ if currentTmInfo["service"] == 1:
+ super().scanForRespectiveTc(currentTmInfo)
+ # Here, the desired event Id or RID can be specified
+ if currentTmInfo["service"] == 5:
+ # mode change
+ if currentTmInfo["EventID"] == 7401 and currentTmInfo["RID"] == 0x4400affe:
+ self.eventCounter = self.eventCounter + 1
+ if currentTmInfo["service"] == 200 and currentTmInfo["subservice"] == 6:
+ # mode change confirmation
+ self.miscCounter = self.miscCounter + 1
+ elif currentTmInfo["service"] == 2 and currentTmInfo["subservice"] == 130:
+ print("subservice 130 detected")
+ self.miscCounter = self.miscCounter + 1
+ elif currentTmInfo["service"] == 2 and currentTmInfo["subservice"] == 131:
+ print("subservice 131 detected")
+ self.miscCounter = self.miscCounter + 1
+ if currentTmInfo["valid"] == 0:
+ self.valid = False
+ assertionDict.update({"MiscCount": self.miscCounter})
+
+
+class TestService5(TestService):
+ @classmethod
+ def setUpClass(cls):
+ super().setUpClass()
+ print("Testing Service 5")
+ # Wait intervals after TC 1,2 and 3 with specified wait times
+ # This is required because the OBSW tasks runs with fixed sequences
+ cls.waitIntervals = [1, 2, 3]
+ cls.waitTime = [1.5, 2.0, 2.0]
+ packService5TestInto(cls.testQueue)
+
+ def test_Service5(self):
+ # analyseTmInfo() and analyseTcInfo are called here
+ super().performService5or17Test()
+
+ def analyseTcInfo(self, tcInfoQueue):
+ super().analyseTcInfo(tcInfoQueue)
+
+ def analyseTmInfo(self, tmInfoQueue, assertionDict):
+ super().analyseService5or17TM(tmInfoQueue, assertionDict)
+
+ @classmethod
+ def tearDownClass(cls):
+ super().tearDownClass()
+
+
+class TestService6(unittest.TestCase):
+ @classmethod
+ def setUpClass(cls):
+ print("Hallo Test 6!")
+
+
+class TestService8(unittest.TestCase):
+ @classmethod
+ def setUpClass(cls):
+ print("Hallo Test 2!")
+
+
+class TestService9(unittest.TestCase):
+ @classmethod
+ def setUpClass(cls):
+ print("Hallo Test 9!")
+
+
+class TestService17(TestService):
+ @classmethod
+ def setUpClass(cls):
+ super().setUpClass()
+ print("Testing Service 17")
+ cls.waitIntervals = [2]
+ cls.waitTime = 2
+ cls.tmTimeout = g.tmTimeout
+ packService17TestInto(cls.testQueue)
+
+ def test_Service17(self):
+ super().performService5or17Test()
+
+ def analyseTmTcInfo(self, tmInfoQueue, tcInfoQueue):
+ assertionDict = super().analyseTmTcInfo(tmInfoQueue, tcInfoQueue)
+ # add anything elsee other than tc verification counter and ssc that is needed for tm analysis
+ return assertionDict
+
+ def analyseTcInfo(self, tcInfoQueue):
+ super().analyseTcInfo(tcInfoQueue)
+
+ def analyseTmInfo(self, tmInfoQueue, assertionDict):
+ super().analyseService5or17TM(tmInfoQueue, assertionDict)
+
+ @classmethod
+ def tearDownClass(cls):
+ print("Testing Service 17 finished")
+ super().tearDownClass()
\ No newline at end of file
diff --git a/test/OBSW_UnitTest.py b/test/OBSW_UnitTest.py
index 821f06235eafe366b8849bc296e751a4f26358b1..be89bd250f115bf156ef5d3f67d78f7dcf211452 100644
--- a/test/OBSW_UnitTest.py
+++ b/test/OBSW_UnitTest.py
@@ -10,7 +10,9 @@ Set up the TMTC client as specified in the header comment and use the unit testi
For Developers:
TestService is the template method, analyseTmInfo and analyseTcInfo are implemented for
specific services or devices by setting up assertionDict entries which count received packets
-based on subservice or entry values.
+based on subservice or entry values. There is a default implementation provided for analyseTcInfo.
+The log files generated by individual service test or the software test mode when providing
+the -p flag are very useful for extending the unit tester
Example Service 17:
TC[17,1] and TC[17,128] are sent, TM[1,1] and TM[1,7] (TC verification) are expected twice, TM[17,2] (ping reply)
@@ -26,7 +28,8 @@ import unittest
import queue
-from tc.OBSW_TcPacker import packService17TestInto, packService5TestInto, packDummyDeviceTestInto
+from tc.OBSW_TcPacker import packDummyDeviceTestInto
+from abc import abstractmethod
from sendreceive.OBSW_MultipleCommandsSenderReceiver import MultipleCommandSenderReceiver
from utility.OBSW_TmTcPrinter import TmtcPrinter
from comIF.OBSW_Ethernet_ComIF import EthernetComIF
@@ -34,9 +37,13 @@ from comIF.OBSW_Serial_ComIF import SerialComIF
from config import OBSW_Config as g
+# documentation: https://docs.python.org/3/library/unittest.html#unittest.TestCase.setUpClass
class TestService(unittest.TestCase):
testQueue = queue.Queue()
+ communicationInterface = 0
+ # A class method which is run before every individually class module is run
+ # must be decorated by the @classmethod fixture
@classmethod
def setUpClass(cls):
cls.displayMode = "long"
@@ -57,12 +64,14 @@ class TestService(unittest.TestCase):
cls.tmtcPrinter = TmtcPrinter(cls.displayMode, cls.printFile, True)
if g.comIF == 0:
cls.communicationInterface = EthernetComIF(cls.tmtcPrinter, g.tmTimeout, g.tcSendTimeoutFactor,
- g.sockSend, g.sockReceive, g.sendAddress)
+ g.sockSend, g.sockReceive, g.sendAddress)
else:
baudRate = 115200
cls.communicationInterface = SerialComIF(cls.tmtcPrinter, g.comPort, baudRate, g.serialTimeout)
# connectToBoard()
+ # This function should be called in each individual test to send the actual telecommands
+ # which are stored inside testQueue
def performTestingAndGenerateAssertionDict(self):
UnitTester = MultipleCommandSenderReceiver(self.communicationInterface, self.tmtcPrinter, self.testQueue,
self.tmTimeout, self.waitIntervals, self.waitTime, self.printTm,
@@ -81,6 +90,7 @@ class TestService(unittest.TestCase):
self.tcVerifyCounter = 0
self.tcVerifyStepCounter = 0
# child test implements this
+ # default implementation provided
self.analyseTcInfo(tcInfoQueue)
# These values are incremented in the analyseTmInfo function
@@ -118,6 +128,11 @@ class TestService(unittest.TestCase):
self.tcServiceArray.append(currentTcInfo["service"])
self.tcSubserviceArray.append(currentTcInfo["subservice"])
+ # must be implemented by child test !
+ @abstractmethod
+ def analyseTmInfo(self, tmInfoQueue, assertionDict):
+ pass
+
# this function looks whether the tc verification SSC matched
# a SSC of the sent TM
def scanForRespectiveTc(self, currentTmInfo):
@@ -131,17 +146,24 @@ class TestService(unittest.TestCase):
elif currentSubservice == 7:
self.tcVerifiedCompletion = self.tcVerifiedCompletion + 1
- # these tests are identical
- def performService5or17Test(self):
- assertionDict = self.performTestingAndGenerateAssertionDict()
- self.eventExpected = 1
- self.miscExpected = 2
+ def performGenericAssertionTest(self, assertionDict):
+ if assertionDict is None:
+ print("Performing Generic Assertion Test: "
+ "Configuratrion error, asseretion dictionary was not passed properly")
+ exit()
self.assertEqual(assertionDict["TcStartCount"], self.tcVerifyCounter)
self.assertEqual(assertionDict["TcCompletionCount"], self.tcVerifyCounter)
self.assertEqual(assertionDict["EventCount"], self.eventExpected)
self.assertEqual(assertionDict["MiscCount"], self.miscExpected)
self.assertTrue(assertionDict["Valid"])
+ # these tests are identical
+ def performService5or17Test(self):
+ assertionDict = self.performTestingAndGenerateAssertionDict()
+ self.eventExpected = 1
+ self.miscExpected = 2
+ self.performGenericAssertionTest(assertionDict)
+
def analyseService5or17TM(self, tmInfoQueue, assertionDict):
self.miscCounter = 0
while not tmInfoQueue.empty():
@@ -167,83 +189,6 @@ class TestService(unittest.TestCase):
if g.comIF == 1:
cls.communicationInterface.serial.close()
- def analyseTmInfo(self, tmInfoQueue, assertionDict):
- pass
-
-
-class TestService5(TestService):
-
- @classmethod
- def setUpClass(cls):
- super().setUpClass()
- print("Testing Service 5")
- cls.waitIntervals = [1, 2, 3]
- cls.waitTime = [1.5, 2.0, 2.0]
- packService5TestInto(cls.testQueue)
-
- def test_Service5(self):
-
- # analyseTmInfo() and analyseTcInfo are called here
- super().performService5or17Test()
-
- def analyseTcInfo(self, tcInfoQueue):
- super().analyseTcInfo(tcInfoQueue)
-
- def analyseTmInfo(self, tmInfoQueue, assertionDict):
- super().analyseService5or17TM(tmInfoQueue, assertionDict)
-
- @classmethod
- def tearDownClass(cls):
- super().tearDownClass()
-
-
-class TestService6(unittest.TestCase):
- @classmethod
- def setUpClass(cls):
- print("Hallo Test 6!")
-
-
-class TestService8(unittest.TestCase):
- @classmethod
- def setUpClass(cls):
- print("Hallo Test 2!")
-
-
-class TestService9(unittest.TestCase):
- @classmethod
- def setUpClass(cls):
- print("Hallo Test 9!")
-
-
-class TestService17(TestService):
- @classmethod
- def setUpClass(cls):
- super().setUpClass()
- print("Testing Service 17")
- cls.waitIntervals = [2]
- cls.waitTime = 2
- cls.tmTimeout = g.tmTimeout
- packService17TestInto(cls.testQueue)
-
- def test_Service17(self):
- super().performService5or17Test()
-
- def analyseTmTcInfo(self, tmInfoQueue, tcInfoQueue):
- assertionDict = super().analyseTmTcInfo(tmInfoQueue, tcInfoQueue)
- # add anything elsee other than tc verification counter and ssc that is needed for tm analysis
- return assertionDict
-
- def analyseTcInfo(self, tcInfoQueue):
- super().analyseTcInfo(tcInfoQueue)
-
- def analyseTmInfo(self, tmInfoQueue, assertionDict):
- super().analyseService5or17TM(tmInfoQueue, assertionDict)
-
- @classmethod
- def tearDownClass(cls):
- print("Testing Service 17 finished")
- super().tearDownClass()
-
class TestDummyDevice(TestService):
@classmethod
@@ -252,6 +197,9 @@ class TestDummyDevice(TestService):
print("Testing Dummy Device")
packDummyDeviceTestInto(cls.testQueue)
+ def analyseTmInfo(self, tmInfoQueue, assertionDict):
+ pass
+
if __name__ == '__main__':
unittest.main()