Add primitive topic handling for mcap files

Signed-off-by: Pierre R. Mai <pmai@pmsf.de>

Author: pmai

qc_ositrace/checks/deserialization/deserialization_checker.py

@@ -23,7 +23,10 @@ def run_checks(config: Configuration, result: Result) -> None:
     expected_type = OSITrace.map_message_type(expected_type_name)
 
     trace = OSITrace(
-        config.get_config_param("InputFile"), config.get_config_param("osiType")
+        config.get_config_param("InputFile"),
+        config.get_config_param("osiType"),
+        False,
+        config.get_config_param("osiTopic"),
     )
 
     result.register_checker(

qc_ositrace/checks/osirules/osirules_checker.py

@@ -445,7 +445,12 @@ def run_checks(config: Configuration, result: Result) -> None:
     )
     expected_type_name = config.get_config_param("osiType")
 
-    trace = OSITrace(config.get_config_param("InputFile"), expected_type_name)
+    trace = OSITrace(
+        config.get_config_param("InputFile"),
+        expected_type_name,
+        False,
+        config.get_config_param("osiTopic"),
+    )
 
     result.register_checker(
         checker_bundle_name=constants.BUNDLE_NAME,

qc_ositrace/main.py

@@ -28,6 +28,11 @@ def args_entrypoint() -> argparse.Namespace:
         type=pathlib.Path,
         help="Path to the input OSI Trace file.",
     )
+    parser.add_argument(
+        "--osiTopic",
+        type=str,
+        help="Channel topic of a multi-trace OSI Trace file to select.",
+    )
     parser.add_argument(
         "--osiType",
         type=str,
@@ -101,6 +106,9 @@ def main():
         logging.info("Setting input file: %s", args.input_file)
         config.set_config_param("InputFile", str(args.input_file))
 
+    if args.osiTopic:
+        logging.info("Setting OSI Topic: %s", args.osiTopic)
+        config.set_config_param("osiTopic", args.osiTopic)
     if args.osiType:
         logging.info("Setting OSI Type: %s", args.osiType)
         config.set_config_param("osiType", args.osiType)

tests/test_deserialization_checks.py

@@ -58,3 +58,37 @@ def test_deserialization_expected_version(
     launch_main(monkeypatch)
     check_issues(rule_uid, issue_count, issue_severity)
     cleanup_files()
+
+
+@pytest.mark.parametrize(
+    "target_file,target_topic,target_type,target_version,issue_count",
+    [
+        ("360", "MySensorView", "SensorView", "3.5.0", 547),
+        ("360", "MySensorView", "SensorView", "3.6.0", 0),
+        ("360", "Foo", "SensorView", "3.6.0", -1),
+    ],
+)
+def test_deserialization_mcap_topic(
+    target_file: str,
+    target_topic: str,
+    target_type: str,
+    target_version: str,
+    issue_count: int,
+    monkeypatch,
+) -> None:
+    base_path = "tests/data/deserialization_expected_version/"
+    target_file_name = f"deserialization_expected_version_{target_file}.mcap"
+    rule_uid = "asam.net:osi:3.0.0:deserialization.expected_version"
+    issue_severity = IssueSeverity.ERROR
+
+    target_file_path = os.path.join(base_path, target_file_name)
+    create_test_config(
+        target_file_path, target_type, target_version, None, target_topic
+    )
+    if issue_count < 0:
+        with pytest.raises(ValueError):
+            launch_main(monkeypatch)
+    else:
+        launch_main(monkeypatch)
+        check_issues(rule_uid, issue_count, issue_severity)
+    cleanup_files()

tests/test_setup.py

@@ -3,6 +3,8 @@
 
 from typing import Optional
 
+from contextlib import suppress
+
 import qc_ositrace.main as main
 
 from qc_ositrace import constants
@@ -17,6 +19,7 @@ def create_test_config(
     target_file_type: str,
     target_file_version: Optional[str] = None,
     target_file_rules: Optional[str] = None,
+    target_file_topic: Optional[str] = None,
 ):
     test_config = Configuration()
     test_config.set_config_param(name="InputFile", value=target_file_path)
@@ -25,6 +28,8 @@ def create_test_config(
         test_config.set_config_param(name="osiVersion", value=target_file_version)
     if target_file_rules is not None:
         test_config.set_config_param(name="osiRulesFile", value=target_file_rules)
+    if target_file_topic is not None:
+        test_config.set_config_param(name="osiTopic", value=target_file_topic)
     test_config.register_checker_bundle(checker_bundle_name=constants.BUNDLE_NAME)
     test_config.set_checker_bundle_param(
         checker_bundle_name=constants.BUNDLE_NAME,
@@ -61,5 +66,6 @@ def launch_main(monkeypatch):
 
 
 def cleanup_files():
-    os.remove(REPORT_FILE_PATH)
-    os.remove(CONFIG_FILE_PATH)
+    with suppress(Exception):
+        os.remove(REPORT_FILE_PATH)
+        os.remove(CONFIG_FILE_PATH)