Each requires confirmation and is recorded in the audit log.
@@ -706,6 +707,7 @@
+ Flash companion firmware
+
+ Upload a .bin and flash the USB MeshCore or Meshtastic companion.
+ MeshCore capture pauses during the flash and auto-reconnects afterward.
+ Admin only; recorded in the audit log.
+
+
+ Firmware file (.bin)
+
+
+ Serial port
+
+
+
+ Baud
+
+
+ Offset
+
+
+
+
+ Flash firmware
+ Clear log
+
+
+ `;
+
+ this._fileInput = this._host.querySelector('[data-fw-file]');
+ this._portInput = this._host.querySelector('[data-fw-port]');
+ this._baudInput = this._host.querySelector('[data-fw-baud]');
+ this._offsetInput = this._host.querySelector('[data-fw-offset]');
+ this._flashBtn = this._host.querySelector('[data-fw-flash]');
+ this._uploadStatus = this._host.querySelector('[data-fw-upload-status]');
+ this._logEl = this._host.querySelector('[data-fw-log]');
+
+ this._fileInput.addEventListener('change', () => this._onFileSelected());
+ this._flashBtn.addEventListener('click', () => this._onFlashClick());
+ this._host.querySelector('[data-fw-clear-log]').addEventListener('click', () => {
+ this._logLines = [];
+ this._paintLog();
+ });
+
+ this._loadDefaults();
+ this._connectLogWs();
+ }
+
+ async _loadDefaults() {
+ try {
+ const res = await fetch('/api/firmware/defaults', { credentials: 'same-origin' });
+ if (!res.ok) return;
+ const data = await res.json();
+ if (data.serial_port && this._portInput) {
+ this._portInput.value = data.serial_port;
+ }
+ if (data.baud_rate != null && this._baudInput) {
+ this._baudInput.value = data.baud_rate;
+ }
+ if (data.partition_offset && this._offsetInput) {
+ this._offsetInput.value = data.partition_offset;
+ }
+ } catch (_e) { /* best-effort */ }
+ }
+
+ _connectLogWs() {
+ if (this._ws) return;
+ const proto = location.protocol === 'https:' ? 'wss:' : 'ws:';
+ const url = `${proto}//${location.host}/api/firmware/ws/flash-log`;
+ try {
+ this._ws = new WebSocket(url);
+ this._ws.onmessage = (event) => {
+ this._appendLog(event.data);
+ };
+ this._ws.onclose = () => {
+ this._ws = null;
+ setTimeout(() => this._connectLogWs(), 5000);
+ };
+ } catch (_e) {
+ this._appendLog('[flasher] WebSocket unavailable');
+ }
+ }
+
+ async _onFileSelected() {
+ const file = this._fileInput?.files?.[0];
+ if (!file) return;
+ if (!file.name.toLowerCase().endsWith('.bin')) {
+ this._setUploadStatus('error', 'Only .bin files are accepted.');
+ this._uploadId = null;
+ this._flashBtn.disabled = true;
+ return;
+ }
+ this._setUploadStatus('pending', 'Uploading…');
+ const form = new FormData();
+ form.append('firmware_file', file);
+ try {
+ const res = await fetch('/api/firmware/upload', {
+ method: 'POST',
+ credentials: 'same-origin',
+ body: form,
+ });
+ if (!res.ok) {
+ const body = await res.json().catch(() => ({}));
+ this._setUploadStatus('error', body.detail || `Upload failed (${res.status}).`);
+ this._flashBtn.disabled = true;
+ return;
+ }
+ const data = await res.json();
+ this._uploadId = data.upload_id;
+ this._filename = data.filename || file.name;
+ this._setUploadStatus(
+ 'success',
+ `Ready: ${this._filename} (${this._formatBytes(data.size_bytes)})`,
+ );
+ this._flashBtn.disabled = false;
+ } catch (_e) {
+ this._setUploadStatus('error', 'Upload failed (network error).');
+ this._flashBtn.disabled = true;
+ }
+ }
+
+ async _onFlashClick() {
+ if (!this._uploadId) return;
+ const port = (this._portInput?.value || '').trim();
+ const baud = Number(this._baudInput?.value || 460800);
+ const offset = (this._offsetInput?.value || '0x10000').trim();
+ const ok = await this._modal.confirm({
+ label: 'flash firmware',
+ command: `Flash ${this._filename}`,
+ description:
+ `This halts MeshCore USB on ${port} for ~20–30 seconds while esptool writes the image. `
+ + 'The companion will reboot and reconnect automatically.',
+ });
+ if (!ok) return;
+
+ this._appendLog(`[ui] Starting flash on ${port}…`);
+ this._flashBtn.disabled = true;
+ try {
+ const res = await fetch('/api/firmware/flash', {
+ method: 'POST',
+ credentials: 'same-origin',
+ headers: { 'Content-Type': 'application/json' },
+ body: JSON.stringify({
+ upload_id: this._uploadId,
+ serial_port: port,
+ baud_rate: baud,
+ partition_offset: offset,
+ }),
+ });
+ const body = await res.json().catch(() => ({}));
+ if (res.status === 409) {
+ this._appendLog(`[ui] ${body.detail || 'Flash already in progress.'}`);
+ } else if (!res.ok) {
+ this._appendLog(`[ui] Flash request failed: ${body.detail || res.status}`);
+ } else {
+ this._appendLog(`[ui] Queued — watch log below for esptool output.`);
+ this._uploadId = null;
+ this._fileInput.value = '';
+ this._setUploadStatus('', '');
+ }
+ } catch (_e) {
+ this._appendLog('[ui] Flash request failed (network error).');
+ } finally {
+ this._flashBtn.disabled = !this._uploadId;
+ }
+ }
+
+ _appendLog(line) {
+ this._logLines.push(String(line));
+ if (this._logLines.length > 500) {
+ this._logLines = this._logLines.slice(-500);
+ }
+ this._paintLog();
+ }
+
+ _paintLog() {
+ if (!this._logEl) return;
+ this._logEl.textContent = this._logLines.join('\n');
+ this._logEl.scrollTop = this._logEl.scrollHeight;
+ }
+
+ _setUploadStatus(kind, message) {
+ if (!this._uploadStatus) return;
+ this._uploadStatus.dataset.kind = kind || '';
+ this._uploadStatus.textContent = message || '';
+ }
+
+ static _formatBytes(n) {
+ const v = Number(n) || 0;
+ if (v < 1024) return `${v} B`;
+ if (v < 1024 * 1024) return `${(v / 1024).toFixed(1)} KB`;
+ return `${(v / (1024 * 1024)).toFixed(2)} MB`;
+ }
+}
+
+window.CompanionFlashCard = CompanionFlashCard;
diff --git a/requirements.txt b/requirements.txt
index 89b9c0b1..83f8b445 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -13,3 +13,6 @@ meshcore>=2.1.0
paho-mqtt>=2.1.0
bcrypt>=4.2.0
PyJWT>=2.10.0
+pyserial>=3.5
+httpx>=0.27.0
+esptool>=4.7.0
diff --git a/src/api/routes/firmware_routes.py b/src/api/routes/firmware_routes.py
new file mode 100644
index 00000000..c02fffae
--- /dev/null
+++ b/src/api/routes/firmware_routes.py
@@ -0,0 +1,232 @@
+"""USB companion firmware upload, flash, and live log WebSocket (PR 14)."""
+from __future__ import annotations
+
+import asyncio
+import logging
+import os
+import tempfile
+from pathlib import Path
+from typing import Awaitable, Callable
+
+from fastapi import (
+ APIRouter,
+ Depends,
+ File,
+ HTTPException,
+ UploadFile,
+ WebSocket,
+ WebSocketDisconnect,
+)
+from pydantic import BaseModel, Field
+
+from src.api.audit import AuditLogWriter
+from src.api.audit.dependencies import get_audit_writer
+from src.api.auth.dependencies import require_admin
+from src.api.auth.jwt_session import ROLE_ADMIN, JwtSessionService, SessionClaims
+from src.api.auth.ws_guard import WS_AUTH_CLOSE_CODE, authenticate_websocket
+from src.firmware.flasher import FlashJob, get_port_lock, run_flash_job
+from src.firmware.log_broadcast import FlashLogBroadcaster
+from src.firmware.upload_store import FirmwareUploadStore
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/api/firmware", tags=["firmware"])
+
+MAX_UPLOAD_BYTES = 10 * 1024 * 1024
+DEFAULT_BAUD = 460800
+DEFAULT_OFFSET = "0x10000"
+
+_upload_store = FirmwareUploadStore()
+_broadcaster = FlashLogBroadcaster()
+_jwt_service: JwtSessionService | None = None
+_suspend_meshcore: Callable[[str], Awaitable[None]] | None = None
+_default_port: str = "/dev/ttyUSB0"
+
+
+def init_routes(
+ *,
+ jwt_service: JwtSessionService | None,
+ suspend_meshcore: Callable[[str], Awaitable[None]] | None = None,
+ default_serial_port: str | None = None,
+) -> None:
+ global _jwt_service, _suspend_meshcore, _default_port
+ _jwt_service = jwt_service
+ _suspend_meshcore = suspend_meshcore
+ if default_serial_port:
+ _default_port = default_serial_port
+
+
+def reset_routes() -> None:
+ global _jwt_service, _suspend_meshcore
+ _jwt_service = None
+ _suspend_meshcore = None
+
+
+class FlashRequest(BaseModel):
+ upload_id: str = Field(..., min_length=8, max_length=64)
+ serial_port: str = Field(default="/dev/ttyUSB0", min_length=1, max_length=128)
+ baud_rate: int = Field(default=DEFAULT_BAUD, ge=9600, le=921600)
+ partition_offset: str = Field(default=DEFAULT_OFFSET, min_length=3, max_length=16)
+
+
+@router.websocket("/ws/flash-log")
+async def flash_log_ws(websocket: WebSocket) -> None:
+ if _jwt_service is None:
+ await websocket.accept()
+ await websocket.close(code=WS_AUTH_CLOSE_CODE)
+ return
+
+ claims = authenticate_websocket(websocket, _jwt_service)
+ if claims is None or claims.role != ROLE_ADMIN:
+ await websocket.accept()
+ await websocket.close(code=WS_AUTH_CLOSE_CODE)
+ return
+
+ await websocket.accept()
+ await _broadcaster.subscribe(websocket)
+ try:
+ while True:
+ await websocket.receive_text()
+ except WebSocketDisconnect:
+ pass
+ except Exception:
+ pass
+ finally:
+ await _broadcaster.unsubscribe(websocket)
+
+
+@router.post("/upload")
+async def upload_firmware(
+ firmware_file: UploadFile = File(...),
+ _claims: SessionClaims = Depends(require_admin),
+):
+ filename = (firmware_file.filename or "").strip()
+ if not filename.lower().endswith(".bin"):
+ raise HTTPException(status_code=400, detail="Only .bin firmware files are accepted.")
+
+ content = await firmware_file.read()
+ if len(content) > MAX_UPLOAD_BYTES:
+ raise HTTPException(
+ status_code=400,
+ detail=f"Firmware file exceeds {MAX_UPLOAD_BYTES // (1024 * 1024)} MB limit.",
+ )
+ if not content:
+ raise HTTPException(status_code=400, detail="Empty firmware file.")
+
+ tmp = tempfile.NamedTemporaryFile(
+ suffix=".bin",
+ prefix="mp_fw_",
+ delete=False,
+ )
+ try:
+ tmp.write(content)
+ tmp.flush()
+ tmp.close()
+ upload_id = _upload_store.store(
+ Path(tmp.name),
+ filename,
+ len(content),
+ )
+ except Exception as exc:
+ try:
+ os.unlink(tmp.name)
+ except OSError:
+ pass
+ raise HTTPException(status_code=500, detail=f"Failed to store upload: {exc}") from exc
+
+ return {
+ "upload_id": upload_id,
+ "filename": filename,
+ "size_bytes": len(content),
+ }
+
+
+@router.get("/defaults")
+async def firmware_defaults(_claims: SessionClaims = Depends(require_admin)):
+ return {
+ "serial_port": _default_port,
+ "baud_rate": DEFAULT_BAUD,
+ "partition_offset": DEFAULT_OFFSET,
+ "max_upload_bytes": MAX_UPLOAD_BYTES,
+ }
+
+
+@router.post("/flash")
+async def flash_companion(
+ req: FlashRequest,
+ claims: SessionClaims = Depends(require_admin),
+ audit: AuditLogWriter = Depends(get_audit_writer),
+):
+ record = _upload_store.pop(req.upload_id)
+ if record is None:
+ raise HTTPException(status_code=404, detail="Unknown or expired upload_id.")
+
+ port = req.serial_port.strip()
+ if not port:
+ raise HTTPException(status_code=400, detail="serial_port is required.")
+
+ port_lock = get_port_lock(port)
+ if port_lock.locked():
+ _upload_store.store(record.path, record.filename, record.size_bytes)
+ raise HTTPException(
+ status_code=409,
+ detail=f"A flash operation is already in progress on {port}.",
+ )
+
+ async def _task() -> None:
+ async with port_lock:
+ exit_code: int | None = None
+ success = False
+ with audit.timed_action(
+ user=claims.subject,
+ action="firmware_flash",
+ params={
+ "filename": record.filename,
+ "port": port,
+ "baud": req.baud_rate,
+ "offset": req.partition_offset,
+ "size_bytes": record.size_bytes,
+ },
+ ) as ctx:
+ try:
+ if _suspend_meshcore is not None:
+ await _broadcaster.broadcast(
+ f"[flasher] releasing MeshCore serial on {port}…"
+ )
+ await _suspend_meshcore(port)
+
+ job = FlashJob(
+ port=port,
+ baud=req.baud_rate,
+ offset=req.partition_offset,
+ bin_path=record.path,
+ log_callback=_broadcaster.broadcast,
+ )
+ success = await run_flash_job(job)
+ exit_code = 0 if success else 1
+ ctx.params["exit_code"] = exit_code
+ ctx.params["success"] = success
+ if not success:
+ ctx.set_result("error")
+ except Exception as exc:
+ logger.exception("Firmware flash task failed")
+ await _broadcaster.broadcast(f"[flasher] EXCEPTION: {exc}")
+ ctx.set_result("error")
+ ctx.params["error"] = str(exc)
+ finally:
+ try:
+ os.unlink(record.path)
+ except OSError:
+ pass
+ await _broadcaster.broadcast(
+ "[flasher] MeshCore will auto-reconnect when the port is free."
+ )
+
+ asyncio.create_task(_task(), name="firmware-flash")
+
+ return {
+ "status": "queued",
+ "port": port,
+ "filename": record.filename,
+ "ws_path": "/api/firmware/ws/flash-log",
+ }
diff --git a/src/api/server.py b/src/api/server.py
index 3bd4fb06..9d7f1a28 100644
--- a/src/api/server.py
+++ b/src/api/server.py
@@ -41,6 +41,7 @@
dangerous_routes,
device,
device_config_routes,
+ firmware_routes,
gps_status,
identity_routes,
messages,
@@ -259,6 +260,7 @@ async def lifespan(app: FastAPI):
app.include_router(auth_config_routes.router)
app.include_router(public_radar_routes.router)
app.include_router(terminal_routes.router)
+ app.include_router(firmware_routes.router)
app.include_router(update_routes.router)
app.include_router(dangerous_routes.router)
@@ -1266,6 +1268,29 @@ def _init_routes(
device_config_routes.init_routes(config=config, identity=identity)
gps_status.init_routes(location_source=coord.location_source)
system_config_routes.init_routes(config=config)
+
+ async def _suspend_meshcore_for_flash(port: str) -> None:
+ src = _find_meshcore_source(coord)
+ if src is None:
+ return
+ resolved = getattr(src, "_resolved_port", None) or port
+ configured = getattr(src, "_configured_port", None)
+ if resolved != port and configured != port:
+ logger.info(
+ "MeshCore USB port %s does not match active source %s — skip release",
+ port,
+ resolved,
+ )
+ return
+ await src.release_serial_for_flash()
+
+ mc_usb = config.capture.meshcore_usb
+ default_mc_port = (mc_usb.serial_port or "/dev/ttyUSB0").strip()
+ firmware_routes.init_routes(
+ jwt_service=auth_subsystem.jwt_service,
+ suspend_meshcore=_suspend_meshcore_for_flash,
+ default_serial_port=default_mc_port,
+ )
meshcore_config_routes.init_routes(config=config, tx_service=tx_service)
diff --git a/src/capture/meshcore_usb_source.py b/src/capture/meshcore_usb_source.py
index 28e0479e..d393ec5b 100644
--- a/src/capture/meshcore_usb_source.py
+++ b/src/capture/meshcore_usb_source.py
@@ -208,6 +208,10 @@ async def _connect(self, port: str) -> None:
)
self._connected = False
+ async def release_serial_for_flash(self) -> None:
+ """Disconnect so an external tool (esptool) can open the serial port."""
+ await self._disconnect()
+
async def _disconnect(self) -> None:
self._connected = False
if self._meshcore:
diff --git a/src/firmware/__init__.py b/src/firmware/__init__.py
new file mode 100644
index 00000000..452a8f40
--- /dev/null
+++ b/src/firmware/__init__.py
@@ -0,0 +1 @@
+"""USB companion firmware flashing (esptool) for field operations."""
diff --git a/src/firmware/flasher.py b/src/firmware/flasher.py
new file mode 100644
index 00000000..8ec6d922
--- /dev/null
+++ b/src/firmware/flasher.py
@@ -0,0 +1,82 @@
+"""Async esptool subprocess runner for companion firmware flashes."""
+from __future__ import annotations
+
+import asyncio
+import logging
+import sys
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Awaitable, Callable
+
+logger = logging.getLogger(__name__)
+
+_port_locks: dict[str, asyncio.Lock] = {}
+
+
+def get_port_lock(port: str) -> asyncio.Lock:
+ if port not in _port_locks:
+ _port_locks[port] = asyncio.Lock()
+ return _port_locks[port]
+
+
+@dataclass
+class FlashJob:
+ port: str
+ baud: int
+ offset: str
+ bin_path: Path
+ log_callback: Callable[[str], Awaitable[None]]
+
+
+async def run_flash_job(job: FlashJob) -> bool:
+ """Run esptool; stream stdout to log_callback. Caller holds port lock."""
+ command = [
+ sys.executable,
+ "-m",
+ "esptool",
+ "--port",
+ job.port,
+ "--baud",
+ str(job.baud),
+ "write_flash",
+ "-z",
+ job.offset,
+ str(job.bin_path),
+ ]
+
+ await job.log_callback(f"[flasher] {' '.join(command)}")
+
+ try:
+ proc = await asyncio.create_subprocess_exec(
+ *command,
+ stdout=asyncio.subprocess.PIPE,
+ stderr=asyncio.subprocess.STDOUT,
+ )
+ if proc.stdout is None:
+ await job.log_callback("[flasher] ERROR: no subprocess stdout")
+ return False
+
+ async for raw in proc.stdout:
+ line = raw.decode("utf-8", errors="replace").rstrip()
+ if line:
+ await job.log_callback(line)
+
+ await proc.wait()
+ success = proc.returncode == 0
+ await job.log_callback(
+ "[flasher] SUCCESS"
+ if success
+ else f"[flasher] FAILED (exit {proc.returncode})"
+ )
+ return success
+
+ except FileNotFoundError:
+ await job.log_callback(
+ "[flasher] ERROR: esptool not found — "
+ "install with: pip install esptool"
+ )
+ return False
+ except Exception as exc:
+ logger.exception("Flash job failed")
+ await job.log_callback(f"[flasher] EXCEPTION: {exc}")
+ return False
diff --git a/src/firmware/log_broadcast.py b/src/firmware/log_broadcast.py
new file mode 100644
index 00000000..9a6f3771
--- /dev/null
+++ b/src/firmware/log_broadcast.py
@@ -0,0 +1,33 @@
+"""Authenticated WebSocket subscribers for live firmware flash logs."""
+from __future__ import annotations
+
+import asyncio
+import logging
+
+from fastapi import WebSocket
+
+logger = logging.getLogger(__name__)
+
+
+class FlashLogBroadcaster:
+ def __init__(self) -> None:
+ self._subscribers: set[WebSocket] = set()
+ self._lock = asyncio.Lock()
+
+ async def subscribe(self, websocket: WebSocket) -> None:
+ async with self._lock:
+ self._subscribers.add(websocket)
+
+ async def unsubscribe(self, websocket: WebSocket) -> None:
+ async with self._lock:
+ self._subscribers.discard(websocket)
+
+ async def broadcast(self, message: str) -> None:
+ async with self._lock:
+ dead: set[WebSocket] = set()
+ for ws in self._subscribers:
+ try:
+ await ws.send_text(message)
+ except Exception:
+ dead.add(ws)
+ self._subscribers.difference_update(dead)
diff --git a/src/firmware/upload_store.py b/src/firmware/upload_store.py
new file mode 100644
index 00000000..28f8ed06
--- /dev/null
+++ b/src/firmware/upload_store.py
@@ -0,0 +1,61 @@
+"""In-memory staging for uploaded firmware binaries."""
+from __future__ import annotations
+
+import secrets
+import time
+from dataclasses import dataclass
+from pathlib import Path
+from threading import Lock
+
+
+@dataclass
+class UploadRecord:
+ upload_id: str
+ path: Path
+ filename: str
+ size_bytes: int
+ created_at: float
+
+
+class FirmwareUploadStore:
+ """Short-lived upload registry keyed by opaque upload_id."""
+
+ def __init__(self, *, ttl_seconds: float = 3600.0) -> None:
+ self._ttl = ttl_seconds
+ self._records: dict[str, UploadRecord] = {}
+ self._lock = Lock()
+
+ def store(self, path: Path, filename: str, size_bytes: int) -> str:
+ self._prune_expired()
+ upload_id = secrets.token_urlsafe(16)
+ record = UploadRecord(
+ upload_id=upload_id,
+ path=path,
+ filename=filename,
+ size_bytes=size_bytes,
+ created_at=time.monotonic(),
+ )
+ with self._lock:
+ self._records[upload_id] = record
+ return upload_id
+
+ def get(self, upload_id: str) -> UploadRecord | None:
+ self._prune_expired()
+ with self._lock:
+ return self._records.get(upload_id)
+
+ def pop(self, upload_id: str) -> UploadRecord | None:
+ self._prune_expired()
+ with self._lock:
+ return self._records.pop(upload_id, None)
+
+ def _prune_expired(self) -> None:
+ cutoff = time.monotonic() - self._ttl
+ with self._lock:
+ expired = [
+ uid
+ for uid, rec in self._records.items()
+ if rec.created_at < cutoff
+ ]
+ for uid in expired:
+ self._records.pop(uid, None)
diff --git a/tests/test_firmware_flasher.py b/tests/test_firmware_flasher.py
new file mode 100644
index 00000000..8fa5543a
--- /dev/null
+++ b/tests/test_firmware_flasher.py
@@ -0,0 +1,114 @@
+"""USB companion firmware flasher (PR 14)."""
+
+from __future__ import annotations
+
+import asyncio
+import unittest
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from src.firmware.flasher import FlashJob, get_port_lock, run_flash_job
+from src.firmware.upload_store import FirmwareUploadStore
+
+
+def _run(coro):
+ return asyncio.run(coro)
+
+
+class _AsyncStdout:
+ def __init__(self, lines: list[bytes]) -> None:
+ self._lines = iter(lines)
+
+ def __aiter__(self):
+ return self
+
+ async def __anext__(self):
+ try:
+ return next(self._lines)
+ except StopIteration:
+ raise StopAsyncIteration from None
+
+
+class TestRunFlashJob(unittest.TestCase):
+ def _make_job(self, callback=None):
+ cb = callback or AsyncMock()
+ job = FlashJob(
+ port="/dev/ttyUSB0",
+ baud=460800,
+ offset="0x10000",
+ bin_path=Path("/tmp/test_fw.bin"),
+ log_callback=cb,
+ )
+ return job, cb
+
+ def test_success_returns_true(self):
+ job, _cb = self._make_job()
+ mock_proc = MagicMock()
+ mock_proc.returncode = 0
+ mock_proc.stdout = _AsyncStdout([b"Writing at 0x10000...\n", b"Hash OK\n"])
+ mock_proc.wait = AsyncMock()
+
+ with patch("asyncio.create_subprocess_exec", return_value=mock_proc):
+ result = _run(run_flash_job(job))
+ self.assertTrue(result)
+
+ def test_failure_returns_false(self):
+ job, _cb = self._make_job()
+ mock_proc = MagicMock()
+ mock_proc.returncode = 2
+ mock_proc.stdout = _AsyncStdout([b"Failed to connect\n"])
+ mock_proc.wait = AsyncMock()
+
+ with patch("asyncio.create_subprocess_exec", return_value=mock_proc):
+ result = _run(run_flash_job(job))
+ self.assertFalse(result)
+
+ def test_missing_esptool_returns_false(self):
+ job, _cb = self._make_job()
+ with patch(
+ "asyncio.create_subprocess_exec",
+ side_effect=FileNotFoundError("esptool not found"),
+ ):
+ result = _run(run_flash_job(job))
+ self.assertFalse(result)
+
+ def test_log_callback_receives_stdout_lines(self):
+ lines_received: list[str] = []
+
+ async def capture(msg: str) -> None:
+ lines_received.append(msg)
+
+ job, _ = self._make_job(callback=capture)
+ mock_proc = MagicMock()
+ mock_proc.returncode = 0
+ mock_proc.stdout = _AsyncStdout([b"line one\n", b"line two\n"])
+ mock_proc.wait = AsyncMock()
+
+ with patch("asyncio.create_subprocess_exec", return_value=mock_proc):
+ _run(run_flash_job(job))
+
+ joined = "\n".join(lines_received)
+ self.assertIn("line one", joined)
+ self.assertIn("line two", joined)
+
+ def test_port_lock_is_per_port(self):
+ lock_a = get_port_lock("/dev/ttyUSB0")
+ lock_b = get_port_lock("/dev/ttyACM0")
+ lock_a2 = get_port_lock("/dev/ttyUSB0")
+ self.assertIs(lock_a, lock_a2)
+ self.assertIsNot(lock_a, lock_b)
+
+
+class TestFirmwareUploadStore(unittest.TestCase):
+ def test_store_and_pop(self):
+ store = FirmwareUploadStore()
+ path = Path("/tmp/fake.bin")
+ upload_id = store.store(path, "fw.bin", 1024)
+ record = store.pop(upload_id)
+ self.assertIsNotNone(record)
+ self.assertEqual(record.filename, "fw.bin")
+ self.assertIsNone(store.pop(upload_id))
+
+
+if __name__ == "__main__":
+ unittest.main()