"""Hardware-free protocol and sensor lifecycle regression tests.""" import io import sys import types import unittest from contextlib import redirect_stderr from pathlib import Path from unittest.mock import patch SERIAL_DIR = Path(__file__).resolve().parents[1] / "serial" sys.path.insert(0, str(SERIAL_DIR)) import example_uart as uart from finger_api import FingerSensorBank class ByteSerial: def __init__(self, data=b""): self.input = bytearray(data) self.output = [] @property def in_waiting(self): return len(self.input) def read(self, size=1): result = bytes(self.input[:size]) del self.input[:size] return result def readline(self): end = self.input.find(b"\n") return self.read(end + 1 if end >= 0 else len(self.input)) def write(self, data): self.output.append(data.decode().rstrip("\n")) return len(data) def flush(self): pass class FakeTactile: info = types.SimpleNamespace(FNORMAL=1, FSHEARX=2, FSHEARY=3) force_units = ("raw", "raw", "raw") def compute_deformation(self): pass def compute_contact(self): pass def read_info(self, *args): return 1.25, -2.5, 3.75 def reset(self): pass class FakeRuntime: def __init__(self, *, warmup_error=False, warmup_ok=True, grab_error=False): self.tactile = FakeTactile() self.closed = False self.warmup_error = warmup_error self.warmup_ok = warmup_ok self.grab_error = grab_error self.capture = types.SimpleNamespace(set=lambda *args: True) self.frames = 1 def warmup(self, **kwargs): if self.warmup_error: raise RuntimeError("warmup failed") return self.warmup_ok, None def grab_and_process(self, **kwargs): if self.grab_error: raise RuntimeError("processing failed") if self.frames: self.frames -= 1 return True, None return False, None def close(self): self.closed = True class HardwareFreeTest(unittest.TestCase): def setUp(self): stderr = redirect_stderr(io.StringIO()) stderr.__enter__() self.addCleanup(stderr.__exit__, None, None, None) finger = types.ModuleType("orisys.finger") finger.FingerRuntime = types.SimpleNamespace(open=lambda **kwargs: FakeRuntime()) capture = types.ModuleType("orisys.capture") capture.is_live_camera = lambda source: isinstance(source, int) or str(source).startswith("/dev/video") capture.list_cameras = lambda **kwargs: [] package = types.ModuleType("orisys") package.finger, package.capture = finger, capture modules = patch.dict(sys.modules, { "orisys": package, "orisys.finger": finger, "orisys.capture": capture, }) modules.start() self.addCleanup(modules.stop) class ExistingRegressionTests(HardwareFreeTest): def test_partial_command_waits_for_terminator(self): ser = ByteSerial(b"st") controller = uart.UartFingerController(ser) self.assertIsNone(controller._poll_command_line(blocking=False)) ser.input.extend(b"op\n") self.assertEqual(controller._poll_command_line(blocking=False), "stop") def test_cr_only_terminal_commands(self): controller = uart.UartFingerController(ByteSerial(b"stop\rrun\r")) self.assertEqual(controller._poll_command_line(blocking=False), "stop") self.assertEqual(controller._poll_command_line(blocking=False), "run") def test_failed_warmup_releases_runtime(self): runtime = FakeRuntime(warmup_error=True) ser = ByteSerial() controller = uart.UartFingerController(ser) with patch("orisys.finger.FingerRuntime.open", return_value=runtime): controller.cmd_connect([]) self.assertTrue(runtime.closed) self.assertTrue(ser.output[-1].startswith("ERR connect")) def test_unsuccessful_warmup_is_not_connected(self): runtime = FakeRuntime(warmup_ok=False) ser = ByteSerial() controller = uart.UartFingerController(ser) with patch("orisys.finger.FingerRuntime.open", return_value=runtime): controller.cmd_connect([]) self.assertTrue(ser.output[-1].startswith("ERR connect")) self.assertTrue(runtime.closed) class MultiSensorTests(HardwareFreeTest): def test_autostart_service_streams_one_through_four_sensors_then_stops(self): for count in range(1, 5): with self.subTest(count=count): ser = ByteSerial() controller = uart.UartFingerController(ser, sources=list(range(count)), autostart=True) original_write = ser.write def write(data, original_write=original_write, ser=ser, count=count): size = original_write(data) if data.startswith(b"force,") and sum(line.startswith("force,") for line in ser.output) == count: ser.input.extend(b"stop all\nquit\n") return size ser.write = write controller.serve() self.assertEqual(sum(line.startswith("force,") for line in ser.output), count) self.assertEqual(ser.output[-1], "OK quit") self.assertFalse(controller._streaming) controller.close_runtime() def test_one_through_four_sensors_have_identifiable_readings(self): for count in range(1, 5): with self.subTest(count=count): ser = ByteSerial() controller = uart.UartFingerController(ser, sources=list(range(count))) controller.cmd_connect(["all"]) controller._dispatch("read all") forces = [line for line in ser.output if line.startswith("force,")] expected = ["force,1.250000,-2.500000,3.750000"] if count == 1 else [ f"force,{i},1.250000,-2.500000,3.750000" for i in range(1, count + 1) ] self.assertEqual(forces, expected) self.assertFalse(controller._streaming) controller.close_runtime() def test_selective_stop_and_reconnect_leave_other_sensor_running(self): controller = uart.UartFingerController(ByteSerial(), sources=[0, 2, 4, 6]) controller.cmd_connect(["all"]) controller._dispatch("run all") controller._dispatch("stop 2") self.assertEqual(controller._streaming, {1, 3, 4}) old = controller.bank.runtimes[3] controller._dispatch("connect 3 8") self.assertTrue(old.closed) self.assertEqual(controller._streaming, {1, 4}) self.assertEqual(controller.bank.sources[3], 8) def test_failed_sensor_does_not_stop_healthy_sensor(self): ser = ByteSerial() controller = uart.UartFingerController(ser, sources=[0, 2]) controller.cmd_connect(["all"]) controller.bank.runtimes[1].grab_error = True controller.cmd_run([]) for sensor_id in sorted(controller._streaming): controller._read_one(sensor_id, context="run") self.assertIn("ERR run 1 processing failed", ser.output) self.assertIn("force,2,1.250000,-2.500000,3.750000", ser.output) self.assertEqual(controller._streaming, {2}) def test_connect_all_retains_successes_when_one_source_fails(self): ser = ByteSerial() controller = uart.UartFingerController(ser, sources=[0, 2, 4]) def open_runtime(**kwargs): if kwargs["video"] == 2: raise RuntimeError("camera unavailable") return FakeRuntime() with patch("orisys.finger.FingerRuntime.open", side_effect=open_runtime): controller.cmd_connect(["all"]) self.assertEqual(set(controller.bank.runtimes), {1, 3}) self.assertIn("ERR connect 2 camera unavailable", ser.output) def test_rejects_duplicate_live_source_without_closing_other_camera(self): controller = uart.UartFingerController(ByteSerial(), sources=[0, 2]) controller.cmd_connect(["all"]) original = controller.bank.runtimes[2] controller._dispatch("connect 2 0") self.assertIs(controller.bank.runtimes[2], original) self.assertFalse(original.closed) def test_config_is_selected_per_sensor(self): bank = FingerSensorBank([0, 2], configs=["left.json", "right.json"]) seen = [] def open_runtime(**kwargs): seen.append(kwargs["config"]) return FakeRuntime() with patch("orisys.finger.FingerRuntime.open", side_effect=open_runtime): bank.connect(1) bank.connect(2) self.assertEqual(seen, ["left.json", "right.json"]) self.assertEqual(bank.read(1).units, ("raw", "raw", "raw")) def test_invalid_sensor_count_and_config_count(self): for sources, configs in [([], ["finger"]), (range(5), ["finger"]), ([0, 2], ["a", "b", "c"])]: with self.assertRaises(ValueError): FingerSensorBank(sources, configs=configs) def test_invalid_command_id_and_unconnected_run_are_errors(self): ser = ByteSerial() controller = uart.UartFingerController(ser, sources=[0, 2]) for command in ["run 0", "read 3", "stop -1", "connect 3 4", "run all", "run 2 extra"]: self.assertTrue(controller._dispatch(command)) self.assertTrue(ser.output[-1].startswith("ERR ")) class ProtocolTests(HardwareFreeTest): def test_repeated_run_is_rejected_without_recursion(self): ser = ByteSerial() controller = uart.UartFingerController(ser) controller.cmd_connect([]) controller._dispatch("run") controller._dispatch("run") self.assertEqual(ser.output[-1], "ERR run already running") self.assertEqual(controller._streaming, {1}) def test_camera_search_failure_is_reported_without_exiting(self): ser = ByteSerial() controller = uart.UartFingerController(ser) with patch("orisys.capture.list_cameras", side_effect=OSError("camera scan failed")): self.assertTrue(controller._dispatch("search")) self.assertEqual(ser.output[-1], "ERR search camera scan failed") def test_legacy_connect_and_batched_stop_quit(self): ser = ByteSerial(b"connect 2\r\nrun\nstop\nquit\n") controller = uart.UartFingerController(ser) controller.serve() self.assertEqual(ser.output, ["OK ready", "OK connect 2 (camera)", "OK run", "OK stop", "OK quit"]) self.assertFalse(controller._streaming) def test_stop_is_processed_before_next_frame(self): ser = ByteSerial(b"stop\nquit\n") controller = uart.UartFingerController(ser) controller.cmd_connect([]) controller.cmd_run([]) controller.serve() self.assertFalse(any(line.startswith("force,") for line in ser.output)) def test_read_only_autostarts_and_ignores_host_commands(self): ser = ByteSerial(b"quit\nstop\nconnect 9\n" + b"a" * 1100 + b"\n") controller = uart.UartFingerController(ser, read_only=True) original_write = ser.write def write(data): size = original_write(data) if data.startswith(b"force,"): raise KeyboardInterrupt() return size ser.write = write with self.assertRaises(KeyboardInterrupt): controller.serve() self.assertIn("OK run", ser.output) self.assertIn("force,1.250000,-2.500000,3.750000", ser.output) self.assertNotIn("OK quit", ser.output) self.assertFalse(any(line.startswith("ERR ") for line in ser.output)) def test_long_line_is_discarded_until_delimiter(self): ser = ByteSerial(b"a" * 1025) controller = uart.UartFingerController(ser) self.assertIsNone(controller._poll_command_line(blocking=False)) ser.input.extend(b"quit\nstop\r") self.assertEqual(controller._poll_command_line(blocking=False), "stop") self.assertEqual(ser.output, ["ERR command line too long (max 1024 bytes)"]) self.assertFalse(controller._should_quit) def test_quoted_paths_and_malformed_quotes(self): controller = uart.UartFingerController(ByteSerial(), sources=[0, 2]) with patch.object(controller.bank, "connect", return_value="/tmp/a b.mp4") as connect: controller._dispatch('connect 2 "/tmp/a b.mp4"') connect.assert_called_once_with(2, "/tmp/a b.mp4") controller._dispatch('connect "broken') self.assertTrue(controller.ser.output[-1].startswith("ERR command")) def test_non_finite_measurements_are_not_published(self): ser = ByteSerial() controller = uart.UartFingerController(ser) controller.cmd_connect([]) controller.bank.runtimes[1].tactile.read_info = lambda *args: (float("nan"), 0, 0) controller._dispatch("read") self.assertEqual(ser.output[-1], "ERR read non-finite force measurement") self.assertFalse(any(line.startswith("force,") for line in ser.output)) def test_serial_short_write_is_a_transport_failure(self): ser = ByteSerial() ser.write = lambda data: len(data) - 1 with self.assertRaises(OSError): uart.UartFingerController(ser).reply("OK ready") def test_error_newlines_cannot_inject_protocol_records(self): ser = ByteSerial() uart.UartFingerController(ser).reply("ERR read bad\nOK quit\r\n") self.assertEqual(ser.output, ["ERR read bad OK quit"]) def test_empty_looping_video_retries_once_then_reports_eof(self): ser = ByteSerial() controller = uart.UartFingerController(ser) runtime = FakeRuntime() runtime.frames = 0 controller.bank.sources[1] = "empty.mp4" controller.bank.runtimes[1] = runtime controller._dispatch("read") self.assertEqual(ser.output, ["OK eof"]) def test_close_attempts_every_sensor_even_when_one_close_fails(self): bank = FingerSensorBank([0, 2]) bank.connect(1) bank.connect(2) second = bank.runtimes[2] bank.runtimes[1].close = lambda: (_ for _ in ()).throw(RuntimeError("close failed")) with self.assertRaisesRegex(RuntimeError, "close failed"): bank.close() self.assertTrue(second.closed) self.assertEqual(bank.runtimes, {}) class CliTests(HardwareFreeTest): def test_open_serial_has_finite_write_timeout_and_no_flow_control(self): serial = types.ModuleType("serial") serial.EIGHTBITS, serial.PARITY_NONE, serial.STOPBITS_ONE = 8, "N", 1 serial.Serial = types.SimpleNamespace() with patch.dict(sys.modules, {"serial": serial}), patch.object(serial, "Serial") as factory: uart.open_serial("/dev/ttyHS1", 115200, 0.1, 0.5) factory.assert_called_once_with( port="/dev/ttyHS1", baudrate=115200, bytesize=8, parity="N", stopbits=1, timeout=0.1, write_timeout=0.5, exclusive=True, ) def test_invalid_options_do_not_open_uart(self): for args in [["--timeout", "0"], ["--write-timeout", "nan"], ["--sources", "0", "1", "2", "3", "4"], ["--sources", "0", "2", "--config", "a", "--config", "b", "--config", "c"]]: with self.subTest(args=args), patch.object(uart, "open_serial") as open_serial: with self.assertRaises(SystemExit) as exit_status: uart.main(args) self.assertEqual(exit_status.exception.code, 2) open_serial.assert_not_called() def test_transport_failure_returns_nonzero_and_closes_all_resources(self): ser = ByteSerial() ser.closed = False ser.close = lambda: setattr(ser, "closed", True) def serve(controller): controller.cmd_connect(["all"]) self.runtimes = list(controller.bank.runtimes.values()) raise OSError("UART disconnected") with patch.object(uart, "open_serial", return_value=ser), patch.object(uart.UartFingerController, "serve", serve): self.assertEqual(uart.main(["--sources", "0", "2"]), 1) self.assertTrue(ser.closed) self.assertTrue(all(runtime.closed for runtime in self.runtimes)) if __name__ == "__main__": unittest.main()