ci+test: Python quality baseline for the shared bridge (ruff + pytest), fix ActionEvent parser crash

.github/workflows/python-quality.yml

@@ -0,0 +1,49 @@
+name: Python Quality
+
+on:
+  pull_request:
+    paths:
+      - "bridge/common/zenoh_bridge/**"
+      - "scripts/**"
+      - "pyproject.toml"
+      - ".github/workflows/python-quality.yml"
+  push:
+    branches: [main]
+    paths:
+      - "bridge/common/zenoh_bridge/**"
+      - "scripts/**"
+      - "pyproject.toml"
+      - ".github/workflows/python-quality.yml"
+  workflow_dispatch:
+
+concurrency:
+  group: python-quality-${{ github.ref }}
+  cancel-in-progress: true
+
+permissions:
+  contents: read
+
+jobs:
+  quality:
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+
+      - name: Install tooling
+        run: python -m pip install --quiet ruff pytest
+
+      - name: Lint (ruff check)
+        run: python -m ruff check .
+
+      - name: Format (ruff format --check)
+        run: python -m ruff format --check .
+
+      - name: Unit tests (pytest)
+        run: python -m pytest -v

bridge/common/zenoh_bridge/tests/test_action_event.py

@@ -0,0 +1,71 @@
+import json
+
+import action_event
+import pytest
+from action_event import ActionEvent, parse_action_event
+
+
+class TestParseActionEvent:
+    def test_parses_valid_event(self):
+        raw = json.dumps(
+            {
+                "payload": {"action": "move_forward", "params": {"speed": 0.5}},
+                "transaction_details": {"hash": "0xabc"},
+                "timestamp": "2026-01-01T00:00:00Z",
+            }
+        ).encode()
+        event = parse_action_event(raw)
+        assert isinstance(event, ActionEvent)
+        assert event.action == "move_forward"
+        assert event.params == {"speed": 0.5}
+        assert event.timestamp == "2026-01-01T00:00:00Z"
+
+    def test_defaults_action_to_stop_when_payload_empty(self):
+        raw = json.dumps({"payload": {}}).encode()
+        event = parse_action_event(raw)
+        assert event == ActionEvent(action="stop", params={}, timestamp="")
+
+    def test_defaults_when_payload_field_missing(self):
+        raw = json.dumps({"transaction_details": {}}).encode()
+        event = parse_action_event(raw)
+        assert event == ActionEvent(action="stop", params={}, timestamp="")
+
+    def test_preserves_params_when_provided(self):
+        raw = json.dumps({"payload": {"params": {"x": 1, "y": 2}}}).encode()
+        event = parse_action_event(raw)
+        assert event.params == {"x": 1, "y": 2}
+        assert event.action == "stop"
+
+    def test_timestamp_defaults_to_empty_string(self):
+        raw = json.dumps({"payload": {"action": "sit"}}).encode()
+        event = parse_action_event(raw)
+        assert event.timestamp == ""
+
+    def test_timestamp_is_taken_from_top_level_event(self):
+        raw = json.dumps(
+            {"payload": {"action": "stand"}, "timestamp": "2026-02-02T00:00:00Z"}
+        ).encode()
+        event = parse_action_event(raw)
+        assert event.timestamp == "2026-02-02T00:00:00Z"
+
+    @pytest.mark.parametrize("raw", [b"", b"{", b"not json", b"[]", b"null", b"42"])
+    def test_malformed_or_non_object_json_returns_none(self, raw):
+        assert parse_action_event(raw) is None
+
+    def test_non_dict_payload_returns_none(self):
+        raw = json.dumps({"payload": "oops"}).encode()
+        assert parse_action_event(raw) is None
+
+    def test_non_dict_params_default_to_empty_dict(self):
+        raw = json.dumps({"payload": {"action": "wave", "params": []}}).encode()
+        event = parse_action_event(raw)
+        assert event == ActionEvent(action="wave", params={}, timestamp="")
+
+    def test_none_payload_is_treated_as_empty(self):
+        raw = json.dumps({"payload": None}).encode()
+        event = parse_action_event(raw)
+        assert event == ActionEvent(action="stop", params={}, timestamp="")
+
+    def test_exports_match_module_public_names(self):
+        assert action_event.ActionEvent is ActionEvent
+        assert action_event.parse_action_event is parse_action_event

bridge/common/zenoh_bridge/tests/test_utils.py

