From c939005bf79f5a88377b27f2991e64efb3f9c851 Mon Sep 17 00:00:00 2001 From: David Montero Crespo Date: Fri, 8 May 2026 18:39:22 -0300 Subject: [PATCH] test(esp32): integration tests for QEMU examples MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- .../component-metadata-bmp280.test.ts | 14 + .../integration/test_esp32_examples_qemu.py | 541 ++++++++++++++++++ 2 files changed, 555 insertions(+) create mode 100644 frontend/src/__tests__/component-metadata-bmp280.test.ts create mode 100644 test/backend/integration/test_esp32_examples_qemu.py diff --git a/frontend/src/__tests__/component-metadata-bmp280.test.ts b/frontend/src/__tests__/component-metadata-bmp280.test.ts new file mode 100644 index 00000000..e7c3ba8a --- /dev/null +++ b/frontend/src/__tests__/component-metadata-bmp280.test.ts @@ -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>).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); + }); +}); diff --git a/test/backend/integration/test_esp32_examples_qemu.py b/test/backend/integration/test_esp32_examples_qemu.py new file mode 100644 index 00000000..bc8edda9 --- /dev/null +++ b/test/backend/integration/test_esp32_examples_qemu.py @@ -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 + +#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 + +#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 + +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 +#include + +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 +#include +#include +#include + +#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)