"""Shared fakes. Nothing here needs a gamepad, an arm, a browser or a daemon.""" import json import pytest from dora_teleop.config import Config from dora_teleop.gamepad import Sample class RecordingNode: """A dora node that records outputs *with* their metadata. `dora.testing.MockNode` drops the metadata dict on `send_output`, and the metadata is where `request_id` rides. So the tests record it themselves. """ def __init__(self): self.sent = [] def send_output(self, output_id, data, metadata=None): self.sent.append((output_id, data, dict(metadata or {}))) def outputs(self, output_id): """Return every message sent on ``output_id`` as (value, metadata) pairs.""" return [(data[0].as_py(), metadata) for name, data, metadata in self.sent if name == output_id] def jogs(self): """Return every velocity sent to the arm, in order.""" return [row["velocity"] for row, _ in self.outputs("jog")] def services(self): """Return the name of every service called, in order.""" return [row["service"] for row, _ in self.outputs("request")] def ui(self): """Return every object handed to the bridge, parsed.""" return [json.loads(value) for value, _ in self.outputs("ui_state")] def clear(self): self.sent.clear() class FakeGamepad: """Stands in for the reader thread: hands back whatever the test set.""" def __init__(self, sample=None): self.sample = sample or Sample() self.stopped = False def read(self): return self.sample def shutdown(self): self.stopped = True def stick(axes=None, buttons=None, present=True): """Build one gamepad reading, padded to a full six axes.""" values = list(axes or []) values += [0.0] * (6 - len(values)) # A trigger at rest reads -1, and a test that forgot to say so would look # like both triggers held down. for index in (2, 5): if axes is None or index >= len(axes): values[index] = -1.0 return Sample(axes=values, buttons=list(buttons or []), present=present) def input_event(event_id, value, metadata=None): """Build the event dict dora hands a node for one input.""" return {"type": "INPUT", "id": event_id, "value": value, "metadata": dict(metadata or {})} @pytest.fixture(autouse=True) def clean_env(monkeypatch): """Run every test against a bare environment, whatever the shell set.""" for name in ( "SPEED_MM_S", "SPEED_DEG_S", "DEADZONE", "EXPO", "JOYSTICK_INDEX", "JOG_TTL_S", "GAMEPAD_IDLE_RELEASE_S", "STATE_POLL_S", ): monkeypatch.delenv(name, raising=False) @pytest.fixture def config(): """Return the default configuration, with expo off so a test can do the arithmetic.""" built = Config.from_env() built.expo = 0.0 return built