Add reusable InT test scripts

Ref/config.json

@@ -0,0 +1,39 @@
+{
+    "Drv.BlockDriver" : "Ref.blockDrv",
+    "Svc.ActiveRateGroup" : "Ref.rateGroup1Comp",
+    "Svc.CommandDispatcher" : "CDHCore.cmdDisp",
+    "Svc.CmdSequencer" : "Ref.cmdSeq",
+    "Svc.FileDownlink" : "Ref.fileDownlink",
+    "Svc.FileManager" : "Ref.fileManager",
+    "Svc.FileUplink" : "Ref.fileUplink",
+    "Ref.PingReceiver" : "Ref.pingRcvr",
+    "Svc.ActiveLogger" : "CDHCore.events",
+    "Svc.TlmChan" : "Ref.tlmSend",
+    "Svc.PrmDb" : "Ref.prmDb",
+    "Svc.PrmDb.filename" : "/tmp/PrmDb.dat",
+    "Svc.DpCatalog" : "Ref.dpCat",
+    "Svc.DpManager" : "Ref.dpMgr",
+    "Svc.DpWriter" : "Ref.dpWriter",
+    "Svc.ComQueue" : "Ref.comQueue",
+    "Ref.TypeDemo" : "Ref.typeDemo",
+    "Svc.Health" : "CDHCore.health",
+    "Ref.SignalGen" : "Ref.SG1",
+    "Ref.SendBuff" : "Ref.sendBuffComp",
+    "Drv.TcpClient" : "Ref.comDriver",
+    "Svc.AssertFatalAdapter" : "Ref.fatalAdapter",
+    "Svc.FatalHandler" : "Ref.fatalHandler",
+    "Svc.BufferManager" : "Ref.commsBufferManager",
+    "Svc.PosixTime" : "Ref.posixTime",
+    "Svc.RateGroupDriver" : "Ref.rateGroupDriverComp",
+    "Ref.RecvBuff" : "Ref.recvBuffComp",
+    "Svc.Version" : "CDHCore.version",
+    "Svc.PassiveTextLogger" : "Ref.textLogger",
+    "Svc.SystemResources" : "Ref.systemResources",
+    "Svc.BufferManager" : "Ref.dpBufferManager",
+    "Svc.FrameAccumulator" : "Ref.frameAccumulator",
+    "Svc.FprimeDeframer" : "Ref.deframer",
+    "Svc.FprimeRouter" : "Ref.fprimeRouter",
+    "Svc.FprimeFramer" : "Ref.fprimeFramer",
+    "Svc.ComStub" : "Ref.comStub"
+}
+

Ref/test/int/test_cmd_parameter.py

@@ -0,0 +1,73 @@
+""" 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
+
+
+
+
+
+
+

Svc/ActiveLogger/test/int/test_cmd_ActiveLogger.py

@@ -0,0 +1,47 @@
+""" test_cmd_ActiveLogger.py:
+
+Test the command ActiveLogger 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.ActiveLogger') + '.' + '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.ActiveLogger') + '.' + '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.ActiveLogger') + '.' + 'DUMP_FILTER_STATE', max_delay=1)
+
+    ### Disable all ACTIVITY_HI
+    fprime_test_api.send_and_assert_command(fprime_test_api.get_mnemonic('Svc.ActiveLogger') + '.' + '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.ActiveLogger') + '.' + 'DUMP_FILTER_STATE', max_delay=1)    
+
+    ## Put back default value
+    fprime_test_api.send_and_assert_command(fprime_test_api.get_mnemonic('Svc.ActiveLogger') + '.' + 'SET_EVENT_FILTER',["ACTIVITY_HI", "ENABLED"], max_delay=1)    
+    fprime_test_api.send_and_assert_command(fprime_test_api.get_mnemonic('Svc.ActiveLogger') + '.' + 'SET_ID_FILTER', ["0x507", "DISABLED"], max_delay=1)
+    fprime_test_api.send_and_assert_command(fprime_test_api.get_mnemonic('Svc.ActiveLogger') + '.' + 'DUMP_FILTER_STATE', max_delay=1)
+
+

Svc/CmdDispatcher/test/int/test_cmd_dispatcher.py

@@ -7,6 +7,16 @@
 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)
+
+
+

Svc/CmdSequencer/test/int/test_cmd_sequencer.py

@@ -0,0 +1,168 @@
+""" 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=1)
+
+    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),
+                "/tmp/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),
+                "/tmp/ref_test_seq_wait.bin",
+            ]
+        ).returncode
+        == 0
+    ), "Failed to run fprime-seqgen"
+
+    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 newfile1   ######
+#    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=1)    
+    # 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=1)    
+
+    ######    ######    ######
+    ## Toggle Manual/AUTO
+    fprime_test_api.send_and_assert_command(fprime_test_api.get_mnemonic('Svc.CmdSequencer') + '.' + 'CS_MANUAL', max_delay=1)
+    fprime_test_api.send_and_assert_command(fprime_test_api.get_mnemonic('Svc.CmdSequencer') + '.' + 'CS_AUTO', max_delay=1)
+
+    ######    ######    ######    
+    ## Manual Mode testing Start/Cancel
+    fprime_test_api.send_and_assert_command(fprime_test_api.get_mnemonic('Svc.CmdSequencer') + '.' + 'CS_MANUAL', max_delay=1)
+
+    # 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=1)    
+
+    fprime_test_api.send_and_assert_command(fprime_test_api.get_mnemonic('Svc.CmdSequencer') + '.' + 'CS_CANCEL', max_delay=1)        
+
+    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=1)
+
+    fprime_test_api.send_and_assert_command(fprime_test_api.get_mnemonic('Svc.CmdSequencer') + '.' + 'CS_VALIDATE',["/tmp/ref_test_seq.bin"], max_delay=1)
+
+    fprime_test_api.send_and_assert_command(fprime_test_api.get_mnemonic('Svc.CmdSequencer') + '.' + 'CS_AUTO', max_delay=1)
+
+    ######    ######    ######    
+    fprime_test_api.send_and_assert_command(fprime_test_api.get_mnemonic('Svc.CmdSequencer') + '.' + 'CS_MANUAL', max_delay=1)
+    ## 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=1)
+    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=1)
+    fprime_test_api.send_and_assert_command(fprime_test_api.get_mnemonic('Svc.CommandDispatcher') + '.' + 'CMD_NO_OP_STRING',["manually START command 0"], max_delay=1)
+
+
+    # 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=1)
+    time.sleep(130)    
+    fprime_test_api.send_and_assert_command(fprime_test_api.get_mnemonic('Svc.CmdSequencer') + '.' + 'CS_STEP', max_delay=3)    
+
+   ######    ######    ######              

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!";

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 !"
+;

Svc/FileDownlink/test/int/test_cmd_fileDownlink.py

@@ -0,0 +1,24 @@
+""" 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/ref_test_seq_wait.bin   ,   Dest = /tmp/<user>/fprime-downlink/DL.log
+    fprime_test_api.send_and_assert_command(fprime_test_api.get_mnemonic('Svc.FileDownlink') + '.' + 'SendFile', ["/tmp/ref_test_seq_wait.bin", "DL.log"], max_delay=30)
+
+    ## source = /tmp/1MiB.txt   ,   Dest = /tmp/<user>/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)
+