Renaming Filter Fixtures

Author: troygibb

README.md

@@ -54,7 +54,7 @@ Each replay test can be authored into its own file, like `my_replay_test.py`. We
 
 Replay testing has three distinct phases, **all of which are required to run a replay test**:
 
-### Fixtures `@fixtures`
+### Filter Fixtures `@fixtures`
 
 For collecting and preparing your fixtures to be run against your launch specification. Duties include:
 - Provides a mechanism for specifying your input fixtures (e.g. `lidar_data.mcap`)
@@ -67,9 +67,9 @@ Here is how you use it:
 
 ```python
 @fixtures.parameterize([McapFixture(path="/tmp/mcap/my_data.mcap")])
-class Fixtures:
-    input_topics = ["/vehicle/cmd_vel"]
-    output_topics = ["/user/cmd_vel"]
+class FilterFixtures:
+    required_input_topics = ["/vehicle/cmd_vel"]
+    expected_output_topics = ["/user/cmd_vel"]
 ```
 
 ### Run `@run`

replay_testing/cli.py

@@ -94,7 +94,7 @@ def run(parser, args):
 
     runner = ReplayTestingRunner(test_module)
 
-    runner.fixtures()
+    runner.filter_fixtures()
     runner.run()
     exit_code, junit_xml_path = runner.analyze()
 

replay_testing/decorators/fixtures.py

@@ -14,6 +14,9 @@
 #
 
 from ..models import ReplayTestingPhase, McapFixture
+from ..logging_config import get_logger
+
+_logger_ = get_logger()
 
 
 class fixtures:
@@ -23,28 +26,25 @@ def __init__(self, *args, **kwargs):
         # If args/kwargs are provided, treat them as parameters
         self.fixture_list = kwargs.get("fixture_list", None)
 
-    def __call__(self, cls):
-        if not hasattr(cls, "input_topics"):
-            raise TypeError(f"Class {cls.__name__} must define a 'input_topics' attribute.")
-
-        if not isinstance(cls.input_topics, list):
-            raise TypeError(f"Class {cls.__name} 'input_topics' attribute must be a list.")
-
-        if not all(isinstance(topic, str) for topic in cls.input_topics):
-            raise TypeError(
-                f"Class {cls.__name} 'input_topics' attribute must be a list of strings."
+    def validate_class_variable(self, cls, prop: str, deprecated_variable: str):
+        if hasattr(cls, deprecated_variable):
+            _logger_.warning(
+                f"Class {cls.__name__} '{prop}' attribute is deprecated. See docs for updated usage."
             )
+            return
 
-        if not hasattr(cls, "output_topics"):
-            raise TypeError(f"Class {cls.__name__} must define a 'output_topics' attribute.")
+        if not hasattr(cls, prop):
+            raise TypeError(f"Class {cls.__name__} must define a '{prop}' attribute.")
 
-        if not isinstance(cls.output_topics, list):
-            raise TypeError(f"Class {cls.__name} 'output_topics' attribute must be a list.")
+        if not isinstance(getattr(cls, prop), list):
+            raise TypeError(f"Class {cls.__name} '{prop}' attribute must be a list.")
 
-        if not all(isinstance(topic, str) for topic in cls.output_topics):
-            raise TypeError(
-                f"Class {cls.__name} 'input_topics' attribute must be a list of strings."
-            )
+        if not all(isinstance(topic, str) for topic in getattr(cls, prop)):
+            raise TypeError(f"Class {cls.__name} '{prop}' attribute must be a list of strings.")
+
+    def __call__(self, cls):
+        self.validate_class_variable(cls, "required_input_topics", "input_topics")
+        self.validate_class_variable(cls, "expected_output_topics", "output_topics")
 
         cls.fixture_list = self.fixture_list
         cls.__annotations__["replay_testing_phase"] = ReplayTestingPhase.FIXTURES

replay_testing/replay_runner.py

@@ -128,7 +128,7 @@ def _create_run_launch_description(
             ]
         )
 
-    def fixtures(self) -> str:
+    def filter_fixtures(self) -> str:
         self._log_stage_start(ReplayTestingPhase.FIXTURES)
 
         # todo: Add exception handling
@@ -153,13 +153,24 @@ def fixtures(self) -> str:
 
             topic_types = reader.get_all_topics_and_types()
 
