Add tests for --content-filter option on echo, hz, and bw

Integration tests:
- test_echo_content_filter: matching and non-matching DDS content
  filter expressions
- test_echo_content_filter_once: --content-filter combined with --once
- test_echo_content_filter_combined_with_filter: DDS content filter
  combined with Python --filter (dual filtering)
- test_hz_content_filter / test_bw_content_filter: happy path
- test_hz_content_filter_no_match / test_bw_content_filter_no_match:
  non-matching filter produces no rate/bandwidth output

Contract tests (test_content_filter_contract.py):
- Argument parsing for --content-filter and --content-filter-params
  across echo, hz, and bw verbs
- ContentFilterOptions construction from parsed arguments
- None propagation when --content-filter is not provided

Author: PavelGuzenfeld

ros2topic/test/test_bw_delay_hz.py

@@ -43,6 +43,8 @@
 from rclpy.qos import ReliabilityPolicy
 from rclpy.utilities import get_rmw_implementation_identifier
 
+from std_msgs.msg import String
+
 
 # Skip cli tests on Windows while they exhibit pathological behavior
 # https://github.com/ros2/build_farmer/issues/248
@@ -504,3 +506,197 @@ def test_bw_both_all_and_topics_error(self, launch_service, proc_info, proc_outp
         assert 'Cannot specify both --all/-a and topic names' in command.output, (
             'bw command did not print expected error message'
         )
+
+    @launch_testing.markers.retry_on_failure(times=5)
+    def test_hz_content_filter(self, launch_service, proc_info, proc_output):
+        topic = '/clitest/topic/hz_content_filter'
+        publisher = self.node.create_publisher(String, topic, 10)
+        assert publisher
+
+        def publish_message():
+            publisher.publish(String(data='hello'))
+
+        publish_timer = self.node.create_timer(0.5, publish_message)
+
+        # Wait for the publisher to be discovered
+        publisher_count = 0
+        timeout_count = 0
+        while publisher_count == 0 and timeout_count < 10:
+            self.executor.spin_once(timeout_sec=0.1)
+            publisher_count = self.node.count_publishers(topic)
+            timeout_count += 1
+        assert publisher_count > 0, 'Publisher was not discovered'
+
+        try:
+            command_action = ExecuteProcess(
+                cmd=['ros2', 'topic', 'hz',
+                     '--content-filter', "data = 'hello'",
+                     topic],
+                additional_env={
+                    'PYTHONUNBUFFERED': '1'
+                },
+                output='screen'
+            )
+            with launch_testing.tools.launch_process(
+                launch_service, command_action, proc_info, proc_output,
+                output_filter=launch_testing_ros.tools.basic_output_filter(
+                    filtered_rmw_implementation=get_rmw_implementation_identifier()
+                )
+            ) as command:
+                # The future won't complete - we will hit the timeout
+                self.executor.spin_until_future_complete(
+                    rclpy.task.Future(), timeout_sec=5
+                )
+            command.wait_for_shutdown(timeout=10)
+            assert command.output, 'hz with content filter printed no output'
+            assert re.search(
+                r'^average rate: [0-9\.]+$', command.output, flags=re.MULTILINE
+            ), 'hz with content filter did not print expected rate'
+        finally:
+            self.node.destroy_timer(publish_timer)
+            self.node.destroy_publisher(publisher)
+
+    @launch_testing.markers.retry_on_failure(times=5)
+    def test_bw_content_filter(self, launch_service, proc_info, proc_output):
+        topic = '/clitest/topic/bw_content_filter'
+        publisher = self.node.create_publisher(String, topic, 10)
+        assert publisher
+
+        def publish_message():
+            publisher.publish(String(data='hello'))
+
+        publish_timer = self.node.create_timer(0.5, publish_message)
+
+        # Wait for the publisher to be discovered
+        publisher_count = 0
+        timeout_count = 0
+        while publisher_count == 0 and timeout_count < 10:
+            self.executor.spin_once(timeout_sec=0.1)
+            publisher_count = self.node.count_publishers(topic)
+            timeout_count += 1
+        assert publisher_count > 0, 'Publisher was not discovered'
+
+        try:
+            command_action = ExecuteProcess(
+                cmd=['ros2', 'topic', 'bw',
+                     '--content-filter', "data = 'hello'",
+                     topic],
+                additional_env={
+                    'PYTHONUNBUFFERED': '1'
+                },
+                output='screen'
+            )
+            with launch_testing.tools.launch_process(
+                launch_service, command_action, proc_info, proc_output,
+                output_filter=launch_testing_ros.tools.basic_output_filter(
+                    filtered_rmw_implementation=get_rmw_implementation_identifier()
+                )
+            ) as command:
+                # The future won't complete - we will hit the timeout
+                self.executor.spin_until_future_complete(
+                    rclpy.task.Future(), timeout_sec=5
+                )
+            command.wait_for_shutdown(timeout=10)
+            assert command.output, 'bw with content filter printed no output'
+            assert re.search(
+                r'^[0-9]+ B/s from [0-9]+ messages$', command.output, flags=re.MULTILINE
+            ), 'bw with content filter did not print expected bandwidth'
+        finally:
+            self.node.destroy_timer(publish_timer)
+            self.node.destroy_publisher(publisher)
+
+    @launch_testing.markers.retry_on_failure(times=5)
+    def test_hz_content_filter_no_match(self, launch_service, proc_info, proc_output):
+        topic = '/clitest/topic/hz_cfilter_nomatch'
+        publisher = self.node.create_publisher(String, topic, 10)
+        assert publisher
+
+        def publish_message():
+            publisher.publish(String(data='hello'))
+
+        publish_timer = self.node.create_timer(0.5, publish_message)
+
+        # Wait for the publisher to be discovered
+        publisher_count = 0
+        timeout_count = 0
+        while publisher_count == 0 and timeout_count < 10:
+            self.executor.spin_once(timeout_sec=0.1)
+            publisher_count = self.node.count_publishers(topic)
+            timeout_count += 1
+        assert publisher_count > 0, 'Publisher was not discovered'
+
+        try:
+            command_action = ExecuteProcess(
+                cmd=['ros2', 'topic', 'hz',
+                     '--content-filter', "data = 'NOMATCH'",
+                     topic],
+                additional_env={
+                    'PYTHONUNBUFFERED': '1'
+                },
+                output='screen'
+            )
+            with launch_testing.tools.launch_process(
+                launch_service, command_action, proc_info, proc_output,
+                output_filter=launch_testing_ros.tools.basic_output_filter(
+                    filtered_rmw_implementation=get_rmw_implementation_identifier()
+                )
+            ) as command:
+                self.executor.spin_until_future_complete(
+                    rclpy.task.Future(), timeout_sec=5
+                )
+            command.wait_for_shutdown(timeout=10)
+            # No messages should match, so no rate should be reported
+            assert not re.search(
+                r'^average rate:', command.output, flags=re.MULTILINE
+            ), 'hz should not report rate when content filter rejects all messages'
+        finally:
+            self.node.destroy_timer(publish_timer)
+            self.node.destroy_publisher(publisher)
+
+    @launch_testing.markers.retry_on_failure(times=5)
+    def test_bw_content_filter_no_match(self, launch_service, proc_info, proc_output):
+        topic = '/clitest/topic/bw_cfilter_nomatch'
+        publisher = self.node.create_publisher(String, topic, 10)
+        assert publisher
+
+        def publish_message():
+            publisher.publish(String(data='hello'))
+
+        publish_timer = self.node.create_timer(0.5, publish_message)
+
+        # Wait for the publisher to be discovered
+        publisher_count = 0
+        timeout_count = 0
+        while publisher_count == 0 and timeout_count < 10:
+            self.executor.spin_once(timeout_sec=0.1)
+            publisher_count = self.node.count_publishers(topic)
+            timeout_count += 1
+        assert publisher_count > 0, 'Publisher was not discovered'
+
+        try:
+            command_action = ExecuteProcess(
+                cmd=['ros2', 'topic', 'bw',
+                     '--content-filter', "data = 'NOMATCH'",
+                     topic],
+                additional_env={
+                    'PYTHONUNBUFFERED': '1'
+                },
+                output='screen'
+            )
+            with launch_testing.tools.launch_process(
+                launch_service, command_action, proc_info, proc_output,
+                output_filter=launch_testing_ros.tools.basic_output_filter(
+                    filtered_rmw_implementation=get_rmw_implementation_identifier()
+                )
+            ) as command:
+                self.executor.spin_until_future_complete(
+                    rclpy.task.Future(), timeout_sec=5
+                )
+            command.wait_for_shutdown(timeout=10)
+            # No messages should match, so no bandwidth should be reported
+            assert not re.search(
+                r'^[0-9]+ B/s', command.output, flags=re.MULTILINE
+            ), 'bw should not report bandwidth when content filter rejects all messages'
+        finally:
+            self.node.destroy_timer(publish_timer)
+            self.node.destroy_publisher(publisher)

ros2topic/test/test_content_filter_contract.py

@@ -0,0 +1,127 @@
+# Copyright 2026 Open Source Robotics Foundation, Inc.
+#
+# 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.
+
+"""Contract tests for --content-filter argument handling.
+
+These are lightweight unit tests that verify the interface contract between
+CLI argument parsing and ContentFilterOptions construction, without requiring
+a running ROS system.
+"""
+
+import argparse
+import unittest
+from unittest.mock import MagicMock
+from unittest.mock import patch
+
+from rclpy.subscription_content_filter_options import ContentFilterOptions
+
+
+class TestContentFilterArgParsing(unittest.TestCase):
+    """Verify that each verb correctly parses --content-filter args."""
+
+    def _parse_echo_args(self, args_list):
+        from ros2topic.verb.echo import EchoVerb
+        verb = EchoVerb()
+        parser = argparse.ArgumentParser()
+        verb.add_arguments(parser, 'ros2 topic')
+        return parser.parse_args(args_list)
+
+    def _parse_hz_args(self, args_list):
+        from ros2topic.verb.hz import HzVerb
+        verb = HzVerb()
+        parser = argparse.ArgumentParser()
+        verb.add_arguments(parser, 'ros2 topic')
+        return parser.parse_args(args_list)
+
+    def _parse_bw_args(self, args_list):
+        from ros2topic.verb.bw import BwVerb
+        verb = BwVerb()
+        parser = argparse.ArgumentParser()
+        verb.add_arguments(parser, 'ros2 topic')
+        return parser.parse_args(args_list)
+
+    def test_echo_content_filter_arg_present(self):
+        args = self._parse_echo_args([
+            '/topic', '--content-filter', "data = 'hello'"])
+        assert args.content_filter_expr == "data = 'hello'"
+        assert args.content_filter_params == []
+
+    def test_echo_content_filter_arg_absent(self):
+        args = self._parse_echo_args(['/topic'])
+        assert args.content_filter_expr is None
+        assert args.content_filter_params == []
+
+    def test_echo_content_filter_with_params(self):
+        args = self._parse_echo_args([
+            '/topic', '--content-filter', 'data = %0',
+            '--content-filter-params', 'hello'])
+        assert args.content_filter_expr == 'data = %0'
+        assert args.content_filter_params == ['hello']
+
+    def test_echo_content_filter_with_multiple_params(self):
+        args = self._parse_echo_args([
+            '/topic', '--content-filter', 'data BETWEEN %0 AND %1',
+            '--content-filter-params', '10', '20'])
+        assert args.content_filter_expr == 'data BETWEEN %0 AND %1'
+        assert args.content_filter_params == ['10', '20']
+
+    def test_hz_content_filter_arg_present(self):
+        args = self._parse_hz_args([
+            '/topic', '--content-filter', "data = 'hello'"])
+        assert args.content_filter_expr == "data = 'hello'"
+        assert args.content_filter_params == []
+
+    def test_hz_content_filter_arg_absent(self):
+        args = self._parse_hz_args(['/topic'])
+        assert args.content_filter_expr is None
+        assert args.content_filter_params == []
+
+    def test_bw_content_filter_arg_present(self):
+        args = self._parse_bw_args([
+            '/topic', '--content-filter', "data = 'hello'"])
+        assert args.content_filter_expr == "data = 'hello'"
+        assert args.content_filter_params == []
+
+    def test_bw_content_filter_arg_absent(self):
+        args = self._parse_bw_args(['/topic'])
+        assert args.content_filter_expr is None
+        assert args.content_filter_params == []
+
+
+class TestContentFilterOptionsConstruction(unittest.TestCase):
+    """Verify ContentFilterOptions is constructed correctly from parsed args."""
+
+    def test_options_with_expression_only(self):
+        opts = ContentFilterOptions(
+            filter_expression="data = 'hello'",
+            expression_parameters=[])
+        assert opts.filter_expression == "data = 'hello'"
+        assert opts.expression_parameters == []
+
+    def test_options_with_expression_and_params(self):
+        opts = ContentFilterOptions(
+            filter_expression='data = %0',
+            expression_parameters=['hello'])
+        assert opts.filter_expression == 'data = %0'
+        assert opts.expression_parameters == ['hello']
+
+    def test_options_none_when_no_filter(self):
+        """Contract: when content_filter_expr is None, no options are created."""
+        content_filter_expr = None
+        content_filter_options = None
+        if content_filter_expr:
+            content_filter_options = ContentFilterOptions(
+                filter_expression=content_filter_expr,
+                expression_parameters=[])
+        assert content_filter_options is None