Run configured test and return results. This function handles the streaming JSONL response from the device. The device sends: 1. REMOTE: {"success": true, "streamMode": true} - RPC acknowledgement 2. RESULT: {"type": "test_start", ...} - Test start event 3. R
(self)
| 311 | return self.send_rpc("setStripSizes", [{"small": small, "large": large}]) |
| 312 | |
| 313 | def run_test(self) -> TestResult: |
| 314 | """Run configured test and return results. |
| 315 | |
| 316 | This function handles the streaming JSONL response from the device. |
| 317 | The device sends: |
| 318 | 1. REMOTE: {"success": true, "streamMode": true} - RPC acknowledgement |
| 319 | 2. RESULT: {"type": "test_start", ...} - Test start event |
| 320 | 3. RESULT: {"type": "test_complete", ...} - Test complete event |
| 321 | |
| 322 | Returns: |
| 323 | TestResult object with per-lane details |
| 324 | |
| 325 | Raises: |
| 326 | RuntimeError: If test execution fails |
| 327 | TimeoutError: If test does not complete within timeout |
| 328 | """ |
| 329 | # Send RPC command |
| 330 | cmd: dict[str, str | list[Any]] = {"method": "runTest", "params": []} |
| 331 | cmd_str = json.dumps(cmd, separators=(",", ":")) |
| 332 | |
| 333 | # Send command |
| 334 | self._monitor.write(cmd_str + "\n") |
| 335 | |
| 336 | # Wait for REMOTE: response (RPC acknowledgement) |
| 337 | response = self._wait_for_response() |
| 338 | |
| 339 | if not response.get("success"): |
| 340 | raise RuntimeError(f"Test failed: {response.get('error')}") |
| 341 | |
| 342 | # Check if using streaming mode |
| 343 | if not response.get("streamMode"): |
| 344 | # Legacy mode - response contains testState directly |
| 345 | # This path is for backward compatibility if needed |
| 346 | state = response.get("testState", {}) |
| 347 | results = state.get("results", {}) |
| 348 | |
| 349 | lane_results: list[LaneResult] = [] |
| 350 | for detail in results.get("details", []): |
| 351 | lane_idx = detail["lane"] |
| 352 | lane_results.append( |
| 353 | LaneResult( |
| 354 | lane=lane_idx, |
| 355 | led_count=detail.get("ledCount", 0), |
| 356 | pin=detail.get("pin", lane_idx), |
| 357 | passed=detail["passed"], |
| 358 | bit_errors=detail["bitErrors"], |
| 359 | timing=detail["timing"], |
| 360 | expected_bytes=detail.get("expectedBytes"), |
| 361 | received_bytes=detail.get("receivedBytes"), |
| 362 | ) |
| 363 | ) |
| 364 | |
| 365 | return TestResult( |
| 366 | passed=results["passed"], |
| 367 | total_tests=results["totalTests"], |
| 368 | passed_tests=results["passedTests"], |
| 369 | failed_tests=results["failedTests"], |
| 370 | duration_ms=state.get("timing", {}).get("durationMs", 0), |
no test coverage detected