From 64b94f9841ee401f9ab837daa381169e268ea6c8 Mon Sep 17 00:00:00 2001 From: pepisg Date: Fri, 9 Jan 2026 20:57:52 +0000 Subject: [PATCH 1/5] allow passing extra args to the test --- replay_testing/cli.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/replay_testing/cli.py b/replay_testing/cli.py index 9ee8e70..0004ee8 100644 --- a/replay_testing/cli.py +++ b/replay_testing/cli.py @@ -123,7 +123,11 @@ def add_arguments(parser): def parse_arguments(): parser = argparse.ArgumentParser(description='replay integration testing tool.') add_arguments(parser) - return parser, parser.parse_args() + # Use parse_known_args to allow extra arguments to pass through to the test file + args, unknown = parser.parse_known_args() + # Store unknown args for potential use by test files + args.unknown_args = unknown + return parser, args def run(parser, args): From 62ca20c2eab1dc0bdc83130246f23c03d8430603 Mon Sep 17 00:00:00 2001 From: pepisg Date: Fri, 6 Feb 2026 19:01:36 +0000 Subject: [PATCH 2/5] pass the run fixture pathto the analyzer --- replay_testing/replay_runner.py | 1 + 1 file changed, 1 insertion(+) diff --git a/replay_testing/replay_runner.py b/replay_testing/replay_runner.py index 6644e9f..2c316be 100644 --- a/replay_testing/replay_runner.py +++ b/replay_testing/replay_runner.py @@ -263,6 +263,7 @@ class AnalyzeWithReader(analyze_cls): def setUp(inner_self): super().setUp() # Call original setUp if it exists inner_self.reader = reader + inner_self.run_fixture_path = run_fixture.path inner_self.suite_classname = analyze_cls.__name__ suite = unittest.TestLoader().loadTestsFromTestCase(AnalyzeWithReader) From 9f98339f2b11d9abe383b7d3d84e64ccdfac781d Mon Sep 17 00:00:00 2001 From: pepisg Date: Fri, 6 Feb 2026 19:22:19 +0000 Subject: [PATCH 3/5] zstd compression for fast rosbags --- replay_testing/replay_runner.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/replay_testing/replay_runner.py b/replay_testing/replay_runner.py index 2c316be..329ed9c 100644 --- a/replay_testing/replay_runner.py +++ b/replay_testing/replay_runner.py @@ -142,7 +142,7 @@ def _create_run_launch_description( # Launch description ld = LaunchDescription([ ExecuteProcess( - cmd=['ros2', 'bag', 'record', '-s', 'mcap', '-o', str(run_fixture.path), '--all'], + cmd=['ros2', 'bag', 'record', '-s', 'mcap', '-o', str(run_fixture.path), '--all', '--storage-preset-profile', 'zstd_fast'], output='screen', ), test_ld, From b0cafccd45e84a030ad682633cb4f464e3096985 Mon Sep 17 00:00:00 2001 From: Sebastian Solarte Date: Thu, 24 Sep 2026 15:33:00 +0000 Subject: [PATCH 4/5] Add some detected improvements to implement a new CM test --- replay_testing/__init__.py | 3 + replay_testing/cli.py | 23 +++- replay_testing/models.py | 6 ++ replay_testing/regenerated_sources.py | 97 +++++++++++++++++ replay_testing/replay_runner.py | 140 +++++++++++++++++++------ replay_testing/wait_for_stack.py | 110 ++++++++++++++++++++ setup.cfg | 1 + test/test_regenerated_sources.py | 144 ++++++++++++++++++++++++++ test/test_replay_runner.py | 42 ++++++++ test/test_wait_for_stack.py | 91 ++++++++++++++++ 10 files changed, 626 insertions(+), 31 deletions(-) create mode 100644 replay_testing/regenerated_sources.py create mode 100644 replay_testing/wait_for_stack.py create mode 100644 test/test_regenerated_sources.py create mode 100644 test/test_wait_for_stack.py diff --git a/replay_testing/__init__.py b/replay_testing/__init__.py index c9b129f..2856c46 100644 --- a/replay_testing/__init__.py +++ b/replay_testing/__init__.py @@ -21,6 +21,7 @@ from .logging_config import get_logger from .models import ReplayRunParams, RunnerArgs from .reader import get_sequential_mcap_reader, read_messages +from .regenerated_sources import RegeneratedSource, ReplaySourcePlan from .replay_runner import ReplayTestingRunner # Alias for backward compatibility. Should be removed in future versions. @@ -33,6 +34,8 @@ 'ReplayTestingRunner', 'get_sequential_mcap_reader', 'read_messages', + 'RegeneratedSource', + 'ReplaySourcePlan', 'ReplayRunParams', 'RunnerArgs', 'unittest_results_to_xml', diff --git a/replay_testing/cli.py b/replay_testing/cli.py index 0004ee8..ca8401a 100644 --- a/replay_testing/cli.py +++ b/replay_testing/cli.py @@ -25,6 +25,10 @@ _logger_ = get_logger() +#: Where --output-dir points, exported for the test module to find. A test that +#: writes artifacts of its own reads this to put them with the results. +OUTPUT_DIR_ENV_VAR = 'REPLAY_TESTING_OUTPUT_DIR' + def _load_python_file_as_module(test_module_name, python_file_path): """Load a given Python replay file (by path) as a Python module.""" @@ -111,6 +115,15 @@ def add_arguments(parser): help='Do write xUnit reports to specified path.', ) + parser.add_argument( + '--output-dir', + action='store', + dest='output_dir', + default=None, + help='Directory to write replay results into. Defaults to test_results under CI ' + 'and the system temp directory otherwise.', + ) + parser.add_argument( '--env', action='store', @@ -148,9 +161,17 @@ def run(parser, args): if not args.package_name: args.package_name = args.replay_test_file.stem + # A test can generate its own artifacts - rewritten parameter files, say - + # and those belong next to the run they describe, not in the system temp + # directory where nothing ties them back to it. The module is imported + # before the runner exists, so the destination is passed through the + # environment, which is the only channel available this early. + if args.output_dir: + os.environ[OUTPUT_DIR_ENV_VAR] = str(Path(args.output_dir).absolute()) + test_module = _load_python_file_as_module(args.package_name, args.replay_test_file.absolute()) - runner = ReplayTestingRunner(test_module, run_id=args.analyze) + runner = ReplayTestingRunner(test_module, run_id=args.analyze, output_dir=args.output_dir) if not args.analyze: runner.filter_fixtures() diff --git a/replay_testing/models.py b/replay_testing/models.py index 4054874..ac72394 100644 --- a/replay_testing/models.py +++ b/replay_testing/models.py @@ -30,6 +30,12 @@ class ReplayTestingPhase(Enum): class RunnerArgs(BaseModel): use_clock: bool = True playback_rate: float = 1.0 + # Hold playback until the nodes under test are publishing. Without it they + # race the player, and on a short fixture the bag can finish before they + # advertise anything. + wait_for_stack: bool = True + wait_for_stack_timeout: float = 30.0 + wait_for_stack_grace: float = 2.0 class ReplayRunParams(BaseModel): diff --git a/replay_testing/regenerated_sources.py b/replay_testing/regenerated_sources.py new file mode 100644 index 0000000..199d937 --- /dev/null +++ b/replay_testing/regenerated_sources.py @@ -0,0 +1,97 @@ +# Copyright (c) 2025-present Polymath Robotics, 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. +# + +"""Declare topics that need to be rebuilt during replay. + +Some topics can't be directly replayed from a bag due to missing data, mismatched types, or differing rates. If the bag contains the underlying inputs, and the transformation node is deterministic, these can be reconstructed offline. +This helper lets tests specify such regenerate-required topics in one place, avoiding manual keeping of three separate lists: recorded inputs needed for rebuild, outputs to strip from the bag, and launch flags to start the rebuilding nodes. Each is derived from a single, annotated declaration. +""" + +from dataclasses import dataclass + + +@dataclass(frozen=True) +class RegeneratedSource: + """An input that must be rebuilt during replay instead of replayed. + + ``topic`` is what the replayed node subscribes to, ``recorded_topics`` what the + bag must carry so it can be rebuilt, and ``launch_argument`` the flag on the + replay launch that starts the nodes rebuilding it. A rebuild can need more + than one recorded input, so this takes a tuple rather than a single topic. + """ + + topic: str + recorded_topics: tuple[str, ...] + launch_argument: str + reason: str + strip_from_replay: bool = True + + +class ReplaySourcePlan: + """Work out what a bag must carry, and what has to be rebuilt. + + Given the topics the replayed nodes subscribe to, this splits them into + topics the bag can replay directly and topics that have to be regenerated + from something else the bag does carry. + """ + + def __init__( + self, + source_topics, + regenerated: tuple[RegeneratedSource, ...] = (), + ): + self.source_topics = set(source_topics) + self._catalogue = {source.topic: source for source in regenerated} + + @property + def regenerated(self) -> list[RegeneratedSource]: + """The sources among the requested ones that must be rebuilt.""" + return [source for topic, source in self._catalogue.items() if topic in self.source_topics] + + @property + def required_topics(self) -> list[str]: + """Topics the bag must carry to answer the question. + + Regenerated sources are swapped for the recorded input they are built + from, since the bag is not expected to carry the derived topic. + """ + topics = set(self.source_topics) + for source in self.regenerated: + topics.discard(source.topic) + topics.update(source.recorded_topics) + return sorted(topics) + + @property + def topics_to_strip(self) -> list[str]: + """Regenerated topics that must not be replayed from the bag. + + A topic that is both replayed and rebuilt would reach the node twice, + mixing the recorded rate with the regenerated one. Stripping a topic the + bag does not carry is a no-op, so every regenerated source is listed. + """ + return sorted(source.topic for source in self.regenerated if source.strip_from_replay) + + @property + def launch_arguments(self) -> dict[str, str]: + """Replay launch flags, one per regeneration, as launch string bools.""" + enabled = {source.launch_argument for source in self.regenerated} + return {source.launch_argument: str(source.launch_argument in enabled) for source in self._catalogue.values()} + + def describe(self) -> list[str]: + """One human readable line per regeneration, for the run log.""" + return [ + f'{source.topic}: {source.reason}; rebuilt from {", ".join(source.recorded_topics)}' + for source in self.regenerated + ] diff --git a/replay_testing/replay_runner.py b/replay_testing/replay_runner.py index 329ed9c..67ef6a8 100644 --- a/replay_testing/replay_runner.py +++ b/replay_testing/replay_runner.py @@ -45,7 +45,7 @@ class ReplayTestingRunner: _replay_results_directory: Path _replay_fixtures: list[ReplayFixture] - def __init__(self, test_module, *, run_id: Optional[str] = None): + def __init__(self, test_module, *, run_id: Optional[str] = None, output_dir: Optional[str] = None): self._replay_fixtures = [] self._test_module = test_module @@ -55,9 +55,21 @@ def __init__(self, test_module, *, run_id: Optional[str] = None): else: self._test_run_uuid = uuid.uuid4() - result_base = Path('test_results') if os.environ.get('CI') else Path(tempfile.gettempdir()) + if output_dir: + result_base = Path(output_dir) + elif os.environ.get('CI'): + result_base = Path('test_results') + else: + result_base = Path(tempfile.gettempdir()) self._replay_directory = result_base / 'replay_testing' + # os.access() is False for a path that does not exist yet, so without + # this the probe below would silently redirect every run to ./test_results. + try: + self._replay_directory.mkdir(parents=True, exist_ok=True) + except OSError as e: + _logger_.error(f'Could not create {self._replay_directory}: {e}') + if not os.access(self._replay_directory, os.W_OK): _logger_.error( f'No write permission for directory: {self._replay_directory}. Setting to ./test_results/replay_testing' @@ -114,7 +126,13 @@ def _get_prev_run_fixtures(self) -> list[ReplayFixture]: return replay_fixture_list def _create_run_launch_description( - self, filtered_fixture, run_fixture, test_ld: launch.LaunchDescription, run, params: ReplayRunParams + self, + filtered_fixture, + run_fixture, + test_ld: launch.LaunchDescription, + run, + params: ReplayRunParams, + expected_output_topics: Optional[list] = None, ) -> launch.LaunchDescription: # Define the process action for playing the MCAP file cmd = [ @@ -139,15 +157,34 @@ def _create_run_launch_description( output='screen', ) - # Launch description - ld = LaunchDescription([ - ExecuteProcess( - cmd=['ros2', 'bag', 'record', '-s', 'mcap', '-o', str(run_fixture.path), '--all', '--storage-preset-profile', 'zstd_fast'], - output='screen', - ), - test_ld, - player_action, # Add the MCAP playback action - ]) + recorder_action = ExecuteProcess( + cmd=[ + 'ros2', + 'bag', + 'record', + '-s', + 'mcap', + '-o', + str(run_fixture.path), + '--all', + '--storage-preset-profile', + 'zstd_fast', + ], + output='screen', + ) + + # Hold the player until the stack under test is publishing, so a short + # fixture cannot finish before its output topics even exist. + gate = self._wait_for_stack_action(expected_output_topics, params) + if gate is not None: + start_playback = [ + gate, + RegisterEventHandler(OnProcessExit(target_action=gate, on_exit=[player_action])), + ] + else: + start_playback = [player_action] + + ld = LaunchDescription([recorder_action, test_ld, *start_playback]) if not params.ignore_playback_finish: # Event handler to gracefully exit when the process finishes @@ -163,6 +200,29 @@ def _create_run_launch_description( return ld + @staticmethod + def _wait_for_stack_action(expected_output_topics, params: ReplayRunParams): + """The process that gates playback, or None when not gating.""" + runner_args = params.runner_args + if not runner_args.wait_for_stack: + return None + + return ExecuteProcess( + cmd=[ + 'ros2', + 'run', + 'replay_testing', + 'replay_test_wait_for_stack', + *expected_output_topics, + '--timeout', + str(runner_args.wait_for_stack_timeout), + '--grace', + str(runner_args.wait_for_stack_grace), + ], + name='wait_for_stack', + output='screen', + ) + def filter_fixtures(self) -> list[ReplayFixture]: self._log_stage_start(ReplayTestingPhase.FIXTURES) @@ -193,23 +253,18 @@ def filter_fixtures(self) -> list[ReplayFixture]: fixture.expected_output_topics if hasattr(fixture, 'expected_output_topics') else fixture.output_topics ) - input_topics_present = [] - for topic_type in topic_types: - if topic_type.name in required_input_topics: - input_topics_present.append(topic_type.name) - - if set(input_topics_present) != set(required_input_topics): - missing_topics = set(required_input_topics) - set(input_topics_present) - extra_topics = set(input_topics_present) - set(required_input_topics) + declared = {topic_type.name for topic_type in topic_types} + missing_topics = set(required_input_topics) - declared - error_msg = 'Input topics do not match:' - if missing_topics: - error_msg += f'\n Missing topics: {sorted(missing_topics)}' - if extra_topics: - error_msg += f'\n Extra topics: {sorted(extra_topics)}' + if missing_topics: + _logger_.error(f'Required input topics not in the bag: {sorted(missing_topics)}') + raise AssertionError('Required input topics missing. Check logs for more information') - _logger_.error(error_msg) - raise AssertionError('Input topics do not match. Check logs for more information') + # A topic may appear in the bag without messages. This isn't fatal: + # it means a silent sensor was correctly recorded. Only later analysis decides if that's an issue. + empty_topics = self._topics_without_messages(replay_fixture, sorted(set(required_input_topics) & declared)) + if empty_topics: + _logger_.warning(f'Required input topics declared but empty: {sorted(empty_topics)}') replay_fixture.filter_input(expected_output_topics) @@ -219,12 +274,32 @@ def filter_fixtures(self) -> list[ReplayFixture]: return self._replay_fixtures + @staticmethod + def _topics_without_messages(replay_fixture: ReplayFixture, topics: list[str]) -> list[str]: + """Which of these topics are declared in the bag but carry no messages.""" + if not topics: + return [] + + counts = dict.fromkeys(topics, 0) + reader = replay_fixture.get_reader(FixtureType.INPUT) + while reader.has_next(): + topic_name, _, _ = reader.read_next() + if topic_name in counts: + counts[topic_name] += 1 + + return [topic for topic, count in counts.items() if count == 0] + def run(self): self._log_stage_start(ReplayTestingPhase.RUN) run_cls = self._get_stage_class(ReplayTestingPhase.RUN) run = run_cls() + fixture_cls = self._get_stage_class(ReplayTestingPhase.FIXTURES) + expected_output_topics = getattr( + fixture_cls, 'expected_output_topics', getattr(fixture_cls, 'output_topics', []) + ) + for replay_fixture in self._replay_fixtures: if len(run.parameters) == 0: raise ValueError('No parameters found for run') @@ -238,7 +313,12 @@ def run(self): test_launch_description = run.generate_launch_description(param) ld = self._create_run_launch_description( - replay_fixture.filtered_fixture, run_fixture, test_launch_description, run, param + replay_fixture.filtered_fixture, + run_fixture, + test_launch_description, + run, + param, + expected_output_topics, ) launch_service = launch.LaunchService() launch_service.include_launch_description(ld) @@ -257,12 +337,12 @@ def analyze(self, *, write_junit: bool = True) -> tuple[int, Path]: analyze_cls: type[Any] = self._get_stage_class(ReplayTestingPhase.ANALYZE) for run_fixture in replay_fixture.run_fixtures: - reader = get_sequential_mcap_reader(run_fixture.path) class AnalyzeWithReader(analyze_cls): def setUp(inner_self): super().setUp() # Call original setUp if it exists - inner_self.reader = reader + # Each test needs a fresh reader since readers can't rewind. + inner_self.reader = get_sequential_mcap_reader(run_fixture.path) inner_self.run_fixture_path = run_fixture.path inner_self.suite_classname = analyze_cls.__name__ diff --git a/replay_testing/wait_for_stack.py b/replay_testing/wait_for_stack.py new file mode 100644 index 0000000..de96aaf --- /dev/null +++ b/replay_testing/wait_for_stack.py @@ -0,0 +1,110 @@ +# Copyright (c) 2025-present Polymath Robotics, 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. +# + +"""Block until the stack under test is up, then let the replay start. + +The recorder, test launch, and player all start together. If the stack under test isn't ready, the bag may finish before key topics exist—causing empty output and hard-to-diagnose failures. + +We can't just wait for publishers on the expected output topics. Nodes running with sim time won't publish until `/clock` arrives (which only happens after the player starts), creating a potential deadlock. However, these nodes subscribe to `/clock` as soon as they're up, signaling readiness. + +This function waits for the first of: +* All expected output topics have publishers (stack uses wall-clock time, already running). +* A `/clock` subscription appears (stack uses sim time, waiting for the player). We then briefly wait for any remaining publishers. +""" + +import argparse +import sys +import time + +import rclpy +from rclpy.node import Node + +from .logging_config import get_logger + +_logger_ = get_logger() + +CLOCK_TOPIC = '/clock' + + +def wait_for_stack( + topics: list[str], + timeout: float, + grace: float = 2.0, + poll_period: float = 0.1, +) -> tuple[bool, list[str]]: + """Wait until the stack under test is up. + + Returns (saw_clock_subscriber, topics still without a publisher). + """ + node = Node('replay_testing_wait_for_stack') + pending = list(topics) + saw_clock = False + + def poll(): + # The graph view only refreshes while the node is spun, so polling it + # without spinning never sees anything appear. + rclpy.spin_once(node, timeout_sec=poll_period) + return [t for t in pending if not node.get_publishers_info_by_topic(t)] + + try: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + pending = poll() + if topics and not pending: + return False, [] + if node.get_subscriptions_info_by_topic(CLOCK_TOPIC): + saw_clock = True + break + + if not saw_clock: + return False, pending + + # Sim-time nodes are up; give any wall-clock outputs a moment to appear. + grace_deadline = min(deadline, time.monotonic() + grace) + while pending and time.monotonic() < grace_deadline: + pending = poll() + finally: + node.destroy_node() + + return saw_clock, pending + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('topics', nargs='*', help='Topics expected to be published') + parser.add_argument('--timeout', type=float, default=30.0) + parser.add_argument('--grace', type=float, default=2.0) + args = parser.parse_args(argv) + + rclpy.init() + try: + saw_clock, missing = wait_for_stack(args.topics, args.timeout, args.grace) + finally: + if rclpy.ok(): + rclpy.shutdown() + + if not saw_clock and missing: + # Not fatal: a test may legitimately declare an output it only reaches + # once the replay is under way, and failing here would hide that behind + # a launch error instead of the analyze assertion that explains it. + _logger_.warning( + f'Starting playback without a ready stack; no publisher on {sorted(missing)} ' + f'and nothing subscribed to {CLOCK_TOPIC}' + ) + return 0 + + +if __name__ == '__main__': + sys.exit(main()) diff --git a/setup.cfg b/setup.cfg index ac81dc7..a1338bd 100644 --- a/setup.cfg +++ b/setup.cfg @@ -1,3 +1,4 @@ [options.entry_points] console_scripts = replay_test = replay_testing.cli:main + replay_test_wait_for_stack = replay_testing.wait_for_stack:main diff --git a/test/test_regenerated_sources.py b/test/test_regenerated_sources.py new file mode 100644 index 0000000..fa79cd0 --- /dev/null +++ b/test/test_regenerated_sources.py @@ -0,0 +1,144 @@ +# Copyright (c) 2025-present Polymath Robotics, 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. +# + +"""Unit tests for declaring topics that must be rebuilt during a replay.""" + +import pytest + +from replay_testing import RegeneratedSource, ReplaySourcePlan + +# A stand-in for a topic the bag does not carry, rebuilt from one that it does. +DERIVED = RegeneratedSource( + topic='/filtered_cloud', + recorded_topics=('/raw_cloud',), + launch_argument='regenerate_filtered_cloud', + reason='not recorded; only the raw cloud is', +) + +# A stand-in for one that needs several recorded inputs and whose recorded copy +# must survive the replay, because it is the reference the rebuild is judged on. +REBUILT = RegeneratedSource( + topic='/derived_map', + recorded_topics=('/image', '/depth'), + launch_argument='regenerate_derived_map', + reason='recorded too slowly to drive the node', + strip_from_replay=False, +) + +CATALOGUE = (DERIVED, REBUILT) + + +class TestDirectlyReplayable: + def test_plain_topics_are_required_as_they_are(self): + plan = ReplaySourcePlan(['/scan', '/odom'], regenerated=CATALOGUE) + + assert plan.required_topics == ['/odom', '/scan'] + assert plan.regenerated == [] + assert plan.topics_to_strip == [] + + def test_every_known_flag_is_reported_off(self): + # The launch declares them all, so all of them must be passed rather + # than left to defaults that could drift. + plan = ReplaySourcePlan(['/scan'], regenerated=CATALOGUE) + + assert plan.launch_arguments == { + 'regenerate_filtered_cloud': 'False', + 'regenerate_derived_map': 'False', + } + + def test_nothing_to_describe(self): + assert ReplaySourcePlan(['/scan'], regenerated=CATALOGUE).describe() == [] + + def test_an_empty_catalogue_leaves_everything_alone(self): + plan = ReplaySourcePlan(['/filtered_cloud']) + + assert plan.required_topics == ['/filtered_cloud'] + assert plan.launch_arguments == {} + + +class TestRebuiltFromOneInput: + def test_the_recorded_input_is_required_instead(self): + plan = ReplaySourcePlan(['/filtered_cloud', '/scan'], regenerated=CATALOGUE) + + assert plan.required_topics == ['/raw_cloud', '/scan'] + + def test_the_derived_topic_is_stripped_from_the_replay(self): + # Replaying it as well would let the recorded copy shadow the rebuilt one. + plan = ReplaySourcePlan(['/filtered_cloud'], regenerated=CATALOGUE) + + assert plan.topics_to_strip == ['/filtered_cloud'] + + def test_only_its_own_flag_turns_on(self): + plan = ReplaySourcePlan(['/filtered_cloud'], regenerated=CATALOGUE) + + assert plan.launch_arguments['regenerate_filtered_cloud'] == 'True' + assert plan.launch_arguments['regenerate_derived_map'] == 'False' + + def test_describe_names_both_ends(self): + (line,) = ReplaySourcePlan(['/filtered_cloud'], regenerated=CATALOGUE).describe() + + assert '/filtered_cloud' in line + assert '/raw_cloud' in line + assert 'not recorded' in line + + +class TestRebuiltFromSeveralInputs: + def test_all_recorded_inputs_are_required(self): + plan = ReplaySourcePlan(['/derived_map'], regenerated=CATALOGUE) + + assert plan.required_topics == ['/depth', '/image'] + + def test_a_reference_copy_is_kept_in_the_replay(self): + # strip_from_replay=False: the recorded copy is what the rebuild is + # measured against, so it has to survive. + plan = ReplaySourcePlan(['/derived_map'], regenerated=CATALOGUE) + + assert plan.topics_to_strip == [] + + +class TestCombined: + def test_both_rebuilds_coexist(self): + plan = ReplaySourcePlan(['/filtered_cloud', '/derived_map', '/scan'], regenerated=CATALOGUE) + + assert plan.required_topics == ['/depth', '/image', '/raw_cloud', '/scan'] + assert plan.topics_to_strip == ['/filtered_cloud'] + assert plan.launch_arguments == { + 'regenerate_filtered_cloud': 'True', + 'regenerate_derived_map': 'True', + } + assert len(plan.describe()) == 2 + + def test_duplicate_requests_collapse(self): + plan = ReplaySourcePlan( + ['/scan', '/scan', '/filtered_cloud', '/filtered_cloud'], + regenerated=CATALOGUE, + ) + + assert plan.required_topics == ['/raw_cloud', '/scan'] + assert len(plan.regenerated) == 1 + + +@pytest.mark.parametrize('strip', [True, False]) +def test_strip_flag_is_carried_through(strip): + source = RegeneratedSource( + topic='/x', + recorded_topics=('/y',), + launch_argument='f', + reason='r', + strip_from_replay=strip, + ) + plan = ReplaySourcePlan(['/x'], regenerated=(source,)) + + assert plan.topics_to_strip == (['/x'] if strip else []) diff --git a/test/test_replay_runner.py b/test/test_replay_runner.py index 2863dfb..e5c89ea 100644 --- a/test/test_replay_runner.py +++ b/test/test_replay_runner.py @@ -36,6 +36,8 @@ cmd_vel_only_fixture = fixtures_dir / 'cmd_vel_only.mcap' cmd_vel_only_2_fixture = fixtures_dir / 'cmd_vel_only_2.mcap' +# Carries /vehicle/cmd_vel plus /sensor/silent, declared with no messages. +cmd_vel_and_empty_fixture = fixtures_dir / 'cmd_vel_and_empty.mcap' pub_cmd_vel = [ 'ros2', @@ -399,3 +401,43 @@ def test_cmd_vel(self): assert exit_code == 1 return + + +def test_a_declared_but_empty_input_is_reported_not_fatal(): + """A recorded topic that produced nothing is still a faithful recording. + + A sensor can legitimately fall silent for the whole window, and replaying + that silence is what the robot saw. Only the test knows which of its inputs + are worthless when empty, so this stage reports it and carries on. + """ + test_module = types.ModuleType('test_module') + + @fixtures.parameterize([LocalFixture(path=cmd_vel_and_empty_fixture)]) + class Fixtures: + required_input_topics = ['/vehicle/cmd_vel', '/sensor/silent'] + expected_output_topics = [] + + test_module.Fixtures = Fixtures + runner = ReplayTestingRunner(test_module) + + replay_fixtures = runner.filter_fixtures() + + assert len(replay_fixtures) == 1 + reader = get_sequential_mcap_reader(replay_fixtures[0].filtered_fixture.path) + assert '/vehicle/cmd_vel' in [topic.name for topic in reader.get_all_topics_and_types()] + + +def test_an_input_missing_from_the_bag_still_raises(): + """Absent is different from silent: the bag cannot answer the question.""" + test_module = types.ModuleType('test_module') + + @fixtures.parameterize([LocalFixture(path=cmd_vel_and_empty_fixture)]) + class Fixtures: + required_input_topics = ['/vehicle/cmd_vel', '/not_in_the_bag'] + expected_output_topics = [] + + test_module.Fixtures = Fixtures + runner = ReplayTestingRunner(test_module) + + with pytest.raises(AssertionError): + runner.filter_fixtures() diff --git a/test/test_wait_for_stack.py b/test/test_wait_for_stack.py new file mode 100644 index 0000000..1502f91 --- /dev/null +++ b/test/test_wait_for_stack.py @@ -0,0 +1,91 @@ +# Copyright (c) 2025-present Polymath Robotics, 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. +# + +import time + +import pytest +import rclpy +from geometry_msgs.msg import Twist +from rclpy.node import Node +from rosgraph_msgs.msg import Clock + +from replay_testing.wait_for_stack import CLOCK_TOPIC, wait_for_stack + + +@pytest.fixture +def ros(): + rclpy.init() + yield + if rclpy.ok(): + rclpy.shutdown() + + +@pytest.fixture +def helper(ros): + """A node standing in for the stack under test.""" + node = Node('wait_for_stack_helper') + yield node + node.destroy_node() + + +def test_returns_when_outputs_are_published(helper): + """A stack on wall-clock time is ready as soon as its outputs exist.""" + helper.create_publisher(Twist, '/stack_output', 1) + + start = time.monotonic() + saw_clock, pending = wait_for_stack(['/stack_output'], timeout=10.0) + + assert pending == [] + assert saw_clock is False + assert time.monotonic() - start < 10.0 + + +def test_returns_when_a_node_subscribes_to_the_clock(helper): + """A stack on sim time is ready once it is waiting for the player. + + Its outputs cannot appear first: nothing publishes until `/clock` does, + and `/clock` only arrives with the player this wait is gating. + """ + helper.create_subscription(Clock, CLOCK_TOPIC, lambda _: None, 1) + + start = time.monotonic() + saw_clock, pending = wait_for_stack(['/never_published'], timeout=10.0, grace=0.5) + elapsed = time.monotonic() - start + + assert saw_clock is True + assert pending == ['/never_published'] + # The clock subscriber releases the wait; only the grace period is spent + # on the output that will never appear before playback. + assert elapsed < 10.0 + + +def test_times_out_when_nothing_comes_up(ros): + """Neither signal: report the missing outputs instead of hanging.""" + start = time.monotonic() + saw_clock, pending = wait_for_stack(['/absent'], timeout=1.0) + + assert saw_clock is False + assert pending == ['/absent'] + assert time.monotonic() - start >= 1.0 + + +def test_waits_for_the_clock_when_no_outputs_are_declared(helper): + """With nothing declared, the clock subscription is the only signal.""" + helper.create_subscription(Clock, CLOCK_TOPIC, lambda _: None, 1) + + saw_clock, pending = wait_for_stack([], timeout=10.0, grace=0.1) + + assert saw_clock is True + assert pending == [] From 67f427e1392ab4a0d8be550617585ac5ede45548 Mon Sep 17 00:00:00 2001 From: Sebastian Solarte Date: Thu, 24 Sep 2026 16:12:17 +0000 Subject: [PATCH 5/5] Add extra mcap for tests --- test/fixtures/cmd_vel_and_empty.mcap | Bin 0 -> 10104 bytes 1 file changed, 0 insertions(+), 0 deletions(-) create mode 100644 test/fixtures/cmd_vel_and_empty.mcap diff --git a/test/fixtures/cmd_vel_and_empty.mcap b/test/fixtures/cmd_vel_and_empty.mcap new file mode 100644 index 0000000000000000000000000000000000000000..e6509251e3b5e706748f38938b873f719f45afa3 GIT binary patch literal 10104 zcmeI2dvp|69mh8hAZ!v|MV@WBfg~I-*$s)MX@H~jp=iY*$$=JZm)V`WyCb_Z%giKc z^i*SymeZmImA15{Vqa8?6qJ`xrLO>@typWZP*0DZ7W#5K%xOEOfedrr`bgp zc6_W&ygPpZFgi_iW-U+1>k6*)Jn4yUnR8N>-zzMeo?lAg6%5K%(sfo!n>>9-SiUE6 zmMsl8$eUqjx^i?~FkIh@MWah(%6Hrrq2ipgRmA^K_S91#gRbNZN}C?tpgi&uzB`BX z;HoUBI6Rp*Y~M;5xm>R>t#nYgzVNe#O}7^uDux(EpI=OgQqA=ky(D%j>$uVs21l2| zajlGHbL6F2DnCQ`u3>v=$IT1k7==R4qR3RJc&Y~B*aYQ!UB@?k%dw3dHL85UvFY*O zaTh&tI_DVv><@|SidVhGtIxe(~<}<%PvT8P0dvR0GtFik(N@Pd8HIqBf5vSZO?mzDD^JHhF=Y3d1Hyp#HJ zn4h4@6zenW?ZIf!leJP=B2g6w8Rn|?JnC1J9WzElfNR;AaCk^q=KEDq|K-_H?ldBK zMurEhVo`g7MpY4H(3ENF;wBAGu56=Qvc6yN+Tw9jdRE4c@o*h;+)O;EgB%r0ipGtV z2`az@mvt=b=W}6~qQX+8TWrM*qT9$>rf|BcYN+0MTxn@$&n|8lp+$Egfp=8Nxvqdj zs4WD&g@zX?mNzwsS@Mz@v1nJAin=vgzC?}wC6SWID7ub!ORA-ujHmKuf|_U<`K3&^ zg=8%KoS@U2iB*o5D7a4AqCAp!WF-B3r(mTL+=>%f!^^VwkR<<;D+&Jb27AO#xJj=;mHJzp(}&f|J&aH)Z9~tf|nhXQGX`X(X(F*-jfYCm09t{L@0{P_R{1#0YF*ACH)x@FYa00jl= z#p$buYDOLVeZvNxW)pe~*`IjR{I zzig z`;L#$jJme6@7s)m0`;t{9HkjGr*!8%jDiBS^X_@0HKV#7dFft8L4i8+r~DYrsOC+h z?`ISgsQS`P=jlcrxa@%dr4*>O_Z}Im8Fl@J?#+yX0<~~=$vDj@Z~W#583hIEM0?A4 z&8QjcfBhXsL4i6l(wv|fb!>0NLyUp~^_$XLCu&CZ>~7n_C@4_BT>XozG&Hnno-HO-rmM2C{T00xn_!H z)B|Jdzt1QrPpK_)1?v29Vwz@Dddm|#83hHZ zra&8Y1?f7!(-C{QO}=&RF=Nn|%k5N#d=DzvKnVL~O{ih#c6cnfv4}bB4no(14p7vu#L4msT z8`~N*qjvUp?q?JfsEx~i*Qgma_s+GCG71XR^u(mNX4KxLdmdvH6sX&;xuQuk>c+*d zKh7v9P=7q;&(e&#{ff~~G71XRidXJv){LsSZ9zYypg^_WbEHKx>Xxa!2N(qf>W1#2 z7imT<`pkC^G71XRjR%`&YevXXpHs zQBa_^?K*UcW)#lXpg_%g>aF*@%Qi8Rp_Hd3z~8v0_lbGNdwC0WDAbmXva z_*G?NbMK6@u>+T^2;*a$>}h3VFROKU7>*66`;?7cpql*C`td^S>(_tLHSInUZT(OWbFajdkeU5n(tMiw`OrE4nef zpG^+SuIrVJwb84}#sa2;x6a97m9byhSPIM-9fo7RzFygwRKKlk%zQsPCXA2i>2u1) z9P%UQh2fa!ZBaI6W8-+Qo*br3Us5*aH7Avg$;Z+Ugz+&$ctY8DjW_b~GC90!Hz*q~ z#Io^WINn;evhiv;rfiJ=mrn@eW02maY>cFn)EPL2yRRr4W6~*QW8hdmDNGN&`=GMX z1!t+F?vZ>b4mWttjWaJC`VF3Qdwh~Sc+UMFKePxxml!WIMnn>BIDPibI^w;#Dx_A0l&Wb?1P}0xiZJzXTjx#cd%J%DgP$tD literal 0 HcmV?d00001