@@ -0,0 +1,23 @@
+import utils
+
+
+class TestClamp:
+    def test_returns_value_within_bounds(self):
+        assert utils.clamp(0.5, 0.0, 1.0) == 0.5
+
+    def test_clamps_below_lower_bound(self):
+        assert utils.clamp(-1.0, 0.0, 1.0) == 0.0
+
+    def test_clamps_above_upper_bound(self):
+        assert utils.clamp(2.0, 0.0, 1.0) == 1.0
+
+    def test_lower_bound_is_inclusive(self):
+        assert utils.clamp(0.0, 0.0, 1.0) == 0.0
+
+    def test_upper_bound_is_inclusive(self):
+        assert utils.clamp(1.0, 0.0, 1.0) == 1.0
+
+    def test_negative_bounds(self):
+        assert utils.clamp(-5.0, -10.0, -2.0) == -5.0
+        assert utils.clamp(-20.0, -10.0, -2.0) == -10.0
+        assert utils.clamp(0.0, -10.0, -2.0) == -2.0

bridge/common/zenoh_bridge/zenoh_bridge/action_event.py

@@ -1,13 +1,14 @@
 """Parse Fabric tunnel Action Event payloads."""
+
 import json
 from dataclasses import dataclass, field
-from typing import Any, Dict, Optional
+from typing import Any, Optional
 
 
 @dataclass
 class ActionEvent:
     action: str
-    params: Dict[str, Any] = field(default_factory=dict)
+    params: dict[str, Any] = field(default_factory=dict)
     timestamp: str = ""
 
 
@@ -29,6 +30,9 @@ def parse_action_event(raw: bytes) -> Optional[ActionEvent]:
     except (json.JSONDecodeError, UnicodeDecodeError):
         return None
 
+    if not isinstance(event, dict):
+        return None
+
     payload = event.get("payload") or {}
     if not isinstance(payload, dict):
         return None

bridge/common/zenoh_bridge/zenoh_bridge/zenoh_subscriber.py

@@ -1,19 +1,20 @@
 """Zenoh session and subscriber helper."""
-from typing import Callable, List
+
+from collections.abc import Callable
+from typing import Any
+
 import zenoh
 
 
 class ZenohSubscriberHelper:
     """Manages a Zenoh session with one or more topic subscriptions."""
 
     def __init__(self, listen_endpoint: str = "tcp/127.0.0.1:7447"):
-        conf = zenoh.Config.from_json5(
-            f'{{"listen":{{"endpoints":["{listen_endpoint}"]}}}}'
-        )
+        conf = zenoh.Config.from_json5(f'{{"listen":{{"endpoints":["{listen_endpoint}"]}}}}')
         self._session = zenoh.open(conf)
-        self._subs: List = []
+        self._subs: list[Any] = []
 
-    def subscribe(self, topic: str, callback: Callable) -> None:
+    def subscribe(self, topic: str, callback: Callable[..., Any]) -> None:
         """Subscribe to a Zenoh topic with the given callback."""
         sub = self._session.declare_subscriber(topic, callback)
         self._subs.append(sub)

pyproject.toml

@@ -0,0 +1,31 @@
+[tool.ruff]
+# RoboPay Python quality baseline.
+#
+# Scope: this covers the pure-logic modules that can be linted and tested
+# without a ROS 2 runtime. ROS 2 packages (isaac_sim_bridge, launch files,
+# setup.py) and the command_mapper (which imports geometry_msgs) are
+# deliberately excluded so the baseline is green today and grows as bridges
+# land. Extend `include` as new pure modules are added.
+line-length = 100
+target-version = "py39"
+src = ["bridge", "scripts"]
+
+include = [
+    "bridge/common/zenoh_bridge/zenoh_bridge/action_event.py",
+    "bridge/common/zenoh_bridge/zenoh_bridge/utils.py",
+    "bridge/common/zenoh_bridge/zenoh_bridge/zenoh_subscriber.py",
+    "bridge/common/zenoh_bridge/tests/*.py",
+    "scripts/*.py",
+]
+
+[tool.ruff.lint]
+select = ["E", "F", "I", "UP", "B", "RUF"]
+
+[tool.ruff.lint.per-file-ignores]
+"bridge/common/zenoh_bridge/tests/*.py" = ["B008"]
+
+[tool.pytest.ini_options]
+testpaths = ["bridge/common/zenoh_bridge/tests"]
+# Import the pure modules directly so tests do not pull in ROS 2 runtime deps
+# through the package __init__ (which imports command_mapper/geometry_msgs).
+pythonpath = ["bridge/common/zenoh_bridge/zenoh_bridge"]