Add subscription verification

panther_lights/test/integration/panther_lights.test.py

@@ -19,6 +19,7 @@
 
 import launch
 import launch_testing
+import panther_utils.integration_test_utils as test_utils
 import rclpy
 import rclpy.qos
 from diagnostic_msgs.msg import DiagnosticArray
@@ -134,6 +135,12 @@ def test_msg_publishers(self):
             received_topics_str = ", ".join(wait_for_topics.topics_received())
             print("Received messages from the following topics: [" + received_topics_str + "]")
 
+    def test_msg_subscribers(self):
+        node_info = test_utils.get_node_info("/lights_driver")
+
+        self.assertIn("/lights/channel_1_frame", node_info.subscribers)
+        self.assertIn("/lights/channel_2_frame", node_info.subscribers)
+
 
 @launch_testing.post_shutdown_test()
 class TestProcessOutput(unittest.TestCase):

panther_utils/panther_utils/integration_test_utils.py

@@ -0,0 +1,79 @@
+#!/usr/bin/env python3
+
+# Copyright 2024 Husarion sp. z o.o.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import subprocess
+from typing import List
+
+
+class ROSNodeInfo:
+    """
+    Class representing the ROS node info.
+    """
+
+    def __init__(self):
+        self.subscribers: List[str] = []
+        self.publishers: List[str] = []
+        self.service_servers: List[str] = []
+        self.service_clients: List[str] = []
+        self.action_servers: List[str] = []
+        self.action_clients: List[str] = []
+
+
+def get_node_info(node_name: str) -> ROSNodeInfo:
+    """
+    Executes the command 'ros2 node info <node_name>' and returns the ROSNodeInfo object.
+
+    Args:
+        node_name (str): The name of the ROS 2 node to get information about.
+
+    Returns:
+        ROSNodeInfo: An object representing a complete node info.
+    """
+    node_info = ROSNodeInfo()
+
+    section_map = {
+        "Subscribers:": node_info.subscribers,
+        "Publishers:": node_info.publishers,
+        "Service Servers:": node_info.service_servers,
+        "Service Clients:": node_info.service_clients,
+        "Action Servers:": node_info.action_servers,
+        "Action Clients:": node_info.action_clients,
+    }
+
+    try:
+        # Execute the `ros2 node info` command
+        result = subprocess.run(
+            ["ros2", "node", "info", node_name], capture_output=True, text=True, check=True
+        )
+
+        # Parse the output
+        lines = result.stdout.splitlines()
+        current_section = None
+
+        for line in lines:
+            line = line.strip()
+            if line in section_map:
+                current_section = section_map[line]
+            elif line and current_section is not None:
+                current_section.append(line.split(":")[0].strip())
+            else:
+                current_section = None
+
+    except subprocess.CalledProcessError as e:
+        print(f"Error executing command: {e}")
+        print(f"stderr: {e.stderr}")
+
+    return node_info