#!/usr/bin/env python3 """ Dangerous Pi - Hardware Integration Test Suite Date: 2026-03-03 Runs against a live Pi over WiFi, serial console, and Bluetooth. Usage: source .venv/bin/activate python test_hardware_2026-03-03.py # Run all phases python test_hardware_2026-03-03.py --phase 1 # Run specific phase python test_hardware_2026-03-03.py --safe-only # Skip destructive tests """ import argparse import asyncio import json import subprocess import sys import time from dataclasses import dataclass, field from datetime import datetime from pathlib import Path import httpx import serial # --------------------------------------------------------------------------- # Configuration # --------------------------------------------------------------------------- PI_HOST = "dangerous-pi.local" API_BASE = f"http://{PI_HOST}:8000/api" WS_BASE = f"ws://{PI_HOST}:8000/ws" HTTPS_BASE = f"https://{PI_HOST}" HTTP_UI = f"http://{PI_HOST}" SERIAL_PORT = "/dev/ttyUSB0" SERIAL_BAUD = 115200 BLE_DEVICE_NAME = "DangerousPi" AUTH_USER = "admin" AUTH_PASSWORD = "changeme" REQUEST_TIMEOUT = 10.0 PM3_CMD_TIMEOUT = 30.0 # --------------------------------------------------------------------------- # Test result tracking # --------------------------------------------------------------------------- @dataclass class TestResult: phase: int test_id: str name: str status: str # PASS, FAIL, SKIP, WARN detail: str = "" duration_ms: float = 0 data: dict = field(default_factory=dict) results: list[TestResult] = [] def record(phase: int, test_id: str, name: str, status: str, detail: str = "", duration_ms: float = 0, data: dict | None = None): r = TestResult(phase, test_id, name, status, detail, duration_ms, data or {}) results.append(r) icon = {"PASS": "\033[32m✅", "FAIL": "\033[31m❌", "SKIP": "\033[33m⏭️", "WARN": "\033[33m⚠️"}[status] print(f" {icon} {test_id} {name}: {detail}\033[0m") return r # --------------------------------------------------------------------------- # Serial console helper # --------------------------------------------------------------------------- class SerialConsole: def __init__(self, port=SERIAL_PORT, baud=SERIAL_BAUD): self.port = port self.baud = baud self._ser = None def open(self): try: self._ser = serial.Serial(self.port, self.baud, timeout=3) # Flush any pending data self._ser.reset_input_buffer() return True except Exception as e: print(f" [serial] Cannot open {self.port}: {e}") return False def close(self): if self._ser and self._ser.is_open: self._ser.close() def run_command(self, cmd: str, timeout: float = 10.0) -> str: """Send a command and read output. Expects a logged-in shell.""" if not self._ser or not self._ser.is_open: return "" # Send command self._ser.write(f"{cmd}\n".encode()) self._ser.flush() time.sleep(0.3) # Read output until timeout output = [] deadline = time.time() + timeout while time.time() < deadline: if self._ser.in_waiting: chunk = self._ser.read(self._ser.in_waiting).decode("utf-8", errors="replace") output.append(chunk) time.sleep(0.1) else: time.sleep(0.2) if not self._ser.in_waiting: break return "".join(output) serial_console = SerialConsole() # --------------------------------------------------------------------------- # HTTP helpers # --------------------------------------------------------------------------- def timed_get(client: httpx.Client, url: str, **kwargs) -> tuple[httpx.Response | None, float]: t0 = time.monotonic() try: r = client.get(url, timeout=kwargs.pop("timeout", REQUEST_TIMEOUT), **kwargs) return r, (time.monotonic() - t0) * 1000 except Exception as e: return None, (time.monotonic() - t0) * 1000 def timed_post(client: httpx.Client, url: str, **kwargs) -> tuple[httpx.Response | None, float]: t0 = time.monotonic() try: r = client.post(url, timeout=kwargs.pop("timeout", REQUEST_TIMEOUT), **kwargs) return r, (time.monotonic() - t0) * 1000 except Exception as e: return None, (time.monotonic() - t0) * 1000 # --------------------------------------------------------------------------- # Phase 1: Regression tests # --------------------------------------------------------------------------- def phase1_regression(client: httpx.Client): print("\n\033[1m═══ Phase 1: Regression - Retest Jan 23 Issues ═══\033[0m") # 1.1 WiFi Connect API - just check status, don't actually reconnect r, ms = timed_get(client, f"{API_BASE}/wifi/status") if r and r.status_code == 200: data = r.json() connected = any(i.get("connected") for i in data.get("interfaces", [])) if connected: record(1, "1.1", "WiFi status reports connected", "PASS", f"ssid={data.get('current_ssid')}, ip={data.get('current_ip')}", ms, data) else: record(1, "1.1", "WiFi status reports connected", "WARN", "WiFi reports not connected despite being reachable", ms) else: record(1, "1.1", "WiFi status reports connected", "FAIL", f"HTTP {r.status_code if r else 'timeout'}", ms) # 1.2 UPS detection r, ms = timed_get(client, f"{API_BASE}/system/ups/status") if r and r.status_code == 200: data = r.json() if data.get("is_available"): record(1, "1.2", "UPS detected at boot", "PASS", f"battery={data.get('battery_percentage')}%, source={data.get('power_source')}", ms, data) else: record(1, "1.2", "UPS detected at boot", "FAIL", f"UPS not available: {data.get('error_message', 'unknown')}", ms, data) else: record(1, "1.2", "UPS detected at boot", "FAIL", f"HTTP {r.status_code if r else 'timeout'}", ms) # 1.3 BLE advertising r, ms = timed_get(client, f"{API_BASE}/system/ble/status") if r and r.status_code == 200: data = r.json() if data.get("advertising"): record(1, "1.3", "BLE advertising active", "PASS", f"name={data.get('device_name')}", ms, data) else: record(1, "1.3", "BLE advertising active", "WARN", f"advertising=false, available={data.get('available')}", ms, data) else: record(1, "1.3", "BLE advertising active", "FAIL", f"HTTP {r.status_code if r else 'timeout'}", ms) # 1.4 Pydantic warning (via serial) - only check logs since last boot if serial_console._ser and serial_console._ser.is_open: out = serial_console.run_command( "journalctl -u dangerous-pi --no-pager -b 2>/dev/null | grep -i 'protected namespace' | tail -3" ) if "protected namespace" in out.lower(): record(1, "1.4", "Pydantic namespace warning", "WARN", "Warning still present in logs", 0) else: record(1, "1.4", "Pydantic namespace warning", "PASS", "No warning found in current boot logs", 0) else: record(1, "1.4", "Pydantic namespace warning", "SKIP", "Serial console not available", 0) # --------------------------------------------------------------------------- # Phase 2: Core health & system endpoints # --------------------------------------------------------------------------- def phase2_system(client: httpx.Client): print("\n\033[1m═══ Phase 2: Core Health & System Endpoints ═══\033[0m") # 2.1 Health r, ms = timed_get(client, f"{API_BASE}/health") if r and r.status_code == 200 and r.json().get("status") == "healthy": record(2, "2.1", "Health check", "PASS", f"{ms:.0f}ms", ms, r.json()) else: record(2, "2.1", "Health check", "FAIL", f"HTTP {r.status_code if r else 'timeout'}", ms) # 2.2 Ready r, ms = timed_get(client, f"{API_BASE}/ready") if r and r.status_code == 200: record(2, "2.2", "Readiness check", "PASS", f"{ms:.0f}ms", ms, r.json()) else: record(2, "2.2", "Readiness check", "FAIL", f"HTTP {r.status_code if r else 'timeout'}", ms) # 2.3 System info r, ms = timed_get(client, f"{API_BASE}/system/info") if r and r.status_code == 200: data = r.json() temp = data.get("cpu_temp", data.get("cpu", {}).get("temperature", 0)) has_fields = all(k in data for k in ["hostname", "cpu_temp", "memory_used", "disk_used"]) if has_fields and temp < 85: record(2, "2.3", "System info", "PASS", f"temp={temp}°C, mem={data.get('memory_used',0)//1024//1024}MB", ms, data) else: record(2, "2.3", "System info", "FAIL", f"Missing fields or temp too high ({temp}°C)", ms, data) else: record(2, "2.3", "System info", "FAIL", f"HTTP {r.status_code if r else 'timeout'}", ms) # 2.4 System config r, ms = timed_get(client, f"{API_BASE}/system/config") if r and r.status_code == 200: record(2, "2.4", "System config", "PASS", f"{ms:.0f}ms", ms) else: # Config endpoint might not exist, mark as WARN status = "WARN" if r and r.status_code == 404 else "FAIL" record(2, "2.4", "System config", status, f"HTTP {r.status_code if r else 'timeout'}", ms) # 2.5 Pi model r, ms = timed_get(client, f"{API_BASE}/system/pi/model") if r and r.status_code == 200: record(2, "2.5", "Pi model info", "PASS", f"{r.json()}", ms, r.json()) else: status = "WARN" if r and r.status_code == 404 else "FAIL" record(2, "2.5", "Pi model info", status, f"HTTP {r.status_code if r else 'timeout'}", ms) # 2.6 CPU cores r, ms = timed_get(client, f"{API_BASE}/system/cpu/cores") if r and r.status_code == 200: data = r.json() record(2, "2.6", "CPU cores", "PASS", f"total={data.get('total_cores')}, online={data.get('online_cores')}", ms, data) else: status = "WARN" if r and r.status_code == 404 else "FAIL" record(2, "2.6", "CPU cores", status, f"HTTP {r.status_code if r else 'timeout'}", ms) # 2.7 Power restrictions r, ms = timed_get(client, f"{API_BASE}/system/power/restrictions") if r and r.status_code == 200: record(2, "2.7", "Power restrictions", "PASS", f"{ms:.0f}ms", ms, r.json()) else: record(2, "2.7", "Power restrictions", "FAIL", f"HTTP {r.status_code if r else 'timeout'}", ms) # 2.8 Header widgets r, ms = timed_get(client, f"{API_BASE}/system/widgets") if r and r.status_code == 200: data = r.json() record(2, "2.8", "Header widgets", "PASS", f"{len(data) if isinstance(data, list) else 'dict'} widgets", ms) else: status = "WARN" if r and r.status_code == 404 else "FAIL" record(2, "2.8", "Header widgets", status, f"HTTP {r.status_code if r else 'timeout'}", ms) # 2.9 Response time check times = [] for _ in range(5): _, ms = timed_get(client, f"{API_BASE}/health") times.append(ms) avg = sum(times) / len(times) if avg < 100: record(2, "2.9", "Response time < 100ms avg", "PASS", f"avg={avg:.0f}ms, min={min(times):.0f}ms, max={max(times):.0f}ms", avg) else: record(2, "2.9", "Response time < 100ms avg", "WARN", f"avg={avg:.0f}ms (target < 100ms)", avg) # --------------------------------------------------------------------------- # Phase 3: PM3 Hardware Integration # --------------------------------------------------------------------------- def phase3_pm3(client: httpx.Client): print("\n\033[1m═══ Phase 3: PM3 Hardware Integration ═══\033[0m") # 3.1 Device discovery r, ms = timed_get(client, f"{API_BASE}/pm3/devices") devices = [] device_id = None if r and r.status_code == 200: data = r.json() devices = data if isinstance(data, list) else data.get("devices", []) if devices: device_id = devices[0].get("device_id") record(3, "3.1", "Device discovery", "PASS", f"{len(devices)} device(s), id={device_id}", ms) else: record(3, "3.1", "Device discovery", "FAIL", "No devices found", ms) else: record(3, "3.1", "Device discovery", "FAIL", f"HTTP {r.status_code if r else 'timeout'}", ms) # 3.2 Device status if device_id: r, ms = timed_get(client, f"{API_BASE}/pm3/devices/{device_id}/status") if r and r.status_code == 200: record(3, "3.2", "Device status", "PASS", f"{r.json()}", ms, r.json()) else: record(3, "3.2", "Device status", "WARN", f"HTTP {r.status_code if r else 'timeout'}", ms) else: record(3, "3.2", "Device status", "SKIP", "No device_id", 0) # 3.3 PM3 overall status r, ms = timed_get(client, f"{API_BASE}/pm3/status") if r and r.status_code == 200: data = r.json() record(3, "3.3", "PM3 overall status", "PASS", f"connected={data.get('connected')}", ms, data) else: record(3, "3.3", "PM3 overall status", "FAIL", f"HTTP {r.status_code if r else 'timeout'}", ms) # 3.4 Create session session_id = None r, ms = timed_post(client, f"{API_BASE}/system/session/create", json={"force_takeover": True}) if r and r.status_code == 200: data = r.json() session_id = data.get("session_id") record(3, "3.4", "Create session", "PASS", f"id={session_id}", ms) else: record(3, "3.4", "Create session", "FAIL", f"HTTP {r.status_code if r else 'timeout'}: {r.text if r else ''}", ms) # 3.5 hw version if session_id: payload = {"command": "hw version", "session_id": session_id} if device_id: payload["device_id"] = device_id r, ms = timed_post(client, f"{API_BASE}/pm3/command", json=payload, timeout=PM3_CMD_TIMEOUT) if r and r.status_code == 200: data = r.json() output = data.get("output", "") if "iceman" in output.lower() or "client" in output.lower(): record(3, "3.5", "Execute hw version", "PASS", f"{ms:.0f}ms", ms, {"output_preview": output[:200]}) else: record(3, "3.5", "Execute hw version", "WARN", f"Unexpected output: {output[:100]}", ms) else: record(3, "3.5", "Execute hw version", "FAIL", f"HTTP {r.status_code if r else 'timeout'}", ms) else: record(3, "3.5", "Execute hw version", "SKIP", "No session", 0) # 3.6 hw status if session_id: payload = {"command": "hw status", "session_id": session_id} if device_id: payload["device_id"] = device_id r, ms = timed_post(client, f"{API_BASE}/pm3/command", json=payload, timeout=PM3_CMD_TIMEOUT) if r and r.status_code == 200: record(3, "3.6", "Execute hw status", "PASS", f"{ms:.0f}ms", ms) else: record(3, "3.6", "Execute hw status", "FAIL", f"HTTP {r.status_code if r else 'timeout'}", ms) else: record(3, "3.6", "Execute hw status", "SKIP", "No session", 0) # 3.7 hw tune (slow) if session_id: payload = {"command": "hw tune", "session_id": session_id} if device_id: payload["device_id"] = device_id r, ms = timed_post(client, f"{API_BASE}/pm3/command", json=payload, timeout=PM3_CMD_TIMEOUT) if r and r.status_code == 200: output = r.json().get("output", "") has_tune = "v @" in output.lower() or "khz" in output.lower() or "mhz" in output.lower() if has_tune: record(3, "3.7", "Execute hw tune", "PASS", f"{ms:.0f}ms", ms) else: record(3, "3.7", "Execute hw tune", "WARN", f"No tune data in output ({ms:.0f}ms)", ms) else: record(3, "3.7", "Execute hw tune", "FAIL", f"HTTP {r.status_code if r else 'timeout'}", ms) else: record(3, "3.7", "Execute hw tune", "SKIP", "No session", 0) # 3.8 HF search if session_id: payload = {"command": "hf search", "session_id": session_id} if device_id: payload["device_id"] = device_id r, ms = timed_post(client, f"{API_BASE}/pm3/command", json=payload, timeout=PM3_CMD_TIMEOUT) if r and r.status_code == 200: record(3, "3.8", "HF search", "PASS", f"{ms:.0f}ms", ms) else: record(3, "3.8", "HF search", "FAIL", f"HTTP {r.status_code if r else 'timeout'}", ms) else: record(3, "3.8", "HF search", "SKIP", "No session", 0) # 3.9 LF search if session_id: payload = {"command": "lf search", "session_id": session_id} if device_id: payload["device_id"] = device_id r, ms = timed_post(client, f"{API_BASE}/pm3/command", json=payload, timeout=PM3_CMD_TIMEOUT) if r and r.status_code == 200: record(3, "3.9", "LF search", "PASS", f"{ms:.0f}ms", ms) else: record(3, "3.9", "LF search", "FAIL", f"HTTP {r.status_code if r else 'timeout'}", ms) else: record(3, "3.9", "LF search", "SKIP", "No session", 0) # 3.10 LED identify if device_id: r, ms = timed_post(client, f"{API_BASE}/pm3/devices/{device_id}/identify") if r and r.status_code == 200: record(3, "3.10", "LED identify", "PASS", f"{ms:.0f}ms", ms) else: record(3, "3.10", "LED identify", "WARN", f"HTTP {r.status_code if r else 'timeout'}", ms) else: record(3, "3.10", "LED identify", "SKIP", "No device_id", 0) # 3.11 Command history r, ms = timed_get(client, f"{API_BASE}/pm3/commands/history") if r and r.status_code == 200: data = r.json() count = len(data) if isinstance(data, list) else data.get("count", "?") record(3, "3.11", "Command history", "PASS", f"{count} entries", ms) else: status = "WARN" if r and r.status_code == 404 else "FAIL" record(3, "3.11", "Command history", status, f"HTTP {r.status_code if r else 'timeout'}", ms) # 3.12 Session release if session_id: r, ms = timed_post(client, f"{API_BASE}/system/session/{session_id}/release") if r and r.status_code == 200: record(3, "3.12", "Session release", "PASS", f"{ms:.0f}ms", ms) else: record(3, "3.12", "Session release", "FAIL", f"HTTP {r.status_code if r else 'timeout'}", ms) else: record(3, "3.12", "Session release", "SKIP", "No session", 0) # 3.14 Command execution time (hw version should be < 1s) hw_ver_results = [r for r in results if r.test_id == "3.5"] if hw_ver_results and hw_ver_results[0].status == "PASS": dur = hw_ver_results[0].duration_ms if dur < 1000: record(3, "3.14", "Command time < 1s", "PASS", f"{dur:.0f}ms", dur) else: record(3, "3.14", "Command time < 1s", "WARN", f"{dur:.0f}ms (target < 1000ms)", dur) else: record(3, "3.14", "Command time < 1s", "SKIP", "hw version not tested", 0) # --------------------------------------------------------------------------- # Phase 4: WiFi Manager # --------------------------------------------------------------------------- def phase4_wifi(client: httpx.Client): print("\n\033[1m═══ Phase 4: WiFi Manager ═══\033[0m") # 4.1 WiFi status r, ms = timed_get(client, f"{API_BASE}/wifi/status") wifi_data = {} if r and r.status_code == 200: wifi_data = r.json() has_fields = "mode" in wifi_data and "interfaces" in wifi_data if has_fields: record(4, "4.1", "WiFi status", "PASS", f"mode={wifi_data.get('mode')}, ssid={wifi_data.get('current_ssid')}", ms, wifi_data) else: record(4, "4.1", "WiFi status", "WARN", "Missing expected fields", ms, wifi_data) else: record(4, "4.1", "WiFi status", "FAIL", f"HTTP {r.status_code if r else 'timeout'}", ms) # 4.2 Interface detection interfaces = wifi_data.get("interfaces", []) wlan0 = next((i for i in interfaces if i.get("name") == "wlan0"), None) if wlan0 and wlan0.get("is_up"): record(4, "4.2", "Interface detection (wlan0)", "PASS", f"mac={wlan0.get('mac')}, up=True", 0) elif wlan0: record(4, "4.2", "Interface detection (wlan0)", "WARN", "wlan0 found but not up", 0) else: record(4, "4.2", "Interface detection (wlan0)", "FAIL", "wlan0 not found", 0) # 4.3 Network scan r, ms = timed_get(client, f"{API_BASE}/wifi/scan", timeout=15.0) if r and r.status_code == 200: data = r.json() networks = data if isinstance(data, list) else data.get("networks", []) record(4, "4.3", "Network scan", "PASS", f"{len(networks)} networks found", ms) else: record(4, "4.3", "Network scan", "WARN", f"HTTP {r.status_code if r else 'timeout'} (scan may require AP mode)", ms) # 4.4 Saved networks r, ms = timed_get(client, f"{API_BASE}/wifi/saved") if r and r.status_code == 200: data = r.json() count = len(data) if isinstance(data, list) else data.get("count", "?") record(4, "4.4", "Saved networks", "PASS", f"{count} saved", ms) else: record(4, "4.4", "Saved networks", "WARN", f"HTTP {r.status_code if r else 'timeout'}", ms) # 4.5 Current connection ip = wifi_data.get("current_ip", "") if ip: record(4, "4.5", "Current connection", "PASS", f"ip={ip}", 0) else: record(4, "4.5", "Current connection", "WARN", "No current IP reported", 0) # 4.6 WiFi mode report mode = wifi_data.get("mode", "unknown") record(4, "4.6", "WiFi mode report", "PASS" if mode != "unknown" else "WARN", f"mode={mode}", 0) # --------------------------------------------------------------------------- # Phase 5: HTTPS & SSL # --------------------------------------------------------------------------- def phase5_https(client: httpx.Client): print("\n\033[1m═══ Phase 5: HTTPS & SSL (Phase 4 Feature) ═══\033[0m") # 5.1 SSL info r, ms = timed_get(client, f"{API_BASE}/system/ssl/info") ssl_data = {} if r and r.status_code == 200: ssl_data = r.json() record(5, "5.1", "SSL info endpoint", "PASS", f"{ms:.0f}ms", ms, ssl_data) elif r and r.status_code == 404: record(5, "5.1", "SSL info endpoint", "WARN", "Endpoint not found (404)", ms) return # Skip remaining SSL tests else: record(5, "5.1", "SSL info endpoint", "FAIL", f"HTTP {r.status_code if r else 'timeout'}", ms) return # 5.2 SSL status field if "enabled" in ssl_data: record(5, "5.2", "SSL enabled field present", "PASS", f"enabled={ssl_data['enabled']}", 0) else: record(5, "5.2", "SSL enabled field present", "FAIL", "Missing 'enabled' field", 0) # 5.3 Certificate details sans = ssl_data.get("san", ssl_data.get("subject_alt_names", [])) if sans: has_local = any("dangerous-pi.local" in str(s) for s in sans) record(5, "5.3", "Certificate SANs", "PASS" if has_local else "WARN", f"SANs={sans}", 0) elif ssl_data.get("certificate_exists") is False: record(5, "5.3", "Certificate SANs", "WARN", "No certificate generated yet", 0) else: record(5, "5.3", "Certificate SANs", "WARN", "No SAN data in response", 0) # 5.4 HTTPS access try: with httpx.Client(verify=False) as https_client: r2, ms = timed_get(https_client, f"{HTTPS_BASE}/api/health") if r2 and r2.status_code == 200: record(5, "5.4", "HTTPS access", "PASS", f"{ms:.0f}ms", ms) else: record(5, "5.4", "HTTPS access", "WARN", f"HTTPS not responding (may not be enabled): {r2.status_code if r2 else 'timeout'}", ms) except Exception as e: record(5, "5.4", "HTTPS access", "WARN", f"HTTPS not available: {e}", 0) # 5.5 SSL toggle (read-only check - don't actually toggle) record(5, "5.5", "SSL toggle endpoint", "SKIP", "Skipped to avoid disrupting connection", 0) # 5.6 Cert regeneration (read-only check) record(5, "5.6", "Cert regeneration", "SKIP", "Skipped to avoid disrupting HTTPS state", 0) # 5.7 & 5.8 Serial checks for cert script and nginx if serial_console._ser and serial_console._ser.is_open: out = serial_console.run_command( "test -x /opt/dangerous-pi/scripts/generate-ssl-cert.sh && echo SCRIPTOK || echo NOTFOUND" ) if "SCRIPTOK" in out: record(5, "5.7", "SSL cert script exists", "PASS", "Script at /opt/dangerous-pi/scripts/generate-ssl-cert.sh", 0) else: record(5, "5.7", "SSL cert script exists", "WARN", "Script not at expected path", 0) out = serial_console.run_command("ls /etc/nginx/sites-available/ 2>/dev/null | head -5") if "dangerous" in out.lower() or "nginx" in out.lower(): record(5, "5.8", "nginx config files", "PASS", f"Found configs", 0) else: record(5, "5.8", "nginx config files", "WARN", "No dangerous-pi nginx config found", 0) else: record(5, "5.7", "SSL cert script exists", "SKIP", "Serial not available", 0) record(5, "5.8", "nginx config files", "SKIP", "Serial not available", 0) # --------------------------------------------------------------------------- # Phase 6: Authentication & WebSocket # --------------------------------------------------------------------------- def phase6_auth(client: httpx.Client, safe_only: bool = False): print("\n\033[1m═══ Phase 6: Authentication & WebSocket ═══\033[0m") # 6.1 Auth status (should be disabled by default) r, ms = timed_get(client, f"{API_BASE}/auth/status") auth_enabled = False if r and r.status_code == 200: data = r.json() auth_enabled = data.get("auth_enabled", False) record(6, "6.1", "Auth status check", "PASS", f"auth_enabled={auth_enabled}", ms, data) else: record(6, "6.1", "Auth status check", "FAIL", f"HTTP {r.status_code if r else 'timeout'}", ms) if safe_only: record(6, "6.2", "Enable auth via serial", "SKIP", "Safe mode", 0) record(6, "6.3", "Auth status (enabled)", "SKIP", "Safe mode", 0) record(6, "6.4", "Unauthenticated request rejected", "SKIP", "Safe mode", 0) record(6, "6.5", "Authenticated request", "SKIP", "Safe mode", 0) record(6, "6.6", "Get WS token", "SKIP", "Safe mode", 0) record(6, "6.7", "WebSocket no token rejected", "SKIP", "Safe mode", 0) record(6, "6.8", "WebSocket with token", "SKIP", "Safe mode", 0) record(6, "6.9", "Disable auth via serial", "SKIP", "Safe mode", 0) elif not auth_enabled: # Auth is disabled - test what we can without enabling record(6, "6.2", "Enable auth via serial", "SKIP", "Auth is disabled, skipping auth toggle tests (use --phase 6 to run interactively)", 0) for tid in ["6.3", "6.4", "6.5", "6.6", "6.7", "6.8", "6.9"]: record(6, tid, f"Auth test {tid}", "SKIP", "Auth disabled on Pi", 0) # 6.10 WebSocket event stream (works regardless of auth) try: import websockets import websockets.sync.client ws_url = f"ws://{PI_HOST}:8000/ws/events" try: ws = websockets.sync.client.connect(ws_url, open_timeout=5) # Wait for a message try: msg = ws.recv(timeout=15) data = json.loads(msg) if isinstance(msg, str) else msg record(6, "6.10", "WebSocket event stream", "PASS", f"Received event: {str(data)[:100]}", 0, {"first_event": str(data)[:200]}) except TimeoutError: record(6, "6.10", "WebSocket event stream", "WARN", "Connected but no events received in 15s", 0) finally: ws.close() except Exception as e: record(6, "6.10", "WebSocket event stream", "FAIL", f"Connection failed: {e}", 0) except ImportError: record(6, "6.10", "WebSocket event stream", "SKIP", "websockets not installed", 0) # --------------------------------------------------------------------------- # Phase 7: Plugin System & Hooks # --------------------------------------------------------------------------- def phase7_plugins(client: httpx.Client): print("\n\033[1m═══ Phase 7: Plugin System & Hooks ═══\033[0m") # 7.1 Plugin discovery r, ms = timed_get(client, f"{API_BASE}/plugins/discover") if r and r.status_code == 200: data = r.json() plugins = data.get("plugins", []) has_hello = "hello_world" in str(plugins) record(7, "7.1", "Plugin discovery", "PASS" if has_hello else "WARN", f"Found: {plugins}", ms, data) else: record(7, "7.1", "Plugin discovery", "FAIL", f"HTTP {r.status_code if r else 'timeout'}", ms) # 7.2 Plugin list r, ms = timed_get(client, f"{API_BASE}/plugins/list") if r and r.status_code == 200: data = r.json() count = len(data) if isinstance(data, list) else "?" record(7, "7.2", "Plugin list", "PASS", f"{count} plugin(s)", ms) else: record(7, "7.2", "Plugin list", "FAIL", f"HTTP {r.status_code if r else 'timeout'}", ms) # 7.3 Plugin info r, ms = timed_get(client, f"{API_BASE}/plugins/hello_world") if r and r.status_code == 200: data = r.json() record(7, "7.3", "Plugin info (hello_world)", "PASS", f"version={data.get('metadata', {}).get('version')}", ms, data) else: record(7, "7.3", "Plugin info (hello_world)", "WARN", f"HTTP {r.status_code if r else 'timeout'}", ms) # 7.4 Load plugin r, ms = timed_post(client, f"{API_BASE}/plugins/load", json={"plugin_id": "hello_world"}) if r and r.status_code == 200: record(7, "7.4", "Load plugin", "PASS", f"{ms:.0f}ms", ms) else: record(7, "7.4", "Load plugin", "WARN", f"HTTP {r.status_code if r else 'timeout'}: {r.text[:100] if r else ''}", ms) # 7.5 Enable plugin r, ms = timed_post(client, f"{API_BASE}/plugins/enable", json={"plugin_id": "hello_world"}) if r and r.status_code == 200: record(7, "7.5", "Enable plugin", "PASS", f"{ms:.0f}ms", ms) else: record(7, "7.5", "Enable plugin", "WARN", f"HTTP {r.status_code if r else 'timeout'}: {r.text[:100] if r else ''}", ms) # 7.6 Hook trigger - run a PM3 command and check if hook fires # Create a session, run command, check logs r_sess, _ = timed_post(client, f"{API_BASE}/system/session/create", json={"force_takeover": True}) if r_sess and r_sess.status_code == 200: sid = r_sess.json().get("session_id") r_cmd, ms = timed_post(client, f"{API_BASE}/pm3/command", json={"command": "hw version", "session_id": sid}, timeout=PM3_CMD_TIMEOUT) if r_cmd and r_cmd.status_code == 200: record(7, "7.6", "Hook trigger (pm3_command)", "PASS", "PM3 command executed with plugin enabled", ms) else: record(7, "7.6", "Hook trigger (pm3_command)", "WARN", f"Command failed: {r_cmd.status_code if r_cmd else 'timeout'}", ms) # Release session timed_post(client, f"{API_BASE}/system/session/{sid}/release") else: record(7, "7.6", "Hook trigger (pm3_command)", "SKIP", "Could not create session", 0) # 7.7 Plugin widgets r, ms = timed_get(client, f"{API_BASE}/system/widgets") if r and r.status_code == 200: record(7, "7.7", "Plugin widgets", "PASS", f"{ms:.0f}ms", ms) else: record(7, "7.7", "Plugin widgets", "WARN", f"HTTP {r.status_code if r else 'timeout'}", ms) # 7.8 Disable plugin r, ms = timed_post(client, f"{API_BASE}/plugins/disable", json={"plugin_id": "hello_world"}) if r and r.status_code == 200: record(7, "7.8", "Disable plugin", "PASS", f"{ms:.0f}ms", ms) else: record(7, "7.8", "Disable plugin", "WARN", f"HTTP {r.status_code if r else 'timeout'}", ms) # 7.9 Unload plugin r, ms = timed_post(client, f"{API_BASE}/plugins/unload", json={"plugin_id": "hello_world"}) if r and r.status_code == 200: record(7, "7.9", "Unload plugin", "PASS", f"{ms:.0f}ms", ms) else: record(7, "7.9", "Unload plugin", "WARN", f"HTTP {r.status_code if r else 'timeout'}", ms) # 7.10 Invalid plugin r, ms = timed_get(client, f"{API_BASE}/plugins/nonexistent_plugin_xyz") if r and r.status_code == 404: record(7, "7.10", "Invalid plugin returns 404", "PASS", f"{ms:.0f}ms", ms) elif r: record(7, "7.10", "Invalid plugin returns 404", "WARN", f"Expected 404, got {r.status_code}", ms) else: record(7, "7.10", "Invalid plugin returns 404", "FAIL", "No response", ms) # --------------------------------------------------------------------------- # Phase 8: BLE & UPS Hardware # --------------------------------------------------------------------------- def phase8_hardware(client: httpx.Client): print("\n\033[1m═══ Phase 8: BLE & UPS Hardware ═══\033[0m") # --- BLE --- # 8.1 BLE status API r, ms = timed_get(client, f"{API_BASE}/system/ble/status") ble_data = {} if r and r.status_code == 200: ble_data = r.json() record(8, "8.1", "BLE status API", "PASS", f"enabled={ble_data.get('enabled')}, advertising={ble_data.get('advertising')}, name={ble_data.get('device_name')}", ms, ble_data) else: record(8, "8.1", "BLE status API", "FAIL", f"HTTP {r.status_code if r else 'timeout'}", ms) # 8.3 Start advertising (before scan so we have something to find) r, ms = timed_post(client, f"{API_BASE}/system/ble/advertising/start") if r and r.status_code == 200: record(8, "8.3", "Start BLE advertising", "PASS", f"{ms:.0f}ms", ms) else: record(8, "8.3", "Start BLE advertising", "WARN", f"HTTP {r.status_code if r else 'timeout'}", ms) # 8.2 BLE scan from laptop (after ensuring advertising is started) import time as _time _time.sleep(2) # Give BLE advertisement time to propagate try: scan_result = subprocess.run( ["bluetoothctl", "--timeout", "12", "scan", "le"], capture_output=True, text=True, timeout=20 ) scan_output = scan_result.stdout + scan_result.stderr if BLE_DEVICE_NAME.lower() in scan_output.lower() or "dangerous" in scan_output.lower(): record(8, "8.2", "BLE scan from laptop", "PASS", f"Found {BLE_DEVICE_NAME} in BLE scan", 0) else: record(8, "8.2", "BLE scan from laptop", "WARN", f"{BLE_DEVICE_NAME} not found in scan (may need longer scan)", 0) except FileNotFoundError: record(8, "8.2", "BLE scan from laptop", "SKIP", "bluetoothctl not found", 0) except subprocess.TimeoutExpired: record(8, "8.2", "BLE scan from laptop", "WARN", "Scan timed out", 0) except Exception as e: record(8, "8.2", "BLE scan from laptop", "WARN", f"Scan error: {e}", 0) # 8.4 Stop advertising r, ms = timed_post(client, f"{API_BASE}/system/ble/advertising/stop") if r and r.status_code == 200: record(8, "8.4", "Stop BLE advertising", "PASS", f"{ms:.0f}ms", ms) else: record(8, "8.4", "Stop BLE advertising", "WARN", f"HTTP {r.status_code if r else 'timeout'}", ms) # 8.5 Send notification r, ms = timed_post(client, f"{API_BASE}/system/ble/notify", json={"type": "pm3_status", "message": "Hardware test notification"}) if r and r.status_code == 200: record(8, "8.5", "Send BLE notification", "PASS", f"{ms:.0f}ms", ms) else: record(8, "8.5", "Send BLE notification", "WARN", f"HTTP {r.status_code if r else 'timeout'}", ms) # Restart advertising for normal operation timed_post(client, f"{API_BASE}/system/ble/advertising/start") # --- UPS --- # 8.7 UPS status r, ms = timed_get(client, f"{API_BASE}/system/ups/status") ups_data = {} if r and r.status_code == 200: ups_data = r.json() if ups_data.get("is_available"): record(8, "8.7", "UPS status", "PASS", f"battery={ups_data.get('battery_percentage')}%, voltage={ups_data.get('voltage')}V, source={ups_data.get('power_source')}", ms, ups_data) else: record(8, "8.7", "UPS status", "WARN", f"UPS not available: {ups_data.get('error_message', 'no error info')}", ms, ups_data) else: record(8, "8.7", "UPS status", "FAIL", f"HTTP {r.status_code if r else 'timeout'}", ms) # 8.8 UPS thresholds r, ms = timed_post(client, f"{API_BASE}/system/ups/thresholds", json={"shutdown_threshold": 15, "warning_threshold": 25}) if r and r.status_code == 200: record(8, "8.8", "UPS thresholds", "PASS", f"{ms:.0f}ms", ms) elif r and r.status_code in (404, 503): record(8, "8.8", "UPS thresholds", "WARN", f"HTTP {r.status_code} (UPS may not be available)", ms) else: record(8, "8.8", "UPS thresholds", "WARN", f"HTTP {r.status_code if r else 'timeout'}", ms) # 8.9 Power source source = ups_data.get("power_source", "unknown") if source != "unknown": record(8, "8.9", "Power source detection", "PASS", f"source={source}", 0) else: record(8, "8.9", "Power source detection", "WARN", "Power source unknown (UPS may not be detected)", 0) # 8.10 I2C via serial if serial_console._ser and serial_console._ser.is_open: out = serial_console.run_command( "python3 -c \"import smbus2; bus=smbus2.SMBus(1); print('I2C OK:', hex(bus.read_byte(0x75))); bus.close()\" 2>&1" ) if "i2c ok" in out.lower(): record(8, "8.10", "I2C device at 0x75", "PASS", "PiSugar responds", 0) elif "error" in out.lower() or "errno" in out.lower(): record(8, "8.10", "I2C device at 0x75", "WARN", f"I2C error: {out.strip()[:100]}", 0) else: record(8, "8.10", "I2C device at 0x75", "WARN", f"Output: {out.strip()[:100]}", 0) else: record(8, "8.10", "I2C device at 0x75", "SKIP", "Serial not available", 0) # --------------------------------------------------------------------------- # Phase 9: Frontend & Performance # --------------------------------------------------------------------------- def phase9_frontend(client: httpx.Client): print("\n\033[1m═══ Phase 9: Frontend & Performance ═══\033[0m") pages = [ ("9.1", "/", "Frontend loads", "root"), ("9.2", "/", "Dashboard page", "dashboard"), ("9.3", "/commands", "Commands page", "command"), ("9.4", "/settings", "Settings page", "settings"), ("9.5", "/logs", "Logs page", "log"), ] for tid, path, name, keyword in pages: try: # Try HTTPS first (if enabled), then fall back to HTTP try: r = httpx.get(f"{HTTPS_BASE}{path}", timeout=10.0, verify=False) except Exception: r = httpx.get(f"{HTTP_UI}{path}", timeout=10.0, follow_redirects=True) t_ms = 0 if r.status_code == 200: body = r.text.lower() # Remix apps should have root element and scripts is_app = "root" in body or " str: now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") total = len(results) passed = sum(1 for r in results if r.status == "PASS") failed = sum(1 for r in results if r.status == "FAIL") warned = sum(1 for r in results if r.status == "WARN") skipped = sum(1 for r in results if r.status == "SKIP") lines = [ f"# Dangerous Pi - Hardware Test Results", f"", f"**Test Date**: {now}", f"**Target**: {PI_HOST} ({API_BASE})", f"**Serial**: {SERIAL_PORT} @ {SERIAL_BAUD}", f"**Tester**: Automated (test_hardware_2026-03-03.py)", f"", f"---", f"", f"## Summary", f"", f"| Metric | Count |", f"|--------|-------|", f"| Total Tests | {total} |", f"| **PASS** | {passed} |", f"| **FAIL** | {failed} |", f"| **WARN** | {warned} |", f"| **SKIP** | {skipped} |", f"| **Pass Rate** | {passed/total*100:.1f}% (excl. skips: {passed/max(total-skipped,1)*100:.1f}%) |", f"", f"---", f"", ] # Group by phase phases = { 1: "Phase 1: Regression - Jan 23 Issues", 2: "Phase 2: Core Health & System Endpoints", 3: "Phase 3: PM3 Hardware Integration", 4: "Phase 4: WiFi Manager", 5: "Phase 5: HTTPS & SSL", 6: "Phase 6: Authentication & WebSocket", 7: "Phase 7: Plugin System & Hooks", 8: "Phase 8: BLE & UPS Hardware", 9: "Phase 9: Frontend & Performance", } for phase_num, phase_name in phases.items(): phase_results = [r for r in results if r.phase == phase_num] if not phase_results: continue p_pass = sum(1 for r in phase_results if r.status == "PASS") p_fail = sum(1 for r in phase_results if r.status == "FAIL") p_warn = sum(1 for r in phase_results if r.status == "WARN") p_skip = sum(1 for r in phase_results if r.status == "SKIP") icon = "✅" if p_fail == 0 and p_warn == 0 else ("❌" if p_fail > 0 else "⚠️") lines.append(f"## {icon} {phase_name}") lines.append(f"") lines.append(f"| # | Test | Status | Detail |") lines.append(f"|---|------|--------|--------|") for r in phase_results: status_icon = {"PASS": "✅", "FAIL": "❌", "WARN": "⚠️", "SKIP": "⏭️"}[r.status] detail = r.detail.replace("|", "\\|")[:120] lines.append(f"| {r.test_id} | {r.name} | {status_icon} {r.status} | {detail} |") lines.append(f"") lines.append(f"**Phase Summary**: {p_pass} pass, {p_fail} fail, {p_warn} warn, {p_skip} skip") lines.append(f"") lines.append(f"---") lines.append(f"") # Issues section failures = [r for r in results if r.status == "FAIL"] warnings = [r for r in results if r.status == "WARN"] if failures: lines.append(f"## Issues Found (FAIL)") lines.append(f"") for r in failures: lines.append(f"### {r.test_id}: {r.name}") lines.append(f"- **Detail**: {r.detail}") if r.data: lines.append(f"- **Data**: `{json.dumps(r.data)[:300]}`") lines.append(f"") if warnings: lines.append(f"## Warnings") lines.append(f"") for r in warnings: lines.append(f"- **{r.test_id} {r.name}**: {r.detail}") lines.append(f"") lines.append(f"---") lines.append(f"") lines.append(f"*Generated by test_hardware_2026-03-03.py*") return "\n".join(lines) # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def main(): parser = argparse.ArgumentParser(description="Dangerous Pi Hardware Test Suite") parser.add_argument("--phase", type=int, help="Run specific phase (1-9)") parser.add_argument("--safe-only", action="store_true", help="Skip destructive tests") parser.add_argument("--no-serial", action="store_true", help="Skip serial console tests") parser.add_argument("--report", default="TEST_RESULTS_2026-03-03.md", help="Output report file") args = parser.parse_args() print(f"\033[1m{'='*60}\033[0m") print(f"\033[1m Dangerous Pi - Hardware Integration Test Suite\033[0m") print(f"\033[1m Target: {PI_HOST} | Serial: {SERIAL_PORT}\033[0m") print(f"\033[1m Date: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\033[0m") print(f"\033[1m{'='*60}\033[0m") # Open serial console if not args.no_serial: print(f"\n[*] Opening serial console {SERIAL_PORT}...") if serial_console.open(): print(f" Serial console connected") # Send a newline to ensure prompt serial_console.run_command("", timeout=1) else: print(f" Serial console not available (tests will skip serial checks)") # Create HTTP client client = httpx.Client(timeout=REQUEST_TIMEOUT) # Quick connectivity check print(f"\n[*] Checking connectivity to {PI_HOST}...") r, ms = timed_get(client, f"{API_BASE}/health") if not r or r.status_code != 200: print(f"\033[31m FATAL: Cannot reach {API_BASE}/health\033[0m") print(f" Ensure the Pi is running and accessible") sys.exit(1) print(f" Backend healthy ({ms:.0f}ms)") # Run phases phase_map = { 1: lambda: phase1_regression(client), 2: lambda: phase2_system(client), 3: lambda: phase3_pm3(client), 4: lambda: phase4_wifi(client), 5: lambda: phase5_https(client), 6: lambda: phase6_auth(client, args.safe_only), 7: lambda: phase7_plugins(client), 8: lambda: phase8_hardware(client), 9: lambda: phase9_frontend(client), } if args.phase: if args.phase in phase_map: phase_map[args.phase]() else: print(f"Unknown phase {args.phase}. Valid: 1-9") sys.exit(1) else: for phase_func in phase_map.values(): phase_func() # Cleanup serial_console.close() client.close() # Print summary total = len(results) passed = sum(1 for r in results if r.status == "PASS") failed = sum(1 for r in results if r.status == "FAIL") warned = sum(1 for r in results if r.status == "WARN") skipped = sum(1 for r in results if r.status == "SKIP") print(f"\n\033[1m{'='*60}\033[0m") print(f"\033[1m RESULTS: {passed} pass, {failed} fail, {warned} warn, {skipped} skip / {total} total\033[0m") if failed == 0: print(f"\033[32m All non-skipped tests passed!\033[0m") else: print(f"\033[31m {failed} test(s) FAILED\033[0m") print(f"\033[1m{'='*60}\033[0m") # Generate report report = generate_report() report_path = Path(args.report) report_path.write_text(report) print(f"\n Report written to: {report_path}") # Exit with failure code if any tests failed sys.exit(1 if failed > 0 else 0) if __name__ == "__main__": main()