Improve reusable Integration Tests system (nasa#5187)

* Add InT for FileHandling subtopology: file uplink, file downlink

* Add InT for CdhCore: command, events, telemetry ch

* remove unused imports

* add documentation for reusable InTs

* Prune unused int_config.json entries from Ref

* format python??

* fix incorrect usages of shared int infrastructure

* fix comp instance name

Author: thomas-bc

Svc/Subtopologies/CdhCore/test/int/test_cdh_core.py

@@ -0,0 +1,75 @@
+"""test_cdh_core.py:
+
+Test the core functionality of the CdhCore subtopology with:
+1. Commands
+2. Events
+3. Telemetry channels
+"""
+
+import random
+from fprime_gds.common.testing_fw.api import IntegrationTestAPI
+
+
+def test_basic_command_and_event(fprime_test_api: IntegrationTestAPI):
+    """Test that we can send a command and receive the expected event in response"""
+
+    # Send NO_OP command to FSW and wait for expected event
+    fprime_test_api.send_and_assert_command(
+        f"{fprime_test_api.get_mnemonic('Svc.CommandDispatcher')}.CMD_NO_OP",
+    )
+
+
+def test_command_and_event_with_string_arg(fprime_test_api: IntegrationTestAPI):
+    """Test that we can send a command with arguments and receive the expected event with args in response"""
+
+    TEST_STRING = f"test string {random.random()}"
+
+    test_event = fprime_test_api.get_event_pred("NoOpStringReceived", [TEST_STRING])
+
+    # Send NO_OP command to FSW and wait for expected event
+    fprime_test_api.send_and_assert_event(
+        f"{fprime_test_api.get_mnemonic('Svc.CommandDispatcher')}.CMD_NO_OP_STRING",
+        [TEST_STRING],
+        events=[test_event],
+        timeout=2,
+    )
+
+
+def test_command_and_event_with_many_args(fprime_test_api: IntegrationTestAPI):
+    """Test that we can send a command with arguments and receive the expected event with args in response"""
+
+    # types are (I32, F32, U8) - random float precision is finnicky, so just use a fixed value
+    TEST_ARGS = [
+        random.randint(-(2 ** 31), 2 ** 31 - 1),
+        1.5,
+        random.randint(0, 2 ** 8 - 1),
+    ]
+
+    test_event = fprime_test_api.get_event_pred("TestCmd1Args", TEST_ARGS)
+
+    # Send CMD_1 (no-op with args) command to FSW and wait for expected event
+    fprime_test_api.send_and_assert_event(
+        f"{fprime_test_api.get_mnemonic('Svc.CommandDispatcher')}.CMD_TEST_CMD_1",
+        TEST_ARGS,
+        events=[test_event],
+        timeout=3,
+    )
+
+
+def test_telemetry_update(fprime_test_api: IntegrationTestAPI):
+    """Test that we can receive telemetry updates with expected values"""
+
+    cmd_dispatched_channel = fprime_test_api.get_telemetry_pred("CommandsDispatched")
+
+    # Wait for telemetry update with expected values
+    begin_result = fprime_test_api.await_telemetry(cmd_dispatched_channel, timeout=3)
+    begin_tlm_val = begin_result.val_obj.val
+
+    # Send no op to increase the count of commands dispatched
+    end_result = fprime_test_api.send_and_await_telemetry(
+        f"{fprime_test_api.get_mnemonic('Svc.CommandDispatcher')}.CMD_NO_OP",
+        channels=cmd_dispatched_channel,
+        timeout=3,
+    )
+    # Assert that the telemetry value has increased by 1 after sending the command
+    assert end_result.val_obj.val == begin_tlm_val + 1

Svc/Subtopologies/FileHandling/test/int/test_file_handling.py

@@ -0,0 +1,67 @@
+"""test_file_handling.py:
+
+Test the file handling functionality with:
+1. Uplink a file
+2. Downlink a file
+
+Both these tests assume that FSW and GDS are running on the same system, for simplicity
+Should projects require to run on different systems, they are encouraged to adapt these tests as needed
+"""
+
+import tempfile
+from pathlib import Path
+import random
+
+
+def test_uplink_file(fprime_test_api):
+    """IMPORTANT: These tests assume that FSW and GDS are running on the same system
+
+    Create a file locally, "uplink" it (FSW and GDS are expected to be the same system)
+    and verify the contents are the same after the uplink process."""
+
+    TEST_DATA = f"test uplink data {random.random()}\n"
+
+    tmp_file_in = tempfile.NamedTemporaryFile(mode="w+")
+    tmp_file_out = tempfile.NamedTemporaryFile(mode="r+")
+
+    tmp_file_in.write(TEST_DATA)
+    tmp_file_in.flush()
+
+    # Begin file uplink and wait for completion event
+    fprime_test_api.uplink_file_and_await_completion(
+        tmp_file_in.name, destination=tmp_file_out.name, timeout=15
+    )
+
+    tmp_file_out.seek(0)
+    assert tmp_file_out.readlines() == [TEST_DATA]
+
+
+def test_downlink_file(fprime_test_api):
+    """IMPORTANT: These tests assume that FSW and GDS are running on the same system
+
+    Create a file locally, "downlink" it (FSW and GDS are expected to be the same system)
+    and verify the contents are the same after the downlink process.
+    """
+    TEST_DATA = f"test downlink data {random.random()}\n"
+
+    # Retrieve GDS sandbox area for file downlink
+    down_store = Path(fprime_test_api.pipeline.down_store)
+    if not down_store.exists():
+        raise RuntimeError("GDS sandbox area not found for file downlink test")
+
+    # Add some randomness to avoid collisions in the downlink folder
+    output_filename = f"downlink_test_{random.randint(0, 10000)}.txt"
+    output_file = down_store / output_filename
+
+    # Using dir="/tmp/" to force short filenames which can be a limitation on downlink
+    tmp_file_in = tempfile.NamedTemporaryFile(mode="w+", dir="/tmp/")
+    tmp_file_in.write(TEST_DATA)
+    tmp_file_in.flush()
+
+    # Request file downlink via FSW command
+    fprime_test_api.send_and_assert_command(
+        f"{fprime_test_api.get_mnemonic('Svc.FileDownlink')}.SendFile",
+        [str(tmp_file_in.name), str(output_filename)],
+    )
+
+    assert output_file.read_text() == TEST_DATA

TestDeploymentsProject/Ref/test/int/int_config.json

@@ -1,37 +1,15 @@
 {
-    "Svc.ActiveRateGroup": "Ref.rateGroup1Comp",
     "Svc.CommandDispatcher": "CdhCore.cmdDisp",
     "Svc.CmdSequencer": "Ref.cmdSeq",
     "Svc.FileDownlink": "FileHandling.fileDownlink",
     "Svc.FileManager": "FileHandling.fileManager",
-    "Svc.FileUplink": "Ref.fileUplink",
-    "Ref.PingReceiver": "Ref.pingRcvr",
+    "Svc.FileUplink": "FileHandling.fileUplink",
     "Svc.EventManager": "CdhCore.events",
-    "Svc.TlmChan": "Ref.tlmSend",
+    "Svc.TlmChan": "CdhCore.tlmSend",
     "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": "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.framer",
-    "Svc.ComStub": "Ref.comStub"
+    "Svc.Health": "CdhCore.health"
 }

TestDeploymentsProject/Ref/test/int/test_cmd_parameter.py

@@ -25,93 +25,93 @@ def test_send_parameter(fprime_test_api):
 
     ## setup default-value
     fprime_test_api.send_and_assert_command(
-        fprime_test_api.get_mnemonic("Ref.RecvBuff") + "." + "PARAMETER1_PRM_SET",
+        "Ref.recvBuffComp.PARAMETER1_PRM_SET",
         [1],
         max_delay=5,
     )
     fprime_test_api.send_and_assert_command(
-        fprime_test_api.get_mnemonic("Ref.RecvBuff") + "." + "PARAMETER2_PRM_SET",
+        "Ref.recvBuffComp.PARAMETER2_PRM_SET",
         [2],
         max_delay=5,
     )
     fprime_test_api.send_and_assert_command(
-        fprime_test_api.get_mnemonic("Ref.SendBuff") + "." + "PARAMETER3_PRM_SET",
+        "Ref.sendBuffComp.PARAMETER3_PRM_SET",
         [3],
         max_delay=5,
     )
     fprime_test_api.send_and_assert_command(
-        fprime_test_api.get_mnemonic("Ref.SendBuff") + "." + "PARAMETER4_PRM_SET",
+        "Ref.sendBuffComp.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",
+        "Ref.recvBuffComp.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
+        "Ref.recvBuffComp.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",
+        "Ref.recvBuffComp.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",
+        "Ref.recvBuffComp.PARAMETER2_PRM_SET",
         [20],
         max_delay=5,
     )
 
     param2_change = fprime_test_api.get_telemetry_pred(
-        fprime_test_api.get_mnemonic("Ref.RecvBuff") + "." + "Parameter2", 20
+        "Ref.recvBuffComp.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",
+        "Ref.recvBuffComp.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",
+        "Ref.sendBuffComp.PARAMETER3_PRM_SET",
         [30],
         max_delay=5,
     )
 
     param3_change = fprime_test_api.get_telemetry_pred(
-        fprime_test_api.get_mnemonic("Ref.SendBuff") + "." + "Parameter3", 30
+        "Ref.sendBuffComp.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",
+        "Ref.sendBuffComp.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",
+        "Ref.sendBuffComp.PARAMETER4_PRM_SET",
         [40],
         max_delay=5,
     )
 
     param4_change = fprime_test_api.get_telemetry_pred(
-        fprime_test_api.get_mnemonic("Ref.SendBuff") + "." + "Parameter4", 40
+        "Ref.sendBuffComp.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",
+        "Ref.sendBuffComp.PARAMETER4_PRM_SAVE",
         max_delay=5,
     )
 

docs/user-manual/gds/reusable-integration-tests.md

@@ -0,0 +1,73 @@
+# Reusable Integration Tests
+
+F´ ships portable integration tests that exercise framework components (`Svc/*`) and standard subtopologies (`Svc/Subtopologies/*`) against any deployment. They are written once in F´ and reused unchanged in your project — you only supply a small JSON file mapping framework component names to your deployment's instance names.
+
+This guide shows how to run them in your own deployment.
+
+## How it works
+
+Reusable tests live next to the component or subtopology under `test/int/` (e.g. [Svc/CmdDispatcher/test/int/test_cmd_dispatcher.py](../../../Svc/CmdDispatcher/test/int/test_cmd_dispatcher.py), [Svc/Subtopologies/CdhCore/test/int/test_cdh_core.py](../../../Svc/Subtopologies/CdhCore/test/int/test_cdh_core.py)). Instead of hardcoding instance mnemonics like `Ref.cmdDisp`, they call `fprime_test_api.get_mnemonic(...)` with the qualified component name:
+
+```python
+fprime_test_api.send_and_assert_command(
+    f"{fprime_test_api.get_mnemonic('Svc.CommandDispatcher')}.CMD_NO_OP",
+)
+```
+
+At runtime, `get_mnemonic("Svc.CommandDispatcher")` reads a JSON file (`int_config.json` by convention) supplied via the `--deployment-config` pytest flag and returns the deployment-specific instance mnemonic — e.g. `CdhCore.cmdDisp` for the Ref deployment. If a key is missing, it falls back to the qualified name itself, so a partial config still works.
+
+Source: `IntegrationTestAPI.get_mnemonic` in [fprime-gds api.py](https://github.com/nasa/fprime-gds/blob/devel/src/fprime_gds/common/testing_fw/api.py).
+
+## Writing your `int_config.json`
+
+Create one JSON file in your deployment (typical location: `<MyDeployment>/test/int/int_config.json`) mapping each `Svc.*` (and any other reusable) component name to the instance mnemonic in your topology:
+
+```json
+{
+    "Svc.CommandDispatcher": "CdhCore.cmdDisp",
+    "Svc.CmdSequencer": "Ref.cmdSeq",
+    "Svc.FileDownlink": "FileHandling.fileDownlink",
+    "Svc.FileManager": "FileHandling.fileManager",
+    ... add more as needed per the tests you want to run ...
+}
+```
+
+The full reference example is [TestDeploymentsProject/Ref/test/int/int_config.json](../../../TestDeploymentsProject/Ref/test/int/int_config.json) — copy it and replace the right-hand side with your instance names. Only include the components whose tests you intend to run.
+
+## Running the tests
+
+Build and start your deployment, then start the GDS pointed at your dictionary, and run pytest against the `test/int` directory of any component or subtopology you want to exercise:
+
+```bash
+# 1. Start GDS + flight software (separate terminal, from your deployment dir)
+fprime-gds --dictionary build-artifacts/*/MyDeployment/dict/MyDeploymentTopologyDictionary.json
+
+# 2. Run the reusable tests
+pytest ./lib/fprime/Svc/CmdDispatcher/test/int \
+       ./lib/fprime/Svc/Subtopologies/CdhCore/test/int \
+       --dictionary build-artifacts/*/MyDeployment/dict/MyDeploymentTopologyDictionary.json \
+       --deployment-config ./MyDeployment/test/int/int_config.json
+```
+
+`--deployment-config` is provided by the `fprime-gds` pytest plugin (see [pytest_integration.py](https://github.com/nasa/fprime-gds/blob/devel/src/fprime_gds/common/testing_fw/pytest_integration.py)). All other flags (`--dictionary`, `--logs`, etc.) are the standard GDS pipeline options.
+
+## CI example
+
+The Ref deployment's CI job is the canonical reference — see [`.github/workflows/ref.yml`](../../../.github/workflows/ref.yml):
+
+```yaml
+- name: "Integration Tests"
+  uses: nasa/fprime-actions/run-integration-tests@devel
+  with:
+    test-working-directory: "TestDeploymentsProject/Ref/test/int"
+    binary: "build-artifacts/*/Ref/bin/Ref"
+    gds-args: "--dictionary build-artifacts/*/Ref/dict/RefTopologyDictionary.json"
+    pytest-args: "--deployment-config ${{ github.workspace }}/TestDeploymentsProject/Ref/test/int/int_config.json"
+```
+
+To also run the framework's reusable tests, pass extra paths via `pytest-args` to include the relevant `Svc/*/test/int/test_*.py` test files.
+
+## See also
+
+- [GDS Integration Test API Guide](gds-test-api-guide.md) — full `IntegrationTestAPI` reference
+- [Ref `int_config.json`](../../../TestDeploymentsProject/Ref/test/int/int_config.json) — complete example