"""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 ManualAddTests(HardwareFreeTest): def test_service_starts_empty_without_opening_a_camera(self): ser = ByteSerial(b"quit\n") controller = uart.UartFingerController(ser) with patch("orisys.finger.FingerRuntime.open") as open_runtime: controller.serve() open_runtime.assert_not_called() self.assertEqual(ser.output, ["OK ready 0", "OK quit"]) def test_add_one_through_four_then_start_produces_uniform_output(self): for count in range(1, 5): with self.subTest(count=count): commands = "".join(f"add {i * 2}\n" for i in range(count)) + "start\n" ser = ByteSerial(commands.encode()) controller = uart.UartFingerController(ser) original_write = ser.write def write(data, ser=ser, original_write=original_write, count=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 controller.serve() self.assertEqual( ser.output[: count + 2], ["OK ready 0"] + [f"OK added {i}" for i in range(1, count + 1)] + ["OK start"], ) 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_add_alone_does_not_start_output(self): ser = ByteSerial(b"add 0\nquit\n") uart.UartFingerController(ser).serve() self.assertEqual(ser.output, ["OK ready 0", "OK added 1", "OK quit"]) def test_start_and_read_require_at_least_one_sensor(self): ser = ByteSerial() controller = uart.UartFingerController(ser) for command in ("start", "read"): controller._dispatch(command) self.assertTrue(ser.output[-1].startswith(f"ERR {command}")) self.assertFalse(controller._streaming) def test_add_requires_stop_and_never_interrupts_existing_stream(self): ser = ByteSerial() controller = uart.UartFingerController(ser) controller._dispatch("add 0") controller._dispatch("start") first = controller.bank.runtimes[1] with patch("orisys.finger.FingerRuntime.open") as open_runtime: controller._dispatch("add 2") open_runtime.assert_not_called() self.assertEqual(ser.output[-1], "ERR add stop streaming first") self.assertTrue(controller._streaming) controller._dispatch("stop") controller._dispatch("add 2") self.assertEqual(ser.output[-1], "OK added 2") self.assertIs(controller.bank.runtimes[1], first) self.assertFalse(first.closed) self.assertFalse(controller._streaming) def test_failed_add_preserves_other_sensor_and_reuses_available_id(self): bank = FingerSensorBank() self.assertEqual(bank.add(0), 1) first = bank.runtimes[1] failed = FakeRuntime(warmup_error=True) with ( patch("orisys.finger.FingerRuntime.open", return_value=failed), self.assertRaises(RuntimeError), ): bank.add(2) self.assertTrue(failed.closed) self.assertFalse(first.closed) self.assertEqual(bank.sources, {1: 0}) self.assertEqual(bank.add(4), 2) self.assertEqual(bank.sources, {1: 0, 2: 4}) def test_failed_warmup_without_exception_closes_new_runtime(self): bank = FingerSensorBank() failed = FakeRuntime(warmup_ok=False) with ( patch("orisys.finger.FingerRuntime.open", return_value=failed), self.assertRaisesRegex(RuntimeError, "warmup"), ): bank.add(0) self.assertTrue(failed.closed) self.assertEqual(bank.runtimes, {}) self.assertEqual(bank.sources, {}) def test_duplicate_source_and_fifth_camera_are_rejected_before_open(self): bank = FingerSensorBank() bank.add(0) for source in ("0", "/dev/video0", -1): with ( self.subTest(source=source), patch("orisys.finger.FingerRuntime.open") as open_runtime, ): with self.assertRaises(ValueError): bank.add(source) open_runtime.assert_not_called() for source in (2, 4, 6): bank.add(source) with patch("orisys.finger.FingerRuntime.open") as open_runtime: with self.assertRaisesRegex(ValueError, "4"): bank.add(8) open_runtime.assert_not_called() def test_each_sensor_can_use_its_own_config(self): bank = FingerSensorBank(config="shared.json") seen = [] def open_runtime(**kwargs): seen.append(kwargs["config"]) return FakeRuntime() with patch("orisys.finger.FingerRuntime.open", side_effect=open_runtime): bank.add(0) bank.add(2, config="right.json") self.assertEqual(seen, ["shared.json", "right.json"]) self.assertIsNot(bank.runtimes[1], bank.runtimes[2]) self.assertEqual([r.sensor_id for r in bank.read_all()], [1, 2]) def test_quoted_source_and_config_are_one_argument_each(self): ser = ByteSerial() controller = uart.UartFingerController(ser) with patch.object(controller.bank, "add", return_value=1) as add: controller._dispatch('add "/tmp/test clip.mp4" "my config.json"') add.assert_called_once_with("/tmp/test clip.mp4", config="my config.json", cal_path=None) self.assertEqual(ser.output[-1], "OK added 1") def test_removed_commands_and_arguments_are_rejected(self): ser = ByteSerial() controller = uart.UartFingerController(ser) for command in ("connect", "run", "search", "exit"): controller._dispatch(command) self.assertTrue(ser.output[-1].startswith("ERR unknown")) for command in ( "add", "add 0 a extra", "start 1", "stop all", "read 2", "quit extra", 'add "broken', ): controller._dispatch(command) self.assertTrue(ser.output[-1].startswith("ERR command")) def test_processing_failure_stops_output_without_partial_round(self): ser = ByteSerial() controller = uart.UartFingerController(ser) controller._dispatch("add 0") controller._dispatch("add 2") 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]) class FramingTests(HardwareFreeTest): def test_partial_line_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_lf_crlf_commands_are_separate(self): controller = uart.UartFingerController(ByteSerial(b"stop\rstart\nread\r\n")) 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) 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_short_write_is_transport_failure(self): ser = ByteSerial() ser.write = lambda data: len(data) - 1 with self.assertRaises(OSError): uart.UartFingerController(ser).reply("OK ready 0") def test_error_newlines_do_not_inject_records(self): ser = ByteSerial() uart.UartFingerController(ser).reply("ERR read bad\nOK quit\r\n") self.assertEqual(ser.output, ["ERR read bad OK quit"]) class BankTests(HardwareFreeTest): def test_nonfinite_force_is_not_a_valid_round(self): bank = FingerSensorBank() bank.add(0) 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() bank.add(0) bank.add(2) 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() bank.add(0) bank.add(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, {}) self.assertEqual(bank.sources, {}) class CliTests(HardwareFreeTest): def test_startup_opens_only_uart_and_waits_for_add(self): ser = ByteSerial(b"quit\n") ser.close = lambda: None with ( patch.object(uart, "open_serial", return_value=ser), patch("orisys.finger.FingerRuntime.open") as runtime, ): self.assertEqual(uart.main([]), 0) runtime.assert_not_called() self.assertEqual(ser.output, ["OK ready 0", "OK quit"]) def test_automatic_setup_options_are_removed(self): for args in ( ["--sources", "0"], ["--read-only"], ["--autostart"], ["--camera-id", "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_transport_failure_closes_cameras_added_by_commands(self): ser = ByteSerial(b"add 0\nadd 2\nstart\n") ser.closed = False ser.close = lambda: setattr(ser, "closed", True) opened = [] def open_runtime(**kwargs): runtime = FakeRuntime() opened.append(runtime) return runtime original_write = ser.write def write(data): if data.startswith(b"force,"): raise OSError("UART disconnected") return original_write(data) ser.write = write with ( patch.object(uart, "open_serial", return_value=ser), patch("orisys.finger.FingerRuntime.open", side_effect=open_runtime), ): self.assertEqual(uart.main([]), 1) self.assertEqual(len(opened), 2) self.assertTrue(ser.closed) self.assertTrue(all(r.closed for r in opened)) if __name__ == "__main__": unittest.main()