Integration Tests Workflow

.github/workflows/integration-tests.yml

@@ -0,0 +1,20 @@
+name: Integration-Tests
+
+on:
+  push:
+  pull_request:
+  workflow_dispatch:
+
+jobs:
+    integration-test:
+        runs-on: ubuntu-latest
+        timeout-minutes: 10
+        steps:
+        - name: Checkout workspace
+          uses: actions/checkout@v4
+
+        - name: Run integration tests
+          uses: ./.github/actions/
+          with:
+            script: './scripts/run-integration-tests'
+            run-website: 'true'

.github/workflows/tests.yml

@@ -52,3 +52,5 @@ jobs:
       uses: ./.github/actions/
       with:
         script: './scripts/clang-tidy'
+
+

scripts/build/.built_by

@@ -0,0 +1 @@
+colcon

scripts/install/.colcon_install_layout

@@ -0,0 +1 @@
+merged

scripts/log/build_2025-01-11_09-55-00/logger_all.log

@@ -0,0 +1,2 @@
+[0.673s] DEBUG:colcon:Command line arguments: ['/usr/bin/colcon', 'build', '--packages-ignore', 'virtual_iridium', '--merge-install', '--symlink-install', '--cmake-args', '-DCMAKE_BUILD_TYPE=Debug', '-DSTATIC_ANALYSIS=OFF', '-DUNIT_TEST=ON', '--no-warn-unused-cli']
+[0.673s] DEBUG:colcon:Parsed command line arguments: Namespace(log_base=None, log_level=None, verb_name='build', build_base='build', install_base='install', merge_install=True, symlink_install=True, test_result_base=None, continue_on_error=False, executor='parallel', parallel_workers=8, event_handlers=None, ignore_user_meta=False, metas=['./colcon.meta'], base_paths=['.'], packages_ignore=['virtual_iridium'], packages_ignore_regex=None, paths=None, packages_up_to=None, packages_up_to_regex=None, packages_above=None, packages_above_and_dependencies=None, packages_above_depth=None, packages_select_by_dep=None, packages_skip_by_dep=None, packages_skip_up_to=None, packages_select_build_failed=False, packages_skip_build_finished=False, packages_select_test_failures=False, packages_skip_test_passed=False, packages_select=None, packages_skip=None, packages_select_regex=None, packages_skip_regex=None, packages_start=None, packages_end=None, cmake_args=['-DCMAKE_BUILD_TYPE=Debug', '-DSTATIC_ANALYSIS=OFF', '-DUNIT_TEST=ON', '--no-warn-unused-cli'], cmake_target=None, cmake_target_skip_unavailable=False, cmake_clean_cache=False, cmake_clean_first=False, cmake_force_configure=False, ament_cmake_args=None, catkin_cmake_args=None, catkin_skip_building_tests=False, verb_parser=<colcon_defaults.argument_parser.defaults.DefaultArgumentsDecorator object at 0x7fed30eff820>, verb_extension=<colcon_core.verb.build.BuildVerb object at 0x7fed30efe800>, main=<bound method BuildVerb.main of <colcon_core.verb.build.BuildVerb object at 0x7fed30efe800>>)

scripts/log/latest

@@ -0,0 +1 @@
+latest_build

scripts/log/latest_build

@@ -0,0 +1 @@
+build_2025-01-11_09-55-00

scripts/run-integration-tests.sh

