QUELLCODE / PY
test_hid.py
tests/test_hid.py
1# OpenAss: nur privater, experimenteller Gebrauch; auf eigene Verantwortung.2# Niemals Unternehmensrechner, Produktivsysteme, Industrieanlagen oder3# sicherheitskritische Systeme anschließen oder bedienen.4# Haftungshinweis und gesetzliche Grenzen: ../DISCLAIMER.txt.5"""Safety-focused host HID tests; no USB target or pyserial is required."""67import asyncio8import time9import unittest1011from hardware_box.hid.driver import DryHID, HIDError, RealHID, validate_action12from hardware_box.hid.keyboard import KeyboardLayout, KeyboardMappingError, LEFT_CTRL, LEFT_SHIFT, RIGHT_ALT, normalize_key_name13from hardware_box.hid.mapping import Geometry, MappingError14from hardware_box.hid.protocol import ProtocolError, crc16_ccitt_false, decode_frame, encode_frame, require_ok151617class ProtocolTests(unittest.TestCase):18 def test_ccitt_false_known_vector(self):19 self.assertEqual(crc16_ccitt_false(b"123456789"), 0x29B1)2021 def test_signed_command_roundtrip_and_crc_domain(self):22 raw = encode_frame(4, "REL", (0, 0, -2, 3, 0))23 frame = decode_frame(raw)24 self.assertEqual(frame.arguments, ("0", "0", "-2", "3", "0"))25 body, trailer = raw[:-1].rsplit(b" *", 1)26 self.assertEqual(int(trailer, 16), crc16_ccitt_false(body))2728 def test_corruption_partial_and_injection_rejected(self):29 raw = encode_frame(1, "ABS", (3, 5, 1))30 for bad in (raw[:-1], raw.replace(b"ABS", b"REL"), raw + b"extra", b"x\n"):31 with self.subTest(bad=bad), self.assertRaises(ProtocolError):32 decode_frame(bad)33 for args in ((True,), ("2\nSTOP",), ("one two",)):34 with self.subTest(args=args), self.assertRaises(ProtocolError):35 encode_frame(1, "ABS", args)3637 def test_ack_sequence_and_error(self):38 with self.assertRaises(ProtocolError):39 require_ok(decode_frame(encode_frame(2, "OK")), 1)40 with self.assertRaisesRegex(ProtocolError, "PHYSICAL_ARM_REQUIRED"):41 require_ok(decode_frame(encode_frame(1, "ERR", ("PHYSICAL_ARM_REQUIRED",))), 1)4243 def test_firmware_line_bound(self):44 with self.assertRaises(ProtocolError):45 encode_frame(1, "ERR", ("X" * 150,))464748class GeometryTests(unittest.TestCase):49 def test_scale_does_not_depend_on_native_resolution(self):50 geometry = Geometry(1920, 1080, 3840, 2160)51 self.assertEqual(geometry.to_hid(0, 0), (0, 0))52 self.assertEqual(geometry.to_absolute(1919, 1079), (32767, 32767))53 self.assertEqual(geometry.to_target(1919, 1079), (3839, 2159))54 self.assertEqual(geometry.to_hid(1250, 620), Geometry(1920, 1080, 1920, 1080).to_hid(1250, 620))5556 def test_letterbox_points_rejected_and_endpoints_exact(self):57 geometry = Geometry(1920, 1080, 3840, 2160, (0, 100, 1920, 880))58 self.assertEqual(geometry.to_hid(0, 100), (0, 0))59 self.assertEqual(geometry.to_hid(1919, 979), (32767, 32767))60 for point in ((100, 99), (100, 980), (-1, 200), (1920, 200), (True, 100), (1.0, 100), (float("nan"), 100)):61 with self.subTest(point=point), self.assertRaises(MappingError):62 geometry.to_hid(*point)6364 def test_invalid_configuration(self):65 for args in ((1, 1080, 1920, 1080), (1920, 1080, 1920, 1080, (0, 100, 1920, 1000)), (True, 1080, 1920, 1080)):66 with self.subTest(args=args), self.assertRaises(MappingError):67 Geometry(*args)686970class KeyboardTests(unittest.TestCase):71 def test_supported_printable_ascii_all_layouts(self):72 for name in ("us", "de_windows", "at_linux"):73 self.assertTrue(KeyboardLayout(name).preflight_text("".join(chr(x) for x in range(32, 127))))7475 def test_de_text_and_shortcut_y_z_and_altgr(self):76 layout = KeyboardLayout("de")77 strokes = layout.preflight_text("yZäÄ߀@")78 self.assertEqual(strokes[0].usages, (29,))79 self.assertEqual((strokes[1].modifiers, strokes[1].usages), (LEFT_SHIFT, (28,)))80 self.assertEqual((strokes[-2].modifiers, strokes[-2].usages), (RIGHT_ALT, (8,)))81 self.assertEqual((strokes[-1].modifiers, strokes[-1].usages), (RIGHT_ALT, (20,)))82 combo = layout.keypress(["CTRL", "Y"])83 self.assertEqual((combo.modifiers, combo.usages), (LEFT_CTRL, (29,)))8485 def test_us_uppercase_and_de_dead_keys(self):86 self.assertEqual(KeyboardLayout("us").preflight_text("A")[0].modifiers, LEFT_SHIFT)87 strokes = KeyboardLayout("de_linux").preflight_text("^`")88 self.assertEqual([x.usages for x in strokes], [(53,), (44,), (46,), (44,)])89 self.assertEqual(strokes[2].modifiers, LEFT_SHIFT)9091 def test_unicode_and_mac_profile_rejected(self):92 with self.assertRaisesRegex(KeyboardMappingError, "U\\+1F680"):93 KeyboardLayout("de").preflight_text("valid prefix 🚀")94 with self.assertRaises(KeyboardMappingError):95 KeyboardLayout("us").preflight_text("ä")96 with self.assertRaises(KeyboardMappingError):97 KeyboardLayout("de_mac")9899 def test_combo_aliases_and_locks(self):100 for name in ("META", "SUPER", "CMD", "WIN"):101 self.assertEqual(KeyboardLayout().keypress([name, "C"]).modifiers, 8)102 self.assertEqual(KeyboardLayout().keypress("CTRL+ALT+DELETE").usages, (76,))103 self.assertEqual(KeyboardLayout().keypress("PAGEDOWN").usages, (78,))104 for keys in (["CAPSLOCK"], ["C", "C"], ["UNKNOWN"], ["CTRL", ""]):105 with self.subTest(keys=keys), self.assertRaises(KeyboardMappingError):106 KeyboardLayout().keypress(keys)107108 def test_policy_modifier_canonicalization(self):109 for name in ("CTRL", "CTL", "CONTROL", "LCTRL", "RIGHT_CTRL"):110 self.assertEqual(normalize_key_name(name), "CTRL")111 for name in ("ALT", "OPTION", "RALT", "ALTGR"):112 self.assertEqual(normalize_key_name(name), "ALT")113 for name in ("CMD", "SUPER", "WIN", "RIGHT_META"):114 self.assertEqual(normalize_key_name(name), "META")115 self.assertEqual(normalize_key_name(" del "), "DELETE")116117118class FakeSerial:119 def __init__(self, *args, **kwargs):120 self.timeout = kwargs.get("timeout", 0.35)121 self.commands = []122 self.pending = None123 self.closed = False124 self.fail_operation = None125 self.delay_operation = None126 self.delay = 0.04127128 def write(self, data):129 self.pending = decode_frame(data)130 self.commands.append(self.pending)131 return len(data)132133 def read_until(self, delimiter, size):134 frame = self.pending135 if frame.operation == self.delay_operation:136 time.sleep(self.delay)137 if frame.operation == self.fail_operation:138 self.fail_operation = None139 return b"" # lost acknowledgement; firmware may have executed140 args = ("1",) if frame.operation == "HELLO" else ()141 return encode_frame(frame.sequence, "OK", args)142143 def close(self):144 self.closed = True145146147class DriverTests(unittest.IsolatedAsyncioTestCase):148 async def asyncSetUp(self):149 self.geometry = Geometry(1920, 1080, 3840, 2160)150151 async def make_dry(self, layout="us"):152 driver = DryHID(keyboard_layout=layout)153 await driver.start()154 await driver.arm()155 return driver156157 async def test_unsupported_text_outputs_nothing(self):158 driver = await self.make_dry("de")159 before = list(driver.commands)160 with self.assertRaises(KeyboardMappingError):161 await driver.execute({"type": "type", "text": "prefix🚀"}, self.geometry)162 self.assertEqual(driver.commands, before)163164 async def test_entire_drag_is_preflighted(self):165 driver = await self.make_dry()166 before = list(driver.commands)167 with self.assertRaises(MappingError):168 await driver.execute({"type": "drag", "path": [{"x": 4, "y": 5}, {"x": 9999, "y": 5}]}, self.geometry)169 self.assertEqual(driver.commands, before)170171 async def test_click_mapping_and_release(self):172 driver = await self.make_dry()173 await driver.execute({"type": "click", "x": 1919, "y": 1079}, self.geometry)174 self.assertEqual(driver.commands[-3:], [("ABS", (32767, 32767, 0)), ("ABS", (32767, 32767, 1)), ("RELEASE", ())])175176 async def test_small_scroll_accumulates_pixels(self):177 driver = await self.make_dry()178 for _ in range(2):179 await driver.execute({"type": "scroll", "x": 2, "y": 3, "scroll_y": 60}, self.geometry)180 self.assertEqual([args for op, args in driver.commands if op == "REL"], [(0, 0, -1, 0, 0)])181182 async def test_cancelled_key_hold_releases(self):183 driver = await self.make_dry()184 task = asyncio.create_task(driver.execute({"type": "keypress", "keys": ["CTRL", "C"]}, self.geometry))185 await asyncio.sleep(0.003)186 task.cancel()187 with self.assertRaises(asyncio.CancelledError):188 await task189 self.assertEqual(driver.commands[-1], ("RELEASE", ()))190191 async def test_stop_blocks_next_output(self):192 driver = await self.make_dry()193 await driver.stop()194 with self.assertRaises(HIDError):195 await driver.execute({"type": "click", "x": 1, "y": 2}, self.geometry)196 self.assertFalse(driver.health_snapshot()["armed"])197198 async def test_rearm_does_not_resume_active_or_queued_actions(self):199 driver = await self.make_dry()200 active = asyncio.create_task(driver.execute({"type": "type", "text": "abcdef"}, self.geometry))201 await asyncio.sleep(0.003)202 queued = asyncio.create_task(driver.execute({"type": "click", "x": 1, "y": 2}, self.geometry))203 await asyncio.sleep(0)204 await driver.stop()205 await driver.arm()206 before = len(driver.commands)207 for task in (active, queued):208 with self.assertRaises(HIDError):209 await task210 self.assertEqual(len(driver.commands), before)211 await driver.execute({"type": "keypress", "keys": ["ENTER"]}, self.geometry)212 self.assertEqual(driver.commands[-1], ("RELEASE", ()))213214 async def test_cancelled_serial_open_is_closed(self):215 serial = FakeSerial()216217 def factory(*args, **kwargs):218 time.sleep(0.03)219 return serial220221 driver = RealHID(serial_factory=factory)222 task = asyncio.create_task(driver.start())223 await asyncio.sleep(0.005)224 task.cancel()225 with self.assertRaises(asyncio.CancelledError):226 await task227 self.assertTrue(serial.closed)228 self.assertFalse(driver.health_snapshot()["connected"])229 self.assertFalse(driver.health_snapshot()["armed"])230231 async def test_real_failure_never_retries_action_and_stops_lease(self):232 serial = FakeSerial()233 driver = RealHID(serial_factory=lambda *a, **kw: serial)234 await driver.start()235 await driver.arm()236 serial.fail_operation = "ABS"237 with self.assertRaises(HIDError):238 await driver.execute({"type": "move", "x": 3, "y": 4}, self.geometry)239 self.assertFalse(driver.health_snapshot()["armed"])240 self.assertTrue(driver.health_snapshot()["fault"])241 self.assertEqual(sum(f.operation == "ABS" for f in serial.commands), 1)242 count = sum(f.operation == "PING" for f in serial.commands)243 await asyncio.sleep(0.15)244 self.assertEqual(sum(f.operation == "PING" for f in serial.commands), count)245 with self.assertRaises(HIDError):246 await driver.arm()247 await driver.start() # explicit recovery: still disarmed248 self.assertFalse(driver.health_snapshot()["armed"])249 self.assertIsNone(driver.health_snapshot()["fault"])250 await driver.close()251252 async def test_real_cancelled_inflight_write_settles_and_disarms(self):253 serial = FakeSerial()254 driver = RealHID(serial_factory=lambda *a, **kw: serial)255 await driver.start()256 await driver.arm()257 serial.delay_operation = "ABS"258 task = asyncio.create_task(driver.execute({"type": "move", "x": 3, "y": 4}, self.geometry))259 await asyncio.sleep(0.01)260 task.cancel()261 with self.assertRaises(asyncio.CancelledError):262 await task263 self.assertFalse(driver.health_snapshot()["armed"])264 self.assertTrue(driver.health_snapshot()["fault"])265 self.assertEqual(sum(f.operation == "ABS" for f in serial.commands), 1)266 self.assertEqual(serial.commands[-1].operation, "STOP")267 await driver.close()268269 async def test_clean_real_stop_does_not_fault(self):270 serial = FakeSerial()271 driver = RealHID(serial_factory=lambda *a, **kw: serial)272 await driver.start()273 await driver.arm()274 await asyncio.sleep(0.005)275 await driver.stop()276 self.assertFalse(driver.health_snapshot()["armed"])277 self.assertIsNone(driver.health_snapshot()["fault"])278 await driver.close()279280 async def test_stop_during_arm_cannot_rearm(self):281 serial = FakeSerial()282 driver = RealHID(serial_factory=lambda *a, **kw: serial)283 await driver.start()284 serial.delay_operation = "ARM"285 task = asyncio.create_task(driver.arm())286 await asyncio.sleep(0.01)287 await driver.stop()288 with self.assertRaises(HIDError):289 await task290 self.assertFalse(driver.health_snapshot()["armed"])291 self.assertEqual(serial.commands[-1].operation, "STOP")292 await driver.close()293294295if __name__ == "__main__":296 unittest.main()