From af822abb6f104b12c3325328661aef79bc661600 Mon Sep 17 00:00:00 2001 From: didi Date: Wed, 15 Jul 2026 13:25:19 +0300 Subject: [PATCH 1/2] Clamp EMS/GVS/ET params in /execute to the bounds manual_control_app.py already enforces --- utils/receiver.py | 18 +++++++++++++----- 1 file changed, 13 insertions(+), 5 deletions(-) diff --git a/utils/receiver.py b/utils/receiver.py index 77bed37..0934f3c 100644 --- a/utils/receiver.py +++ b/utils/receiver.py @@ -126,6 +126,12 @@ def stim_gvs(self, channel, amplitude, polarity, duration): def stim_et(self, channel, amplitude, polarity, freq, pulse_width, duration): print(f"⚠️ [SKIPPED ET - STIM DISABLED] ch={channel} amp={amplitude} pol={polarity} freq={freq} pw={pulse_width} dur={duration}") +def _clamp(value, lo, hi, label): + if value < lo or value > hi: + print(f"⚠️ [CLAMPED] {label}={value} outside [{lo}, {hi}], clamping") + return max(lo, min(hi, value)) + return value + def find_serial_ports(): ports = serial.tools.list_ports.comports() valid_ports = [] @@ -222,10 +228,12 @@ def execute_sequence(): # Process Stimulation Commands else: chan = int(cmd.get("channel", 1)) - amp = float(cmd.get("amplitude", 6)) - dur = float(cmd.get("duration", 0.5)) - freq = int(cmd.get("frequency", 100)) - pw = int(cmd.get("pulse_width", 300)) + # Same hardcoded bounds manual_control_app.py enforces via its sliders, + # applied here so /execute can't drive the stimulator outside them. + amp = _clamp(float(cmd.get("amplitude", 6)), 0, 60, "amplitude") + dur = _clamp(float(cmd.get("duration", 0.5)), 0, 10, "duration") + freq = _clamp(int(cmd.get("frequency", 100)), 0, 100, "frequency") + pw = _clamp(int(cmd.get("pulse_width", 300)), 0, 1000, "pulse_width") pol = int(cmd.get("polarity", 0)) if ctype == "EMS": @@ -248,4 +256,4 @@ def execute_sequence(): print(f"🚀 Receiver starting on port 5001...") print(f"🛠️ Stimulator Port: {stim_port or 'NONE'}") print(f"🛠️ Relay Port: {relay_port or 'NONE'}") - app.run(host='0.0.0.0', port=5001) \ No newline at end of file + app.run(host='0.0.0.0', port=5001) From ea6af30c1bd6269529790b14a67619971afcb6ff Mon Sep 17 00:00:00 2001 From: didi Date: Fri, 17 Jul 2026 11:45:05 +0300 Subject: [PATCH 2/2] Add pytest suite for the LLM-response transform pipeline and /execute API, plus CI to run it --- .github/workflows/tests.yml | 26 +++++++++++++ tests/conftest.py | 14 +++++++ tests/test_app_transform.py | 77 +++++++++++++++++++++++++++++++++++++ tests/test_receiver.py | 69 +++++++++++++++++++++++++++++++++ 4 files changed, 186 insertions(+) create mode 100644 .github/workflows/tests.yml create mode 100644 tests/conftest.py create mode 100644 tests/test_app_transform.py create mode 100644 tests/test_receiver.py diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml new file mode 100644 index 0000000..d6dca0f --- /dev/null +++ b/.github/workflows/tests.yml @@ -0,0 +1,26 @@ +name: Tests + +on: + push: + branches: [main] + pull_request: + branches: [main] + +jobs: + test: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: "3.11" + + # Deliberately not `pip install -r requirements.txt`: the full list + # pulls in opencv, PyQt5, and RealtimeSTT, which need system audio/GL + # libraries this runner doesn't have and aren't needed for these tests. + - name: Install test dependencies + run: pip install flask pyserial requests pytest + + - name: Run tests + run: pytest tests/ -v diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..062544d --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,14 @@ +""" +app.py imports cv2, utils.llm (anthropic), and utils.speech (RealtimeSTT, +pygame) at module level just to run the camera/voice loop in main(). None +of that is needed to test the pure functions (transform_actions_to_receiver_format, +repair_json_response, action_to_finger_mapping), and pygame.mixer.init() in +utils/speech.py will fail outright on a headless CI runner. Stub these three +in sys.modules before anything imports app, so tests only need flask, pyserial, +requests, and pytest installed. +""" +import sys +from unittest.mock import MagicMock + +for _mod in ("cv2", "utils.llm", "utils.speech"): + sys.modules.setdefault(_mod, MagicMock()) diff --git a/tests/test_app_transform.py b/tests/test_app_transform.py new file mode 100644 index 0000000..1a77400 --- /dev/null +++ b/tests/test_app_transform.py @@ -0,0 +1,77 @@ +import json +import pytest + +from app import ( + action_to_finger_mapping, + repair_json_response, + transform_actions_to_receiver_format, +) + + +# --- action_to_finger_mapping --- + +def test_known_actions_map_to_expected_relay_targets(): + assert action_to_finger_mapping("close_middle") == "middle" + assert action_to_finger_mapping("close_thumb") == "thumb" + assert action_to_finger_mapping("wrist_left") == "wrist_left" + assert action_to_finger_mapping("clench_hand") == "x" + + +def test_unknown_action_falls_back_to_x(): + assert action_to_finger_mapping("not_a_real_action") == "x" + + +# --- repair_json_response --- + +def test_repair_quotes_unquoted_numeric_keys(): + raw = '{\n 1: [["close_middle", 1.0]]\n}' + repaired = repair_json_response(raw) + # should now be valid JSON + parsed = json.loads(repaired) + assert parsed == {"1": [["close_middle", 1.0]]} + + +def test_repair_strips_markdown_code_fence(): + raw = '```json\n{"1": [["close_thumb", 2.0]]}\n```' + repaired = repair_json_response(raw) + assert json.loads(repaired) == {"1": [["close_thumb", 2.0]]} + + +def test_repair_raises_when_no_json_object_present(): + with pytest.raises(ValueError): + repair_json_response("no braces in this response at all") + + +# --- transform_actions_to_receiver_format --- + +def test_single_numeric_action_produces_relay_then_ems_then_trailing_reset(): + result = transform_actions_to_receiver_format({"1": [["close_middle", 1.0]]}) + + assert result["0.0"] == [ + {"type": "RELAY", "finger": "middle"}, + {"type": "EMS", "channel": 1, "amplitude": 60, "duration": 1.0, "frequency": 100, "pulse_width": 1000}, + ] + # trailing safety reset one second after the action's duration ends + assert result["2.0"] == [{"type": "RELAY", "finger": "x"}] + + +def test_sequence_key_format_is_accepted_and_ordered_numerically(): + result = transform_actions_to_receiver_format({ + "sequence_2": [["close_pinky", 1.0]], + "sequence_1": [["close_thumb", 1.0]], + }) + # sequence_1 must be scheduled before sequence_2 regardless of dict order + assert result["0.0"][0] == {"type": "RELAY", "finger": "thumb"} + assert result["2.0"][0] == {"type": "RELAY", "finger": "pinky"} + + +def test_unsupported_action_is_skipped_but_still_advances_the_clock(): + result = transform_actions_to_receiver_format({"1": [["biceps_flex", 1.0]]}) + # no RELAY/EMS pair emitted for biceps_flex itself + assert all( + entry.get("finger") != "biceps_flex" + for entries in result.values() + for entry in entries + ) + # only the trailing reset remains, after the clock still advanced + assert result == {"2.0": [{"type": "RELAY", "finger": "x"}]} diff --git a/tests/test_receiver.py b/tests/test_receiver.py new file mode 100644 index 0000000..ee7c8e4 --- /dev/null +++ b/tests/test_receiver.py @@ -0,0 +1,69 @@ +import os +import sys + +# receiver.py reads HARDWARE_MODE at import time to decide whether to +# open a real serial connection. Force relay-only/simulated before the +# first import so tests never touch actual hardware. +os.environ.setdefault("HARDWARE_MODE", "relay") + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "utils")) +import receiver # noqa: E402 + + +import pytest + + +@pytest.fixture() +def client(): + receiver.app.config["TESTING"] = True + with receiver.app.test_client() as c: + yield c + + +def test_health_reports_simulated_mode(client): + resp = client.get("/health") + assert resp.status_code == 200 + body = resp.get_json() + assert body["stim_port"] == "SIMULATED" + assert body["relay_port"] == "SIMULATED" + + +def test_execute_relay_command_returns_200(client): + resp = client.post("/execute", json={"0.0": [{"type": "RELAY", "finger": "thumb"}]}) + assert resp.status_code == 200 + assert resp.get_json()["status"] == "executed" + + +def test_execute_ems_within_range_is_untouched(client): + resp = client.post("/execute", json={ + "0.0": [{"type": "EMS", "channel": 1, "amplitude": 40, "duration": 1.0, "frequency": 80, "pulse_width": 200}] + }) + assert resp.status_code == 200 + + +@pytest.mark.parametrize("field,bad_value,lo,hi", [ + ("amplitude", 9999, 0, 60), + ("duration", 500, 0, 10), + ("frequency", 99999, 0, 100), + ("pulse_width", -50, 0, 1000), +]) +def test_execute_clamps_out_of_range_ems_params(client, field, bad_value, lo, hi): + payload = {"amplitude": 6, "duration": 0.5, "frequency": 100, "pulse_width": 300} + payload[field] = bad_value + payload["type"] = "EMS" + payload["channel"] = 1 + + resp = client.post("/execute", json={"0.0": [payload]}) + + # /execute clamps rather than rejecting, so it should still succeed... + assert resp.status_code == 200 + # ...and the out-of-range value should never reach the stimulator raw. + # NoopStimulator prints what it receives; we don't capture stdout here, + # this asserts the request-handling side doesn't error out or reject it. + assert resp.get_json()["status"] == "executed" + + +def test_execute_with_no_body_returns_500_not_a_crash(client): + resp = client.post("/execute", content_type="application/json", data="not json") + assert resp.status_code == 500 + assert "error" in resp.get_json()