"""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 ConnectionFirstTests(HardwareFreeTest): def test_connects_every_source_before_ready_or_force(self): for count in range(1, 5): with self.subTest(count=count): ser = ByteSerial() controller = uart.UartFingerController(ser, sources=list(range(count))) events = [] def open_runtime(events=events, **kwargs): events.append(kwargs["video"]) return FakeRuntime() original_write = ser.write def write(data, events=events, count=count, ser=ser, original_write=original_write): self.assertEqual(events, list(range(count))) size = original_write(data) if data.startswith(b"force,") and sum(x.startswith("force,") for x in ser.output) == count: ser.input.extend(b"quit\n") return size ser.write = write with patch("orisys.finger.FingerRuntime.open", side_effect=open_runtime): controller.bank.connect_all() controller.serve() self.assertEqual(ser.output[0], f"OK ready {count}") self.assertEqual([x for x in ser.output if x.startswith("force,")], [ f"force,{i},1.250000,-2.500000,3.750000" for i in range(1, count + 1) ]) controller.close_runtime() def test_one_failed_connection_rolls_back_every_opened_camera(self): bank = FingerSensorBank([0, 2, 4]) opened = [] def open_runtime(**kwargs): runtime = FakeRuntime(warmup_error=kwargs["video"] == 2) opened.append(runtime) return runtime with (patch("orisys.finger.FingerRuntime.open", side_effect=open_runtime), self.assertRaisesRegex(RuntimeError, "sensor 2")): bank.connect_all() self.assertEqual(bank.runtimes, {}) self.assertTrue(all(r.closed for r in opened)) def test_no_uart_is_opened_if_camera_preparation_fails(self): with (patch("orisys.finger.FingerRuntime.open", return_value=FakeRuntime(warmup_ok=False)), patch.object(uart, "open_serial") as serial): self.assertEqual(uart.main(["--sources", "0", "2"]), 1) serial.assert_not_called() def test_legacy_commands_are_rejected(self): ser = ByteSerial() controller = uart.UartFingerController(ser, sources=[0, 2]) for command in ["connect", "connect 2", "connect all", "run", "run all", "search", "exit"]: self.assertTrue(controller._dispatch(command)) self.assertTrue(ser.output[-1].startswith("ERR unknown")) def test_no_per_sensor_or_all_command_arguments(self): ser = ByteSerial() controller = uart.UartFingerController(ser, sources=[0, 2]) for command in ["start 1", "stop all", "read 2", "quit extra"]: self.assertTrue(controller._dispatch(command)) self.assertTrue(ser.output[-1].startswith("ERR command")) def test_start_stop_and_read_control_entire_connected_bank(self): ser = ByteSerial() controller = uart.UartFingerController(ser, sources=[0, 2]) controller.bank.connect_all() controller._dispatch("stop") controller._dispatch("read") self.assertEqual(ser.output[-2:], [ "force,1,1.250000,-2.500000,3.750000", "force,2,1.250000,-2.500000,3.750000", ]) self.assertFalse(controller._streaming) controller._dispatch("start") self.assertTrue(controller._streaming) self.assertEqual(ser.output[-1], "OK start") def test_read_all_never_publishes_partial_round_on_processing_failure(self): ser = ByteSerial() controller = uart.UartFingerController(ser, sources=[0, 2]) controller.bank.connect_all() controller.bank.runtimes[2].grab_error = True controller._streaming = True controller._dispatch("read") self.assertFalse(controller._streaming) self.assertFalse(any(x.startswith("force,") for x in ser.output)) self.assertIn("sensor 2", ser.output[-1]) def test_serve_rejects_incomplete_bank_before_ready(self): ser = ByteSerial() controller = uart.UartFingerController(ser, sources=[0, 2]) with self.assertRaises(RuntimeError): controller.serve() self.assertEqual(ser.output, []) class FramingTests(HardwareFreeTest): def test_partial_line_waits_for_terminator(self): ser = ByteSerial(b"st") controller = uart.UartFingerController(ser, sources=[0]) 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_lf_crlf_commands_are_separate(self): controller = uart.UartFingerController(ByteSerial(b"stop\rstart\nread\r\n"), sources=[0]) self.assertEqual(controller._poll_command_line(blocking=False), "stop") self.assertEqual(controller._poll_command_line(blocking=False), "start") self.assertEqual(controller._poll_command_line(blocking=False), "read") def test_overlong_command_is_discarded_until_terminator(self): ser = ByteSerial(b"a" * 1025) controller = uart.UartFingerController(ser, sources=[0]) 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.assertTrue(ser.output[-1].startswith("ERR command")) def test_read_only_ignores_every_command_and_still_streams(self): ser = ByteSerial(b"stop\nquit\n" + b"a" * 1100 + b"\n") controller = uart.UartFingerController(ser, sources=[0], read_only=True) controller.bank.connect_all() 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.assertEqual(ser.output, ["OK ready 1", "force,1,1.250000,-2.500000,3.750000"]) def test_short_write_is_transport_failure(self): ser = ByteSerial() ser.write = lambda data: len(data) - 1 with self.assertRaises(OSError): uart.UartFingerController(ser, sources=[0]).reply("OK ready 1") def test_error_newlines_do_not_inject_records(self): ser = ByteSerial() uart.UartFingerController(ser, sources=[0]).reply("ERR read bad\nOK quit\r\n") self.assertEqual(ser.output, ["ERR read bad OK quit"]) class BankTests(HardwareFreeTest): def test_invalid_count_config_count_and_duplicate_sources(self): for sources, configs in [([], ["finger"]), (range(5), ["finger"]), ([0, 2], ["a", "b", "c"]), ([0, "0"], ["finger"]), ([0, "/dev/video0"], ["finger"]), ([-1], ["finger"])]: with self.subTest(sources=sources), self.assertRaises(ValueError): FingerSensorBank(sources, configs=configs) def test_configs_and_runtime_states_are_independent(self): bank = FingerSensorBank([0, 2], configs=["left.json", "right.json"]) configs = [] def open_runtime(**kwargs): configs.append(kwargs["config"]) return FakeRuntime() with patch("orisys.finger.FingerRuntime.open", side_effect=open_runtime): bank.connect_all() self.assertEqual(configs, ["left.json", "right.json"]) self.assertIsNot(bank.runtimes[1], bank.runtimes[2]) readings = bank.read_all() self.assertEqual([r.sensor_id for r in readings], [1, 2]) self.assertEqual(readings[0].units, ("raw", "raw", "raw")) def test_no_frame_during_warmup_closes_runtime(self): runtime = FakeRuntime(warmup_ok=False) bank = FingerSensorBank([0]) with (patch("orisys.finger.FingerRuntime.open", return_value=runtime), self.assertRaisesRegex(RuntimeError, "warmup")): bank.connect_all() self.assertTrue(runtime.closed) self.assertEqual(bank.runtimes, {}) def test_nonfinite_force_is_not_a_valid_round(self): bank = FingerSensorBank([0]) bank.connect_all() bank.runtimes[1].tactile.read_info = lambda *args: (float("nan"), 0, 0) with self.assertRaisesRegex(RuntimeError, "non-finite"): bank.read_all() def test_eof_returns_no_partial_round(self): bank = FingerSensorBank([0, 2]) bank.connect_all() bank.runtimes[2].frames = 0 bank.sources[2] = "ended.mp4" with self.assertRaisesRegex(EOFError, "sensor 2"): bank.read_all() def test_all_resources_are_attempted_even_when_close_fails(self): bank = FingerSensorBank([0, 2]) bank.connect_all() 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_sources_are_required_and_legacy_options_are_removed(self): for args in [[], ["--camera-id", "0"], ["--video", "a.mp4"], ["--sources", "0", "--autostart"], ["--sources", "0", "--max-camera-index", "10"], ["--sources", "0", "1", "2", "3", "4"], ["--sources", "0", "--timeout", "nan"]]: with self.subTest(args=args), patch.object(uart, "open_serial") as serial: with self.assertRaises(SystemExit) as status: uart.main(args) self.assertEqual(status.exception.code, 2) serial.assert_not_called() def test_serial_failure_closes_every_connected_runtime(self): opened = [] def open_runtime(**kwargs): runtime = FakeRuntime() opened.append(runtime) return runtime with (patch("orisys.finger.FingerRuntime.open", side_effect=open_runtime), patch.object(uart, "open_serial", side_effect=OSError("UART unavailable"))): self.assertEqual(uart.main(["--sources", "0", "2"]), 1) self.assertEqual(len(opened), 2) self.assertTrue(all(r.closed for r in opened)) if __name__ == "__main__": unittest.main()