QUELLCODE / PY
test_controller.py
tests/test_controller.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"""Supervised controller integration tests without a real API, camera or HID."""67import asyncio8import io9import tempfile10import time11import unittest1213from PIL import Image1415from hardware_box.config import Settings16from hardware_box.controller.agent import Controller17from hardware_box.hid.driver import DryHID18from hardware_box.hid.mapping import Geometry19from hardware_box.safety.policy import Policy, SafetyError20from hardware_box.vision.frame import Frame212223def computer_response(actions, checks=None):24 return {"id": "script-call", "status": "completed", "output": [25 {"type": "computer_call", "call_id": "call-1", "actions": actions,26 "pending_safety_checks": checks or []}]}272829def result_response():30 return {"id": "script-result", "status": "completed", "output": [31 {"type": "message", "content": [{"type": "output_text", "text": "Visually complete; please verify."}]}]}323334class FakeCapture:35 def __init__(self):36 self.geometry = Geometry(320, 240, 640, 480)37 self.sequence = 038 self.error, self.age = "", 039 self.failure = None40 self.gate = None41 self.entered = asyncio.Event()42 self.set_color("navy")4344 def set_color(self, color):45 output = io.BytesIO()46 Image.new("RGB", (320, 240), color).save(output, "PNG")47 self.png = output.getvalue()4849 async def snapshot(self, after=None):50 self.entered.set()51 if self.gate:52 await self.gate.wait()53 if self.failure:54 raise self.failure55 self.sequence += 156 return Frame(self.png, self.geometry, time.monotonic(), self.sequence)5758 def health(self):59 return {"error": self.error, "age": self.age}606162class ScriptedClient:63 def __init__(self, first):64 self.first = first65 self.observations = []66 self.api_calls = self.total_tokens = 067 self.closed = False6869 async def start(self, task, frame):70 self.api_calls += 171 return self.first7273 async def observe(self, outputs):74 self.observations.append(outputs)75 self.api_calls += 176 return result_response()7778 async def close(self):79 self.closed = True808182class ControllerTests(unittest.IsolatedAsyncioTestCase):83 async def asyncSetUp(self):84 self.temp = tempfile.TemporaryDirectory()85 self.settings = Settings(log_dir=self.temp.name, screenshot_width=320,86 screenshot_height=240, target_width=640,87 target_height=480, settle_seconds=0.1,88 max_actions_per_second=10)89 self.controllers = []9091 async def asyncTearDown(self):92 for controller in self.controllers:93 if controller.worker and not controller.worker.done():94 await controller.stop()95 if controller.monitor:96 controller.monitor.cancel()97 await asyncio.gather(controller.monitor, return_exceptions=True)98 self.temp.cleanup()99100 def make_controller(self, actions, checks=None):101 capture = FakeCapture()102 hid = DryHID(keyboard_layout=self.settings.keyboard_layout)103 client = ScriptedClient(computer_response(actions, checks))104 controller = Controller(self.settings, capture, hid,105 agent_factory=lambda settings, key: client)106 self.controllers.append(controller)107 return controller, capture, hid, client108109 async def until(self, predicate, timeout=1.5):110 deadline = time.monotonic() + timeout111 while not predicate():112 if time.monotonic() >= deadline:113 self.fail("controller did not reach expected state before deadline")114 await asyncio.sleep(0.005)115116 async def finish(self, controller):117 await asyncio.wait_for(asyncio.shield(controller.worker), 1.5)118119 def input_commands(self, hid):120 return [(op, args) for op, args in hid.commands if op in {"ABS", "REL", "KEY"}]121122 async def approve_pending(self, controller):123 pending = controller.pending.copy()124 await controller.confirm(pending["nonce"], pending["screenshot_sha256"], True)125126 async def test_exact_human_approval_precedes_hid_and_result_review(self):127 controller, capture, hid, client = self.make_controller([{"type": "click", "x": 20, "y": 10}])128 await controller.start("Click the shown button")129 await self.until(lambda: controller.pending is not None)130 self.assertEqual(self.input_commands(hid), [])131 pending = controller.pending.copy()132 with self.assertRaises(SafetyError):133 await controller.confirm("wrong nonce", pending["screenshot_sha256"], True)134 with self.assertRaises(SafetyError):135 await controller.confirm(pending["nonce"], "wrong screenshot", True)136 self.assertEqual(self.input_commands(hid), [])137 self.assertFalse(controller._decision.done())138 await self.approve_pending(controller)139 await self.finish(controller)140 self.assertTrue(self.input_commands(hid))141 self.assertEqual(controller.state, "RESULT_REVIEW")142 self.assertFalse(hid.health_snapshot()["armed"])143 self.assertTrue(client.closed)144 events = [event["kind"] for event in controller.audit.events]145 self.assertLess(events.index("approved"), events.index("action_acknowledged"))146 controller.review_result(True)147 self.assertEqual(controller.state, "COMPLETED")148149 async def test_human_denial_outputs_no_hid(self):150 controller, _, hid, client = self.make_controller([{"type": "keypress", "keys": ["ENTER"]}])151 await controller.start("Confirm dialog")152 await self.until(lambda: controller.pending is not None)153 pending = controller.pending.copy()154 await controller.confirm(pending["nonce"], pending["screenshot_sha256"], False)155 await self.finish(controller)156 self.assertEqual(controller.state, "FAULT")157 self.assertIn("abgelehnt", controller.error)158 self.assertEqual(self.input_commands(hid), [])159 self.assertEqual(client.observations, [])160 self.assertFalse(hid.health_snapshot()["armed"])161162 async def test_stop_revokes_nonce_and_disarms(self):163 controller, _, hid, _ = self.make_controller([{"type": "click", "x": 20, "y": 10}])164 await controller.start("Click")165 await self.until(lambda: controller.pending is not None)166 pending = controller.pending.copy()167 await controller.stop()168 self.assertEqual(controller.state, "STOPPED")169 self.assertIsNone(controller.pending)170 with self.assertRaises(SafetyError):171 await controller.confirm(pending["nonce"], pending["screenshot_sha256"], True)172 self.assertEqual(self.input_commands(hid), [])173 self.assertFalse(hid.health_snapshot()["armed"])174175 async def test_full_call_preflight_rejects_text_before_first_click(self):176 controller, _, hid, client = self.make_controller([177 {"type": "click", "x": 20, "y": 10}, {"type": "type", "text": "valid prefix🚀"}])178 await controller.start("Click then type")179 await self.finish(controller)180 self.assertEqual(controller.state, "FAULT")181 self.assertIn("unsupported character", controller.error)182 self.assertEqual(self.input_commands(hid), [])183 self.assertIsNone(controller.pending)184 self.assertEqual(client.observations, [])185186 async def test_changed_screenshot_rejects_approval(self):187 controller, capture, hid, _ = self.make_controller([{"type": "click", "x": 20, "y": 10}])188 await controller.start("Click")189 await self.until(lambda: controller.pending is not None)190 capture.set_color("orange")191 with self.assertRaisesRegex(SafetyError, "Bild seit"):192 await self.approve_pending(controller)193 await self.finish(controller)194 self.assertEqual(controller.state, "FAULT")195 self.assertEqual(self.input_commands(hid), [])196197 async def test_api_safety_checks_require_separate_human_approval(self):198 checks = [{"id": "check-1", "code": "possible_sensitive_action", "message": "Review the operation"}]199 controller, _, hid, client = self.make_controller([{"type": "click", "x": 20, "y": 10}], checks)200 await controller.start("Click")201 await self.until(lambda: controller.pending is not None)202 first_nonce = controller.pending["nonce"]203 self.assertEqual(controller.pending["action"]["type"], "api_safety_check")204 self.assertEqual(client.observations, [])205 self.assertEqual(self.input_commands(hid), [])206 await self.approve_pending(controller)207 await self.until(lambda: controller.pending is not None and controller.pending["nonce"] != first_nonce)208 self.assertEqual(controller.pending["action"]["type"], "click")209 self.assertEqual(client.observations, [])210 self.assertEqual(self.input_commands(hid), [])211 await self.approve_pending(controller)212 await self.finish(controller)213 self.assertEqual(client.observations[0][0]["acknowledged_safety_checks"], checks)214 self.assertEqual(controller.state, "RESULT_REVIEW")215216 async def test_denied_api_safety_checks_are_never_acknowledged(self):217 checks = [{"id": "check-1", "code": "possible_sensitive_action", "message": "Review"}]218 controller, _, hid, client = self.make_controller([{"type": "screenshot"}], checks)219 await controller.start("Inspect")220 await self.until(lambda: controller.pending is not None)221 pending = controller.pending.copy()222 await controller.confirm(pending["nonce"], pending["screenshot_sha256"], False)223 await self.finish(controller)224 self.assertEqual(client.observations, [])225 self.assertEqual(self.input_commands(hid), [])226227 async def test_pause_disarms_and_requires_new_session(self):228 controller, _, hid, _ = self.make_controller([{"type": "type", "text": "hello"}])229 await controller.start("Type")230 await self.until(lambda: controller.pending is not None)231 await controller.stop(pause=True)232 self.assertEqual(controller.state, "PAUSED")233 self.assertIsNone(controller.pending)234 self.assertFalse(hid.health_snapshot()["armed"])235 self.assertEqual(self.input_commands(hid), [])236237 async def test_capture_failure_while_waiting_stops_output(self):238 controller, capture, hid, _ = self.make_controller([{"type": "click", "x": 20, "y": 10}])239 await controller.start("Click")240 await self.until(lambda: controller.pending is not None)241 capture.error = "USB capture disconnected"242 await self.until(lambda: controller.state == "FAULT")243 await self.finish(controller)244 self.assertIn("Capture", controller.error)245 self.assertFalse(hid.health_snapshot()["armed"])246 self.assertEqual(self.input_commands(hid), [])247248 async def test_session_deadline_applies_while_waiting(self):249 controller, _, hid, _ = self.make_controller([{"type": "click", "x": 20, "y": 10}])250 await controller.start("Click")251 await self.until(lambda: controller.pending is not None)252 controller.policy.started = time.monotonic() - self.settings.session_seconds - 1253 await self.until(lambda: controller.state == "FAULT")254 await self.finish(controller)255 self.assertIn("Sitzungslimit", controller.error)256 self.assertFalse(hid.health_snapshot()["armed"])257 self.assertEqual(self.input_commands(hid), [])258259 async def test_replacement_nonce_during_capture_cannot_be_approved(self):260 controller, capture, hid, _ = self.make_controller([])261 controller.state = "WAITING_CONFIRMATION"262 frame = await capture.snapshot()263 old = {"nonce": "old-nonce", "screenshot_sha256": frame.sha256, "expires_at": time.time() + 60}264 controller.pending = old265 old_future = controller._decision = asyncio.get_running_loop().create_future()266 capture.entered.clear()267 capture.gate = asyncio.Event()268 approval = asyncio.create_task(controller.confirm(old["nonce"], frame.sha256, True))269 await capture.entered.wait()270 replacement = {**old, "nonce": "replacement-nonce"}271 replacement_future = asyncio.get_running_loop().create_future()272 controller.pending, controller._decision = replacement, replacement_future273 capture.gate.set()274 with self.assertRaisesRegex(SafetyError, "ungültig"):275 await approval276 self.assertFalse(old_future.done())277 self.assertFalse(replacement_future.done())278 self.assertIs(controller.pending, replacement)279 self.assertEqual(self.input_commands(hid), [])280 old_future.cancel()281 replacement_future.cancel()282283 async def test_approval_expiry_is_rechecked_after_capture(self):284 controller, capture, _, _ = self.make_controller([])285 controller.state = "WAITING_CONFIRMATION"286 frame = await capture.snapshot()287 pending = {"nonce": "n", "screenshot_sha256": frame.sha256, "expires_at": time.time() + 60}288 controller.pending = pending289 decision = controller._decision = asyncio.get_running_loop().create_future()290 capture.entered.clear()291 capture.gate = asyncio.Event()292 approval = asyncio.create_task(controller.confirm("n", frame.sha256, True))293 await capture.entered.wait()294 pending["expires_at"] = time.time() - 1295 capture.gate.set()296 with self.assertRaisesRegex(SafetyError, "abgelaufen"):297 await approval298 self.assertFalse(decision.done())299 decision.cancel()300301 def test_forbidden_chords_checked_by_physical_modifiers(self):302 policy = Policy(self.settings)303 geometry = Geometry(320, 240, 640, 480)304 for keys in (["RIGHT_CTRL", "RALT", "DEL"], ["RIGHT_META", "R"], ["CONTROL", "SHIFT", "ESCAPE"]):305 with self.subTest(keys=keys), self.assertRaises(SafetyError):306 policy.validate({"type": "keypress", "keys": keys}, geometry)307308309if __name__ == "__main__":310 unittest.main()