From 45b52961b305151117fbd053dbd8c5ba9eb26073 Mon Sep 17 00:00:00 2001 From: maciejmajek Date: Thu, 9 Apr 2026 13:34:49 +0200 Subject: [PATCH 1/6] refactor(ROS2BaseConnector): move topic logic to topic api --- .../rai/communication/ros2/api/topic.py | 68 +++++++++++++++++++ .../rai/communication/ros2/connectors/base.py | 50 +++----------- 2 files changed, 77 insertions(+), 41 deletions(-) diff --git a/src/rai_core/rai/communication/ros2/api/topic.py b/src/rai_core/rai/communication/ros2/api/topic.py index cb55002f6..9ae7f1b7c 100644 --- a/src/rai_core/rai/communication/ros2/api/topic.py +++ b/src/rai_core/rai/communication/ros2/api/topic.py @@ -13,6 +13,7 @@ # limitations under the License. import time +from functools import partial from typing import ( Any, Callable, @@ -243,6 +244,73 @@ def _verify_publisher_exists(self, topic: str) -> List[TopicEndpointInfo]: raise ValueError(f"No publisher found for topic: {topic}") return topic_endpoints + def receive_message( + self, + topic: str, + timeout_sec: float = 1.0, + msg_type: Optional[str] = None, + qos_profile: Optional[QoSProfile] = None, + auto_qos_matching: bool = True, + ) -> Any: + """Receive a single message from a ROS2 topic. + + Creates an internal subscriber for the topic if one does not yet exist. + Checks the message cache first; if a sufficiently recent message is found + it is returned immediately. Otherwise polls until a message arrives or the + timeout expires. + + If ``destroy_subscribers`` was set to ``True`` at construction time the + subscriber is destroyed immediately after the first message is received. + This frees resources but may trigger the rclpy executor crash described in + https://github.com/ros2/rclpy/issues/1142, so the default is ``False``. + + Args: + topic: Name of the topic to receive from. + timeout_sec: Maximum time to wait for a message, in seconds. + msg_type: ROS2 message type string. Auto-detected when not provided. + qos_profile: QoS profile to use when creating the subscriber. + auto_qos_matching: Automatically match QoS with existing publishers. + + Returns: + The first raw ROS2 message received from the topic. + + Raises: + TimeoutError: If no message is received within ``timeout_sec``. + """ + if topic not in self._subscriptions: + sub_qos = self._resolve_qos_profile( + topic, auto_qos_matching, qos_profile, for_publisher=False + ) + if msg_type is None: + msg_type = self.get_topic_type(topic) + msg_cls = self.import_message_from_str(msg_type) + subscription = self.node.create_subscription( + topic=topic, + msg_type=msg_cls, + callback=partial(self._generic_callback, topic), + qos_profile=sub_qos, + ) + self._subscriptions[topic] = subscription + else: + # Subscriber already exists; serve from cache if fresh enough. + if topic in self._last_msg: + timestamp, msg = self._last_msg[topic] + if timestamp > time.time() - timeout_sec: + return msg + + start_time = time.time() + while time.time() - start_time < timeout_sec: + if topic in self._last_msg: + _, msg = self._last_msg[topic] + if self._destroy_subscribers and topic in self._subscriptions: + self.node.destroy_subscription(self._subscriptions.pop(topic)) + return msg + time.sleep(0.1) + + raise TimeoutError( + f"Message from {topic} not received in {timeout_sec} seconds" + ) + def shutdown(self) -> None: """Cleanup publishers when object is destroyed.""" for publisher in self._publishers.values(): diff --git a/src/rai_core/rai/communication/ros2/connectors/base.py b/src/rai_core/rai/communication/ros2/connectors/base.py index 4f0accec1..604c0bd48 100644 --- a/src/rai_core/rai/communication/ros2/connectors/base.py +++ b/src/rai_core/rai/communication/ros2/connectors/base.py @@ -16,7 +16,7 @@ import time import uuid from functools import partial -from typing import Any, Callable, Dict, Final, List, Literal, Optional, Tuple, TypeVar +from typing import Any, Callable, Final, List, Literal, Optional, Tuple, TypeVar import rclpy import rclpy.executors @@ -163,9 +163,6 @@ def __init__( self._thread.start() self.last_executor_performance_time = time.time() - # cache for last received messages - self.last_msg: Dict[str, T] = {} - def _executor_performance_callback(self) -> None: """Monitor executor performance and log warnings if it falls behind schedule. @@ -197,18 +194,6 @@ def _executor_performance_callback(self) -> None: else: self.last_executor_performance_time = current_time - def _last_message_callback(self, source: str, msg: T): - """Store the last received message for a given source. - - Parameters - ---------- - source : str - The topic source identifier. - msg : T - The received message. - """ - self.last_msg[source] = msg - def get_topics_names_and_types(self) -> List[Tuple[str, List[str]]]: """Get list of available topics and their message types. @@ -372,31 +357,14 @@ def receive_message( TimeoutError If no message is received within the timeout period. """ - if self._topic_api.subscriber_exists(source): - # trying to hit cache first - if source in self.last_msg: - if self.last_msg[source].timestamp > time.time() - timeout_sec: - return self.last_msg[source] - else: - self._topic_api.create_subscriber( - topic=source, - callback=partial(self.general_callback, source), - msg_type=msg_type, - qos_profile=qos_profile, - auto_qos_matching=auto_qos_matching, - ) - self.register_callback(source, partial(self._last_message_callback, source)) - - start_time = time.time() - # wait for the message to be received - while time.time() - start_time < timeout_sec: - if source in self.last_msg: - return self.last_msg[source] - time.sleep(0.1) - else: - raise TimeoutError( - f"Message from {source} not received in {timeout_sec} seconds" - ) + raw_msg = self._topic_api.receive_message( + topic=source, + timeout_sec=timeout_sec, + msg_type=msg_type, + qos_profile=qos_profile, + auto_qos_matching=auto_qos_matching, + ) + return self.general_callback_preprocessor(raw_msg) @staticmethod def wait_for_transform( From fb4c70adcc0e048d7619465f6e85447b9444b6a6 Mon Sep 17 00:00:00 2001 From: maciejmajek Date: Thu, 9 Apr 2026 13:42:47 +0200 Subject: [PATCH 2/6] chore: bump version --- src/rai_core/pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/rai_core/pyproject.toml b/src/rai_core/pyproject.toml index b6c1c5d3b..612987849 100644 --- a/src/rai_core/pyproject.toml +++ b/src/rai_core/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "rai_core" -version = "2.11.1" +version = "2.11.2" description = "Core functionality for RAI framework" readme = "README.md" requires-python = ">=3.10,<3.13" From 4eedb547da065bd84fe034c28b87dcd23f98664e Mon Sep 17 00:00:00 2001 From: maciejmajek Date: Thu, 9 Apr 2026 14:25:55 +0200 Subject: [PATCH 3/6] test: add new cases --- tests/communication/ros2/test_api.py | 62 ++++++++++++++++++++++++++++ 1 file changed, 62 insertions(+) diff --git a/tests/communication/ros2/test_api.py b/tests/communication/ros2/test_api.py index 64974fb76..17625b6bd 100644 --- a/tests/communication/ros2/test_api.py +++ b/tests/communication/ros2/test_api.py @@ -38,6 +38,7 @@ from std_srvs.srv import SetBool from .helpers import ( + MessagePublisher, MessageSubscriber, ServiceServer, TestActionClient, @@ -51,6 +52,67 @@ _ = ros_setup # Explicitly use the fixture to prevent pytest warnings +def test_ros2_topic_api_receive_message( + ros_setup: None, request: pytest.FixtureRequest +) -> None: + topic_name = f"{request.node.originalname}_topic" # type: ignore + node_name = f"{request.node.originalname}_node" # type: ignore + message_publisher = MessagePublisher(topic_name) + node = Node(node_name) + executors, threads = multi_threaded_spinner([message_publisher, node]) + + try: + topic_api = ROS2TopicAPI(node) + # First call: creates internal subscriber and waits for message + msg = topic_api.receive_message(topic_name, timeout_sec=2.0) + assert msg.data == "Hello, ROS2!" + # Second call: subscriber already exists, message is in cache → cache-hit branch + msg2 = topic_api.receive_message(topic_name, timeout_sec=2.0) + assert msg2.data == "Hello, ROS2!" + finally: + shutdown_executors_and_threads(executors, threads) + + +def test_ros2_topic_api_receive_message_timeout( + ros_setup: None, request: pytest.FixtureRequest +) -> None: + topic_name = f"{request.node.originalname}_topic" # type: ignore + node_name = f"{request.node.originalname}_node" # type: ignore + node = Node(node_name) + executors, threads = multi_threaded_spinner([node]) + + try: + topic_api = ROS2TopicAPI(node) + # No publisher active → subscriber is created but no messages arrive + with pytest.raises(TimeoutError): + topic_api.receive_message( + topic_name, + timeout_sec=0.3, + msg_type="std_msgs/msg/String", + ) + finally: + shutdown_executors_and_threads(executors, threads) + + +def test_ros2_topic_api_receive_message_destroy_subscriber( + ros_setup: None, request: pytest.FixtureRequest +) -> None: + topic_name = f"{request.node.originalname}_topic" # type: ignore + node_name = f"{request.node.originalname}_node" # type: ignore + message_publisher = MessagePublisher(topic_name) + node = Node(node_name) + executors, threads = multi_threaded_spinner([message_publisher, node]) + + try: + topic_api = ROS2TopicAPI(node, destroy_subscribers=True) + msg = topic_api.receive_message(topic_name, timeout_sec=2.0) + assert msg.data == "Hello, ROS2!" + # Subscriber should have been torn down after first message + assert topic_name not in topic_api._subscriptions + finally: + shutdown_executors_and_threads(executors, threads) + + def test_ros2_single_message_publish( ros_setup: None, request: pytest.FixtureRequest ) -> None: From 9c6caba107f9caad5ff0fa5d37d98ce92b2695ab Mon Sep 17 00:00:00 2001 From: maciejmajek Date: Thu, 9 Apr 2026 14:28:58 +0200 Subject: [PATCH 4/6] chore: update lock --- uv.lock | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/uv.lock b/uv.lock index 82cf7b3f7..3b6e18e00 100644 --- a/uv.lock +++ b/uv.lock @@ -4700,7 +4700,7 @@ requires-dist = [ [[package]] name = "rai-core" -version = "2.11.1" +version = "2.11.2" source = { editable = "src/rai_core" } dependencies = [ { name = "coloredlogs" }, From 405fca1f7aebaf442f2571ebcaf5f8e6051baceb Mon Sep 17 00:00:00 2001 From: maciejmajek Date: Thu, 9 Apr 2026 14:43:57 +0200 Subject: [PATCH 5/6] test: add new cases --- tests/communication/ros2/test_api.py | 218 ++++++++++++++++++++ tests/communication/ros2/test_connectors.py | 88 ++++++++ 2 files changed, 306 insertions(+) diff --git a/tests/communication/ros2/test_api.py b/tests/communication/ros2/test_api.py index 17625b6bd..ef21c1a6a 100644 --- a/tests/communication/ros2/test_api.py +++ b/tests/communication/ros2/test_api.py @@ -35,6 +35,8 @@ ) from rclpy.executors import MultiThreadedExecutor from rclpy.node import Node +from rclpy.qos import QoSProfile +from std_msgs.msg import String from std_srvs.srv import SetBool from .helpers import ( @@ -113,6 +115,222 @@ def test_ros2_topic_api_receive_message_destroy_subscriber( shutdown_executors_and_threads(executors, threads) +def test_ros2_topic_api_subscriber_and_publisher_exists( + ros_setup: None, request: pytest.FixtureRequest +) -> None: + topic_name = f"{request.node.originalname}_topic" # type: ignore + node_name = f"{request.node.originalname}_node" # type: ignore + node = Node(node_name) + executors, threads = multi_threaded_spinner([node]) + + try: + topic_api = ROS2TopicAPI(node) + assert not topic_api.subscriber_exists(topic_name) + assert not topic_api.publisher_exists(topic_name) + + topic_api.create_subscriber( + topic_name, + callback=lambda msg: None, + msg_type="std_msgs/msg/String", + ) + assert topic_api.subscriber_exists(topic_name) + + topic_api.create_publisher(topic_name, msg_type="std_msgs/msg/String") + assert topic_api.publisher_exists(topic_name) + finally: + shutdown_executors_and_threads(executors, threads) + + +def test_ros2_topic_api_create_subscriber_explicit_qos( + ros_setup: None, request: pytest.FixtureRequest +) -> None: + topic_name = f"{request.node.originalname}_topic" # type: ignore + node_name = f"{request.node.originalname}_node" # type: ignore + node = Node(node_name) + executors, threads = multi_threaded_spinner([node]) + + try: + topic_api = ROS2TopicAPI(node) + sub = topic_api.create_subscriber( + topic_name, + callback=lambda msg: None, + msg_type="std_msgs/msg/String", + qos_profile=QoSProfile(depth=1), + auto_qos_matching=False, + ) + assert sub is not None + assert topic_api.subscriber_exists(topic_name) + finally: + shutdown_executors_and_threads(executors, threads) + + +def test_ros2_topic_api_create_subscriber_no_qos_raises( + ros_setup: None, request: pytest.FixtureRequest +) -> None: + topic_name = f"{request.node.originalname}_topic" # type: ignore + node_name = f"{request.node.originalname}_node" # type: ignore + node = Node(node_name) + executors, threads = multi_threaded_spinner([node]) + + try: + topic_api = ROS2TopicAPI(node) + with pytest.raises(ValueError, match="Either qos_profile or auto_qos_matching"): + topic_api.create_subscriber( + topic_name, + callback=lambda msg: None, + msg_type="std_msgs/msg/String", + qos_profile=None, + auto_qos_matching=False, + ) + finally: + shutdown_executors_and_threads(executors, threads) + + +def test_ros2_topic_api_create_publisher_explicit_qos( + ros_setup: None, request: pytest.FixtureRequest +) -> None: + topic_name = f"{request.node.originalname}_topic" # type: ignore + node_name = f"{request.node.originalname}_node" # type: ignore + node = Node(node_name) + executors, threads = multi_threaded_spinner([node]) + + try: + topic_api = ROS2TopicAPI(node) + pub = topic_api.create_publisher( + topic_name, + msg_type="std_msgs/msg/String", + qos_profile=QoSProfile(depth=1), + auto_qos_matching=False, + ) + assert pub is not None + assert topic_api.publisher_exists(topic_name) + finally: + shutdown_executors_and_threads(executors, threads) + + +def test_ros2_topic_api_create_publisher_no_qos_raises( + ros_setup: None, request: pytest.FixtureRequest +) -> None: + topic_name = f"{request.node.originalname}_topic" # type: ignore + node_name = f"{request.node.originalname}_node" # type: ignore + node = Node(node_name) + executors, threads = multi_threaded_spinner([node]) + + try: + topic_api = ROS2TopicAPI(node) + with pytest.raises(ValueError, match="Either qos_profile or auto_qos_matching"): + topic_api.create_publisher( + topic_name, + msg_type="std_msgs/msg/String", + qos_profile=None, + auto_qos_matching=False, + ) + finally: + shutdown_executors_and_threads(executors, threads) + + +def test_ros2_topic_api_verify_receive_args( + ros_setup: None, request: pytest.FixtureRequest +) -> None: + node_name = f"{request.node.originalname}_node" # type: ignore + node = Node(node_name) + executors, threads = multi_threaded_spinner([node]) + + try: + topic_api = ROS2TopicAPI(node) + with pytest.raises(ValueError, match="Cannot provide both"): + topic_api._verify_receive_args( + "topic", auto_topic_type=True, msg_type="std_msgs/msg/String" + ) + with pytest.raises(ValueError, match="msg_type must be provided"): + topic_api._verify_receive_args( + "topic", auto_topic_type=False, msg_type=None + ) + finally: + shutdown_executors_and_threads(executors, threads) + + +def test_ros2_topic_api_generic_callback( + ros_setup: None, request: pytest.FixtureRequest +) -> None: + node_name = f"{request.node.originalname}_node" # type: ignore + node = Node(node_name) + executors, threads = multi_threaded_spinner([node]) + + try: + topic_api = ROS2TopicAPI(node) + msg = String(data="test") + topic_api._generic_callback("test_topic", msg) + assert "test_topic" in topic_api._last_msg + timestamp, stored_msg = topic_api._last_msg["test_topic"] + assert stored_msg is msg + assert timestamp <= time.time() + finally: + shutdown_executors_and_threads(executors, threads) + + +def test_ros2_topic_api_is_topic_available( + ros_setup: None, request: pytest.FixtureRequest +) -> None: + topic_name = f"{request.node.originalname}_topic" # type: ignore + node_name = f"{request.node.originalname}_node" # type: ignore + message_publisher = MessagePublisher(topic_name) + node = Node(node_name) + executors, threads = multi_threaded_spinner([message_publisher, node]) + + try: + topic_api = ROS2TopicAPI(node) + assert topic_api._is_topic_available(topic_name, timeout_sec=2.0) + assert not topic_api._is_topic_available( + "/definitely_nonexistent_topic_xyz", timeout_sec=0.3 + ) + finally: + shutdown_executors_and_threads(executors, threads) + + +def test_ros2_topic_api_resolve_qos_profile_branches( + ros_setup: None, request: pytest.FixtureRequest +) -> None: + topic_name = f"{request.node.originalname}_topic" # type: ignore + node_name = f"{request.node.originalname}_node" # type: ignore + node = Node(node_name) + executors, threads = multi_threaded_spinner([node]) + + try: + topic_api = ROS2TopicAPI(node) + qos = QoSProfile(depth=5) + + # auto_qos_matching=True AND qos_profile provided → logs warning, returns qos_profile + result = topic_api._resolve_qos_profile( + topic_name, True, qos, for_publisher=True + ) + assert result is qos + + # auto_qos_matching=False, qos_profile provided → returns qos_profile directly + result2 = topic_api._resolve_qos_profile( + topic_name, False, qos, for_publisher=True + ) + assert result2 is qos + finally: + shutdown_executors_and_threads(executors, threads) + + +def test_ros2_topic_api_verify_publisher_exists_raises( + ros_setup: None, request: pytest.FixtureRequest +) -> None: + topic_name = f"{request.node.originalname}_topic" # type: ignore + node_name = f"{request.node.originalname}_node" # type: ignore + node = Node(node_name) + executors, threads = multi_threaded_spinner([node]) + + try: + topic_api = ROS2TopicAPI(node) + with pytest.raises(ValueError, match="No publisher found"): + topic_api._verify_publisher_exists(topic_name) + finally: + shutdown_executors_and_threads(executors, threads) + + def test_ros2_single_message_publish( ros_setup: None, request: pytest.FixtureRequest ) -> None: diff --git a/tests/communication/ros2/test_connectors.py b/tests/communication/ros2/test_connectors.py index 9fbf0d4e1..ba99bf272 100644 --- a/tests/communication/ros2/test_connectors.py +++ b/tests/communication/ros2/test_connectors.py @@ -19,6 +19,7 @@ from unittest.mock import MagicMock import pytest +import rclpy from nav2_msgs.action import NavigateToPose from PIL import Image from pydub import AudioSegment @@ -35,6 +36,7 @@ ) from std_msgs.msg import String from std_srvs.srv import SetBool +from tf2_ros import LookupException from .helpers import ( HRIMessageSubscriber, @@ -44,6 +46,7 @@ TestActionClient, TestActionServer, TestServiceClient, + TransformPublisher, multi_threaded_spinner, ros_setup, shutdown_executors_and_threads, @@ -359,3 +362,88 @@ def test_ros2_connector_unique_names(ros_setup: None): finally: connector1.shutdown() connector2.shutdown() + + +def test_ros2_connector_auto_init() -> None: + """Connector initializes rclpy automatically when it is not yet running.""" + if rclpy.ok(): + rclpy.shutdown() + connector = None + try: + connector = ROS2Connector() + assert rclpy.ok() + finally: + if connector: + connector.shutdown() + if rclpy.ok(): + rclpy.shutdown() + + +def test_ros2_connector_invalid_executor_type(ros_setup: None) -> None: + with pytest.raises(ValueError, match="Invalid executor type"): + ROS2Connector(executor_type="invalid_type") # type: ignore[arg-type] + + +def test_ros2_connector_executor_performance_warning(ros_setup: None) -> None: + connector = ROS2Connector() + try: + # Simulate the executor running far behind schedule — covers the warning branch + connector.last_executor_performance_time = time.time() - 10.0 + connector._executor_performance_callback() + # Warning goes to the rclpy stderr logger; we just verify it doesn't raise + finally: + connector.shutdown() + + +def test_ros2_connector_register_callback( + ros_setup: None, request: pytest.FixtureRequest +) -> None: + topic_name = f"{request.node.originalname}_topic" # type: ignore + message_publisher = MessagePublisher(topic_name) + executors, threads = multi_threaded_spinner([message_publisher]) + connector = ROS2Connector() + received: List[Any] = [] + try: + callback_id = connector.register_callback( + topic_name, + lambda msg: received.append(msg), + msg_type="std_msgs/msg/String", + ) + assert callback_id is not None + time.sleep(0.5) + assert len(received) > 0 + finally: + connector.shutdown() + shutdown_executors_and_threads(executors, threads) + + +def test_ros2_connector_wait_for_transform_found( + ros_setup: None, request: pytest.FixtureRequest +) -> None: + transform_publisher = TransformPublisher(f"{request.node.originalname}_tf") # type: ignore + executors, threads = multi_threaded_spinner([transform_publisher]) + connector = ROS2Connector() + try: + result = connector.wait_for_transform( + connector._tf_buffer, + target_frame="map", + source_frame="base_link", + timeout_sec=3.0, + ) + assert result is True + finally: + connector.shutdown() + shutdown_executors_and_threads(executors, threads) + + +def test_ros2_connector_get_transform_lookup_exception(ros_setup: None) -> None: + connector = ROS2Connector() + try: + with pytest.raises(LookupException): + connector.get_transform( + target_frame="nonexistent_frame", + source_frame="also_nonexistent", + timeout_sec=0.2, + ) + finally: + connector.shutdown() From ae93608a2636eddac9a4670360e450a350cea266 Mon Sep 17 00:00:00 2001 From: maciejmajek Date: Thu, 9 Apr 2026 14:54:32 +0200 Subject: [PATCH 6/6] refactor(ROS2TopicAPI): move receive_message logic from connector, fix resource leak and housekeeping --- .../rai/communication/ros2/connectors/base.py | 11 ++++++----- tests/communication/ros2/test_connectors.py | 2 +- 2 files changed, 7 insertions(+), 6 deletions(-) diff --git a/src/rai_core/rai/communication/ros2/connectors/base.py b/src/rai_core/rai/communication/ros2/connectors/base.py index 604c0bd48..35e14001a 100644 --- a/src/rai_core/rai/communication/ros2/connectors/base.py +++ b/src/rai_core/rai/communication/ros2/connectors/base.py @@ -156,12 +156,14 @@ def __init__( elif self._executor_type == "single_threaded": self._executor = SingleThreadedExecutor() else: + self._tf_listener.unregister() + self._node.destroy_node() raise ValueError(f"Invalid executor type: {self._executor_type}") self._executor.add_node(self._node) self._thread = threading.Thread(target=self._executor.spin) self._thread.start() - self.last_executor_performance_time = time.time() + self._last_executor_performance_time = time.time() def _executor_performance_callback(self) -> None: """Monitor executor performance and log warnings if it falls behind schedule. @@ -173,7 +175,7 @@ def _executor_performance_callback(self) -> None: current_time = time.time() time_behind = ( current_time - - self.last_executor_performance_time + - self._last_executor_performance_time - self._executor_performance_time_delta ) threshold = ( @@ -190,9 +192,7 @@ def _executor_performance_callback(self) -> None: f"{self._executor.__class__.__name__} is {time_behind:.2f} seconds behind. " f"If you see this message frequently, consider switching to {', '.join(alternative_executors)}." ) - self.last_executor_performance_time = current_time - else: - self.last_executor_performance_time = current_time + self._last_executor_performance_time = current_time def get_topics_names_and_types(self) -> List[Tuple[str, List[str]]]: """Get list of available topics and their message types. @@ -531,6 +531,7 @@ def shutdown(self): 4. Shuts down the topic API 5. Shuts down the executor 6. Joins the executor thread + """ self._tf_listener.unregister() self._node.destroy_node() diff --git a/tests/communication/ros2/test_connectors.py b/tests/communication/ros2/test_connectors.py index ba99bf272..102c65163 100644 --- a/tests/communication/ros2/test_connectors.py +++ b/tests/communication/ros2/test_connectors.py @@ -388,7 +388,7 @@ def test_ros2_connector_executor_performance_warning(ros_setup: None) -> None: connector = ROS2Connector() try: # Simulate the executor running far behind schedule — covers the warning branch - connector.last_executor_performance_time = time.time() - 10.0 + connector._last_executor_performance_time = time.time() - 10.0 connector._executor_performance_callback() # Warning goes to the rclpy stderr logger; we just verify it doesn't raise finally: