test(aircraft): pure ros2 topic info parser with unit tests

aircraft/aircraft_resources/scripts/inspect_topics.py

@@ -6,6 +6,8 @@
 """
 import subprocess
 
+from inspect_topics_parse import parse_topic_info_lines
+
 def inspect_topics():
     try:
         topics = subprocess.check_output(['ros2', 'topic', 'list']).decode().split()
@@ -22,25 +24,10 @@ def inspect_topics():
         except Exception:
             continue
 
-        msg_type = "Unknown"
-        pubs, subs = [], []
-        current_section = None
-
-        for line in info:
-            line = line.strip()
-            if line.startswith("Type:"):
-                msg_type = line.split("Type:")[1].strip()
-            elif line.startswith("Publisher count:"):
-                current_section = "pub"
-            elif line.startswith("Subscription count:"):
-                current_section = "sub"
-            elif line.startswith("Node name:"):
-                node_name = line.split("Node name:")[1].strip()
-                if node_name != "UNKNOWN" and not node_name.startswith("_ros2cli"):
-                    if current_section == "pub":
-                        pubs.append(node_name)
-                    elif current_section == "sub":
-                        subs.append(node_name)
+        parsed = parse_topic_info_lines(info)
+        msg_type = parsed["msg_type"]
+        pubs = parsed["pubs"]
+        subs = parsed["subs"]
 
         pub_str = ", ".join(pubs) if pubs else "None"
         sub_str = ", ".join(subs) if subs else "None"

aircraft/aircraft_resources/scripts/inspect_topics_parse.py

@@ -0,0 +1,38 @@
+"""Pure parser for `ros2 topic info -v` line output (offline-testable)."""
+
+from __future__ import annotations
+
+
+def parse_topic_info_lines(lines):
+    """Parse verbose ros2 topic info lines into type/pubs/subs.
+
+    Parameters
+    ----------
+    lines : iterable of str
+        Output of ``ros2 topic info -v <topic>`` split into lines.
+
+    Returns
+    -------
+    dict with keys: msg_type (str), pubs (list[str]), subs (list[str])
+    """
+    msg_type = "Unknown"
+    pubs, subs = [], []
+    current_section = None
+
+    for raw in lines:
+        line = raw.strip()
+        if line.startswith("Type:"):
+            msg_type = line.split("Type:", 1)[1].strip() or "Unknown"
+        elif line.startswith("Publisher count:"):
+            current_section = "pub"
+        elif line.startswith("Subscription count:"):
+            current_section = "sub"
+        elif line.startswith("Node name:"):
+            node_name = line.split("Node name:", 1)[1].strip()
+            if node_name != "UNKNOWN" and not node_name.startswith("_ros2cli"):
+                if current_section == "pub":
+                    pubs.append(node_name)
+                elif current_section == "sub":
+                    subs.append(node_name)
+
+    return {"msg_type": msg_type, "pubs": pubs, "subs": subs}

aircraft/aircraft_resources/scripts/test_inspect_topics_parse.py

@@ -0,0 +1,41 @@
+#!/usr/bin/env python3
+import unittest
+
+from inspect_topics_parse import parse_topic_info_lines
+
+
+SAMPLE = """
+Type: sensor_msgs/msg/Image
+Publisher count: 1
+Node name: cam_node
+Node namespace: /
+Subscription count: 2
+Node name: viewer
+Node namespace: /
+Node name: _ros2cli_123
+Node namespace: /
+Node name: UNKNOWN
+Node namespace: /
+""".strip().splitlines()
+
+
+class TestInspectTopicsParse(unittest.TestCase):
+    def test_sample(self):
+        parsed = parse_topic_info_lines(SAMPLE)
+        self.assertEqual(parsed["msg_type"], "sensor_msgs/msg/Image")
+        self.assertEqual(parsed["pubs"], ["cam_node"])
+        self.assertEqual(parsed["subs"], ["viewer"])
+
+    def test_empty(self):
+        parsed = parse_topic_info_lines([])
+        self.assertEqual(parsed["msg_type"], "Unknown")
+        self.assertEqual(parsed["pubs"], [])
+        self.assertEqual(parsed["subs"], [])
+
+    def test_type_only(self):
+        parsed = parse_topic_info_lines(["Type: std_msgs/msg/String"])
+        self.assertEqual(parsed["msg_type"], "std_msgs/msg/String")
+
+
+if __name__ == "__main__":
+    unittest.main()