+            required_input_topics = (
+                fixture.required_input_topics
+                if hasattr(fixture, "required_input_topics")
+                else fixture.input_topics
+            )
+            expected_output_topics = (
+                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 fixture.input_topics:
+                if topic_type.name in required_input_topics:
                     input_topics_present.append(topic_type.name)
 
-            if set(input_topics_present) != set(fixture.input_topics):
-                diff = difflib.ndiff(input_topics_present, fixture.input_topics)
+            if set(input_topics_present) != set(required_input_topics):
+                diff = difflib.ndiff(input_topics_present, required_input_topics)
                 diff_str = "\n".join(diff)
                 _logger_.error(f"Input topics do not match. Diff: {diff_str}")
                 raise AssertionError("Input topics do not match. Check logs for more information")
@@ -174,7 +185,7 @@ def fixtures(self) -> str:
             filter_mcap(
                 replay_fixture.input_fixture.path,
                 replay_fixture.filtered_fixture.path,
-                fixture.output_topics,
+                expected_output_topics,
             )
 
             self._replay_fixtures.append(replay_fixture)

test/replay_tests/basic_replay.py

@@ -32,8 +32,8 @@
 
 @fixtures.parameterize([McapFixture(path=cmd_vel_only_fixture)])
 class Fixtures:
-    input_topics = ["/vehicle/cmd_vel"]
-    output_topics = ["/user/cmd_vel"]
+    required_input_topics = ["/vehicle/cmd_vel"]
+    expected_output_topics = ["/user/cmd_vel"]
 
 
 @run.default()

test/replay_tests/basic_replay_failure.py

@@ -31,9 +31,9 @@
 
 
 @fixtures.parameterize([McapFixture(path=cmd_vel_only_fixture)])
-class Fixtures:
-    input_topics = ["/vehicle/cmd_vel"]
-    output_topics = ["/user/cmd_vel"]
+class FilterFixtures:
+    required_input_topics = ["/vehicle/cmd_vel"]
+    expected_output_topics = ["/user/cmd_vel"]
 
 
 @run.default()

test/replay_tests/basic_replay_multiple_fixtures_and_params.py

@@ -41,8 +41,8 @@
     ]
 )
 class Fixtures:
-    input_topics = ["/vehicle/cmd_vel"]
-    output_topics = ["/user/cmd_vel"]
+    required_input_topics = ["/vehicle/cmd_vel"]
+    expected_output_topics = ["/user/cmd_vel"]
 
 
 @run.parameterize(

test/replay_tests/nexus_replay.py

@@ -22,8 +22,8 @@
 
 @fixtures.parameterize([NexusFixture(path="generic/cmd_vel_only.mcap")])
 class Fixtures:
-    input_topics = ["/vehicle/cmd_vel"]
-    output_topics = ["/user/cmd_vel"]
+    required_input_topics = ["/vehicle/cmd_vel"]
+    expected_output_topics = ["/user/cmd_vel"]
 
 
 @run.default()

test/test_cli.py

@@ -64,38 +64,3 @@ def test_cli_xml_output_success(capsys):
             assert "test_cmd_vel" in tescase.get("name")
             assert tescase.get("classname") == "AnalyzeBasicReplay"
             assert tescase.find("failure") is None
-
-
-@pytest.skip("File created causes colcon test-result to fail")
-def test_cli_xml_output_failure(capsys):
-    xml_path = os.path.join("/tmp", "test.xml")
-    replay_file_path = os.path.join(
-        replay_testing_dir, "test", "replay_tests", "basic_replay_failure.py"
-    )
-
-    # Mock sys.argv for the CLI arguments
-    sys.argv = ["replay_test", replay_file_path, "--junit-xml", xml_path, "--verbose"]
-
-    with pytest.raises(SystemExit) as pytest_wrapped_e:
-        main()
-
-    # Check that exit code is 0 (indicating success)
-    assert pytest_wrapped_e.type is SystemExit
-    assert pytest_wrapped_e.value.code == 1
-
-    junit_xml = ET.parse(xml_path)
-    # Remove file so it doesn't conflict with colcon test-result since this is an expected failure
-    os.remove(xml_path)
-
-    root = junit_xml.getroot()
-
-    # Assert top level name
-    assert root.tag == "testsuites"
-
-    # Assert on test suites within <testsuites> in a loop
-    for testsuite in root.iter("testsuite"):
-        assert testsuite.get("tests") == "1"
-        for tescase in testsuite.iter("testcase"):
-            assert "test_expected_failure" in tescase.get("name")
-            assert tescase.get("classname") == "AnalyzeBasicReplay"
-            assert tescase.find("failure") is not None