From 993c7544d0a030db72f29c78103663ff89ec3c1b Mon Sep 17 00:00:00 2001
From: "Robin.Mueller" <robin.mueller.m@gmail.com>
Date: Fri, 10 Apr 2020 13:18:50 +0200
Subject: [PATCH] sender receiver objects simplified
---
config/OBSW_Config.py | 9 +-
obsw_tmtc_client.py | 22 ++--
sendreceive/obsw_command_sender_receiver.py | 50 +++++---
.../obsw_multiple_commands_sender_receiver.py | 40 +++----
.../obsw_sequential_sender_receiver.py | 23 ++--
.../obsw_single_command_sender_receiver.py | 9 +-
tc/OBSW_TcService2.py | 2 +-
tc/OBSW_TcService3.py | 2 +-
tc/OBSW_TcService8.py | 2 +-
tc/obsw_pus_tc_packer.py | 4 +-
.../{OBSW_UnitTest.py => obsw_module_test.py} | 111 ++++++++----------
...erviceTest.py => obsw_pus_service_test.py} | 36 +++---
utility/obsw_args_parser.py | 4 +-
utility/obsw_exit_handler.py | 6 +-
14 files changed, 156 insertions(+), 164 deletions(-)
rename test/{OBSW_UnitTest.py => obsw_module_test.py} (68%)
rename test/{OBSW_PusServiceTest.py => obsw_pus_service_test.py} (77%)
diff --git a/config/OBSW_Config.py b/config/OBSW_Config.py
index ae1d4d2..67ae5b3 100644
--- a/config/OBSW_Config.py
+++ b/config/OBSW_Config.py
@@ -77,9 +77,9 @@ G_PRINT_RAW_TM = False
"""
These objects are set for the Unit Test, no better solution found yet
"""
-tmListener = None
-comInterface = None
-tmtcPrinter = None
+G_TM_LISTENER = None
+G_COM_INTERFACE = None
+G_TMTC_PRINTER = None
# noinspection PyUnusedLocal
@@ -135,9 +135,8 @@ def setGlobals(args):
G_SEND_ADDRESS = sendAddressToSet
G_MODE_ID = G_MODE_ID
G_PRINT_HK_DATA = args.print_hk
- G_PRINT_TO_FILE = args.printFile
+ G_PRINT_TO_FILE = args.print_file
G_PRINT_RAW_TM = args.rawDataPrint
- G_SERVICE = G_SERVICE
G_COM_PORT = args.com_port
G_TM_TIMEOUT = args.tm_timeout
diff --git a/obsw_tmtc_client.py b/obsw_tmtc_client.py
index ac39fc0..909736e 100644
--- a/obsw_tmtc_client.py
+++ b/obsw_tmtc_client.py
@@ -59,7 +59,7 @@ import logging
import sys
from collections import deque
-from test import OBSW_PusServiceTest
+from test import obsw_pus_service_test
from config import OBSW_Config as g
from config.OBSW_Config import setGlobals
from tc.obsw_pus_tc_packer import PusTelecommand, createTotalTcQueue, service_test_select
@@ -116,32 +116,30 @@ def main():
pus_packet_tuple = command_preparation()
sender_and_receiver = SingleCommandSenderReceiver(
comInterface=communication_interface, tmtcPrinter=tmtc_printer, tmListener=tm_listener,
- pusPacketTuple=pus_packet_tuple, tmTimeout=g.G_TM_TIMEOUT,
- tcTimeoutFactor=g.G_TC_SEND_TIMEOUT_FACTOR)
+ pusPacketTuple=pus_packet_tuple)
sender_and_receiver.send_single_tc_and_receive_tm()
elif g.G_MODE_ID == g.ModeList.ServiceTestMode:
service_queue = deque()
sender_and_receiver = SequentialCommandSenderReceiver(
- comInterface=communication_interface, tmtcPrinter=tmtc_printer, tmTimeout=g.G_TM_TIMEOUT,
- tcQueue=service_test_select(g.G_SERVICE, service_queue),
- tcTimeoutFactor=g.G_TC_SEND_TIMEOUT_FACTOR, tmListener=tm_listener)
+ com_interface=communication_interface, tmtc_printer=tmtc_printer, tm_listener=tm_listener,
+ tc_queue=service_test_select(g.G_SERVICE, service_queue))
sender_and_receiver.send_queue_tc_and_receive_tm_sequentially()
elif g.G_MODE_ID == g.ModeList.SoftwareTestMode:
all_tc_queue = createTotalTcQueue()
sender_and_receiver = SequentialCommandSenderReceiver(
- comInterface=communication_interface, tmtcPrinter=tmtc_printer, tmTimeout=g.G_TM_TIMEOUT,
- tcQueue=all_tc_queue, tcTimeoutFactor=g.G_TC_SEND_TIMEOUT_FACTOR, tmListener=tm_listener)
+ com_interface=communication_interface, tmtc_printer=tmtc_printer,
+ tc_queue=all_tc_queue, tm_listener=tm_listener)
sender_and_receiver.send_queue_tc_and_receive_tm_sequentially()
elif g.G_MODE_ID == g.ModeList.UnitTest:
# Set up test suite and run it with runner. Verbosity specifies detail level
- g.tmListener = tm_listener
- g.comInterface = communication_interface
- g.tmtcPrinter = tmtc_printer
+ g.G_TM_LISTENER = tm_listener
+ g.G_COM_INTERFACE = communication_interface
+ g.G_TMTC_PRINTER = tmtc_printer
# noinspection PyTypeChecker
- suite = unittest.TestLoader().loadTestsFromModule(OBSW_PusServiceTest)
+ suite = unittest.TestLoader().loadTestsFromModule(obsw_pus_service_test)
unittest.TextTestRunner(verbosity=2).run(suite)
else:
diff --git a/sendreceive/obsw_command_sender_receiver.py b/sendreceive/obsw_command_sender_receiver.py
index 571ef36..3e71394 100644
--- a/sendreceive/obsw_command_sender_receiver.py
+++ b/sendreceive/obsw_command_sender_receiver.py
@@ -1,5 +1,5 @@
"""
-Program: OBSW_UnitTest.py
+Program: obsw_module_test.py
Date: 01.11.2019
Description: All functions related to TmTc Sending and Receiving, used by UDP client
@@ -13,35 +13,29 @@ if the first reply has not been received.
"""
import sys
import time
-from comIF.obsw_com_interface import CommunicationInterface, ComIfT
-from utility.obsw_tmtc_printer import TmTcPrinterT, TmTcPrinter
-from sendreceive.obsw_tm_listener import TmListenerT, TmListener
+
+import config.OBSW_Config as g
+from comIF.obsw_com_interface import CommunicationInterface
+from utility.obsw_tmtc_printer import TmTcPrinter
+from sendreceive.obsw_tm_listener import TmListener
from tc.obsw_pus_tc_base import TcQueueEntryT
-# pylint: disable=too-few-public-methods
-# pylint: disable=too-many-arguments
# pylint: disable=too-many-instance-attributes
-# TODO: Maybe replace timeout, tc send timeout and print boolean by just using global values
-# in config file or passing them to the respective public member functions.
class CommandSenderReceiver:
"""
This is the generic CommandSenderReceiver object. All TMTC objects inherit this object,
for example specific implementations (e.g. SingleCommandSenderReceiver)
"""
- def __init__(self, comInterface: ComIfT, tmtcPrinter: TmTcPrinterT, tmListener: TmListenerT,
- tmTimeout: float, tcSendTimeoutFactor: float):
+ def __init__(self, comInterface: CommunicationInterface, tmtcPrinter: TmTcPrinter,
+ tmListener: TmListener):
"""
:param comInterface: CommunicationInterface object. Instantiate the desired one
and pass it here
:param tmtcPrinter: TmTcPrinter object. Instantiate it and pass it here.
- :param tmTimeout: TM Timeout. After each sent telecommand,
- listen for TMs for this time period
- :param tcSendTimeoutFactor: If TM is not received,
- resend TC after G_TM_TIMEOUT * G_TC_SEND_TIMEOUT_FACTOR
"""
- self.tm_timeout = tmTimeout
- self.tc_send_timeout_factor = tcSendTimeoutFactor
+ self._tm_timeout = g.G_TM_TIMEOUT
+ self._tc_send_timeout_factor = g.G_TC_SEND_TIMEOUT_FACTOR
if isinstance(comInterface, CommunicationInterface):
self._com_interface = comInterface
@@ -67,6 +61,24 @@ class CommandSenderReceiver:
self._last_tc = bytearray()
self._last_tc_info = dict()
+ def set_tm_timeout(self, tm_timeout: float = g.G_TM_TIMEOUT):
+ """
+ Set the TM timeout. Usually, the global value set by the args parser is set,
+ but the TM timeout can be reset (e.g. for slower architectures)
+ :param tm_timeout: New TM timeout value as a float value in seconds
+ :return:
+ """
+ self._tm_timeout = tm_timeout
+
+ def set_tc_send_timeout_factor(self, new_factor: float = g.G_TC_SEND_TIMEOUT_FACTOR):
+ """
+ Set the TC resend timeout factor. After self._tm_timeout * new_factor seconds,
+ a telecommand will be resent again.
+ :param new_factor: Factor as a float number
+ :return:
+ """
+ self._tc_send_timeout_factor = new_factor
+
def _check_for_first_reply(self) -> None:
"""
Checks for replies. If no reply is received, send telecommand again in checkForTimeout()
@@ -90,7 +102,7 @@ class CommandSenderReceiver:
queue_entry_is_telecommand = False
if queue_entry_first == "wait":
wait_time = queue_entry_second
- self.tm_timeout = self.tm_timeout + wait_time
+ self._tm_timeout = self._tm_timeout + wait_time
time.sleep(wait_time)
elif queue_entry_first == "print":
print_string = queue_entry_second
@@ -99,7 +111,7 @@ class CommandSenderReceiver:
export_name = queue_entry_second
self._tmtc_printer.print_to_file(export_name, True)
elif queue_entry_first == "timeout":
- self.tm_timeout = queue_entry_second
+ self._tm_timeout = queue_entry_second
else:
self._last_tc, self._last_tc_info = (queue_entry_first, queue_entry_second)
return True
@@ -116,7 +128,7 @@ class CommandSenderReceiver:
sys.exit()
if self._start_time != 0:
self._elapsed_time = time.time() - self._start_time
- if self._elapsed_time > self.tm_timeout * self.tc_send_timeout_factor:
+ if self._elapsed_time > self._tm_timeout * self._tc_send_timeout_factor:
print("Command Sender Receiver: Timeout, sending TC again !")
self._com_interface.send_telecommand(self._last_tc, self._last_tc_info)
self._timeout_counter = self._timeout_counter + 1
diff --git a/sendreceive/obsw_multiple_commands_sender_receiver.py b/sendreceive/obsw_multiple_commands_sender_receiver.py
index 55bfafd..77b38a3 100644
--- a/sendreceive/obsw_multiple_commands_sender_receiver.py
+++ b/sendreceive/obsw_multiple_commands_sender_receiver.py
@@ -11,8 +11,8 @@ from typing import Union, Deque
from collections import deque
from sendreceive.obsw_sequential_sender_receiver import SequentialCommandSenderReceiver
-from comIF.obsw_com_interface import ComIfT
-from utility.obsw_tmtc_printer import TmTcPrinterT
+from comIF.obsw_com_interface import CommunicationInterface
+from utility.obsw_tmtc_printer import TmTcPrinter
from sendreceive.obsw_tm_listener import TmListenerT
import config.OBSW_Config as g
@@ -23,25 +23,23 @@ class MultipleCommandSenderReceiver(SequentialCommandSenderReceiver):
Wait intervals can be specified with wait time between the send bursts.
This is generally done in the separate test classes in UnitTest
"""
- def __init__(self, comInterface: ComIfT, tmtcPrinter: TmTcPrinterT, tcQueue: Deque,
- tmListener: TmListenerT, tmTimeout: float, waitIntervals: list,
- waitTime: Union[float, list], printTm: bool, tcTimeoutFactor: float):
+ def __init__(self, comInterface: CommunicationInterface, tmtcPrinter: TmTcPrinter,
+ tcQueue: Deque, tmListener: TmListenerT, waitIntervals: list,
+ waitTime: Union[float, list], printTm: bool):
"""
TCs are sent in burst when applicable. Wait intervals can be specified by supplying
respective arguments
:param comInterface:
:param tmtcPrinter:
:param tcQueue:
- :param tmTimeout:
- :param waitIntervals: List of pause intervals. For example [1,3] means that a waitTime
+ :param waitIntervals: List of pause intervals. For example [1,3] means that a wait_time
is applied after
sendinf the first and the third telecommand
:param waitTime: List of wait times or uniform wait time as float
:param printTm:
- :param tcTimeoutFactor:
"""
- super().__init__(comInterface=comInterface, tmtcPrinter=tmtcPrinter, tmListener=tmListener,
- tmTimeout=tmTimeout, tcQueue=tcQueue, tcTimeoutFactor=tcTimeoutFactor)
+ super().__init__(com_interface=comInterface, tmtc_printer=tmtcPrinter,
+ tm_listener=tmListener, tc_queue=tcQueue)
self.waitIntervals = waitIntervals
self.waitTime = waitTime
self.printTm = printTm
@@ -56,9 +54,9 @@ class MultipleCommandSenderReceiver(SequentialCommandSenderReceiver):
self._tm_listener.modeId = g.ModeList.UnitTest
self._tm_listener.modeChangeEvent.set()
# TC info queue is set in this function
- self.sendAllQueue()
- self.handleLastRepliesListening(self.tm_timeout / 1.5)
- self.tmInfoQueue = self.retrieveListenerTmInfoQueue()
+ self.__send_all_queue()
+ self.wait_for_last_replies_listening(self._tm_timeout / 1.5)
+ self.tmInfoQueue = self.__retrieve_listener_tm_info_queue()
self._tm_listener.modeOpFinished.set()
if g.G_PRINT_TO_FILE:
self._tmtc_printer.print_to_file()
@@ -75,19 +73,19 @@ class MultipleCommandSenderReceiver(SequentialCommandSenderReceiver):
self._start_time = time.time()
self._check_for_timeout()
- def sendAllQueue(self):
+ def __send_all_queue(self):
while not self._tc_queue.__len__() == 0:
- self.sendAndPrintTc()
+ self.__send_and_print_tc()
- def sendAndPrintTc(self):
+ def __send_and_print_tc(self):
tc_queue_tuple = self._tc_queue.pop()
if self.check_queue_entry(tc_queue_tuple):
pus_packet, pus_packet_info = tc_queue_tuple
self.tcInfoQueue.append(pus_packet_info)
self._com_interface.send_telecommand(pus_packet, pus_packet_info)
- self.handleWaiting()
+ self.__handle_waiting()
- def handleWaiting(self):
+ def __handle_waiting(self):
self.waitCounter = self.waitCounter + 1
if self.waitCounter in self.waitIntervals:
if isinstance(self.waitTime, list):
@@ -95,18 +93,18 @@ class MultipleCommandSenderReceiver(SequentialCommandSenderReceiver):
else:
time.sleep(self.waitTime)
- def retrieveListenerTmInfoQueue(self):
+ def __retrieve_listener_tm_info_queue(self):
if self._tm_listener.replyEvent.is_set():
return self._tm_listener.tmInfoQueue
else:
print("Multiple Command SenderReceiver: Configuration error,"
" reply event not set in TM listener")
- def clearListenerTmInfoQueue(self):
+ def __clear_listener_tm_info_queue(self):
self._tm_listener.tmInfoQueue.clear()
@staticmethod
- def handleLastRepliesListening(waitTime: float):
+ def wait_for_last_replies_listening(waitTime: float):
elapsed_time_seconds = 0
start_time = time.time()
while elapsed_time_seconds < waitTime:
diff --git a/sendreceive/obsw_sequential_sender_receiver.py b/sendreceive/obsw_sequential_sender_receiver.py
index 9f9afcc..b486490 100644
--- a/sendreceive/obsw_sequential_sender_receiver.py
+++ b/sendreceive/obsw_sequential_sender_receiver.py
@@ -10,7 +10,7 @@ import time
import config.OBSW_Config as g
from sendreceive.obsw_command_sender_receiver import CommandSenderReceiver
from sendreceive.obsw_tm_listener import TmListenerT
-from comIF.obsw_com_interface import ComIfT
+from comIF.obsw_com_interface import CommunicationInterface
from utility.obsw_tmtc_printer import TmTcPrinterT
from tc.obsw_pus_tc_base import TcQueueT
@@ -19,21 +19,18 @@ class SequentialCommandSenderReceiver(CommandSenderReceiver):
"""
Specific implementation of CommandSenderReceiver to send multiple telecommands in sequence
"""
- def __init__(self, comInterface: ComIfT, tmtcPrinter: TmTcPrinterT, tmListener: TmListenerT,
- tmTimeout: float, tcQueue: TcQueueT, tcTimeoutFactor: float):
+ def __init__(self, com_interface: CommunicationInterface, tmtc_printer: TmTcPrinterT,
+ tm_listener: TmListenerT, tc_queue: TcQueueT):
"""
- :param comInterface: CommunicationInterface object, passed on to CommandSenderReceiver
- :param tmListener: TmListener object which runs in the background and receives all Telemetry
- :param tmtcPrinter: TmTcPrinter object, passed on to CommandSenderReceiver
- :param tmTimeout: TM Timeout. After each sent telecommand, listen for TMs
+ :param com_interface: CommunicationInterface object, passed on to CommandSenderReceiver
+ :param tm_listener: TmListener object which runs in the background and receives
+ all Telemetry
+ :param tmtc_printer: TmTcPrinter object, passed on to CommandSenderReceiver
for this time period
- :param tcTimeoutFactor: If TM is not received, resend TC after
- G_TM_TIMEOUT * G_TC_SEND_TIMEOUT_FACTOR
"""
- super().__init__(
- comInterface=comInterface, tmtcPrinter=tmtcPrinter, tmListener=tmListener,
- tmTimeout=tmTimeout, tcSendTimeoutFactor=tcTimeoutFactor)
- self._tc_queue = tcQueue
+ super().__init__(comInterface=com_interface, tmtcPrinter=tmtc_printer,
+ tmListener=tm_listener)
+ self._tc_queue = tc_queue
self.__first_reply_received = False
self.__all_replies_received = False
self.__mode_op_finished = False
diff --git a/sendreceive/obsw_single_command_sender_receiver.py b/sendreceive/obsw_single_command_sender_receiver.py
index 9918ebf..c80116e 100644
--- a/sendreceive/obsw_single_command_sender_receiver.py
+++ b/sendreceive/obsw_single_command_sender_receiver.py
@@ -22,18 +22,13 @@ class SingleCommandSenderReceiver(CommandSenderReceiver):
This object can be used by instantiating it and calling sendSingleTcAndReceiveTm()
"""
def __init__(self, comInterface: ComIfT, tmtcPrinter: TmTcPrinterT, tmListener: TmListenerT,
- pusPacketTuple: tuple, tmTimeout: float, tcTimeoutFactor: float):
+ pusPacketTuple: tuple):
"""
:param comInterface: CommunicationInterface object, passed on to CommandSenderReceiver
:param tmListener: TmListener object which runs in the background and receives all TM
:param tmtcPrinter: TmTcPrinter object, passed on to CommandSenderReceiver
- :param tmTimeout: TM Timeout. After each sent telecommand,
- listen for TMs for this time period
- :param tcTimeoutFactor: If TM is not received,
- resend TC after G_TM_TIMEOUT * G_TC_SEND_TIMEOUT_FACTOR
"""
- super().__init__(comInterface=comInterface, tmListener=tmListener, tmtcPrinter=tmtcPrinter,
- tmTimeout=tmTimeout, tcSendTimeoutFactor=tcTimeoutFactor)
+ super().__init__(comInterface=comInterface, tmListener=tmListener, tmtcPrinter=tmtcPrinter)
self.faulty_input = False
try:
self.pus_packet, self.pus_packet_info = pusPacketTuple
diff --git a/tc/OBSW_TcService2.py b/tc/OBSW_TcService2.py
index 305a757..cdc1826 100644
--- a/tc/OBSW_TcService2.py
+++ b/tc/OBSW_TcService2.py
@@ -1,6 +1,6 @@
# -*- coding: utf-8 -*-
"""
-Program: OBSW_UnitTest.py
+Program: obsw_module_test.py
Date: 01.11.2019
Description: PUS Custom Service 8: Device Access, Native low-level commanding
diff --git a/tc/OBSW_TcService3.py b/tc/OBSW_TcService3.py
index d42e71a..1c9de0a 100644
--- a/tc/OBSW_TcService3.py
+++ b/tc/OBSW_TcService3.py
@@ -1,6 +1,6 @@
# -*- coding: utf-8 -*-
"""
-Program: OBSW_UnitTest.py
+Program: obsw_module_test.py
Date: 01.11.2019
Description: PUS Custom Service 8: Device Access, Native low-level commanding
diff --git a/tc/OBSW_TcService8.py b/tc/OBSW_TcService8.py
index ea008b4..98e8d2c 100644
--- a/tc/OBSW_TcService8.py
+++ b/tc/OBSW_TcService8.py
@@ -1,6 +1,6 @@
# -*- coding: utf-8 -*-
"""
-Program: OBSW_UnitTest.py
+Program: obsw_module_test.py
Date: 01.11.2019
Description: PUS Custom Service 8: Function Management, High-Level Commanding
diff --git a/tc/obsw_pus_tc_packer.py b/tc/obsw_pus_tc_packer.py
index 9675f14..5ec05f6 100644
--- a/tc/obsw_pus_tc_packer.py
+++ b/tc/obsw_pus_tc_packer.py
@@ -1,6 +1,6 @@
# -*- coding: utf-8 -*-
"""
-Program: OBSW_UnitTest.py
+Program: obsw_module_test.py
Date: 01.11.2019
Description: Packs the TC queue for specific G_SERVICE or device testing
@@ -213,5 +213,5 @@ def createTotalTcQueue() -> TcQueueT:
tcQueue = packService200TestInto(tcQueue)
tcQueue = packDummyDeviceTestInto(tcQueue)
# objectId = bytearray([0x44, 0x00, 0x1F, 0x00])
- # tcQueue = packGpsTestInto(objectId, tcQueue)
+ # tc_queue = packGpsTestInto(objectId, tc_queue)
return tcQueue
diff --git a/test/OBSW_UnitTest.py b/test/obsw_module_test.py
similarity index 68%
rename from test/OBSW_UnitTest.py
rename to test/obsw_module_test.py
index 08904c0..3223607 100644
--- a/test/OBSW_UnitTest.py
+++ b/test/obsw_module_test.py
@@ -1,10 +1,11 @@
"""
-Program: OBSW_UnitTest.py
+Program: obsw_module_test.py
Date: 01.11.2019
-Description: Unit/High-Level test of on-board software, used by the TMTC client in software test mode
-It is very difficult to execute Unit Tests on embedded hardware. The most significant functonalities for satellites
-are commanding and TM reception. As such, these functionalities will be tested by sending TCs and checking
-the expected Telemetry.
+Description: Module/High-Level test of on-board software, used by the TMTC client in
+software test mode. It is very difficult to execute Unit Tests on embedded hardware.
+The most significant functonalities for satellites are commanding and TM reception.
+As such, these functionalities will be tested by sending TCs and checking the expected Telemetry.
+Still, we make use of the Unit Testing Framework provided by Python (with some hacks involved).
Manual:
Set up the TMTC client as specified in the header comment and use the unit testing mode
@@ -19,11 +20,13 @@ To check the dictionary keys of the telecommand and telemetry information queue
go look up the packTmInformation() and packTcInformation() implementations in the TM and TC folder.
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)
-is expected twice and TM[5,1] (test event) is expected once. Note that the TC verification counting is
-done by the template class by comparing with TC source sequence count. For PUS standalone services (PUS Service Base),
-only the start and completion success need to be asserted. For PUS gateway services (PUS Commanding Service Base),
-step verification needs to be asserted additionally. If there are commands with multiple steps, this needs
+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) is expected twice and TM[5,1] (test event) is expected once.
+Note that the TC verification counting is done by the template class by comparing with
+TC source sequence count. For PUS standalone services (PUS Service Base),
+only the start and completion success need to be asserted. For PUS gateway services
+(PUS Commanding Service Base), step verification needs to be asserted additionally.
+If there are commands with multiple steps, this needs
to be specified in the analyseTcInfo method of the child test.
@author: R. Mueller
@@ -45,7 +48,7 @@ TmInfoQueueService1T = Deque[pusPacketInfoService1T]
class TestService(unittest.TestCase):
"""
- Generic G_SERVICE test class.
+ Generic service test class.
See documentation: https://docs.python.org/3/library/unittest.html#unittest.TestCase.setUpClass
"""
@classmethod
@@ -58,38 +61,42 @@ class TestService(unittest.TestCase):
cls._displayMode = "long"
# wait intervals between tc send bursts.
# Example: [2,4] sends to send 2 tc from queue and wait, then sends another 2 and wait again
- cls.waitIntervals = []
- cls.waitTime = 2
- cls.printTm = True
- cls.printTc = True
+ cls.wait_intervals = []
+ cls.wait_time = 5.0
+ cls.print_tc = True
# default wait time between tc send bursts
- cls.waitTime = 5.0
- cls.testQueue = deque()
+
+ cls.test_queue = deque()
# Extremely ugly solution so that we can use the Python Unit Test Framework
- # default timeout for receiving TM, set in subclass manually
- cls.tmTimeout = g.G_TM_TIMEOUT
- cls.tcTimeoutFactor = g.G_TC_SEND_TIMEOUT_FACTOR
- cls.printFile = g.G_PRINT_TO_FILE
- cls.tmtcPrinter = g.tmtcPrinter
- cls.tmListener = g.tmListener
- cls.communicationInterface = g.comInterface
+ cls.tm_timeout = g.G_TM_TIMEOUT
+ cls.tc_timeout_factor = g.G_TC_SEND_TIMEOUT_FACTOR
+ cls.print_file = g.G_PRINT_TO_FILE
+ cls.tmtc_printer = g.G_TMTC_PRINTER
+ cls.tm_listener = g.G_TM_LISTENER
+ cls.communication_interface = g.G_COM_INTERFACE
# connectToBoard()
- # This function should be called in each individual test to send the actual telecommands
- # which are stored inside testQueue
- def performTestingAndGenerateAssertionDict(self):
- # TODO: Maybe we should instantiate this once in the main and then reuse it instead
- # of calling the constructor over and over again.
- UnitTester = MultipleCommandSenderReceiver(
- comInterface=self.communicationInterface, tmtcPrinter=self.tmtcPrinter,
- tmListener=self.tmListener, tcQueue=self.testQueue, tmTimeout=self.tmTimeout,
- waitIntervals=self.waitIntervals, waitTime=self.waitTime, printTm=self.printTm,
- tcTimeoutFactor=self.tcTimeoutFactor)
- (tcInfoQueue, tmInfoQueue) = UnitTester.send_tc_queue_and_return_info()
- assertionDict = self.analyseTmTcInfo(tmInfoQueue, tcInfoQueue)
- return assertionDict
-
- def analyseTmTcInfo(self, tmInfoQueue: PusTmInfoQueueT, tcInfoQueue: PusTcInfoQueueT) -> dict:
+ def perform_testing_and_generate_assertion_dict(self):
+ """
+ This function should be called in each individual test to send the actual telecommands
+ which are stored inside test_queue
+ TODO: Maybe we should instantiate this once in the main and then reuse it instead
+ of calling the constructor over and over again.
+ If done so we need a setter for: wait_time, wait_intervals, printTm,
+ tc_timeout_factor and the tc.queue. Furthermore, changing parameters should
+ only be allowed as long as the commander/receiver is not running by checking a flag
+ :return:
+ """
+ module_tester = MultipleCommandSenderReceiver(
+ comInterface=self.communication_interface, tmtcPrinter=self.tmtc_printer,
+ tmListener=self.tm_listener, tcQueue=self.test_queue, waitIntervals=self.wait_intervals,
+ waitTime=self.wait_time, printTm=g.G_PRINT_RAW_TM)
+ (tc_info_queue, tm_info_queue) = module_tester.send_tc_queue_and_return_info()
+ assertion_dict = self._analyse_tm_tc_info(tm_info_queue, tc_info_queue)
+ return assertion_dict
+
+ def _analyse_tm_tc_info(
+ self, tm_info_queue: PusTmInfoQueueT, tcInfoQueue: PusTcInfoQueueT) -> dict:
self.tcSscArray = []
self.tcServiceArray = []
self.tcSubserviceArray = []
@@ -114,7 +121,7 @@ class TestService(unittest.TestCase):
assertionDict = {}
# child test implements this and can add additional dict entries
- self.analyseTmInfo(tmInfoQueue, assertionDict)
+ self.analyseTmInfo(tm_info_queue, assertionDict)
assertionDict.update({
"TcStartCount": self.tcVerifiedStart,
@@ -155,7 +162,7 @@ class TestService(unittest.TestCase):
elif currentSubservice == 7:
self.tcVerifiedCompletion = self.tcVerifiedCompletion + 1
- def performGenericAssertionTest(self, assertionDict: dict):
+ def _perform_generic_assertion_test(self, assertionDict: dict):
if assertionDict is None:
print("Performing Generic Assertion Test: "
"Configuratrion error, asseretion dictionary was not passed properly")
@@ -168,10 +175,10 @@ class TestService(unittest.TestCase):
# these tests are identical
def performService5or17Test(self):
- assertionDict = self.performTestingAndGenerateAssertionDict()
+ assertionDict = self.perform_testing_and_generate_assertion_dict()
self.eventExpected = 1
self.miscExpected = 2
- self.performGenericAssertionTest(assertionDict)
+ self._perform_generic_assertion_test(assertionDict)
def analyseService5or17TM(self, tmInfoQueue: PusTmInfoQueueT, assertionDict: dict):
self.miscCounter = 0
@@ -195,7 +202,7 @@ class TestService(unittest.TestCase):
cls.eventCounter = 0
cls.tcVerifyCounter = 0
# noinspection PyUnresolvedReferences
- cls.testQueue.clear()
+ cls.test_queue.clear()
class TestDummyDevice(TestService):
@@ -203,7 +210,7 @@ class TestDummyDevice(TestService):
def setUpClass(cls):
super().setUpClass()
print("Testing Dummy Device")
- packDummyDeviceTestInto(super().testQueue)
+ packDummyDeviceTestInto(super().test_queue)
def analyseTmInfo(self, tmInfoQueue, assertionDict):
pass
@@ -211,17 +218,3 @@ class TestDummyDevice(TestService):
if __name__ == '__main__':
unittest.main()
-
-
-
-
-
-
-
-
-
-
-
-
-
-
diff --git a/test/OBSW_PusServiceTest.py b/test/obsw_pus_service_test.py
similarity index 77%
rename from test/OBSW_PusServiceTest.py
rename to test/obsw_pus_service_test.py
index f173e2e..310d547 100644
--- a/test/OBSW_PusServiceTest.py
+++ b/test/obsw_pus_service_test.py
@@ -1,5 +1,5 @@
import unittest
-from test.OBSW_UnitTest import TestService, PusTmInfoQueueT
+from test.obsw_module_test import TestService, PusTmInfoQueueT
from tc.obsw_pus_tc_base import PusTcInfoQueueT
from tc.obsw_pus_tc_packer import packService17TestInto, packService5TestInto, packService2TestInto
import config.OBSW_Config as g
@@ -7,19 +7,19 @@ import config.OBSW_Config as g
class TestService2(TestService):
@classmethod
- def setUpClass(cls):
+ def setUpClass(cls: TestService):
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, 3]
- packService2TestInto(cls.testQueue)
+ cls.wait_intervals = [1, 2, 3, 4]
+ cls.wait_time = [2, 2, 2, 3]
+ packService2TestInto(cls.test_queue)
def test_Service2(self):
- assertionDict = self.performTestingAndGenerateAssertionDict()
+ assertionDict = self.perform_testing_and_generate_assertion_dict()
self.eventExpected = 1
self.miscExpected = 4
- super().performGenericAssertionTest(assertionDict)
+ super()._perform_generic_assertion_test(assertionDict)
def analyseTmInfo(self, tmInfoQueue: PusTmInfoQueueT, assertionDict: dict):
while not tmInfoQueue.__len__() == 0:
@@ -47,14 +47,14 @@ class TestService2(TestService):
class TestService5(TestService):
@classmethod
- def setUpClass(cls):
+ def setUpClass(cls: TestService):
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)
+ cls.wait_intervals = [1, 2, 3]
+ cls.wait_time = [1.5, 2.0, 2.0]
+ packService5TestInto(cls.test_queue)
def test_Service5(self):
# analyseTmInfo() and analyseTcInfo are called here
@@ -91,19 +91,19 @@ class TestService9(unittest.TestCase):
class TestService17(TestService):
@classmethod
- def setUpClass(cls):
+ def setUpClass(cls: TestService):
super().setUpClass()
print("Testing Service 17")
- cls.waitIntervals = [2]
- cls.waitTime = 2
- cls.tmTimeout = g.G_TM_TIMEOUT
- packService17TestInto(cls.testQueue)
+ cls.wait_intervals = [2]
+ cls.wait_time = 2
+ cls.tm_timeout = g.G_TM_TIMEOUT
+ packService17TestInto(cls.test_queue)
def test_Service17(self):
super().performService5or17Test()
- def analyseTmTcInfo(self, tmInfoQueue: PusTmInfoQueueT, tcInfoQueue: PusTcInfoQueueT):
- assertionDict = super().analyseTmTcInfo(tmInfoQueue=tmInfoQueue, tcInfoQueue=tcInfoQueue)
+ def _analyse_tm_tc_info(self, tmInfoQueue: PusTmInfoQueueT, tcInfoQueue: PusTcInfoQueueT):
+ assertionDict = super()._analyse_tm_tc_info(tm_info_queue=tmInfoQueue, tcInfoQueue=tcInfoQueue)
# add anything elsee other than tc verification counter
# and ssc that is needed for tm analysis
return assertionDict
diff --git a/utility/obsw_args_parser.py b/utility/obsw_args_parser.py
index cd80b89..71ab6a6 100644
--- a/utility/obsw_args_parser.py
+++ b/utility/obsw_args_parser.py
@@ -34,7 +34,7 @@ def parse_input_arguments():
'-t', '--tm_timeout', type=float, help='TM Timeout when listening to verification sequence.'
' Default: 12, 6(Serial)', default=6.0)
arg_parser.add_argument(
- '--np', dest='printFile', help='Supply --np to suppress print output to log files.',
+ '--np', dest='print_file', help='Supply --np to suppress print output to log files.',
action='store_false')
arg_parser.add_argument(
'-o', '--tc_timeout_factor', type=float, help='TC Timeout Factor. Multiplied with '
@@ -91,7 +91,7 @@ def handle_unspecified_args(args) -> None:
print("2: Single Command Mode with manual command")
print("3: Service Mode, Commands specified in tc folder")
print("4: Software Mode, runs all command specified in obsw_pus_tc_packer.py")
- print("5: Unit Test, runs unit test specified in OBSW_UnitTest.py")
+ print("5: Unit Test, runs unit test specified in obsw_module_test.py")
args.mode = input("Please enter Mode: ")
if args.mode == 1 and args.service is None:
args.service = input("No Service specified for Service Mode. "
diff --git a/utility/obsw_exit_handler.py b/utility/obsw_exit_handler.py
index 118ba24..38232e9 100644
--- a/utility/obsw_exit_handler.py
+++ b/utility/obsw_exit_handler.py
@@ -1,5 +1,5 @@
import signal
-from comIF.obsw_com_interface import ComIfT
+from comIF.obsw_com_interface import CommunicationInterface
class GracefulKiller:
@@ -14,9 +14,9 @@ class GracefulKiller:
print("I was killed")
-def keyboard_interrupt_handler(comInterface: ComIfT):
+def keyboard_interrupt_handler(comInterface: CommunicationInterface):
print("Disconnect registered")
# Unit Test closes Serial Port at the end
# We could do some optional stuff here
if comInterface is not None:
- comInterface.sendTelecommand(bytearray([0, 0, 0, 0, 0]))
+ comInterface.send_telecommand(bytearray([0, 0, 0, 0, 0]))
--
GitLab