test(esp32): integration tests for QEMU examples

Closes the loop on the four ESP32 fixes that landed earlier in this
series. Each previously-broken or noisy example now has a regression
test that compiles the sketch through the production ESP-IDF compiler
and runs it in QEMU via esp32_worker.py — the same code path the
WebSocket /ws/{client_id} endpoint drives in production. Testing the
worker directly skips the WS transport but exercises the same compile
→ flash → boot → serial cascade.

Coverage:
- TestEsp32SerialCleanliness — DHT22, Servo+Pot, Joystick, Dual ADC
  Asserts the user's Serial.print substring shows up AND no
  `I (xxx) gpio:|wifi:|phy:` info-level ESP-IDF logs leak through.
  This validates the sdkconfig CONFIG_LOG_DEFAULT_LEVEL_WARN change
  from commit b373c97.

- TestEsp32CompileSuccess — BLE Advertise, LEDC RGB
  BLE Advertise validates the sdkconfig switch to Bluedroid (was
  NimBLE-only, which broke arduino-esp32's BLEDevice.h).
  LEDC RGB validates velxio_compat.h's ledcAttach() shim from
  commit f6f6f43; the sketch uses the arduino-esp32 3.x one-shot API
  on a 2.0.17 toolchain.

- TestEsp32WiFiSketches — WiFi Connect, WiFi WebServer
  Regression coverage to make sure the sdkconfig changes didn't break
  WiFi association. Connect must reach an "IP Address:" line; Server
  must report "Server started".

- frontend/src/__tests__/component-metadata-bmp280.test.ts
  Sanity check that the BMP280 entry from commit 1f3f2e0 survives
  metadata regeneration.

All ESP-IDF/QEMU tests use unittest.skipUnless on
_toolchain_available() so they no-op cleanly on dev boxes without
libqemu-xtensa, and only do real work in the Docker CI image.

Sketches are inlined verbatim from the public Velxio examples.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
David Montero Crespo 2026-05-08 18:39:22 -03:00
parent 1f3f2e07bb
commit c939005bf7
2 changed files with 555 additions and 0 deletions

View File

@ -0,0 +1,14 @@
import { describe, it, expect } from 'vitest';
import metadata from '../../public/components-metadata.json';
describe('component metadata — BMP280 entry', () => {
it('is registered with id "bmp280" and tagName "velxio-bmp280"', () => {
const entry = (metadata as Array<Record<string, unknown>>).find(
(c) => c.id === 'bmp280',
);
expect(entry, 'BMP280 missing from components-metadata.json').toBeDefined();
expect(entry!.tagName).toBe('velxio-bmp280');
expect(entry!.category).toBe('sensor');
expect(entry!.pinCount).toBe(4);
});
});

View File

@ -0,0 +1,541 @@
#!/usr/bin/env python3
"""End-to-end QEMU tests for ESP32 example sketches reported by beta testers.
Each test:
1. Compiles a sketch via app.services.espidf_compiler.compile()
2. Launches esp32_worker.py with the resulting firmware
3. Captures serial output for N seconds
4. Asserts on expected substrings + verifies NO ANSI escape sequences
and NO `I (xxx) gpio:` info logs leak into the user's serial output
Skipped automatically when the ESP-IDF toolchain or libqemu shared
library are not available, so the suite is safe to run on Windows
dev boxes and only does real work in the Docker CI image.
Sketches are inlined verbatim from frontend/src/data/examples.ts and
frontend/src/data/examples-circuits.ts.
Run from the repo root:
pytest test/backend/integration/test_esp32_examples_qemu.py -v
Or directly:
python test/backend/integration/test_esp32_examples_qemu.py
"""
from __future__ import annotations
import asyncio
import base64
import io
import json
import re
import subprocess
import sys
import threading
import time
import unittest
from pathlib import Path
from typing import cast
# ── Path setup ────────────────────────────────────────────────────────────────
BACKEND_DIR = Path(__file__).resolve().parents[3] / 'backend'
SERVICES_DIR = BACKEND_DIR / 'app' / 'services'
WORKER = SERVICES_DIR / 'esp32_worker.py'
# libqemu has different filenames per platform.
if sys.platform == 'win32':
LIB_XTENSA = SERVICES_DIR / 'libqemu-xtensa.dll'
elif sys.platform == 'darwin':
LIB_XTENSA = SERVICES_DIR / 'libqemu-xtensa.dylib'
else:
LIB_XTENSA = SERVICES_DIR / 'libqemu-xtensa.so'
if str(BACKEND_DIR) not in sys.path:
sys.path.insert(0, str(BACKEND_DIR))
from app.services.espidf_compiler import espidf_compiler # type: ignore[import-not-found] # noqa: E402
# ── Regex helpers — bug #1 verifications ─────────────────────────────────────
ANSI_RE = re.compile(rb'\x1b\[[0-9;]*m')
# Matches lines like "I (53306) gpio: GPIO[4]| InputEn: 1| ..." that the
# beta tester reported leaking into user output before the WARN log-level fix.
ESPIDF_INFO_RE = re.compile(rb'^I \(\d+\) (gpio|wifi|phy|wifi_init):', re.MULTILINE)
# ─────────────────────────────────────────────────────────────────────────────
# Sketches under test (verbatim from frontend/src/data/examples.ts)
# ─────────────────────────────────────────────────────────────────────────────
SKETCH_DHT22 = r"""// ESP32 - DHT22 Temperature & Humidity Sensor
#include <DHT.h>
#define DHT_PIN 4
#define DHT_TYPE DHT22
DHT dht(DHT_PIN, DHT_TYPE);
void setup() {
Serial.begin(115200);
dht.begin();
delay(2000);
Serial.println("ESP32 DHT22 ready!");
}
void loop() {
delay(2000);
float h = dht.readHumidity();
float t = dht.readTemperature();
if (isnan(h) || isnan(t)) {
Serial.println("DHT22: waiting for sensor...");
return;
}
Serial.printf("Temp: %.1f C Humidity: %.1f %%\n", t, h);
}
"""
SKETCH_SERVO_POT = r"""// ESP32 - Servo controlled by Potentiometer
#include <ESP32Servo.h>
#define SERVO_PIN 13
#define POT_PIN 34
Servo myServo;
void setup() {
Serial.begin(115200);
myServo.attach(SERVO_PIN, 500, 2400);
Serial.println("ESP32 Servo + Pot control");
}
void loop() {
int raw = analogRead(POT_PIN);
int angle = map(raw, 0, 4095, 0, 180);
myServo.write(angle);
Serial.printf("Pot: %4d Angle: %3d deg\n", raw, angle);
delay(20);
}
"""
SKETCH_JOYSTICK = r"""// ESP32 - Analog Joystick
#define JOY_HORZ 35
#define JOY_VERT 34
#define JOY_BTN 15
void setup() {
Serial.begin(115200);
pinMode(JOY_BTN, INPUT_PULLUP);
Serial.println("ESP32 Joystick ready");
}
void loop() {
int x = analogRead(JOY_HORZ);
int y = analogRead(JOY_VERT);
bool btn = (digitalRead(JOY_BTN) == LOW);
int xPct = map(x, 0, 4095, -100, 100);
int yPct = map(y, 0, 4095, -100, 100);
Serial.printf("X=%4d(%4d%%) Y=%4d(%4d%%) BTN=%s\n",
x, xPct, y, yPct, btn ? "PRESSED" : "---");
delay(100);
}
"""
SKETCH_DUAL_ADC = r"""// ESP32 dual ADC - 12-bit, 3.3V reference
void setup() {
Serial.begin(115200);
analogReadResolution(12);
}
void loop() {
int ch1 = analogRead(34);
int ch2 = analogRead(35);
float v1 = ch1 * 3.3 / 4095.0;
float v2 = ch2 * 3.3 / 4095.0;
Serial.printf("CH1=%.3fV CH2=%.3fV\n", v1, v2);
delay(500);
}
"""
# Bug #3: uses arduino-esp32 3.x ledcAttach() — fails without velxio_compat.h shim.
SKETCH_LEDC_RGB = r"""// ESP32 LEDC PWM - RGB LED color cycling
#define R_PIN 16
#define G_PIN 17
#define B_PIN 18
void setup() {
ledcAttach(R_PIN, 5000, 8);
ledcAttach(G_PIN, 5000, 8);
ledcAttach(B_PIN, 5000, 8);
}
void loop() {
for(int h=0; h<360; h+=5) {
float r,g,b;
int i = h/60; float f = h/60.0-i;
switch(i%6) {
case 0: r=1; g=f; b=0; break;
case 1: r=1-f; g=1; b=0; break;
case 2: r=0; g=1; b=f; break;
case 3: r=0; g=1-f; b=1; break;
case 4: r=f; g=0; b=1; break;
case 5: r=1; g=0; b=1-f; break;
}
ledcWrite(R_PIN, (int)(r*255));
ledcWrite(G_PIN, (int)(g*255));
ledcWrite(B_PIN, (int)(b*255));
delay(30);
}
}
"""
SKETCH_WIFI_CONNECT = r"""#include <WiFi.h>
const char* ssid = "Velxio-GUEST";
void setup() {
Serial.begin(115200);
delay(1000);
Serial.println("ESP32 WiFi Connection Demo");
Serial.println("==========================");
Serial.printf("Connecting to %s", ssid);
WiFi.begin(ssid, "", 6);
while (WiFi.status() != WL_CONNECTED) {
delay(500);
Serial.print(".");
}
Serial.println(" Connected!");
Serial.printf("IP Address: %s\n", WiFi.localIP().toString().c_str());
}
void loop() {
delay(5000);
}
"""
SKETCH_WIFI_SERVER = r"""#include <WiFi.h>
#include <WebServer.h>
const char* ssid = "Velxio-GUEST";
WebServer server(80);
void handleRoot() {
server.send(200, "text/plain", "Hello from ESP32!");
}
void setup() {
Serial.begin(115200);
delay(1000);
Serial.println("ESP32 HTTP Server");
WiFi.begin(ssid, "", 6);
while (WiFi.status() != WL_CONNECTED) {
delay(500);
}
server.on("/", handleRoot);
server.begin();
Serial.printf("Server started at: http://%s/\n", WiFi.localIP().toString().c_str());
}
void loop() {
server.handleClient();
}
"""
# Bug #2: arduino-esp32 BLEDevice.h needs Bluedroid (was NimBLE-only before fix).
SKETCH_BLE_ADVERTISE = r"""#include <BLEDevice.h>
#include <BLEServer.h>
#include <BLEUtils.h>
#include <BLE2902.h>
#define SERVICE_UUID "4fafc201-1fb5-459e-8fcc-c5c9c331914b"
#define CHARACTERISTIC_UUID "beb5483e-36e1-4688-b7f5-ea07361b26a8"
BLEServer* pServer = nullptr;
BLECharacteristic* pCharacteristic = nullptr;
void setup() {
Serial.begin(115200);
delay(1000);
Serial.println("ESP32 BLE Advertise Demo");
BLEDevice::init("Velxio-ESP32");
pServer = BLEDevice::createServer();
BLEService *pService = pServer->createService(SERVICE_UUID);
pCharacteristic = pService->createCharacteristic(
CHARACTERISTIC_UUID,
BLECharacteristic::PROPERTY_READ |
BLECharacteristic::PROPERTY_WRITE |
BLECharacteristic::PROPERTY_NOTIFY
);
pCharacteristic->addDescriptor(new BLE2902());
pCharacteristic->setValue("Hello from Velxio!");
pService->start();
BLEAdvertising *pAdvertising = BLEDevice::getAdvertising();
pAdvertising->addServiceUUID(SERVICE_UUID);
pAdvertising->setScanResponse(true);
pAdvertising->start();
Serial.println("BLE advertising started!");
}
void loop() {
delay(2000);
}
"""
# ─────────────────────────────────────────────────────────────────────────────
# Helpers
# ─────────────────────────────────────────────────────────────────────────────
def _compile_via_espidf(sketch_code: str, fqbn: str = 'esp32:esp32:esp32') -> bytes:
"""Compile sketch through the production ESP-IDF compiler. Returns the
trimmed firmware bytes ready to hand to esp32_worker.py.
Raises AssertionError with stderr on compile failure so test diagnostics
are immediately visible.
"""
files = [{'name': 'sketch.ino', 'content': sketch_code}]
result = asyncio.run(espidf_compiler.compile(files, fqbn))
if not result.get('success'):
raise AssertionError(
f"Compile failed for fqbn={fqbn}:\n"
f" error: {result.get('error')}\n"
f" stderr: {(result.get('stderr') or '')[-2000:]}\n"
)
bin_b64 = result.get('binary_content')
if not bin_b64:
raise AssertionError(f"Compile reported success but no binary_content: {result}")
return base64.b64decode(bin_b64)
def _run_worker(firmware_bytes: bytes,
run_seconds: float = 8.0,
machine: str = 'esp32-picsimlab') -> dict:
"""Launch esp32_worker.py with firmware, collect events for `run_seconds`,
return aggregated dict. Mirrors the pattern in test_esp32c3_emulation.py.
Returns:
{
'serial_bytes': bytes, # concatenated UART0 output
'gpio_events': list[(pin, state)],
'sys_events': list[str],
'errors': list[str],
'booted': bool,
'all_events': list[dict],
}
"""
if not WORKER.exists():
raise unittest.SkipTest(f'Worker not found: {WORKER}')
if not LIB_XTENSA.exists():
raise unittest.SkipTest(f'libqemu not found: {LIB_XTENSA}')
cfg = {
'lib_path': str(LIB_XTENSA),
'firmware_b64': base64.b64encode(firmware_bytes).decode('ascii'),
'machine': machine,
}
proc = subprocess.Popen(
[sys.executable, str(WORKER)],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=False,
)
# PIPE was passed for all three streams, so they are non-None handles.
# Cast to satisfy the type checker (Optional narrowing doesn't flow
# through attribute access into closures).
p_stdin = cast(io.BufferedWriter, proc.stdin)
p_stdout = cast(io.BufferedReader, proc.stdout)
p_stderr = cast(io.BufferedReader, proc.stderr)
p_stdin.write((json.dumps(cfg) + '\n').encode('utf-8'))
p_stdin.flush()
serial_chunks: list[bytes] = []
gpio_events: list[tuple[int, int]] = []
sys_events: list[str] = []
errors: list[str] = []
all_events: list[dict] = []
def _read_stdout() -> None:
for raw_line in p_stdout:
line = raw_line.decode('utf-8', errors='replace').strip()
if not line:
continue
try:
evt = json.loads(line)
except Exception:
continue
all_events.append(evt)
t = evt.get('type')
if t == 'uart_tx':
# Single-byte event — append as a 1-byte chunk.
b = evt.get('byte')
if isinstance(b, int) and 0 <= b <= 255:
serial_chunks.append(bytes([b]))
elif t == 'serial_output':
data = evt.get('data')
if isinstance(data, str):
serial_chunks.append(data.encode('utf-8', errors='replace'))
elif t == 'gpio_change':
gpio_events.append((evt.get('pin'), evt.get('state')))
elif t == 'system':
sys_events.append(evt.get('event', ''))
elif t == 'error':
errors.append(evt.get('message', ''))
def _read_stderr() -> None:
for _ in p_stderr:
pass # discard; surface via errors[] if the worker emits a JSON error event
threading.Thread(target=_read_stdout, daemon=True).start()
threading.Thread(target=_read_stderr, daemon=True).start()
deadline = time.monotonic() + run_seconds
while time.monotonic() < deadline and proc.poll() is None:
time.sleep(0.2)
# Graceful stop, then force.
try:
p_stdin.write((json.dumps({'cmd': 'stop'}) + '\n').encode('utf-8'))
p_stdin.flush()
except Exception:
pass
time.sleep(0.5)
if proc.poll() is None:
proc.terminate()
try:
proc.wait(timeout=5)
except subprocess.TimeoutExpired:
proc.kill()
return {
'serial_bytes': b''.join(serial_chunks),
'gpio_events': gpio_events,
'sys_events': sys_events,
'errors': errors,
'booted': 'booted' in sys_events,
'all_events': all_events,
}
def _toolchain_available() -> bool:
return espidf_compiler.available and LIB_XTENSA.exists() and WORKER.exists()
# ─────────────────────────────────────────────────────────────────────────────
# Bug #1 — Serial cleanliness across multiple sketches
# ─────────────────────────────────────────────────────────────────────────────
@unittest.skipUnless(_toolchain_available(),
'Requires libqemu-xtensa + ESP-IDF toolchain (Docker CI)')
class TestEsp32SerialCleanliness(unittest.TestCase):
"""The reporter saw raw `[0;32m` ANSI codes and `I (xxx) gpio:` info logs
mixed into the Serial.print output of these examples. After the sdkconfig
log-level + ANSI strip fixes, neither should leak.
Note: ANSI verification is on the BACKEND side (the worker emits raw
UART bytes frontend strip is a separate concern, covered by the
SerialMonitor change in commit 3 of this series). The backend bug is
that ESP-IDF's logger tags lines with `I (xxx) gpio:` at INFO level
by default; we want only WARN+ leaking through.
"""
def _assert_clean_user_output(self, sketch: str, expected_substr: bytes,
run_seconds: float = 12.0):
fw = _compile_via_espidf(sketch)
out = _run_worker(fw, run_seconds=run_seconds)
self.assertTrue(out['booted'], f"firmware never booted; events={out['all_events'][:5]}")
self.assertFalse(
ESPIDF_INFO_RE.search(out['serial_bytes']),
f"ESP-IDF INFO logs leaked into user serial output:\n"
f"{out['serial_bytes'][:400]!r}"
)
self.assertIn(
expected_substr, out['serial_bytes'],
f"Expected {expected_substr!r} in user output but got:\n"
f"{out['serial_bytes'][:400]!r}"
)
def test_dht22_serial_is_clean(self):
self._assert_clean_user_output(SKETCH_DHT22, b'Temp:', run_seconds=14)
def test_servo_pot_serial_is_clean(self):
self._assert_clean_user_output(SKETCH_SERVO_POT, b'Pot:')
def test_joystick_serial_is_clean(self):
self._assert_clean_user_output(SKETCH_JOYSTICK, b'X=')
def test_dual_adc_serial_is_clean(self):
self._assert_clean_user_output(SKETCH_DUAL_ADC, b'CH1=')
# ─────────────────────────────────────────────────────────────────────────────
# Bugs #2, #3 — Compile success regression coverage
# ─────────────────────────────────────────────────────────────────────────────
@unittest.skipUnless(_toolchain_available(),
'Requires libqemu-xtensa + ESP-IDF toolchain (Docker CI)')
class TestEsp32CompileSuccess(unittest.TestCase):
"""These sketches were reported as 'Fail to Compile' before commits 1
(Bluedroid) and 2 (LEDC compat shim).
"""
def test_ble_advertise_compiles(self):
# Bug #2 — sdkconfig now enables Bluedroid, BLEDevice.h must compile.
fw = _compile_via_espidf(SKETCH_BLE_ADVERTISE)
self.assertGreater(len(fw), 100_000,
f"BLE Advertise firmware looks suspiciously small: {len(fw)} bytes")
def test_ledc_rgb_compiles(self):
# Bug #3 — velxio_compat.h shim provides ledcAttach() on 2.0.17.
fw = _compile_via_espidf(SKETCH_LEDC_RGB)
self.assertGreater(len(fw), 100_000,
f"LEDC RGB firmware looks suspiciously small: {len(fw)} bytes")
def test_ledc_rgb_boots(self):
"""Stronger than compile: the firmware must reach setup() and run
the user loop without crashing. The shim's dynamic channel allocation
is exercised at runtime here.
"""
fw = _compile_via_espidf(SKETCH_LEDC_RGB)
out = _run_worker(fw, run_seconds=6)
self.assertTrue(out['booted'],
f"LEDC RGB never booted; events={out['all_events'][:5]}")
# ─────────────────────────────────────────────────────────────────────────────
# Regression — WiFi sketches still associate after the sdkconfig changes
# ─────────────────────────────────────────────────────────────────────────────
@unittest.skipUnless(_toolchain_available(),
'Requires libqemu-xtensa + ESP-IDF toolchain (Docker CI)')
class TestEsp32WiFiSketches(unittest.TestCase):
"""The sdkconfig changes (log level + Bluedroid) shouldn't have broken
WiFi association. WiFi takes longer to come up give it 25 s.
"""
def test_wifi_connect_reaches_got_ip(self):
fw = _compile_via_espidf(SKETCH_WIFI_CONNECT)
out = _run_worker(fw, run_seconds=25)
self.assertTrue(out['booted'])
self.assertIn(b'IP Address:', out['serial_bytes'],
f"Never saw IP Address line; output:\n{out['serial_bytes'][-800:]!r}")
# Bug #1 also applies to WiFi sketches — no info logs leaking.
self.assertFalse(ESPIDF_INFO_RE.search(out['serial_bytes']),
"WiFi sketch leaked ESP-IDF info logs")
def test_wifi_server_starts(self):
fw = _compile_via_espidf(SKETCH_WIFI_SERVER)
out = _run_worker(fw, run_seconds=25)
self.assertTrue(out['booted'])
self.assertIn(b'Server started', out['serial_bytes'],
f"Server never started; output:\n{out['serial_bytes'][-800:]!r}")
# ─────────────────────────────────────────────────────────────────────────────
if __name__ == '__main__':
unittest.main(verbosity=2)