a9acdb85ce
Replace PM3 compile-from-source in pi-gen with pre-built tarball extraction (saves 43-58 min). Merge stagePM3 into stageDangerousPi as 02-pm3-install substage, renumber all subsequent substages. Switch CI PM3 build to native ARM64 runner (ubuntu-24.04-arm64) eliminating QEMU overhead. Add weekly base-image workflow for pre-baking stages 0-2. Support PM3_TARBALL, BASE_IMAGE, and APT_PROXY env vars in build-image.sh. Also includes prior Phase 5 work: theme system, design system integration, component update system, OS updates, CI build pipeline, and test results. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
1202 lines
49 KiB
Python
1202 lines
49 KiB
Python
#!/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 "<script" in body or "remix" in body
|
||
if is_app:
|
||
record(9, tid, name, "PASS",
|
||
f"{len(r.text)} bytes, has app markup", t_ms)
|
||
else:
|
||
record(9, tid, name, "WARN",
|
||
f"Got HTML but no app markers ({len(r.text)} bytes)", t_ms)
|
||
else:
|
||
record(9, tid, name, "FAIL", f"HTTP {r.status_code}", t_ms)
|
||
except Exception as e:
|
||
record(9, tid, name, "FAIL", f"Error: {e}", 0)
|
||
|
||
# 9.6 Response time benchmark
|
||
times = []
|
||
for _ in range(10):
|
||
_, ms = timed_get(client, f"{API_BASE}/health")
|
||
times.append(ms)
|
||
avg = sum(times) / len(times)
|
||
p95 = sorted(times)[int(len(times) * 0.95)]
|
||
record(9, "9.6", "Response time (10 requests)", "PASS" if avg < 100 else "WARN",
|
||
f"avg={avg:.0f}ms, p95={p95:.0f}ms, min={min(times):.0f}ms, max={max(times):.0f}ms", avg)
|
||
|
||
# 9.7 Memory usage
|
||
r, _ = timed_get(client, f"{API_BASE}/system/info")
|
||
if r and r.status_code == 200:
|
||
data = r.json()
|
||
mem_mb = data.get("memory_used", 0) / 1024 / 1024
|
||
record(9, "9.7", "Memory usage", "PASS" if mem_mb < 200 else "WARN",
|
||
f"system total used: {mem_mb:.0f}MB", 0, data)
|
||
else:
|
||
record(9, "9.7", "Memory usage", "SKIP", "Could not get system info", 0)
|
||
|
||
# 9.8 CPU temperature
|
||
if r and r.status_code == 200:
|
||
data = r.json()
|
||
temp = data.get("cpu_temp", data.get("cpu", {}).get("temperature", 0))
|
||
record(9, "9.8", "CPU temperature", "PASS" if temp < 70 else "WARN",
|
||
f"{temp}°C (target < 70°C idle)", 0)
|
||
else:
|
||
record(9, "9.8", "CPU temperature", "SKIP", "Could not get system info", 0)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Report generator
|
||
# ---------------------------------------------------------------------------
|
||
def generate_report() -> 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()
|