Something went wrong on our end
Forked from an inaccessible project.
-
Robin.Mueller authoredRobin.Mueller authored
OBSW_PusServiceTest.py 4.15 KiB
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
# wiretapping messages
elif currentTmInfo["service"] == 2 and currentTmInfo["subservice"] == 130:
self.miscCounter = self.miscCounter + 1
elif currentTmInfo["service"] == 2 and currentTmInfo["subservice"] == 131:
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()