@@ -0,0 +1,25 @@
+#!/bin/bash
+set -e
+
+if [[ $LOCAL_RUN != "true" ]]; then
+    # give user permissions, required for GitHub Actions
+    sudo chown -R $(whoami):$(whoami) $ROS_WORKSPACE
+
+    source /opt/ros/${ROS_DISTRO}/setup.bash
+    ./scripts/setup.sh
+    ./scripts/build.sh
+    ./scripts/run_virtual_iridium.sh
+    source $ROS_WORKSPACE/install/setup.bash
+fi
+
+echo "Integration tests started"
+cd $ROS_WORKSPACE/src/integration_tests
+
+for FILE in $ROS_WORKSPACE/src/integration_tests/testplans/*; do
+    if [ -f $FILE ]; then
+        echo "Running $FILE test plan"
+        ros2 run integration_tests run --ros-args -p testplan:=$FILE
+    fi
+done
+
+echo "Integration tests finished"

src/integration_tests/integration_tests/integration_test_node.py

@@ -1,23 +1,24 @@
 import builtins
 import functools
+import json
 import os
 import signal
 import subprocess
 import sys
 import time
 import traceback
+import urllib.request
 from dataclasses import dataclass, field
-from typing import Any, Optional, Tuple, Type, TypeVar, Union
+from typing import Any, Optional, Tuple, Type, Union
 
-import rclpy
-import rclpy.node
-import std_msgs.msg
+import custom_interfaces.msg  # type: ignore
+import rclpy  # type: ignore
+import rclpy.node  # type: ignore
+import std_msgs.msg  # type: ignore
 import yaml
-from rclpy.impl.rcutils_logger import RcutilsLogger
+from rclpy.impl.rcutils_logger import RcutilsLogger  # type: ignore
 from rclpy.node import MsgType, Node
 
-import custom_interfaces.msg
-
 MIN_SETUP_DELAY_S = 1  # Minimum 1 second delay between setting up the test and sending inputs
 DEFAULT_TIMEOUT_SEC = 3  # Number of seconds that the test has to run
 
@@ -30,9 +31,13 @@
 NON_ROS_PACKAGES = ["virtual_iridium", "website"]
 
 # TODO: TestMsgType needs to encompass/inherit whatever type we use for HTTP messages
-HTTP_MSG_PLACEHOLDER_TYPE = TypeVar("HTTP_MSG_PLACEHOLDER_TYPE")
+HTTP_MSG_PLACEHOLDER_TYPE = builtins.dict[str, Any]
 TestMsgType = Union[rclpy.node.MsgType, HTTP_MSG_PLACEHOLDER_TYPE]
 
+GLOBAL_PATH_API_NAME = "globalpath"
+POST_URL = "http://localhost:8081/global-path"
+PULL_URL = "http://localhost:3005/api/"
+
 
 def main(args=None):
     rclpy.init(args=args)
@@ -284,7 +289,7 @@ def get_ros_dtype(dtype: str) -> Tuple[Union[builtins.type, MsgType], Type[MsgTy
                 return (
                     getattr(custom_interfaces.msg, dtype)(),
                     getattr(custom_interfaces.msg, dtype),
-                )
+                )  # type: ignore
             except AttributeError:
                 raise TypeError(f"INVALID TYPE: {dtype}")
 
@@ -355,6 +360,24 @@ def parse_ros_data(data: dict) -> Tuple[Union[None, rclpy.node.MsgType], rclpy.n
     return msg, msg_type
 
 
+def parse_http_data(
+    data: dict,
+) -> Tuple[Union[None, HTTP_MSG_PLACEHOLDER_TYPE], HTTP_MSG_PLACEHOLDER_TYPE]:
+    """Parses
+
+    Args:
+        data (dict): Stores the expected HTTP output
+
+    Returns:
+        Tuple[Union[None, HTTP_MSG_PLACEHOLDER_TYPE], HTTP_MSG_PLACEHOLDER_TYPE]: _description_
+    """
+
+    msg_type = data["dtype"]
+    data.pop("dtype")
+
+    return data, msg_type
+
+
 @dataclass
 class IOEntry:
     """Represents IO data"""
@@ -396,7 +419,7 @@ def __set_inputs(self, inputs: list[dict]):
             inputs (list[dict]): list of inputs
 
         Raises:
-            NotImplementedError: If an input of type HTTP is given. It's on the TODO list :)
+            NotImplementedError: GlobalPath messages are not yet supported
             KeyError: If an invalid input type is given. Valid input types are ROS and HTTP.
         """
         for input_dict in inputs:
@@ -408,7 +431,17 @@ def __set_inputs(self, inputs: list[dict]):
                 self.__ros_inputs.append(new_input)
 
             elif input_dict["type"] == "HTTP":
-                raise NotImplementedError("HTTP support is a WIP")
+                data = input_dict["data"]
+                if GLOBAL_PATH_API_NAME == input_dict["name"]:
+                    # This is a GlobalPath message
+                    msg, msg_type = parse_http_data(data)  # type: ignore
+                    new_input = IOEntry(name=input_dict["name"], msg_type=msg_type, msg=msg)
+                else:
+                    msg, msg_type = parse_ros_data(data)  # type: ignore
+                    new_input = IOEntry(name=input_dict["name"], msg_type=msg_type, msg=msg)
+
+                self.__http_inputs.append(new_input)
+
             else:
                 raise KeyError(f"Invalid input type: {input_dict['type']}")
 
@@ -431,7 +464,12 @@ def __set_expected_outputs(self, outputs: list[dict]):
                 self.__ros_e_outputs.append(new_output)
 
             elif output["type"] == "HTTP":
-                raise NotImplementedError("HTTP support is a WIP")
+                data = output["data"]
+
+                msg, msg_type = parse_http_data(data)  # type: ignore
+                new_output = IOEntry(name=output["name"], msg_type=msg_type, msg=msg)
+                self.__http_e_outputs.append(new_output)
+
             else:
                 raise KeyError(f"Invalid output type: {output['type']}")
 
@@ -585,6 +623,9 @@ def __init__(self) -> None:
 
         testplan_file = self.get_parameter("testplan").get_parameter_value().string_value
 
+        self.__http_outputs: list[dict[str, Any]] = []
+        self.__web_fail = 0
+
         try:
             self.__test_inst = IntegrationTestSequence(testplan_file)
             try:
@@ -615,13 +656,32 @@ def __init__(self) -> None:
                     )
                     self.__ros_subs.append(sub)
 
-                # TODO: HTTP
+                self.__http_inputs: list[ROSInputEntry] = []
+                self.__global_path_pub = False
+                for http_input in self.__test_inst.http_inputs():
+                    if http_input.name != GLOBAL_PATH_API_NAME:
+                        pub = self.create_publisher(
+                            msg_type=http_input.msg_type,
+                            topic=http_input.name,
+                            qos_profile=10,  # change
+                        )
+                        self.__http_inputs.append(ROSInputEntry(pub=pub, msg=http_input.msg))
+                    else:
+                        # This is a GlobalPath message
+                        self.__global_path_pub = self.pub_global_path(http_input.msg)
+
+                        if self.__global_path_pub:
+                            self.get_logger().info("Published GlobalPath to database")
+                        else:
+                            self.__web_fail = 1
 
                 # IMPORTANT: MAKE SURE EXPECTED OUTPUTS ARE SETUP BEFORE SENDING INPUTS
                 time.sleep(MIN_SETUP_DELAY_S)
                 self.drive_inputs()
 
-                self.timeout = self.create_timer(self.__test_inst.timeout_sec(), self.__timeout_cb)
+                self.timeout = self.create_timer(
+                    self.__test_inst.timeout_sec(), self.__timeout_cb()  # type: ignore
+                )
             except Exception as e:
                 # At this point, the test instance has successfully started all package processes.
                 # This except block is a failsafe to kill the processes in case anything crashes
@@ -651,14 +711,91 @@ def __pub_ros(self) -> None:
             pub.publish(msg)
             self.get_logger().info(f'Published to topic: "{pub.topic}", with msg: "{msg}"')
 
+    def __pub_http(self) -> None:
+        """Publish to all registered ROS input topics that interact with HTTP endpoints"""
+        for web_input in self.__http_inputs:
+            pub = web_input.pub
+            msg = web_input.msg
+            pub.publish(msg)
+            self.get_logger().info(f'Published to topic: "{pub.topic}", with msg: "{msg}"')
+
+    def pub_global_path(self, msg: Union[dict[str, Any], None]) -> bool:
+        """Publish to the GlobalPath HTTP endpoint
+
+        Args:
+            msg (HTTP_MSG_PLACEHOLDER_TYPE): Message to publish
+        """
+        data = json.dumps(msg).encode("utf8")
+
+        try:
+            urllib.request.urlopen(
+                POST_URL,
+                data,
+            )
+            return True
+        except urllib.error.URLError as e:
+            self.get_logger().error(
+                f"Failed to publish Global path to the remote transceiver: {e}"
+            )
+            return False
+        except urllib.error.HTTPError as e:
+            self.get_logger().error(
+                f"Failed to publish Global path to the remote transceiver: {e}"
+            )
+            return False
+
+    def http_evaluate(self, http_outputs: list[dict[str, Any]]) -> int:
+
+        num_fail = 0
+
+        for count, http_e_output in enumerate(self.__test_inst.http_expected_outputs()):
+            topic = http_e_output.name
+            data = http_outputs[count]
+
+            if data != http_e_output.msg:
+                self._logger.error(
+                    f"HTTP output from {topic} does not match expected output: {data}"
+                )
+                num_fail += 1
+
+        return num_fail
+
+    def get_http_outputs(self) -> int:
+
+        num_fail = 0
+
+        for e_http_output in self.__test_inst.http_expected_outputs():
+            topic = e_http_output.name
+            try:
+                contents = urllib.request.urlopen(PULL_URL + topic)
+            except urllib.error.HTTPError as e:
+                self._logger.error(f"HTTPError: {e}")
+                return False
+
+            data_dict = json.load(contents)
+            try:
+                data = (data_dict["data"])[0]
+            except IndexError:
+                self._logger.error(f"No data found for {topic}")
+                return False
+
+            data.pop(
+                "timestamp"
+            )  # Remove timestamp from data as it is not relevant for comparison
+            self.__http_outputs.append(data)
+
+        num_fail = self.http_evaluate(self.__http_outputs)
+
+        return num_fail
+
     def __timeout_cb(self) -> None:
         """Callback for when the test times out. Stops all test processes, evaluates correctness,
         and exits
         """
         self.__test_inst.finish()  # Stop tests
 
         num_fail, num_warn = self.__monitor.evaluate(self.get_logger())
-
+        num_fail += self.__web_fail + self.get_http_outputs()
         if num_warn > 0:
             self.get_logger().warn(
                 (
@@ -680,6 +817,7 @@ def drive_inputs(self) -> None:
         """Drive all registered test inputs"""
         self.__pub_ros()
         # TODO: add HTTP
+        self.__pub_http()
 
 
 if __name__ == "__main__":