Add integration test.

Signed-off-by: Jelmer de Wolde <[email protected]>

Author: Jelmerdw

.github/workflows/ros-workspace.yml

@@ -12,17 +12,27 @@ on:
 jobs:
   ros-workspace:
     runs-on: ubuntu-latest
-    container: 
+    container:
       image: rcdt/robotics
       options: --user root
     steps:
       - name: checkout
         uses: actions/checkout@v4
         with:
-          submodules: 'true'
-      - name: build
-        shell: bash -ieo pipefail {0} 
+          submodules: "true"
+      - name: build workspace
+        shell: bash -ieo pipefail {0}
         run: |
-            source /home/rcdt/.bashrc
-            cd ros2_ws
-            colcon build
+          source /home/rcdt/.bashrc
+          cd ros2_ws
+          colcon build --symlink-install
+      - name: run tests
+        shell: bash -ieo pipefail {0}
+        run: |
+          source /home/rcdt/.bashrc
+          GZ_SIM_RESOURCE_PATH="$GZ_SIM_RESOURCE_PATH:/$PWD/ros2_ws/src"
+          GZ_SIM_RESOURCE_PATH="$GZ_SIM_RESOURCE_PATH:/$PWD/ros2_ws/src/rcdt_gz_worlds/models"
+          GZ_SIM_RESOURCE_PATH="$GZ_SIM_RESOURCE_PATH:/$PWD/ros2_ws/src/rcdt_gz_worlds/worlds"
+          cd ros2_ws
+          source install/setup.bash
+          pytest-3 src/

dockerfiles/install_scripts/dev_packages.sh

@@ -4,6 +4,9 @@
 
 set -e
 
+# Install pytest-timeout:
+pip install pytest-timeout
+
 # Specifying numpy range to work with ROS and Ultralytics:
 pip install "numpy>=1.23.0,<2.0"
 

pyproject.toml

@@ -4,4 +4,10 @@
 
 [tool.ruff.lint]
 select = ["I", "F", "E", "W", "B", "N", "A", "C4", "PL", "Q000", "SIM", "ANN", "ARG", "ERA"]
-ignore = ["E501", "ANN204"]
+ignore = ["E501", "ANN204"]
+
+[tool.pytest.ini_options]
+filterwarnings = [
+    "ignore:invalid escape sequence",
+]
+timeout = 90

ros2_ws/src/rcdt_franka/test/__init__.py

@@ -0,0 +1,3 @@
+# SPDX-FileCopyrightText: Alliander N. V.
+#
+# SPDX-License-Identifier: Apache-2.0

ros2_ws/src/rcdt_franka/test/test_integration.py

@@ -0,0 +1,25 @@
+import launch_pytest
+import pytest
+from launch import LaunchDescription
+from launch.actions import IncludeLaunchDescription
+from rcdt_utilities.launch_utils import assert_for_message, get_file_path
+from sensor_msgs.msg import JointState
+
+
+@launch_pytest.fixture(scope="module")
+def launch_description() -> LaunchDescription:
+    franka = IncludeLaunchDescription(
+        get_file_path("rcdt_franka", ["launch"], "franka.launch.py")
+    )
+
+    return LaunchDescription(
+        [
+            franka,
+            launch_pytest.actions.ReadyToTest(),
+        ]
+    )
+
+
+@pytest.mark.launch(fixture=launch_description)
+def test_joint_states_published() -> None:
+    assert_for_message(JointState, "/joint_states", 10)

ros2_ws/src/rcdt_utilities/rcdt_utilities/launch_utils.py

@@ -4,7 +4,6 @@
 
 import ast
 import os
-import time
 from typing import List
 
 import rclpy
@@ -16,6 +15,7 @@
 from launch.event_handlers import OnProcessExit
 from launch.events.process import ProcessExited
 from launch.substitutions import LaunchConfiguration
+from launch_testing_ros.wait_for_topics import WaitForTopics
 from rclpy.executors import Executor
 from rclpy.logging import get_logger
 from rclpy.node import Node
@@ -102,10 +102,17 @@ def on_completion(
         if event.returncode == 0:
             return complete
         else:
-            while rclpy.ok:
-                logger.error("Target did not start successfully. Please restart.")
-                time.sleep(WAIT)
+            logger.error("Target did not start successfully. Please restart.")
 
     return RegisterEventHandler(
         OnProcessExit(target_action=target, on_exit=on_completion)
     )
+
+
+def assert_for_message(message_type: type, topic: str, timeout: int) -> bool:
+    wait_for_topics = WaitForTopics([(topic, message_type)], timeout)
+    received = wait_for_topics.wait()
+    wait_for_topics.shutdown()
+    assert received, (
+        f"No message received of type {message_type.__name__} on topic {topic} within {timeout} seconds."
+    )