diff --git a/Ref/PrmDb.dat b/Ref/PrmDb.dat deleted file mode 100644 index aa1b20d9f02..00000000000 Binary files a/Ref/PrmDb.dat and /dev/null differ diff --git a/Ref/config.json b/Ref/test/int/int_config.json similarity index 67% rename from Ref/config.json rename to Ref/test/int/int_config.json index 731811a9678..30ad6c737aa 100644 --- a/Ref/config.json +++ b/Ref/test/int/int_config.json @@ -1,21 +1,21 @@ { - "Drv.BlockDriver" : "Ref.blockDrv", "Svc.ActiveRateGroup" : "Ref.rateGroup1Comp", - "Svc.CommandDispatcher" : "Ref.cmdDisp", + "Svc.CommandDispatcher" : "CdhCore.cmdDisp", "Svc.CmdSequencer" : "Ref.cmdSeq", - "Svc.FileDownlink" : "Ref.fileDownlink", - "Svc.FileManager" : "Ref.fileManager", + "Svc.FileDownlink" : "FileHandling.fileDownlink", + "Svc.FileManager" : "FileHandling.fileManager", "Svc.FileUplink" : "Ref.fileUplink", "Ref.PingReceiver" : "Ref.pingRcvr", - "Svc.ActiveLogger" : "Ref.eventLogger", + "Svc.EventManager" : "CdhCore.events", "Svc.TlmChan" : "Ref.tlmSend", - "Svc.PrmDb" : "Ref.prmDb", - "Svc.DpCatalog" : "Ref.dpCat", - "Svc.DpManager" : "Ref.dpMgr", - "Svc.DpWriter" : "Ref.dpWriter", + "Svc.PrmDb" : "FileHandling.prmDb", + "Svc.PrmDb.filename" : "/tmp/PrmDb.dat", + "Svc.DpCatalog" : "DataProducts.dpCat", + "Svc.DpManager" : "DataProducts.dpMgr", + "Svc.DpWriter" : "DataProducts.dpWriter", "Svc.ComQueue" : "Ref.comQueue", "Ref.TypeDemo" : "Ref.typeDemo", - "Svc.Health" : "Ref.health", + "Svc.Health" : "CdhCore.health", "Ref.SignalGen" : "Ref.SG1", "Ref.SendBuff" : "Ref.sendBuffComp", "Drv.TcpClient" : "Ref.comDriver", @@ -25,7 +25,7 @@ "Svc.PosixTime" : "Ref.posixTime", "Svc.RateGroupDriver" : "Ref.rateGroupDriverComp", "Ref.RecvBuff" : "Ref.recvBuffComp", - "Svc.Version" : "Ref.version", + "Svc.Version" : "CdhCore.version", "Svc.PassiveTextLogger" : "Ref.textLogger", "Svc.SystemResources" : "Ref.systemResources", "Svc.BufferManager" : "Ref.dpBufferManager", @@ -35,3 +35,4 @@ "Svc.FprimeFramer" : "Ref.fprimeFramer", "Svc.ComStub" : "Ref.comStub" } + diff --git a/Ref/test/int/logs/.gitignore b/Ref/test/int/logs/.gitignore deleted file mode 100644 index 33031faf94e..00000000000 --- a/Ref/test/int/logs/.gitignore +++ /dev/null @@ -1,5 +0,0 @@ -# output files from pytest and test API. -*.xlsx -*.xlsx# -assets -*.html diff --git a/Ref/test/int/test_cmd_parameter.py b/Ref/test/int/test_cmd_parameter.py new file mode 100644 index 00000000000..2e828596166 --- /dev/null +++ b/Ref/test/int/test_cmd_parameter.py @@ -0,0 +1,118 @@ +"""test_parameter.py: + +Test the command dispatcher with basic integration tests. +""" + + +def test_send_parameter(fprime_test_api): + """Test that commands may be sent + + Tests command send, dispatch, and receipt using send_and_assert command with a pair of CmdDispatcher commands. + + recvBuffComp.PARAMETER1_PRM_SET (call telemetry confirm value) + recvBuffComp.PARAMETER1_PRM_SAVE + recvBuffComp.PARAMETER2_PRM_SET (call telemetry confirm value) + recvBuffComp.PARAMETER2_PRM_SAVE + + sendBuffComp.PARAMETER3_PRM_SET (call telemetry confirm value) + sendBuffComp.PARAMETER3_PRM_SAVE + sendBuffComp.PARAMETER4_PRM_SET (call telemetry confirm value) + sendBuffComp.PARAMETER4_PRM_SAVE + + Notes: send a parameterX_set,value=current telemetry channel. + the function assert_telemetry will fail because no_change + """ + + ## setup default-value + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Ref.RecvBuff") + "." + "PARAMETER1_PRM_SET", + [1], + max_delay=5, + ) + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Ref.RecvBuff") + "." + "PARAMETER2_PRM_SET", + [2], + max_delay=5, + ) + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Ref.SendBuff") + "." + "PARAMETER3_PRM_SET", + [3], + max_delay=5, + ) + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Ref.SendBuff") + "." + "PARAMETER4_PRM_SET", + [4], + max_delay=5, + ) + + # Only work if send command PARAMETER1_PRM_SET then check telemetry. Unsigned integer 0..4294967295) + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Ref.RecvBuff") + "." + "PARAMETER1_PRM_SET", + [10], + max_delay=5, + ) + + # Check Telem only will not work + param1_change = fprime_test_api.get_telemetry_pred( + fprime_test_api.get_mnemonic("Ref.RecvBuff") + "." + "Parameter1", 10 + ) + fprime_test_api.assert_telemetry(param1_change, timeout=5) + + # Send PARAMETER1_PRM_SAVE + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Ref.RecvBuff") + "." + "PARAMETER1_PRM_SAVE", + max_delay=1, + ) + + # Send PARAMETER2_PRM_SET (confirm new value / SAVE ) signed integer -32867 and 32767 + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Ref.RecvBuff") + "." + "PARAMETER2_PRM_SET", + [20], + max_delay=5, + ) + + param2_change = fprime_test_api.get_telemetry_pred( + fprime_test_api.get_mnemonic("Ref.RecvBuff") + "." + "Parameter2", 20 + ) + fprime_test_api.assert_telemetry(param2_change, timeout=5) + + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Ref.RecvBuff") + "." + "PARAMETER2_PRM_SAVE", + max_delay=5, + ) + + # Send PARAMETER3_PRM_SET (confirm new value / SAVE ) unsigned integer 0..255 + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Ref.SendBuff") + "." + "PARAMETER3_PRM_SET", + [30], + max_delay=5, + ) + + param3_change = fprime_test_api.get_telemetry_pred( + fprime_test_api.get_mnemonic("Ref.SendBuff") + "." + "Parameter3", 30 + ) + fprime_test_api.assert_telemetry(param3_change, timeout=5) + + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Ref.SendBuff") + "." + "PARAMETER3_PRM_SAVE", + max_delay=5, + ) + + # Send PARAMETER4_PRM_SET (confirm new value / SAVE ) float + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Ref.SendBuff") + "." + "PARAMETER4_PRM_SET", + [40], + max_delay=5, + ) + + param4_change = fprime_test_api.get_telemetry_pred( + fprime_test_api.get_mnemonic("Ref.SendBuff") + "." + "Parameter4", 40 + ) + fprime_test_api.assert_telemetry(param4_change, timeout=5) + + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Ref.SendBuff") + "." + "PARAMETER4_PRM_SAVE", + max_delay=5, + ) + + # Notes: send a xxx_set,value=current will not update telemetry channel. call assert_telemetry will fail diff --git a/Svc/CmdDispatcher/test/int/test_cmd_dispatcher.py b/Svc/CmdDispatcher/test/int/test_cmd_dispatcher.py index 035f9fb9b44..7f55a7513f6 100644 --- a/Svc/CmdDispatcher/test/int/test_cmd_dispatcher.py +++ b/Svc/CmdDispatcher/test/int/test_cmd_dispatcher.py @@ -1,4 +1,4 @@ -""" test_cmd_dispatcher.py: +"""test_cmd_dispatcher.py: Test the command dispatcher with basic integration tests. """ @@ -7,6 +7,31 @@ def test_send_command(fprime_test_api): """Test that commands may be sent - Tests command send, dispatch, and receipt using send_and_assert command with a pair of NO-OP commands. + Tests command send, dispatch, and receipt using send_and_assert command with a pair of CmdDispatcher commands. """ - fprime_test_api.send_and_assert_command("cmdDisp.CMD_NO_OP", max_delay=0.1) + + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.CommandDispatcher") + "." + "CMD_NO_OP", + max_delay=1, + ) + + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.CommandDispatcher") + + "." + + "CMD_CLEAR_TRACKING", + max_delay=1, + ) + + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.CommandDispatcher") + + "." + + "CMD_NO_OP_STRING", + ["test_string_2"], + max_delay=1, + ) + + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.CommandDispatcher") + "." + "CMD_TEST_CMD_1", + [2, 3, 4], + max_delay=1, + ) diff --git a/Svc/CmdSequencer/test/int/test_cmd_sequencer.py b/Svc/CmdSequencer/test/int/test_cmd_sequencer.py new file mode 100644 index 00000000000..3172974eb43 --- /dev/null +++ b/Svc/CmdSequencer/test/int/test_cmd_sequencer.py @@ -0,0 +1,336 @@ +"""test_cmd_sequencer.py: + +A set of integration tests to apply to the CmdSequencer app. This is intended to be a reference of integration testing. +""" + +import subprocess +import time +import os +from pathlib import Path + + +def test_generate_file(fprime_test_api): + """generate seq files before run seqgen""" + + # Create test_seq_new.seq file with Deployment name + with open("test_seq_new.seq", "w") as f: + f.write("; A test sequence " + "\n") + f.write(";\n") + f.write( + "R00:00:00" + + " " + + ( + fprime_test_api.get_mnemonic("Svc.CommandDispatcher") + + "." + + "CMD_NO_OP" + ) + + "\n\n" + ) + + f.write("; Let's try out some commands with arguments " + "\n") + f.write( + "R00:00:01.050" + + " " + + ( + fprime_test_api.get_mnemonic("Svc.CommandDispatcher") + + "." + + "CMD_NO_OP_STRING" + ) + + " " + + '"' + + "Awesome String!" + + '"' + + ";" + + "\n" + ) + + # raise error + try: + with open("test_seq_new.seq", "x") as f: + f.write("Created using exclusive mode.") + except FileExistsError: + print("Already exists.") + + # Create test_seq_wait_new.seq file with Deployment name + with open("test_seq_wait_new.seq", "w") as f2: + f2.write("; A test sequence " + "\n") + f2.write(";\n") + f2.write( + "R00:00:00" + + " " + + ( + fprime_test_api.get_mnemonic("Svc.CommandDispatcher") + + "." + + "CMD_NO_OP" + ) + + "\n\n" + ) + + f2.write("; Let's try out some commands with arguments " + "\n") + f2.write( + "R00:00:01.050" + + " " + + ( + fprime_test_api.get_mnemonic("Svc.CommandDispatcher") + + "." + + "CMD_NO_OP_STRING" + ) + + " " + + '"' + + "SEQ WAIT 2 MINS!" + + '"' + + "\n" + ) + f2.write(";\n") + f2.write( + "R00:02:00.050" + + " " + + ( + fprime_test_api.get_mnemonic("Svc.CommandDispatcher") + + "." + + "CMD_NO_OP_STRING" + ) + + " " + + '"' + + "SEQ after 2mins!" + + '"' + + "\n" + ) + f2.write(";\n") + f2.write( + "R00:00:01.050" + + " " + + ( + fprime_test_api.get_mnemonic("Svc.CommandDispatcher") + + "." + + "CMD_NO_OP_STRING" + ) + + " " + + '"' + + "SEQ DONE !" + + '"' + + "\n" + ) + f2.write(";\n") + + # raise error + try: + with open("test_seq_wait_new.seq", "x") as f2: + f2.write("Created using exclusive mode.") + except FileExistsError: + print("Already exists.") + + +def test_seqgen(fprime_test_api): + """Tests the seqgen can be dispatched (requires localhost testing)""" + + deployment = fprime_test_api.get_deployment() + + print("DEPLOYMENT: %s", deployment) + + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.CmdSequencer") + "." + "CS_AUTO", max_delay=10 + ) + + sequence = Path(__file__).parent / "test_seq_new.seq" + sequence2 = Path(__file__).parent / "test_seq_wait_new.seq" + + assert ( + subprocess.run( + [ + "fprime-seqgen", + "-d", + str(fprime_test_api.pipeline.dictionary_path), + str(sequence), + "ref_test_seq.bin", + ] + ).returncode + == 0 + ), "Failed to run fprime-seqgen" + + assert ( + subprocess.run( + [ + "fprime-seqgen", + "-d", + str(fprime_test_api.pipeline.dictionary_path), + str(sequence2), + "ref_test_seq_wait.bin", + ] + ).returncode + == 0 + ), "Failed to run fprime-seqgen" + + # uplink ref_test_seq.bin and ref_test_seq_wait.bin + fprime_test_api.uplink_file_and_await_completion( + "ref_test_seq.bin", "/tmp/ref_test_seq.bin", timeout=100 + ) + fprime_test_api.uplink_file_and_await_completion( + "ref_test_seq_wait.bin", "/tmp/ref_test_seq_wait.bin", timeout=100 + ) + + # execute sequence + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.CmdSequencer") + "." + "CS_RUN", + args=["/tmp/ref_test_seq.bin", "BLOCK"], + max_delay=5, + ) + + ###### ###### remove new file 1 ###### + # print(f" file1:",sequence) + # print(f" file2:",sequence2) + os.remove(sequence) + os.remove(sequence2) + + +def test_send_seq(fprime_test_api): + """ + CS_RUN + CS_VALIDATE + CS_MANUAL + CS_AUTO + CS_START + CS_CANCEL + CS_STEP + CS_JOINT_WAIT (Need to have more 1 run sequence. Ref don't have more one sequence) + """ + + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.CmdSequencer") + "." + "CS_VALIDATE", + ["/tmp/ref_test_seq.bin"], + max_delay=10, + ) + # sequence execute_2 auto + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.CmdSequencer") + "." + "CS_RUN", + ["/tmp/ref_test_seq.bin", "BLOCK"], + max_delay=5, + ) + + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.CommandDispatcher") + + "." + + "CMD_NO_OP_STRING", + ["test_string_execute_2 auto completed"], + max_delay=10, + ) + + ###### ###### ###### + ## Toggle Manual/AUTO + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.CmdSequencer") + "." + "CS_MANUAL", + max_delay=10, + ) + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.CmdSequencer") + "." + "CS_AUTO", max_delay=10 + ) + + ###### ###### ###### + ## Manual Mode testing Start/Cancel + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.CmdSequencer") + "." + "CS_MANUAL", + max_delay=10, + ) + + # Load Sequence but not execute is current SEQ manual (will load sequence only) + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.CmdSequencer") + "." + "CS_RUN", + ["/tmp/ref_test_seq.bin", "NO_BLOCK"], + max_delay=5, + ) + + # sequence execute_5 manually Command 0 in sequence + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.CmdSequencer") + "." + "CS_START", + max_delay=10, + ) + + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.CmdSequencer") + "." + "CS_CANCEL", + max_delay=10, + ) + + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.CommandDispatcher") + + "." + + "CMD_NO_OP_STRING", + ["manually START/CANCEL after cmd 0"], + max_delay=10, + ) + + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.CmdSequencer") + "." + "CS_VALIDATE", + ["/tmp/ref_test_seq.bin"], + max_delay=10, + ) + + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.CmdSequencer") + "." + "CS_AUTO", max_delay=10 + ) + + ###### ###### ###### + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.CmdSequencer") + "." + "CS_MANUAL", + max_delay=10, + ) + ## Manual Mode testing Start WARNING_LO = No sequence active + # WARNING_LO => No sequence active and EXECUTION_ERROR. No completion (will cause pytest to assert when no completion) use send_command will ignore completion? + + # fprime_test_api.send_and_assert_command(fprime_test_api.get_mnemonic('Svc.CmdSequencer') + '.' + 'CS_START', max_delay=10) + fprime_test_api.send_command( + fprime_test_api.get_mnemonic("Svc.CmdSequencer") + "." + "CS_START" + ) + + # Load Sequence but not execute is current SEQ manual (will load sequence only) + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.CmdSequencer") + "." + "CS_RUN", + ["/tmp/ref_test_seq_wait.bin", "NO_BLOCK"], + max_delay=5, + ) + + # sequence manually mode start Command 0 in sequence + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.CmdSequencer") + "." + "CS_START", + max_delay=10, + ) + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.CommandDispatcher") + + "." + + "CMD_NO_OP_STRING", + ["manually START command 0"], + max_delay=10, + ) + + # sequence manually mode step Next Command 1 in sequence (Only manual mode) + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.CmdSequencer") + "." + "CS_STEP", max_delay=3 + ) + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.CommandDispatcher") + + "." + + "CMD_NO_OP_STRING", + ["manually START/STEP cmd 1 "], + max_delay=3, + ) + + time.sleep(10) + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.CmdSequencer") + "." + "CS_STEP", max_delay=10 + ) + time.sleep(130) + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.CmdSequencer") + "." + "CS_STEP", max_delay=3 + ) + + # cleanup sequence files (sequence create local and uplink files) + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.FileManager") + "." + "RemoveFile", + ["/tmp/ref_test_seq.bin", True], + max_delay=15, + ) + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.FileManager") + "." + "RemoveFile", + ["/tmp/ref_test_seq_wait.bin", True], + max_delay=15, + ) + ###### ###### ###### diff --git a/Svc/CmdSequencer/test/int/test_seq.seq b/Svc/CmdSequencer/test/int/test_seq.seq new file mode 100644 index 00000000000..480ef0b32ff --- /dev/null +++ b/Svc/CmdSequencer/test/int/test_seq.seq @@ -0,0 +1,6 @@ +; A test sequence +; +R00:00:00 cmdDisp.CMD_NO_OP + +; Let's try out some commands with arguments +R00:00:01.050 cmdDisp.CMD_NO_OP_STRING "Awesome string!"; \ No newline at end of file diff --git a/Svc/CmdSequencer/test/int/test_seq_wait.seq b/Svc/CmdSequencer/test/int/test_seq_wait.seq new file mode 100644 index 00000000000..c102d9ae7f8 --- /dev/null +++ b/Svc/CmdSequencer/test/int/test_seq_wait.seq @@ -0,0 +1,11 @@ +; A test sequence +; +R00:00:00 cmdDisp.CMD_NO_OP + +; Let's try out some commands with arguments +R00:00:01.050 cmdDisp.CMD_NO_OP_STRING "SEQ WAIT 2 MINS!" +; +R00:02:00.050 cmdDisp.CMD_NO_OP_STRING "SEQ after 2mins!" +; +R00:00:01.050 cmdDisp.CMD_NO_OP_STRING "SEQ DONE !" +; \ No newline at end of file diff --git a/Svc/DpCatalog/test/int/test_cmd_dpCatalog.py b/Svc/DpCatalog/test/int/test_cmd_dpCatalog.py new file mode 100644 index 00000000000..03c2576b8b2 --- /dev/null +++ b/Svc/DpCatalog/test/int/test_cmd_dpCatalog.py @@ -0,0 +1,29 @@ +"""test_cmd_dpWriter.py: + +Test the command dpWriter with basic integration tests. +""" + + +def test_send_dpWriter_command(fprime_test_api): + """Test that commands may be sent + + Tests command send, dispatch, and receipt using send_and_assert command + """ + + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.DpCatalog") + "." + "CLEAR_CATALOG", + max_delay=10, + ) + + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.DpCatalog") + "." + "BUILD_CATALOG", + max_delay=10, + ) + # wait/no_wait option command fatal + # F fprime_test_api.send_and_assert_command(fprime_test_api.get_mnemonic('Svc.DpCatalog') + '.' + 'START_XMIT_CATALOG', ["NO_WAIT"], max_delay=10) + + # warning_lo bc DpCatalog transmit not active + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.DpCatalog") + "." + "STOP_XMIT_CATALOG", + max_delay=10, + ) diff --git a/Svc/DpManager/test/int/test_cmd_dpMgr.py b/Svc/DpManager/test/int/test_cmd_dpMgr.py new file mode 100644 index 00000000000..a69ecb196ef --- /dev/null +++ b/Svc/DpManager/test/int/test_cmd_dpMgr.py @@ -0,0 +1,16 @@ +"""test_cmd_dpMgr.py: + +Test the command dpMgr with basic integration tests. +""" + + +def test_send_dpMgr_command(fprime_test_api): + """Test that commands may be sent + + Tests command send, dispatch, and receipt using send_and_assert command + """ + + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.DpManager") + "." + "CLEAR_EVENT_THROTTLE", + max_delay=1, + ) diff --git a/Svc/DpWriter/test/int/test_cmd_dpWriter.py b/Svc/DpWriter/test/int/test_cmd_dpWriter.py new file mode 100644 index 00000000000..55c0996c7b3 --- /dev/null +++ b/Svc/DpWriter/test/int/test_cmd_dpWriter.py @@ -0,0 +1,16 @@ +"""test_cmd_dpWriter.py: + +Test the command dpWriter with basic integration tests. +""" + + +def test_send_dpWriter_command(fprime_test_api): + """Test that commands may be sent + + Tests command send, dispatch, and receipt using send_and_assert command + """ + + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.DpWriter") + "." + "CLEAR_EVENT_THROTTLE", + max_delay=10, + ) diff --git a/Svc/EventManager/test/int/test_cmd_eventManager.py b/Svc/EventManager/test/int/test_cmd_eventManager.py new file mode 100644 index 00000000000..3a06d1d0ff2 --- /dev/null +++ b/Svc/EventManager/test/int/test_cmd_eventManager.py @@ -0,0 +1,79 @@ +"""test_cmd_EventManager.py: + +Test the command EventManager with basic integration tests. + eventLogger.DUMP_FILTER_STATE + eventLogger.SET_ID_FILTER + eventLogger.SET_EVENT_FILTER +""" + + +def test_send_command(fprime_test_api): + """Test that commands may be sent + + Tests command send_and_assert command events level and state + """ + # Expected True for (WARNING_HI, WARNING_LO, COMMAND, ACTIVITY_HI, ACTIVITY_LO & False for DIAGNOSTIC) + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.EventManager") + "." + "DUMP_FILTER_STATE", + max_delay=1, + ) + + # EVR Display => Event Id: 0x507, Event Severity: ACTIVITY_HI, Event Description: Received a NO-OP command + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.CommandDispatcher") + "." + "CMD_NO_OP", + max_delay=0.1, + ) + + # enable specific evr Id 0x507 or dec 1287 + # EVR Display => Event Id: 0xb01, Event Severity: ACTIVITY_HI, Event Description: ID 1287 is filtered. + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.EventManager") + "." + "SET_ID_FILTER", + ["0x507", "ENABLED"], + max_delay=1, + ) + + # send noop command. + # Confirm SET_ID_FILTER: NO evr specific Event Id 0x507 report due SET_ID_FILTER is enable + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.CommandDispatcher") + "." + "CMD_NO_OP", + max_delay=0.1, + ) + + # dump_filter_state, Expected True for (WARNING_HI, WARNING_LO, COMMAND, ACTIVITY_HI, ACTIVITY_LO) & False for DIAGNOSTIC) + # Confirm DUMP_FILTER_STATE activity_hi work: include set_id filter enable (activity_hi) + # EVR Display => Event Id: 0xb01, Event Severity: ACTIVITY_HI, Event Description: ID 1287 is filtered. + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.EventManager") + "." + "DUMP_FILTER_STATE", + max_delay=1, + ) + + ### Disable all ACTIVITY_HI + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.EventManager") + "." + "SET_EVENT_FILTER", + ["ACTIVITY_HI", "DISABLED"], + max_delay=1, + ) + + # dump_filter_state, Expected True for (WARNING_HI, WARNING_LO, COMMAND, ACTIVITY_LO) & False for DIAGNOSTIC, ACTIVITY_HI) + # Confirm SET_EVENT_FILTER work: Not include any activity_hi, + # even specific Id 0x507 still set_id_filter enable + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.EventManager") + "." + "DUMP_FILTER_STATE", + max_delay=1, + ) + + ## Put back default value + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.EventManager") + "." + "SET_EVENT_FILTER", + ["ACTIVITY_HI", "ENABLED"], + max_delay=1, + ) + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.EventManager") + "." + "SET_ID_FILTER", + ["0x507", "DISABLED"], + max_delay=1, + ) + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.EventManager") + "." + "DUMP_FILTER_STATE", + max_delay=1, + ) diff --git a/Svc/FileDownlink/test/int/test_cmd_fileDownlink.py b/Svc/FileDownlink/test/int/test_cmd_fileDownlink.py new file mode 100644 index 00000000000..3ca2b5a9da4 --- /dev/null +++ b/Svc/FileDownlink/test/int/test_cmd_fileDownlink.py @@ -0,0 +1,34 @@ +"""test_cmd_fileDownlink.py: + +Test the command FileDownlink with basic integration tests. + fileDownlink.SendFile + fileDownlink.SendPartial + fileDownlink.Cancel + + +""" + + +def test_send_fileDownlink_command(fprime_test_api): + """Test that commands may be sent + + Tests command send, dispatch, and receipt using send_and_assert command with a pair of fileDownlink commands. + """ + + ## source = /tmp/test_seq_wait.seq , Dest = /tmp//fprime-downlink/DL.log + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.FileDownlink") + "." + "SendFile", + ["/tmp/test_seq_wait.seq", "DL.log"], + max_delay=30, + ) + + ## source = /tmp/1MiB.txt , Dest = /tmp//fprime-downlink/DL3.log, start offset 0 , for 100 Bytes + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.FileDownlink") + "." + "SendPartial", + ["/tmp/1MiB.txt", "DL3.log", 0, 100], + max_delay=5, + ) + + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.FileDownlink") + "." + "Cancel", max_delay=5 + ) diff --git a/Svc/FileManager/test/int/test_cmd_fileManager.py b/Svc/FileManager/test/int/test_cmd_fileManager.py new file mode 100644 index 00000000000..5a9a822bcd5 --- /dev/null +++ b/Svc/FileManager/test/int/test_cmd_fileManager.py @@ -0,0 +1,86 @@ +"""test_cmd_FileManager.py: + +Test the command FileManager with basic integration tests. + fileManager.CreateDirectory + fileManager.ShellCommand # Don't use ShellCommand + fileManager.FileSize + fileManager.MoveFile + fileManager.AppendFile + fileManager.RemoveFile + fileManager.RemoveDirectory + +""" + +import time + + +def test_send_fileManager_command(fprime_test_api): + """Test that commands may be sent + + Tests command send, dispatch, and receipt using send_and_assert command with a pair of fileManager commands. + """ + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.FileManager") + "." + "CreateDirectory", + ["/tmp/file"], + max_delay=10, + ) + + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.FileManager") + "." + "CreateDirectory", + ["/tmp/file2"], + max_delay=10, + ) + + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.FileManager") + "." + "FileSize", + ["/tmp/1MiB.txt"], + max_delay=50, + ) + + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.FileManager") + "." + "MoveFile", + ["/tmp/1MiB.txt", "/tmp/file/1MiB.txt"], + max_delay=50, + ) + + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.FileManager") + "." + "FileSize", + ["/tmp/file/1MiB.txt"], + max_delay=50, + ) + # Use small file to AppendFile to work with RaspberryPi + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.FileManager") + "." + "AppendFile", + ["/tmp/test_seq.seq", "/tmp/file/test_seq.seq"], + ) + time.sleep(10) + # put back + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.FileManager") + "." + "MoveFile", + ["/tmp/file/1MiB.txt", "/tmp/1MiB.txt"], + max_delay=50, + ) + + # fail bc directory is not empty (pytest script will stop if use send_and_assert_cmd. Use send_command to avoid stop) No max_delay (expected warning_hi) + fprime_test_api.send_command( + fprime_test_api.get_mnemonic("Svc.FileManager") + "." + "RemoveDirectory", + ["/tmp/file"], + ) + + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.FileManager") + "." + "RemoveDirectory", + ["/tmp/file2"], + max_delay=10, + ) + # Cleanup directory + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.FileManager") + "." + "RemoveFile", + ["/tmp/file/test_seq.seq", True], + max_delay=5, + ) + + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.FileManager") + "." + "RemoveDirectory", + ["/tmp/file"], + max_delay=5, + ) diff --git a/Svc/FileUplink/test/int/1MiB.txt b/Svc/FileUplink/test/int/1MiB.txt new file mode 100644 index 00000000000..44985be31b7 Binary files /dev/null and b/Svc/FileUplink/test/int/1MiB.txt differ diff --git a/Svc/FileUplink/test/int/test_cmd_uplink.py b/Svc/FileUplink/test/int/test_cmd_uplink.py new file mode 100644 index 00000000000..aa904835715 --- /dev/null +++ b/Svc/FileUplink/test/int/test_cmd_uplink.py @@ -0,0 +1,23 @@ +"""test_cmd_uplink.py: + +Test command dispatcher with basic integration tests. +""" + + +def test_send_uplink_command(fprime_test_api): + """Test that commands may be sent + + Tests command send, dispatch, and receipt using send_and_assert command with a pair of CmdDispatcher commands. + """ + # file_path = test_seq.seq and destination = /tmp/test_seq.seq (for fileManager) + fprime_test_api.uplink_file_and_await_completion( + "test_seq.seq", "/tmp/test_seq.seq", timeout=100 + ) + # for fileDownlink + fprime_test_api.uplink_file_and_await_completion( + "test_seq_wait.seq", "/tmp/test_seq_wait.seq", timeout=100 + ) + # for health, fileDownlink, fileManager + fprime_test_api.uplink_file_and_await_completion( + "1MiB.txt", "/tmp/1MiB.txt", timeout=100 + ) diff --git a/Svc/FileUplink/test/int/test_seq.seq b/Svc/FileUplink/test/int/test_seq.seq new file mode 100644 index 00000000000..480ef0b32ff --- /dev/null +++ b/Svc/FileUplink/test/int/test_seq.seq @@ -0,0 +1,6 @@ +; A test sequence +; +R00:00:00 cmdDisp.CMD_NO_OP + +; Let's try out some commands with arguments +R00:00:01.050 cmdDisp.CMD_NO_OP_STRING "Awesome string!"; \ No newline at end of file diff --git a/Svc/FileUplink/test/int/test_seq_wait.seq b/Svc/FileUplink/test/int/test_seq_wait.seq new file mode 100644 index 00000000000..c102d9ae7f8 --- /dev/null +++ b/Svc/FileUplink/test/int/test_seq_wait.seq @@ -0,0 +1,11 @@ +; A test sequence +; +R00:00:00 cmdDisp.CMD_NO_OP + +; Let's try out some commands with arguments +R00:00:01.050 cmdDisp.CMD_NO_OP_STRING "SEQ WAIT 2 MINS!" +; +R00:02:00.050 cmdDisp.CMD_NO_OP_STRING "SEQ after 2mins!" +; +R00:00:01.050 cmdDisp.CMD_NO_OP_STRING "SEQ DONE !" +; \ No newline at end of file diff --git a/Svc/Health/test/int/test_cmd_health.py b/Svc/Health/test/int/test_cmd_health.py new file mode 100644 index 00000000000..3eb1a7b2620 --- /dev/null +++ b/Svc/Health/test/int/test_cmd_health.py @@ -0,0 +1,163 @@ +"""test_cmd_health.py: + +Test the command dispatcher with basic integration tests. +""" + +import time +from fprime_gds.common.testing_fw import predicates +from fprime_gds.common.utils.event_severity import EventSeverity + + +def test_send_health_command(fprime_test_api): + """Test that commands may be sent + + Tests command send, dispatch, and receipt using send_and_assert command with a pair of health commands. + + health.HLTH_ENABLE, [disabled/enabled} + health.HLTH_PING_ENABLE,["FileManager","DISABLED/ENABLED"] + health.HLTH_CHNG_PING,["FileManager",1,1] + """ + + cmd_events = fprime_test_api.get_event_pred(severity=EventSeverity.COMMAND) + actHi_events = fprime_test_api.get_event_pred(severity=EventSeverity.ACTIVITY_HI) + warnLo_events = fprime_test_api.get_event_pred(severity=EventSeverity.WARNING_LO) + + pred = predicates.greater_than(0) + zero = predicates.equal_to(0) + + # Expect number still increment after clear_history + fprime_test_api.clear_histories() # will clear all history (can read telemetry channel again with latest value. otherwise still have old value) + time.sleep(5) + + # Verify disable/enable the Health command + # Command: Disable the health HLTH_ENABLE command + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.Health") + "." + "HLTH_ENABLE", ["DISABLED"] + ) + assert ( + fprime_test_api.get_command_test_history().size() == 1 + ) # current command count + + fprime_test_api.assert_event_count(pred, cmd_events) + fprime_test_api.assert_event_count(pred, actHi_events) + + # Command: Enable health Expect number increment HLTH_Enable command + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.Health") + "." + "HLTH_ENABLE", ["ENABLED"] + ) + assert ( + fprime_test_api.get_command_test_history().size() == 2 + ) # current command count + + fprime_test_api.assert_event_count(4, cmd_events) # Verify event command + fprime_test_api.assert_event_count(2, actHi_events) # Verify event activity_hi + time.sleep(5) + + ## Command: Disabled HLTH_PING_ENABLE command with invalid entry (expected warning_lo) + # use send_command because No completion (will cause pytest to assert when no completion) use send_command will ignore completion) + fprime_test_api.send_command( + fprime_test_api.get_mnemonic("Svc.Health") + "." + "HLTH_PING_ENABLE", + ["Ref_cmdDispatch", "DISABLED"], + ) + assert ( + fprime_test_api.get_command_test_history().size() == 3 + ) # current command count + + time.sleep(3) + cmdErrors = fprime_test_api.await_telemetry( + fprime_test_api.get_mnemonic("Svc.CommandDispatcher") + "." + "CommandErrors", + start="NOW", + ) + # If no constraints are specified on the channels, the predicate will always return true + param_error = fprime_test_api.get_telemetry_pred( + fprime_test_api.get_mnemonic("Svc.CommandDispatcher") + "." + "CommandErrors", + cmdErrors, + ) + + time.sleep(3) + fprime_test_api.assert_event_count(6, cmd_events) # Verify event command + fprime_test_api.assert_event_count(2, actHi_events) # Verify event actHi + fprime_test_api.assert_event_count(1, warnLo_events) # Verify event warning_lo + + ##### Disabled/Enabled HLTH_PING_ENABLE command (PR_NumPings channel will stop when command disabled) and count increment command is enabled + # namespace _health {Svc::Health:PingEntry pingEntries[NUM_PING_ENTRIES ] (look at /build-fprime-automatic-native//top/TopologyAc.cpp + # Get mnemonic of PingReceiver => Ref.pingRcvr (replace . to _ ) + # Notes: + # After send cmd Ref.pingRcvr.PR_StopPings command the channel Ref.pingRcvr.PR_NumPIngs will stop increment + # command health_enable or HLTH_PING_ENABLE (enable option) still not restart this channel ??? + # + # Try append_file 1MiB.txt multi time trigger + # Ref.health.HLTH_PING_WARN WARNING_HI Ping entry Ref_fileManager late warning + + # Verify can disable/enable ping for a particular set of components + # threshold warningValue and Threshold fatalValue + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.Health") + "." + "HLTH_PING_ENABLE", + [fprime_test_api.get_mnemonic("Svc.FileManager").replace(".", "_"), "DISABLED"], + ) + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.Health") + "." + "HLTH_CHNG_PING", + [fprime_test_api.get_mnemonic("Svc.FileManager").replace(".", "_"), 1, 1], + ) + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.Health") + "." + "HLTH_PING_ENABLE", + [fprime_test_api.get_mnemonic("Svc.FileManager").replace(".", "_"), "ENABLED"], + ) + fprime_test_api.send_command( + fprime_test_api.get_mnemonic("Svc.FileManager") + "." + "AppendFile", + ["/tmp/1MiB.txt", "/tmp/2MiB.txt"], + ) + time.sleep(50) + # expected trigger warning hi + fprime_test_api.send_command( + fprime_test_api.get_mnemonic("Svc.FileManager") + "." + "AppendFile", + ["/tmp/2MiB.txt", "/tmp/2MiB.txt"], + ) + time.sleep(10) + fprime_test_api.send_command( + fprime_test_api.get_mnemonic("Svc.FileManager") + "." + "AppendFile", + ["/tmp/2MiB.txt", "/tmp/2MiB.txt"], + ) + time.sleep(10) + fprime_test_api.send_command( + fprime_test_api.get_mnemonic("Svc.FileManager") + "." + "AppendFile", + ["/tmp/2MiB.txt", "/tmp/2MiB.txt"], + ) + time.sleep(10) + fprime_test_api.send_command( + fprime_test_api.get_mnemonic("Svc.FileManager") + "." + "AppendFile", + ["/tmp/2MiB.txt", "/tmp/2MiB.txt"], + ) + time.sleep(10) + fprime_test_api.send_command( + fprime_test_api.get_mnemonic("Svc.FileManager") + "." + "AppendFile", + ["/tmp/2MiB.txt", "/tmp/2MiB.txt"], + ) + time.sleep(10) + fprime_test_api.send_command( + fprime_test_api.get_mnemonic("Svc.FileManager") + "." + "AppendFile", + ["/tmp/2MiB.txt", "/tmp/2MiB.txt"], + ) + time.sleep(10) + fprime_test_api.send_command( + fprime_test_api.get_mnemonic("Svc.FileManager") + "." + "AppendFile", + ["/tmp/2MiB.txt", "/tmp/2MiB.txt"], + ) + time.sleep(10) + fprime_test_api.send_command( + fprime_test_api.get_mnemonic("Svc.FileManager") + "." + "AppendFile", + ["/tmp/2MiB.txt", "/tmp/2MiB.txt"], + ) + time.sleep(120) + + # If no constraints are specified on the channels, the predicate will always return true # confirm PingLateWarnings + WarnHi_error = fprime_test_api.get_telemetry_pred( + fprime_test_api.get_mnemonic("Svc.Health") + "." + "PingLateWarnings", 1 + ) + fprime_test_api.assert_telemetry(WarnHi_error, timeout=5) + + # Cleanup directory + fprime_test_api.send_command( + fprime_test_api.get_mnemonic("Svc.FileManager") + "." + "RemoveFile", + ["/tmp/2MiB.txt", True], + ) diff --git a/Svc/PrmDb/test/int/test_cmd_PrmDb.py b/Svc/PrmDb/test/int/test_cmd_PrmDb.py new file mode 100644 index 00000000000..dd5a5264801 --- /dev/null +++ b/Svc/PrmDb/test/int/test_cmd_PrmDb.py @@ -0,0 +1,30 @@ +"""test_cmd_PrmDb.py: + +Test the command dispatcher with basic integration tests. +""" + +import time + + +def test_send_PrmDb(fprime_test_api): + """Test that commands may be sent + + Tests command send, dispatch, and receipt using send_and_assert command with a pair of CmdDispatcher commands. + + """ + + # send PRM_SAVE_FILE (Wrote 0 records) + + # dir2 = fprime_test_api.get_prm_db_path() + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.PrmDb") + "." + "PRM_SAVE_FILE", max_delay=1 + ) + # Verify PrmDB.dat is empty open where fprime-gds start (ref) /PrmDb.dat (PrmDb.dat file = 0bytes ) + # print ( "DIR2: %s ", dir2) + + # Current PRM_SAVE_FILE saved at where fprime-gds start . Move file to expected location + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.FileManager") + "." + "MoveFile", + ["PrmDb.dat", fprime_test_api.get_mnemonic("Svc.PrmDb.filename")], + max_delay=1, + ) diff --git a/Svc/SystemResources/test/int/test_cmd_systemResources.py b/Svc/SystemResources/test/int/test_cmd_systemResources.py new file mode 100644 index 00000000000..1ee47d3c6b0 --- /dev/null +++ b/Svc/SystemResources/test/int/test_cmd_systemResources.py @@ -0,0 +1,169 @@ +"""test_cmd_systemResources.py: + +Test the command dispatcher with basic integration tests. +""" + +import time + + +def test_send_systemResources_command(fprime_test_api): + """Test that commands may be sent + + Tests command send, dispatch, and receipt using send_and_assert command with a pair of CmdDispatcher commands. + + SystemResources.Enable, (read SystemResources telemetry confirm value stale or stop) + SystemResources.Enable, (read SystemResources telemetry confirm value changing) + + """ + + ## Verify memory usage Mem_total and Mem_used and Non_Volatile_total and Non_Volatile_free greater than a certain value 1KB + mem_total = fprime_test_api.await_telemetry( + fprime_test_api.get_mnemonic("Svc.SystemResources") + "." + "MEMORY_TOTAL", + start="NOW", + ) + mem_used = fprime_test_api.await_telemetry( + fprime_test_api.get_mnemonic("Svc.SystemResources") + "." + "MEMORY_USED", + start="NOW", + ) + nv_total = fprime_test_api.await_telemetry( + fprime_test_api.get_mnemonic("Svc.SystemResources") + + "." + + "NON_VOLATILE_TOTAL", + start="NOW", + ) + nv_free = fprime_test_api.await_telemetry( + fprime_test_api.get_mnemonic("Svc.SystemResources") + "." + "NON_VOLATILE_FREE", + start="NOW", + ) + + mem_total_list = str(mem_total).split() + mem_used_list = str(mem_used).split() + nv_total_list = str(nv_total).split() + nv_free_list = str(nv_free).split() + print("Index my_LIST", mem_total_list) + + # verify memory usage + if ( + int(mem_total_list[3]) >= 1 + and mem_total_list[4] == "KB" + and int(mem_used_list[3]) >= 1 + and mem_used_list[4] == "KB" + and int(nv_total_list[3]) >= 1 + and nv_total_list[4] == "KB" + and int(nv_free_list[3]) >= 1 + and nv_free_list[4] == "KB" + ): + print( + "PASSED: Memory usage MEM_TOTAL, MEM_USED, NV_TOTAL, NV_FREE greater than a certain value 1K" + ) + print( + "LIST VALUE: ", + mem_total_list[3], + mem_used_list[3], + nv_total_list[3], + nv_free_list[3], + ) + else: + print("FAILED to confirm memory usage") + + # Verify a number of CPUs greater than or equal to 1 ((example for 4 CPUs other sometimes is less than 1 percent ) + cpu = fprime_test_api.await_telemetry( + fprime_test_api.get_mnemonic("Svc.SystemResources") + "." + "CPU", start="NOW" + ) + cpu_00 = fprime_test_api.await_telemetry( + fprime_test_api.get_mnemonic("Svc.SystemResources") + "." + "CPU_00", + start="NOW", + ) + cpu_01 = fprime_test_api.await_telemetry( + fprime_test_api.get_mnemonic("Svc.SystemResources") + "." + "CPU_01", + start="NOW", + ) + cpu_02 = fprime_test_api.await_telemetry( + fprime_test_api.get_mnemonic("Svc.SystemResources") + "." + "CPU_02", + start="NOW", + ) + cpu_03 = fprime_test_api.await_telemetry( + fprime_test_api.get_mnemonic("Svc.SystemResources") + "." + "CPU_03", + start="NOW", + ) + + cpu_list = str(cpu).split() + cpu_00_list = str(cpu_00).split() + cpu_01_list = str(cpu_01).split() + cpu_02_list = str(cpu_02).split() + cpu_03_list = str(cpu_03).split() + print("CPU LIST", cpu_list) + + if ( + float(cpu_list[3]) >= 1 + and float(cpu_00_list[3]) >= 1 + and float(cpu_01_list[3]) >= 1 + and float(cpu_02_list[3]) >= 1 + and float(cpu_03_list[3]) >= 1 + ): + print("PASSED : A number of CPUs greater than or equal to 1") + print( + "PERCENT VALUE: ", + cpu_list[3], + cpu_00_list[3], + cpu_01_list[3], + cpu_02_list[3], + cpu_03_list[3], + ) + else: + print("FAILED: cpu") + + # Start command here: + # Current channels before disable + CPU_resources1 = fprime_test_api.await_telemetry( + fprime_test_api.get_mnemonic("Svc.SystemResources") + "." + "CPU", start="NOW" + ) + CPU_percent1 = fprime_test_api.get_telemetry_pred( + fprime_test_api.get_mnemonic("Svc.SystemResources") + "." + "CPU", + CPU_resources1, + ) + print("CPU RESOURCES1: ", CPU_resources1) + print("PERCENT: ", CPU_percent1) + + fprime_test_api.clear_histories() # will clear all history (can read telemetry channel again with latest value. otherwise still have old value) + time.sleep(5) + + # Expect number still changing after clear_history + CPU_resources1A = fprime_test_api.await_telemetry( + fprime_test_api.get_mnemonic("Svc.SystemResources") + "." + "CPU", start="NOW" + ) + CPU_percent1A = fprime_test_api.get_telemetry_pred( + fprime_test_api.get_mnemonic("Svc.SystemResources") + "." + "CPU", + CPU_resources1A, + ) + + ##### Command Disabled SystemResources.ENABLE command (DISABLED) + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.SystemResources") + "." + "ENABLE", + ["DISABLED"], + ) + + time.sleep(3) + # Expect number no change (stale or stop) after Disable + CPU_resources2 = fprime_test_api.await_telemetry( + fprime_test_api.get_mnemonic("Svc.SystemResources") + "." + "CPU", start="NOW" + ) + CPU_percent2 = fprime_test_api.get_telemetry_pred( + fprime_test_api.get_mnemonic("Svc.SystemResources") + "." + "CPU", + CPU_resources2, + ) + time.sleep(5) + + CPU_resources2B = fprime_test_api.await_telemetry( + fprime_test_api.get_mnemonic("Svc.SystemResources") + "." + "CPU", start="NOW" + ) + CPU_percent2B = fprime_test_api.get_telemetry_pred( + fprime_test_api.get_mnemonic("Svc.SystemResources") + "." + "CPU", + CPU_resources2B, + ) + + ##### Command Disabled SystemResources.ENABLE command (ENABLED) + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.SystemResources") + "." + "ENABLE", + ["ENABLED"], + ) diff --git a/Svc/Version/test/int/test_cmd_version.py b/Svc/Version/test/int/test_cmd_version.py new file mode 100644 index 00000000000..5d36de68a5d --- /dev/null +++ b/Svc/Version/test/int/test_cmd_version.py @@ -0,0 +1,98 @@ +"""test_cmd_version.py: + + +Test the command version with basic integration tests. +""" + +from enum import Enum +from fprime_gds.common.testing_fw import predicates +from fprime_gds.common.utils.event_severity import EventSeverity + + +""" +This enum is includes the values of EventSeverity that can be filtered by the ActiveLogger Component +""" +FilterSeverity = Enum( + "FilterSeverity", + "WARNING_HI WARNING_LO COMMAND ACTIVITY_HI ACTIVITY_LO DIAGNOSTIC", +) + + +def test_send_version_command(fprime_test_api): + """Test that commands may be sent + + Tests command send, dispatch, and receipt using send_and_assert command with a pair of version commands. + version.VERSION, [PROJECT,FRAMEWORK,LIBRARY,CUSTOM,ALL] + version.ENABLE, [DISABLED/ENABLED] + """ + + for count, value in enumerate( + ["PROJECT", "FRAMEWORK", "LIBRARY", "CUSTOM", "ALL"], 1 + ): + + pred = predicates.greater_than(0) + + ActLo_events = fprime_test_api.get_event_pred( + severity=EventSeverity.ACTIVITY_LO + ) + + results2 = fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.Version") + "." + "VERSION", + [ + value, + ], + max_delay=5, + # events=events, + ) + + event_cnt = fprime_test_api.assert_event_count(pred, ActLo_events) + # EVR or report MSG_EVR: 2025-07-28T17:36:43.474151: CdhCore.version.ProjectVersion EventSeverity.ACTIVITY_LO : Project Version: [v4.0.0a1-122-g7b6e9a2e1] + + for result in event_cnt: + msg = "{}".format(result.get_str()) + msg_list = str(msg).split() + version = msg_list[6].replace("[", "") + report_ver_value = version.replace("]", "") + + # Channel (Telemetry) History search found the specified item: 2025-07-28T17:36:43.474180: CdhCore.version.ProjectVersion = v4.0.0a1-122-g7b6e9a2e1 + if count == 1: + evr_ver = fprime_test_api.await_telemetry( + fprime_test_api.get_mnemonic("Svc.Version") + "." + "ProjectVersion", + start="NOW", + ) + elif count == 2: + evr_ver = fprime_test_api.await_telemetry( + fprime_test_api.get_mnemonic("Svc.Version") + "." + "FrameworkVersion", + start="NOW", + ) + + if count == 1 or count == 2: + evr_ver_list = str(evr_ver).split() + evr_ver_value = evr_ver_list[3] + + if evr_ver_value == report_ver_value: + print( + "COMPARE EVR vs. TELEMETRY channel and version cmd_option PASS ", + evr_ver_value, + report_ver_value, + value, + ) + else: + print( + "COMPARE EVR vs. TELEMETRY channel and version cmd_option FAIL ", + evr_ver_value, + report_ver_value, + value, + ) + + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.Version") + "." + "ENABLE", + ["ENABLED"], + max_delay=1, + ) + + fprime_test_api.send_and_assert_command( + fprime_test_api.get_mnemonic("Svc.Version") + "." + "ENABLE", + ["DISABLED"], + max_delay=1, + ) diff --git a/ci/tests/fputil.bash b/ci/tests/fputil.bash index 6340f6abf7a..30e75601a7d 100755 --- a/ci/tests/fputil.bash +++ b/ci/tests/fputil.bash @@ -107,7 +107,7 @@ function integration_test_run { then TIMEOUT="gtimeout" # macOS homebrew "coreutils" fi - TOP_CONFIG_ARGS="--deployment-config ${WORKDIR}/config.json" + TOP_CONFIG_ARGS="--deployment-config ${WORKDIR}/test/int/int_config.json" ${TIMEOUT} --kill-after=10s 180s pytest ${DICTIONARY_ARGS} ${TOP_CONFIG_ARGS} ) RET_PYTEST=$?