unit test exnteded

pus service testin separate file

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:

test/OBSW_PusServiceTest.py

+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

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()