test(firmware): gated USB flash round-trip + validated end-to-end flash (#99)

src/flipperzero_mcp/tools/firmware.py

@@ -16,7 +16,11 @@
 from flipperzero_mcp.tools._common import ensure_connected, get_rpc, require_firmware_flash
 
 _PKG_NAME = "mcp-update"
-_RECONNECT_BUDGET_S = 300.0
+# The device is off the bus while the on-device updater applies the bundle and
+# reboots. A larger firmware (Momentum's resources) was observed to re-enumerate
+# just past 300 s, so the old budget raised a false "did not reconnect" on an
+# update that had in fact succeeded; 600 s covers the slow apply.
+_RECONNECT_BUDGET_S = 600.0
 
 
 async def _resolve(source: dict[str, Any], target: str) -> Any:

tests/integration/test_firmware_flash.py

@@ -0,0 +1,78 @@
+"""End-to-end USB firmware-flash round-trip against a real Flipper.
+
+Destructive and slow: each flash pushes a full update bundle, reboots the
+device into the on-device updater, and waits for it to come back (minutes per
+direction). Guarded behind ``FLIPPER_RUN_FLASH_TEST`` so it never runs as part
+of the ordinary ``integration``/``usb`` sweep. The test always attempts to
+flash back to the starting flavor, even when the forward flash fails, so a
+green or assertion-failed run leaves no net change; only a wedged device can
+strand the swap.
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Any
+
+import pytest
+from fastmcp import Client
+
+from flipperzero_mcp.config import FlipperConfig
+from flipperzero_mcp.firmware.flavor import FirmwareFlavor, classify
+from flipperzero_mcp.rpc.client import FlipperClient
+from flipperzero_mcp.server import create_server
+from flipperzero_mcp.transport import get_transport
+
+_FLASH_GATE = "FLIPPER_RUN_FLASH_TEST"
+_OPPOSITE = {FirmwareFlavor.OFFICIAL: "momentum", FirmwareFlavor.MOMENTUM: "official"}
+
+
+async def _read_identity() -> tuple[str, FirmwareFlavor]:
+    """Read the device name and firmware flavor over a throwaway connection."""
+    cfg = FlipperConfig(_env_file=None, transport="usb")  # ty: ignore[unknown-argument]
+    client = FlipperClient(get_transport("usb", cfg.as_transport_config()))
+    if not await client.connect() or client.rpc is None:
+        pytest.skip("No USB Flipper connected")
+    try:
+        info = await client.rpc.get_device_info()
+    finally:
+        await client.disconnect()
+    name = info.get("hardware_name")
+    flavor = classify(info).flavor
+    if not name or flavor not in _OPPOSITE:
+        pytest.skip(f"device identity unusable for a flash: name={name!r} flavor={flavor}")
+    return name, flavor
+
+
+async def _flash(client: Client, flavor: str, confirm: str) -> dict[str, Any]:
+    source = {"flavor": flavor, "channel": "release", "version": "latest"}
+    result = await client.call_tool(
+        "flipperzero_firmware_install", {"source": source, "confirm": confirm}
+    )
+    return result.data
+
+
+@pytest.mark.integration
+@pytest.mark.usb
+async def test_usb_firmware_flash_roundtrip():
+    if not os.environ.get(_FLASH_GATE):
+        pytest.skip(f"set {_FLASH_GATE}=1 to run the destructive flash round-trip")
+
+    name, origin = await _read_identity()
+    other = _OPPOSITE[origin]
+
+    cfg = FlipperConfig(
+        _env_file=None,  # ty: ignore[unknown-argument]
+        transport="usb",
+        enable_write_tools=True,
+        enable_firmware_flash=True,
+    )
+    async with Client(create_server(cfg)) as client:
+        try:
+            forward = await _flash(client, other, name)
+            assert forward["after_confirmed"] is True
+            assert forward["after"]["flavor"] == other
+        finally:
+            back = await _flash(client, origin.value, name)
+        assert back["after_confirmed"] is True
+        assert back["after"]["flavor"] == origin.value