diff --git a/.idea/runConfigurations/OBSW_TmTcClient__Service_200_Serial.xml b/.idea/runConfigurations/tmtcclient__Service_200_Serial.xml
similarity index 80%
rename from .idea/runConfigurations/OBSW_TmTcClient__Service_200_Serial.xml
rename to .idea/runConfigurations/tmtcclient__Service_200_Serial.xml
index 3ece0d3eaf7e29f8e12d3315e5112a8c94130d53..ec2175bfadade9795139b22a62d815dd0ffa752e 100644
--- a/.idea/runConfigurations/OBSW_TmTcClient__Service_200_Serial.xml
+++ b/.idea/runConfigurations/tmtcclient__Service_200_Serial.xml
@@ -1,5 +1,5 @@
<component name="ProjectRunConfigurationManager">
- <configuration default="false" name="OBSW_TmTcClient Service 200 Serial" type="PythonConfigurationType" factoryName="Python" folderName="Serial Communication">
+ <configuration default="false" name="tmtcclient Service 200 Serial" type="PythonConfigurationType" factoryName="Python" folderName="Serial Communication">
<module name="tmtc" />
<option name="INTERPRETER_OPTIONS" value="" />
<option name="PARENT_ENVS" value="true" />
@@ -13,7 +13,7 @@
<option name="ADD_SOURCE_ROOTS" value="true" />
<EXTENSION ID="PythonCoverageRunConfigurationExtension" runner="coverage.py" />
<option name="SCRIPT_NAME" value="$PROJECT_DIR$/obsw_tmtc_client.py" />
- <option name="PARAMETERS" value="-m 3 -s 200 -c 1 --hk -t 5" />
+ <option name="PARAMETERS" value="-m 3 -s 200 -c 1 --hk -t 3" />
<option name="SHOW_COMMAND_LINE" value="false" />
<option name="EMULATE_TERMINAL" value="false" />
<option name="MODULE_MODE" value="false" />
diff --git a/test/obsw_module_test.py b/test/obsw_module_test.py
index 9c1f4681c42b95d1e2884648a1682cc85694f96c..ba476d98049c01f97c47945acc4aed5cc8916396 100644
--- a/test/obsw_module_test.py
+++ b/test/obsw_module_test.py
@@ -35,6 +35,8 @@ to be specified in the analyse_tc_info method of the child test.
@author: R. Mueller
"""
+import struct
+
import sys
import unittest
from enum import Enum
@@ -249,6 +251,20 @@ class TestService(unittest.TestCase):
self.valid = False
assertion_dict.update({"MiscCount": self.misc_counter})
+ def _generic_mode_tm_check(self, current_tm_info: dict,
+ reporter_id: bytearray, event_id_list: list):
+ # mode change
+ if current_tm_info[TmDictionaryKeys.SERVICE] == 5:
+ if (current_tm_info[TmDictionaryKeys.REPORTER_ID] ==
+ struct.unpack('>I', reporter_id)[0]) \
+ and (current_tm_info[TmDictionaryKeys.EVENT_ID] in event_id_list):
+ self.event_counter = self.event_counter + 1
+ if current_tm_info[TmDictionaryKeys.SERVICE] == 200 and \
+ current_tm_info[TmDictionaryKeys.SUBSERVICE] == 6:
+ # mode change confirmation
+ self.misc_counter += 1
+
+
@classmethod
# we could print out everything here.
def tearDownClass(cls):
diff --git a/test/obsw_pus_service_test.py b/test/obsw_pus_service_test.py
index 857013b158a90a76e1add81cc42a917e5612875e..3aa61d557b36c27f7cb1a014500275de692db006 100644
--- a/test/obsw_pus_service_test.py
+++ b/test/obsw_pus_service_test.py
@@ -11,7 +11,7 @@ from typing import Deque
from test.obsw_module_test import TestService, PusTmInfoQueueT, TmDictionaryKeys, AssertionDictKeys
from tc.obsw_pus_tc_base import PusTcInfoQueueT
from tc.obsw_pus_tc_packer import pack_service17_test_into, pack_service5_test_into, \
- pack_service2_test_into, pack_service8_test_into
+ pack_service2_test_into, pack_service8_test_into, pack_service200_test_into
import config.obsw_config as g
from utility.obsw_logger import get_logger
@@ -137,18 +137,10 @@ class TestService8(TestService):
current_tm_info = tm_info_queue.pop()
if current_tm_info[TmDictionaryKeys.SERVICE] == 1:
self.scan_for_respective_tc(current_tm_info)
- if current_tm_info[TmDictionaryKeys.SERVICE] == 5:
- if (current_tm_info[TmDictionaryKeys.REPORTER_ID] ==
- struct.unpack('>I', g.DUMMY_DEVICE_ID)[0]) and \
- (current_tm_info[TmDictionaryKeys.EVENT_ID] == 7401
- or current_tm_info[TmDictionaryKeys.EVENT_ID] == 7400):
- self.event_counter += 1
- if current_tm_info[TmDictionaryKeys.SERVICE] == 200:
- # mode change confirmation
- self.misc_counter += 1
if (current_tm_info[TmDictionaryKeys.SERVICE] == 8 and
current_tm_info[TmDictionaryKeys.SUBSERVICE] == 130):
self.data_reply_count += 1
+ self._generic_mode_tm_check(current_tm_info, g.DUMMY_DEVICE_ID, [7401, 7400])
class TestService9(unittest.TestCase):
@@ -192,3 +184,40 @@ class TestService17(TestService):
def tearDownClass(cls):
print("Testing Service 17 finished")
super().tearDownClass()
+
+
+class TestService200(TestService):
+ @classmethod
+ def setUpClass(cls: TestService):
+ super().setUpClass()
+ LOGGER.info("Testing Service 200")
+ cls.wait_intervals = [1, 2, 3]
+ cls.wait_time = [2, 2, 2]
+ cls.tm_timeout = g.G_TM_TIMEOUT
+ pack_service200_test_into(cls.test_queue)
+
+ def test_Service200(self):
+ assertion_dict = self.perform_testing_and_generate_assertion_dict()
+ # 4 x Mode change with 2 events for each
+ self.event_expected = 8
+ self.misc_expected = 4
+ self._perform_generic_assertion_test(assertion_dict)
+
+ def _analyse_tm_tc_info(self, tm_info_queue: PusTmInfoQueueT, tc_info_queue: PusTcInfoQueueT):
+ assertion_dict = super()._analyse_tm_tc_info(tm_info_queue=tm_info_queue,
+ tc_info_queue=tc_info_queue)
+ # add anything else other than tc verification counter
+ # and ssc that is needed for tm analysis
+ return assertion_dict
+
+ def analyse_tc_info(self, tc_info_queue):
+ super().analyse_tc_info(tc_info_queue)
+
+ def analyse_tm_info(self, tm_info_queue: Deque, assertion_dict: dict):
+ while not tm_info_queue.__len__() == 0:
+ current_tm_info = tm_info_queue.pop()
+ # Tc verification scanning is generic and has been moved to the superclass
+ if current_tm_info[TmDictionaryKeys.SERVICE] == 1:
+ self.scan_for_respective_tc(current_tm_info)
+ self._generic_mode_tm_check(current_tm_info, g.DUMMY_DEVICE_ID, [7401, 7400])
+