diff --git a/.github/workflows/python-quality.yml b/.github/workflows/python-quality.yml new file mode 100644 index 000000000..a5612a04b --- /dev/null +++ b/.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 diff --git a/bridge/common/zenoh_bridge/tests/test_action_event.py b/bridge/common/zenoh_bridge/tests/test_action_event.py new file mode 100644 index 000000000..0c73f000e --- /dev/null +++ b/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 diff --git a/bridge/common/zenoh_bridge/tests/test_utils.py b/bridge/common/zenoh_bridge/tests/test_utils.py new file mode 100644 index 000000000..2393294da --- /dev/null +++ b/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 diff --git a/bridge/common/zenoh_bridge/zenoh_bridge/action_event.py b/bridge/common/zenoh_bridge/zenoh_bridge/action_event.py index 09ee7a9e7..1fec85ebd 100644 --- a/bridge/common/zenoh_bridge/zenoh_bridge/action_event.py +++ b/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 diff --git a/bridge/common/zenoh_bridge/zenoh_bridge/zenoh_subscriber.py b/bridge/common/zenoh_bridge/zenoh_bridge/zenoh_subscriber.py index 661e75de4..955df641b 100644 --- a/bridge/common/zenoh_bridge/zenoh_bridge/zenoh_subscriber.py +++ b/bridge/common/zenoh_bridge/zenoh_bridge/zenoh_subscriber.py @@ -1,5 +1,8 @@ """Zenoh session and subscriber helper.""" -from typing import Callable, List + +from collections.abc import Callable +from typing import Any + import zenoh @@ -7,13 +10,11 @@ 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) diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 000000000..f64a39b37 --- /dev/null +++ b/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"]