Add USB camera and uhubctl support for new test suite. Also included some bug fixes (#10204)
* Add USB camera and uhubctl support for new test suite. Also added some bug fixes * Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> * Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> * Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> * Refactor test messages for clarity and consistency in regex tests --------- Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
Ben Meadors
co-authored by
GitHub
Copilot Autofix powered by AI
GitHub
Copilot Autofix powered by AI
parent
6b15571e14
commit
de23e5199d
@@ -0,0 +1,112 @@
|
||||
"""USB hub power control for tests — thin composition of the `uhubctl`
|
||||
module + `_port_discovery.resolve_port_by_role`.
|
||||
|
||||
Why separate from the production module:
|
||||
- `meshtastic_mcp.uhubctl.cycle` returns as soon as uhubctl exits (VBUS is
|
||||
back on, but the device hasn't finished enumerating as a CDC port yet).
|
||||
- Tests that want to immediately issue a `connect(port=...)` need the NEW
|
||||
`/dev/cu.*` path, which can differ from the pre-cycle path on nRF52
|
||||
boards (CDC re-enumeration assigns a fresh `cu.usbmodemNNNN`).
|
||||
- `resolve_port_by_role` already handles that wait + path-resolution for
|
||||
the `factory_reset` flow. Composing the two gives a one-call helper.
|
||||
|
||||
Also exposes `is_uhubctl_available()` so fixtures can skip cleanly when
|
||||
uhubctl isn't installed — we never want "no uhubctl" to look like a test
|
||||
failure.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
from typing import Any
|
||||
|
||||
from meshtastic_mcp import config as config_mod
|
||||
from meshtastic_mcp import uhubctl as uhubctl_mod
|
||||
|
||||
from ._port_discovery import resolve_port_by_role
|
||||
|
||||
|
||||
def is_uhubctl_available() -> bool:
|
||||
"""Return True iff `config.uhubctl_bin()` resolves AND the binary is callable.
|
||||
|
||||
Soft-fails silently — fixtures use this to `pytest.skip` with an
|
||||
actionable message when the operator hasn't installed uhubctl.
|
||||
"""
|
||||
try:
|
||||
config_mod.uhubctl_bin()
|
||||
except Exception: # noqa: BLE001
|
||||
return False
|
||||
# Do NOT actually invoke uhubctl here — on macOS a non-sudo run would
|
||||
# fail, which is a config issue, not a tool-missing issue. That gets
|
||||
# surfaced to the user when they actually run a recovery action.
|
||||
return True
|
||||
|
||||
|
||||
def power_on(role: str) -> dict[str, Any]:
|
||||
"""Power on the hub port hosting `role`. Does NOT wait for re-enumeration.
|
||||
Use `power_cycle` or follow with `resolve_port_by_role` to block on readiness.
|
||||
"""
|
||||
loc, port = uhubctl_mod.resolve_target(role)
|
||||
return uhubctl_mod.power_on(loc, port)
|
||||
|
||||
|
||||
def power_off(role: str) -> dict[str, Any]:
|
||||
"""Power off the hub port hosting `role`. The device disappears from
|
||||
`list_devices` immediately.
|
||||
"""
|
||||
loc, port = uhubctl_mod.resolve_target(role)
|
||||
return uhubctl_mod.power_off(loc, port)
|
||||
|
||||
|
||||
def power_cycle(
|
||||
role: str,
|
||||
*,
|
||||
delay_s: int = 2,
|
||||
rediscover_timeout_s: float = 30.0,
|
||||
) -> str:
|
||||
"""Cycle the port hosting `role`, wait for re-enumeration, return the
|
||||
new port path.
|
||||
|
||||
On nRF52 the post-cycle path typically matches the pre-cycle path, but
|
||||
macOS may assign a different `/dev/cu.usbmodemNNNN` if the previous
|
||||
CDC endpoint hasn't been fully released. `resolve_port_by_role`
|
||||
handles that transparently.
|
||||
"""
|
||||
loc, port = uhubctl_mod.resolve_target(role)
|
||||
uhubctl_mod.cycle(loc, port, delay_s=delay_s)
|
||||
# After uhubctl exits, VBUS is on but the device may still be in
|
||||
# bootloader init. Give it ~500 ms head-start before polling so we
|
||||
# don't spam list_devices pointlessly.
|
||||
time.sleep(0.5)
|
||||
return resolve_port_by_role(role, timeout_s=rediscover_timeout_s)
|
||||
|
||||
|
||||
def wait_for_absence(role: str, *, timeout_s: float = 10.0) -> None:
|
||||
"""Block until a device matching `role` is NOT in `list_devices`.
|
||||
|
||||
Used by the recovery tier to assert power_off actually took effect.
|
||||
Raises TimeoutError on failure.
|
||||
"""
|
||||
from meshtastic_mcp import devices as devices_mod
|
||||
|
||||
from ._port_discovery import _ROLE_VIDS, _coerce_vid # type: ignore[attr-defined]
|
||||
|
||||
if role not in _ROLE_VIDS:
|
||||
raise ValueError(f"unknown role {role!r}")
|
||||
wanted = _ROLE_VIDS[role]
|
||||
deadline = time.monotonic() + timeout_s
|
||||
while time.monotonic() < deadline:
|
||||
found = devices_mod.list_devices(include_unknown=True)
|
||||
if not any(_coerce_vid(d.get("vid")) in wanted for d in found):
|
||||
return
|
||||
time.sleep(0.3)
|
||||
raise TimeoutError(f"role {role!r} still visible after {timeout_s}s of power_off")
|
||||
|
||||
|
||||
__all__ = [
|
||||
"is_uhubctl_available",
|
||||
"power_cycle",
|
||||
"power_off",
|
||||
"power_on",
|
||||
"wait_for_absence",
|
||||
]
|
||||
@@ -123,15 +123,24 @@ def pytest_collection_modifyitems(
|
||||
return (2, item.nodeid)
|
||||
if "/monitor/" in path or "tests/monitor" in path:
|
||||
return (3, item.nodeid)
|
||||
if "/fleet/" in path or "tests/fleet" in path:
|
||||
# Recovery tier: explicitly cycles device power via uhubctl. Slots
|
||||
# between monitor (read-only) and ui (state-preserving) so any tier
|
||||
# after it starts from a known re-enumerated + re-verified state.
|
||||
if "/recovery/" in path or "tests/recovery" in path:
|
||||
return (4, item.nodeid)
|
||||
# UI tier slots here — read-only w.r.t. mesh state, only mutates
|
||||
# the on-screen UI (BACK×5 guard restores home before each test).
|
||||
if "/ui/" in path or "tests/ui" in path:
|
||||
return (5, item.nodeid)
|
||||
if "/fleet/" in path or "tests/fleet" in path:
|
||||
return (6, item.nodeid)
|
||||
# State-mutating tiers run last.
|
||||
if "/admin/" in path or "tests/admin" in path:
|
||||
return (5, item.nodeid)
|
||||
return (7, item.nodeid)
|
||||
if "/provisioning/" in path or "tests/provisioning" in path:
|
||||
return (6, item.nodeid)
|
||||
return (8, item.nodeid)
|
||||
# Top-level + anything else falls between.
|
||||
return (7, item.nodeid)
|
||||
return (9, item.nodeid)
|
||||
|
||||
items.sort(key=sort_key)
|
||||
|
||||
@@ -156,13 +165,20 @@ def session_seed(request: pytest.FixtureRequest) -> str:
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def test_profile(session_seed: str) -> dict[str, Any]:
|
||||
"""The canonical isolated-mesh test profile for this session."""
|
||||
"""The canonical isolated-mesh test profile for this session.
|
||||
|
||||
`enable_ui_log=True` stamps `USERPREFS_UI_TEST_LOG` so the firmware
|
||||
emits `Screen: frame N/M name=... reason=...` log lines per UI
|
||||
transition — consumed by the `tests/ui/` tier. Harmless on boards
|
||||
without a screen (the `#ifdef` sits behind `HAS_SCREEN`).
|
||||
"""
|
||||
return userprefs.build_testing_profile(
|
||||
psk_seed=session_seed,
|
||||
channel_name="McpTest",
|
||||
channel_num=88,
|
||||
region="US",
|
||||
modem_preset="LONG_FAST",
|
||||
enable_ui_log=True,
|
||||
)
|
||||
|
||||
|
||||
@@ -654,6 +670,7 @@ def pytest_generate_tests(metafunc: pytest.Metafunc) -> None:
|
||||
def baked_single(
|
||||
baked_mesh: dict[str, Any],
|
||||
baked_single_role: str,
|
||||
hub_devices: dict[str, str],
|
||||
) -> dict[str, Any]:
|
||||
"""Function-scoped: a single verified baked device.
|
||||
|
||||
@@ -662,10 +679,75 @@ def baked_single(
|
||||
(e.g. `test_owner_survives_reboot[nrf52]` +
|
||||
`test_owner_survives_reboot[esp32s3]`). Tests never hardcode a role
|
||||
and never skip a device that happens to be connected.
|
||||
|
||||
Auto-recovery: if the baked device fails a pre-test `device_info` probe
|
||||
AND uhubctl is available, power-cycle the port once and retry. Without
|
||||
uhubctl, surface the wedge as a clear skip. This catches "device got
|
||||
stuck between tests" without masking persistent regressions (a second
|
||||
wedge after cycling still skips).
|
||||
"""
|
||||
if baked_single_role not in baked_mesh:
|
||||
pytest.skip(f"role {baked_single_role!r} not present on the hub")
|
||||
return {"role": baked_single_role, **baked_mesh[baked_single_role]}
|
||||
|
||||
entry = baked_mesh[baked_single_role]
|
||||
port = entry.get("port")
|
||||
if port:
|
||||
try:
|
||||
_run_with_timeout(lambda: info.device_info(port=port, timeout_s=3.0), 5.0)
|
||||
except Exception:
|
||||
# Device didn't respond. Try a power-cycle recovery if uhubctl
|
||||
# is installed; otherwise surface a skip that names the root
|
||||
# cause clearly.
|
||||
from tests import _power
|
||||
|
||||
if not _power.is_uhubctl_available():
|
||||
pytest.skip(
|
||||
f"device {baked_single_role!r} unresponsive on {port}; "
|
||||
"install uhubctl (`brew install uhubctl` / `apt install "
|
||||
"uhubctl`) for auto power-cycle recovery"
|
||||
)
|
||||
try:
|
||||
new_port = _power.power_cycle(baked_single_role, delay_s=2)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
pytest.skip(
|
||||
f"device {baked_single_role!r} wedged and power-cycle "
|
||||
f"failed: {exc}"
|
||||
)
|
||||
# Mutate both the session-scoped `hub_devices` map AND the
|
||||
# baked_mesh entry so downstream fixtures see the recovered port.
|
||||
hub_devices[baked_single_role] = new_port
|
||||
baked_mesh[baked_single_role]["port"] = new_port
|
||||
entry = baked_mesh[baked_single_role]
|
||||
return {"role": baked_single_role, **entry}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def power_cycle(
|
||||
hub_devices: dict[str, str],
|
||||
) -> Callable[..., str]:
|
||||
"""Return a callable `(role, delay_s=2) -> new_port` that hard-resets the
|
||||
hub port hosting `role`. Skips the test cleanly when uhubctl isn't
|
||||
installed — never want "no uhubctl" to look like a test failure.
|
||||
|
||||
The callable mutates `hub_devices[role]` in place so subsequent fixture
|
||||
lookups pick up the post-cycle port (mirrors the pattern in
|
||||
provisioning/test_userprefs_survive_factory_reset.py).
|
||||
"""
|
||||
from tests import _power
|
||||
|
||||
if not _power.is_uhubctl_available():
|
||||
pytest.skip(
|
||||
"uhubctl not installed; this test needs it for power control. "
|
||||
"Install via `brew install uhubctl` (macOS) or `apt install "
|
||||
"uhubctl` (Debian/Ubuntu)."
|
||||
)
|
||||
|
||||
def _cycle(role: str, delay_s: int = 2) -> str:
|
||||
new_port = _power.power_cycle(role, delay_s=delay_s)
|
||||
hub_devices[role] = new_port
|
||||
return new_port
|
||||
|
||||
return _cycle
|
||||
|
||||
|
||||
_DEFAULT_ROLE_ENVS = {
|
||||
@@ -960,6 +1042,45 @@ def _run_with_timeout(fn: Callable[[], Any], timeout: float) -> Any:
|
||||
raise TimeoutError(f"operation did not complete within {timeout}s") from exc
|
||||
|
||||
|
||||
def _attach_ui_captures(item: pytest.Item, report: Any) -> None:
|
||||
"""Embed per-step UI captures (PNG + OCR) into the pytest-html extras.
|
||||
|
||||
Runs for every UI-tier test on BOTH pass and fail so the HTML report
|
||||
always shows the image strip + OCR transcript. Silently no-ops if
|
||||
pytest-html isn't installed or the test didn't use `frame_capture`.
|
||||
"""
|
||||
captures = getattr(item, "_ui_captures", None)
|
||||
if not captures:
|
||||
return
|
||||
try:
|
||||
from pytest_html import extras as html_extras # type: ignore[import-untyped]
|
||||
except ImportError:
|
||||
return
|
||||
|
||||
existing = getattr(report, "extras", None) or []
|
||||
extras_list = list(existing)
|
||||
for cap in captures:
|
||||
png_path = cap.get("png_path")
|
||||
label = f"{cap.get('step', '?')}: {cap.get('label', '')}"
|
||||
frame = cap.get("frame") or {}
|
||||
frame_str = (
|
||||
f" — frame {frame.get('idx')} {frame.get('name')!r}" if frame else ""
|
||||
)
|
||||
if png_path:
|
||||
try:
|
||||
with open(png_path, "rb") as fh:
|
||||
import base64
|
||||
|
||||
b64 = base64.b64encode(fh.read()).decode("ascii")
|
||||
extras_list.append(html_extras.png(b64, name=f"{label}{frame_str}"))
|
||||
except OSError:
|
||||
pass
|
||||
ocr = (cap.get("ocr_text") or "").strip()
|
||||
if ocr:
|
||||
extras_list.append(html_extras.text(ocr, name=f"OCR: {label}{frame_str}"))
|
||||
report.extras = extras_list # type: ignore[attr-defined]
|
||||
|
||||
|
||||
@pytest.hookimpl(hookwrapper=True)
|
||||
def pytest_runtest_makereport(item: pytest.Item, call: pytest.CallInfo[Any]) -> Any:
|
||||
"""On test failure, attach serial capture + device state as report artifacts.
|
||||
@@ -967,10 +1088,20 @@ def pytest_runtest_makereport(item: pytest.Item, call: pytest.CallInfo[Any]) ->
|
||||
Hard-bounded by `_run_with_timeout` — if the device is unreachable (stuck
|
||||
port, unbaked firmware, dead board), the dump is skipped rather than
|
||||
hanging the session.
|
||||
|
||||
For UI-tier tests, also embeds per-step camera captures + OCR on every
|
||||
test (pass or fail) so the HTML report shows visual evidence of what
|
||||
the device did.
|
||||
"""
|
||||
outcome = yield
|
||||
report = outcome.get_result()
|
||||
|
||||
# Attach UI captures on any outcome (pass + fail) — these are the whole
|
||||
# point of the UI tier. Do this before the failure-only branch below so
|
||||
# passing tests still get their image strip.
|
||||
if report.when == "call":
|
||||
_attach_ui_captures(item, report)
|
||||
|
||||
if report.when != "call" or report.outcome != "failed":
|
||||
return
|
||||
|
||||
|
||||
@@ -0,0 +1,155 @@
|
||||
"""Isolation test for peer-offline-then-back mid-conversation.
|
||||
|
||||
Verifies the mesh stack's behavior when a peer is physically powered
|
||||
off mid-send via uhubctl, then powered back on.
|
||||
|
||||
Flow (parametrized over every directed mesh_pair):
|
||||
1. Bilateral PKI warmup (same pattern as test_direct_with_ack).
|
||||
2. TX sends a broadcast text "msg-1" — RX confirms receipt via pubsub.
|
||||
3. Power OFF RX via uhubctl. The RX device disappears from the OS.
|
||||
4. TX sends a directed text "msg-2" with wantAck=True. Firmware retries
|
||||
internally for ~30s before giving up. Assertion: the packet object
|
||||
was accepted by the TX stack (non-None) — we don't assert an ACK
|
||||
since there's no peer to send one.
|
||||
5. Power ON RX. Wait for re-enumeration + boot.
|
||||
6. Bilateral PKI re-nudge — RX's in-RAM PKI cache was wiped on reboot,
|
||||
so the first directed send may err=35 without a fresh NodeInfo ping.
|
||||
7. TX sends a directed "msg-3" — RX receives it via pubsub, confirming
|
||||
the mesh recovered.
|
||||
|
||||
Skips cleanly if uhubctl isn't installed (via the `power_cycle` fixture's
|
||||
auto-skip). Skips for pair directions where RX isn't power-controllable
|
||||
(e.g. a USB-IF hub that doesn't support PPPS for its port).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from meshtastic_mcp.connection import connect
|
||||
from tests import _power
|
||||
from tests._port_discovery import resolve_port_by_role
|
||||
|
||||
from ._receive import ReceiveCollector, nudge_nodeinfo
|
||||
|
||||
|
||||
@pytest.mark.timeout(360)
|
||||
def test_peer_offline_then_recovers(
|
||||
mesh_pair: dict[str, Any],
|
||||
power_cycle, # noqa: ARG001 — forces uhubctl-availability skip
|
||||
hub_devices: dict[str, str],
|
||||
) -> None:
|
||||
tx_port = mesh_pair["tx"]["port"]
|
||||
rx_node_num = mesh_pair["rx"]["my_node_num"]
|
||||
tx_role = mesh_pair["tx_role"]
|
||||
rx_role = mesh_pair["rx_role"]
|
||||
|
||||
unique_pre = f"peer-offline-pre-{tx_role}-to-{rx_role}-{int(time.time())}"
|
||||
unique_post = f"peer-offline-post-{tx_role}-to-{rx_role}-{int(time.time())}"
|
||||
|
||||
# Step 1 + 2: warm up + confirm baseline delivery works before the test.
|
||||
with ReceiveCollector(
|
||||
mesh_pair["rx"]["port"], topic="meshtastic.receive.text"
|
||||
) as rx:
|
||||
rx.broadcast_nodeinfo_ping()
|
||||
with connect(port=tx_port) as tx_iface:
|
||||
nudge_nodeinfo(tx_iface)
|
||||
# Wait for bilateral PKI (RX pubkey in TX's nodesByNum).
|
||||
deadline = time.monotonic() + 45.0
|
||||
last_nudge = time.monotonic()
|
||||
while time.monotonic() < deadline:
|
||||
rec = (tx_iface.nodesByNum or {}).get(rx_node_num, {})
|
||||
if rec.get("user", {}).get("publicKey"):
|
||||
break
|
||||
if time.monotonic() - last_nudge > 15.0:
|
||||
rx.broadcast_nodeinfo_ping()
|
||||
nudge_nodeinfo(tx_iface)
|
||||
last_nudge = time.monotonic()
|
||||
time.sleep(1.0)
|
||||
else:
|
||||
pytest.skip(
|
||||
f"bilateral PKI never completed ({tx_role}→{rx_role}); "
|
||||
"can't run the offline test without a warm baseline"
|
||||
)
|
||||
|
||||
tx_iface.sendText(unique_pre, destinationId=rx_node_num, wantAck=True)
|
||||
got = rx.wait_for(
|
||||
lambda pkt: pkt.get("decoded", {}).get("text") == unique_pre,
|
||||
timeout=30,
|
||||
)
|
||||
assert got is not None, (
|
||||
f"baseline directed send ({tx_role}→{rx_role}) didn't land — "
|
||||
"skipping offline test to avoid false positive"
|
||||
)
|
||||
|
||||
# Step 3: power off RX. uhubctl skips the test with a clear message if
|
||||
# the RX role isn't on a controllable hub.
|
||||
try:
|
||||
_power.power_off(rx_role)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
pytest.skip(f"can't power-control {rx_role!r}: {exc}")
|
||||
|
||||
try:
|
||||
_power.wait_for_absence(rx_role, timeout_s=10.0)
|
||||
except TimeoutError:
|
||||
_power.power_on(rx_role) # restore hub state before failing
|
||||
resolve_port_by_role(rx_role, timeout_s=30.0)
|
||||
pytest.fail(f"{rx_role!r} didn't disappear after power_off")
|
||||
|
||||
# Step 4: send to a peer that isn't there. Firmware will retry
|
||||
# internally. We don't wait for an ACK (there won't be one); we just
|
||||
# confirm TX's stack accepts the packet without crashing.
|
||||
try:
|
||||
with connect(port=tx_port) as tx_iface:
|
||||
packet = tx_iface.sendText(
|
||||
f"while-offline-{rx_role}",
|
||||
destinationId=rx_node_num,
|
||||
wantAck=True,
|
||||
)
|
||||
assert packet is not None
|
||||
# Give firmware a moment to do a retry or two while RX is down.
|
||||
time.sleep(5.0)
|
||||
except Exception as exc: # noqa: BLE001 — TX should survive the peer being gone
|
||||
# Restore RX before reraising so the bench state is sane.
|
||||
_power.power_on(rx_role)
|
||||
resolve_port_by_role(rx_role, timeout_s=30.0)
|
||||
raise AssertionError(f"TX crashed when sending to offline peer: {exc}") from exc
|
||||
|
||||
# Step 5: power RX back on + rediscover.
|
||||
_power.power_on(rx_role)
|
||||
time.sleep(0.5)
|
||||
new_rx_port = resolve_port_by_role(rx_role, timeout_s=30.0)
|
||||
hub_devices[rx_role] = new_rx_port
|
||||
|
||||
# Step 6 + 7: bilateral re-warmup + directed send that should now work.
|
||||
with ReceiveCollector(new_rx_port, topic="meshtastic.receive.text") as rx:
|
||||
# RX rebooted → its PKI cache is gone. Re-warm.
|
||||
rx.broadcast_nodeinfo_ping()
|
||||
with connect(port=tx_port) as tx_iface:
|
||||
nudge_nodeinfo(tx_iface)
|
||||
time.sleep(3.0)
|
||||
|
||||
got = None
|
||||
for _attempt in range(3):
|
||||
packet = tx_iface.sendText(
|
||||
unique_post,
|
||||
destinationId=rx_node_num,
|
||||
wantAck=True,
|
||||
)
|
||||
assert packet is not None
|
||||
got = rx.wait_for(
|
||||
lambda pkt: pkt.get("decoded", {}).get("text") == unique_post,
|
||||
timeout=30,
|
||||
)
|
||||
if got is not None:
|
||||
break
|
||||
rx.broadcast_nodeinfo_ping()
|
||||
nudge_nodeinfo(tx_iface)
|
||||
time.sleep(5.0)
|
||||
|
||||
assert got is not None, (
|
||||
f"post-recovery directed send {unique_post!r} ({tx_role}→{rx_role}) "
|
||||
"never landed — recovery path may be broken"
|
||||
)
|
||||
@@ -0,0 +1,6 @@
|
||||
"""Recovery tier — exercises `uhubctl` power control end-to-end.
|
||||
|
||||
Requires `uhubctl` installed AND at least one connected device on a
|
||||
PPPS-capable hub. The whole tier skips cleanly via
|
||||
`tests/recovery/conftest.py::_recovery_tier_guard` when either is missing.
|
||||
"""
|
||||
@@ -0,0 +1,44 @@
|
||||
"""Recovery-tier gating + shared helpers.
|
||||
|
||||
Session-scoped guard skips the whole tier when uhubctl isn't installed.
|
||||
Tests under this directory assume uhubctl is callable AND that at least
|
||||
one hub role is detected on a PPPS-capable port.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture(scope="session", autouse=True)
|
||||
def _recovery_tier_guard() -> None:
|
||||
"""Skip the tier when uhubctl is unavailable OR no device is on a
|
||||
PPPS-capable hub. Prints the specific reason so operators know what
|
||||
to fix."""
|
||||
from tests import _power
|
||||
|
||||
if not _power.is_uhubctl_available():
|
||||
pytest.skip(
|
||||
"uhubctl not installed; recovery tier needs it. "
|
||||
"Install via `brew install uhubctl` or `apt install uhubctl`.",
|
||||
allow_module_level=True,
|
||||
)
|
||||
|
||||
# Probe: can we even list hubs? (A macOS user without sudo gets a
|
||||
# permission error here — we'd rather find out once at tier-start than
|
||||
# 6 tests later.)
|
||||
from meshtastic_mcp import uhubctl
|
||||
|
||||
try:
|
||||
hubs = uhubctl.list_hubs()
|
||||
except uhubctl.UhubctlError as exc:
|
||||
pytest.skip(
|
||||
f"uhubctl list failed: {exc}. Try the udev rules or `sudo` as a fallback.",
|
||||
allow_module_level=True,
|
||||
)
|
||||
|
||||
if not any(h["ppps"] for h in hubs):
|
||||
pytest.skip(
|
||||
"no PPPS-capable hubs detected — recovery tier has nothing to exercise.",
|
||||
allow_module_level=True,
|
||||
)
|
||||
@@ -0,0 +1,43 @@
|
||||
"""Smoke test: `uhubctl_list` returns a well-formed structure.
|
||||
|
||||
No destructive action. Runs first in the tier as a sanity check that the
|
||||
tier's dependencies (uhubctl binary + permissions) are actually satisfied.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from meshtastic_mcp import uhubctl
|
||||
|
||||
|
||||
@pytest.mark.timeout(30)
|
||||
def test_list_hubs_returns_at_least_one_ppps_hub() -> None:
|
||||
hubs = uhubctl.list_hubs()
|
||||
assert hubs, "uhubctl found no hubs at all — is a USB hub connected?"
|
||||
assert any(h["ppps"] for h in hubs), (
|
||||
"no PPPS-capable hubs detected; power control won't work. "
|
||||
"Check that the hub supports Per-Port Power Switching."
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.timeout(30)
|
||||
def test_list_hubs_structure(hub_devices: dict[str, str]) -> None:
|
||||
hubs = uhubctl.list_hubs()
|
||||
for hub in hubs:
|
||||
assert "location" in hub and hub["location"]
|
||||
assert "ports" in hub and isinstance(hub["ports"], list)
|
||||
for port in hub["ports"]:
|
||||
assert "port" in port and isinstance(port["port"], int)
|
||||
assert "status" in port
|
||||
|
||||
# At least one of the detected Meshtastic roles should show up in some
|
||||
# port's device_vid — otherwise the recovery tier can't drive them.
|
||||
seen_vids = {
|
||||
p["device_vid"] for h in hubs for p in h["ports"] if p["device_vid"] is not None
|
||||
}
|
||||
expected_any = {0x239A, 0x303A, 0x10C4} & seen_vids
|
||||
assert expected_any or not hub_devices, (
|
||||
f"hub_devices detected roles {list(hub_devices)} but uhubctl sees "
|
||||
f"VIDs {sorted(hex(v) for v in seen_vids)} — the devices may be on "
|
||||
"a hub that uhubctl can't see (e.g. built-in laptop ports)."
|
||||
)
|
||||
@@ -0,0 +1,60 @@
|
||||
"""Hard reset via uhubctl must NOT wipe NVS. Verify the test profile's
|
||||
region + channel survive a power-cycle.
|
||||
|
||||
Guards against a regression where a firmware change treats unexpected
|
||||
power loss as a factory-reset trigger (e.g. bad EEPROM wear-leveling,
|
||||
erase-on-boot-for-safety). Such a regression would be catastrophic for
|
||||
field deployments.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
|
||||
import pytest
|
||||
from meshtastic_mcp import admin, info
|
||||
from tests import _power
|
||||
from tests._port_discovery import resolve_port_by_role
|
||||
|
||||
|
||||
@pytest.mark.timeout(180)
|
||||
def test_lora_config_survives_power_cycle(
|
||||
baked_single: dict[str, object],
|
||||
test_profile: dict[str, object],
|
||||
) -> None:
|
||||
role = baked_single["role"]
|
||||
pre_port = baked_single["port"]
|
||||
|
||||
pre_config = admin.get_config(section="lora", port=pre_port)["config"]["lora"]
|
||||
pre_region = pre_config.get("region")
|
||||
pre_preset = pre_config.get("modem_preset")
|
||||
assert pre_region, f"lora.region not set pre-cycle on {role}"
|
||||
|
||||
# Hard power-cycle.
|
||||
_power.power_cycle(role, delay_s=2)
|
||||
time.sleep(0.5)
|
||||
new_port = resolve_port_by_role(role, timeout_s=30.0)
|
||||
# Let the firmware complete boot before admin reads.
|
||||
time.sleep(2.0)
|
||||
# Quick readiness probe.
|
||||
probe = info.device_info(port=new_port, timeout_s=10.0)
|
||||
assert (
|
||||
probe.get("my_node_num") is not None
|
||||
), f"device {role!r} didn't respond after power-cycle"
|
||||
|
||||
post_config = admin.get_config(section="lora", port=new_port)["config"]["lora"]
|
||||
post_region = post_config.get("region")
|
||||
post_preset = post_config.get("modem_preset")
|
||||
|
||||
assert post_region == pre_region, (
|
||||
f"lora.region wiped by power-cycle on {role}: "
|
||||
f"pre={pre_region!r} post={post_region!r}"
|
||||
)
|
||||
assert post_preset == pre_preset, (
|
||||
f"lora.modem_preset wiped by power-cycle on {role}: "
|
||||
f"pre={pre_preset!r} post={post_preset!r}"
|
||||
)
|
||||
|
||||
# Channel-0 name should also match the test profile.
|
||||
pri_ch = admin.get_channel_url(port=new_port)
|
||||
assert pri_ch.get("url"), f"channel URL empty after power-cycle on {role}"
|
||||
@@ -0,0 +1,61 @@
|
||||
"""Full power-cycle round-trip: off → verify gone → on → verify identity
|
||||
preserved.
|
||||
|
||||
Parametrized over every connected role. Validates both the uhubctl
|
||||
plumbing AND that the device survives a hard reset with the same
|
||||
`my_node_num` (no firmware-level identity regeneration).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
|
||||
import pytest
|
||||
from meshtastic_mcp import info
|
||||
from tests import _power
|
||||
from tests._port_discovery import resolve_port_by_role
|
||||
|
||||
|
||||
@pytest.mark.timeout(180)
|
||||
def test_power_cycle_preserves_node_identity(
|
||||
baked_single: dict[str, object],
|
||||
) -> None:
|
||||
role = baked_single["role"]
|
||||
pre_port = baked_single["port"]
|
||||
pre_node_num = baked_single["my_node_num"]
|
||||
pre_fw = baked_single.get("firmware_version")
|
||||
|
||||
# Record pre-cycle state.
|
||||
pre_info = info.device_info(port=pre_port, timeout_s=5.0)
|
||||
assert pre_info.get("my_node_num") == pre_node_num
|
||||
|
||||
# Power off; confirm the device actually disappears from list_devices.
|
||||
_power.power_off(role)
|
||||
try:
|
||||
_power.wait_for_absence(role, timeout_s=10.0)
|
||||
except TimeoutError:
|
||||
# If it didn't disappear, power it back on so we don't leave the
|
||||
# hub in a weird state for the next test.
|
||||
_power.power_on(role)
|
||||
resolve_port_by_role(role, timeout_s=30.0)
|
||||
pytest.fail(f"device {role!r} stayed visible after power_off")
|
||||
|
||||
# Power back on + re-discover port.
|
||||
_power.power_on(role)
|
||||
time.sleep(0.5) # head-start before polling
|
||||
new_port = resolve_port_by_role(role, timeout_s=30.0)
|
||||
|
||||
# Give the firmware a moment to finish boot before we hit it with admin.
|
||||
time.sleep(2.0)
|
||||
|
||||
post_info = info.device_info(port=new_port, timeout_s=10.0)
|
||||
assert post_info.get("my_node_num") == pre_node_num, (
|
||||
f"my_node_num changed across power-cycle: pre={pre_node_num:#x} "
|
||||
f"post={post_info.get('my_node_num'):#x}"
|
||||
)
|
||||
# Firmware version must match (same bake, not a re-flash).
|
||||
if pre_fw:
|
||||
assert post_info.get("firmware_version") == pre_fw, (
|
||||
f"firmware changed across cycle: pre={pre_fw} "
|
||||
f"post={post_info.get('firmware_version')}"
|
||||
)
|
||||
@@ -73,6 +73,13 @@ _TOOL_MAP: dict[str, tuple[str, str]] = {
|
||||
"reboot": ("meshtastic_mcp.admin", "reboot"),
|
||||
"shutdown": ("meshtastic_mcp.admin", "shutdown"),
|
||||
"factory_reset": ("meshtastic_mcp.admin", "factory_reset"),
|
||||
"send_input_event": ("meshtastic_mcp.admin", "send_input_event"),
|
||||
# `capture_screen` in server.py calls camera.get_camera — instrument that.
|
||||
"capture_screen": ("meshtastic_mcp.camera", "get_camera"),
|
||||
# USB power control via uhubctl.
|
||||
"uhubctl_list": ("meshtastic_mcp.uhubctl", "list_hubs"),
|
||||
"uhubctl_power": ("meshtastic_mcp.uhubctl", "power_on"),
|
||||
"uhubctl_cycle": ("meshtastic_mcp.uhubctl", "cycle"),
|
||||
# USERPREFS
|
||||
"userprefs_manifest": ("meshtastic_mcp.userprefs", "build_manifest"),
|
||||
"userprefs_get": ("meshtastic_mcp.userprefs", "read_state"),
|
||||
|
||||
@@ -0,0 +1,7 @@
|
||||
"""UI tier — input-broker-driven screen navigation tests.
|
||||
|
||||
Only runs when a screen-bearing role (esp32s3/heltec-v3) is present on the
|
||||
hub AND the firmware was baked with `enable_ui_log=True` (so the
|
||||
`Screen: frame N/M name=... reason=...` log lines are emitted). The
|
||||
`tests/ui/conftest.py` fixture forces that bake stamp.
|
||||
"""
|
||||
@@ -0,0 +1,176 @@
|
||||
"""Parse `Screen: frame N/M name=X reason=Y` log lines from `_debug_log_buffer`.
|
||||
|
||||
The firmware emits one line per frame transition when
|
||||
`USERPREFS_UI_TEST_LOG` is defined (see src/graphics/Screen.cpp). Tests use
|
||||
these helpers to assert which frame is shown / to wait for a transition to
|
||||
settle before taking a camera capture.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from typing import Iterable, Iterator
|
||||
|
||||
FRAME_RE = re.compile(
|
||||
r"Screen: frame (?P<idx>\d+)/(?P<count>\d+) name=(?P<name>\S+) reason=(?P<reason>\S+)"
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class FrameEvent:
|
||||
idx: int
|
||||
count: int
|
||||
name: str
|
||||
reason: str
|
||||
raw: str
|
||||
|
||||
@classmethod
|
||||
def parse(cls, line: str) -> "FrameEvent | None":
|
||||
m = FRAME_RE.search(line)
|
||||
if not m:
|
||||
return None
|
||||
return cls(
|
||||
idx=int(m["idx"]),
|
||||
count=int(m["count"]),
|
||||
name=m["name"],
|
||||
reason=m["reason"],
|
||||
raw=line,
|
||||
)
|
||||
|
||||
|
||||
def iter_frame_events(lines: Iterable[str]) -> Iterator[FrameEvent]:
|
||||
for line in lines:
|
||||
evt = FrameEvent.parse(line)
|
||||
if evt is not None:
|
||||
yield evt
|
||||
|
||||
|
||||
def get_current_frame(lines: list[str]) -> FrameEvent | None:
|
||||
"""Return the most recent FrameEvent in `lines`, or None if none found."""
|
||||
for line in reversed(lines):
|
||||
evt = FrameEvent.parse(line)
|
||||
if evt is not None:
|
||||
return evt
|
||||
return None
|
||||
|
||||
|
||||
def wait_for_frame(
|
||||
lines: list[str],
|
||||
expected_name: str,
|
||||
*,
|
||||
timeout_s: float = 5.0,
|
||||
poll_interval_s: float = 0.1,
|
||||
reason: str | None = None,
|
||||
) -> FrameEvent:
|
||||
"""Poll `lines` (the `_debug_log_buffer`) until a FrameEvent with
|
||||
`name=expected_name` appears after the call started. Raises TimeoutError
|
||||
with context if it doesn't arrive in `timeout_s`.
|
||||
|
||||
`reason` optionally filters to events matching a specific cause
|
||||
(e.g. `"fn_f1"`, `"next"`, `"rebuild"`).
|
||||
"""
|
||||
start_idx = len(lines)
|
||||
deadline = time.monotonic() + timeout_s
|
||||
last: FrameEvent | None = None
|
||||
while time.monotonic() < deadline:
|
||||
# Scan only lines appended since we started waiting.
|
||||
for line in lines[start_idx:]:
|
||||
evt = FrameEvent.parse(line)
|
||||
if evt is None:
|
||||
continue
|
||||
last = evt
|
||||
if evt.name == expected_name and (reason is None or evt.reason == reason):
|
||||
return evt
|
||||
time.sleep(poll_interval_s)
|
||||
|
||||
seen = [e.name for e in iter_frame_events(lines[start_idx:])]
|
||||
raise TimeoutError(
|
||||
f"frame name={expected_name!r} reason={reason!r} not seen in {timeout_s}s; "
|
||||
f"saw {len(seen)} transition(s): {seen!r}; last={last!r}"
|
||||
)
|
||||
|
||||
|
||||
def wait_for_any_frame(
|
||||
lines: list[str],
|
||||
*,
|
||||
timeout_s: float = 5.0,
|
||||
poll_interval_s: float = 0.1,
|
||||
) -> FrameEvent:
|
||||
"""Wait for ANY frame transition to appear after call-start. Useful for
|
||||
`no-op` tests that want to confirm a transition did NOT happen (via
|
||||
TimeoutError) vs. one that did.
|
||||
"""
|
||||
start_idx = len(lines)
|
||||
deadline = time.monotonic() + timeout_s
|
||||
while time.monotonic() < deadline:
|
||||
for line in lines[start_idx:]:
|
||||
evt = FrameEvent.parse(line)
|
||||
if evt is not None:
|
||||
return evt
|
||||
time.sleep(poll_interval_s)
|
||||
raise TimeoutError(f"no frame transition in {timeout_s}s")
|
||||
|
||||
|
||||
def wait_for_reason(
|
||||
lines: list[str],
|
||||
reason: str,
|
||||
*,
|
||||
timeout_s: float = 5.0,
|
||||
poll_interval_s: float = 0.1,
|
||||
) -> FrameEvent:
|
||||
"""Wait for a frame event with `reason=<reason>` after call-start.
|
||||
|
||||
Matches only on `reason` — useful when the caller knows *why* a
|
||||
transition should happen (e.g. `fn_f1`, `rebuild`) but not which named
|
||||
frame the firmware will land on for this particular board.
|
||||
"""
|
||||
start_idx = len(lines)
|
||||
deadline = time.monotonic() + timeout_s
|
||||
last: FrameEvent | None = None
|
||||
while time.monotonic() < deadline:
|
||||
for line in lines[start_idx:]:
|
||||
evt = FrameEvent.parse(line)
|
||||
if evt is None:
|
||||
continue
|
||||
last = evt
|
||||
if evt.reason == reason:
|
||||
return evt
|
||||
time.sleep(poll_interval_s)
|
||||
raise TimeoutError(
|
||||
f"no frame with reason={reason!r} in {timeout_s}s; last={last!r}"
|
||||
)
|
||||
|
||||
|
||||
def assert_no_frame_change(
|
||||
lines: list[str],
|
||||
*,
|
||||
wait_s: float = 2.0,
|
||||
) -> None:
|
||||
"""Assert that NO new FrameEvent lines arrive within `wait_s`.
|
||||
|
||||
Used by idempotency / no-op tests (e.g. BACK on home frame).
|
||||
"""
|
||||
start_idx = len(lines)
|
||||
time.sleep(wait_s)
|
||||
new = [
|
||||
e for e in (FrameEvent.parse(ln) for ln in lines[start_idx:]) if e is not None
|
||||
]
|
||||
if new:
|
||||
raise AssertionError(
|
||||
f"expected no frame change in {wait_s}s, but saw {len(new)} event(s): "
|
||||
f"{[(e.reason, e.name) for e in new]!r}"
|
||||
)
|
||||
|
||||
|
||||
__all__ = [
|
||||
"FRAME_RE",
|
||||
"FrameEvent",
|
||||
"assert_no_frame_change",
|
||||
"get_current_frame",
|
||||
"iter_frame_events",
|
||||
"wait_for_any_frame",
|
||||
"wait_for_frame",
|
||||
"wait_for_reason",
|
||||
]
|
||||
@@ -0,0 +1,381 @@
|
||||
"""UI-tier fixtures: camera lifecycle, OCR warmup, per-test frame capture,
|
||||
and a `ui_home_state` autouse guard that resets to the home frame before
|
||||
every test (prevents state bleed if a prior test exited inside a menu).
|
||||
|
||||
The camera + OCR modules live in `meshtastic_mcp/{camera,ocr}.py` (production
|
||||
code, so the `capture_screen` MCP tool can share them). These fixtures wire
|
||||
them into pytest + write per-test captures to `tests/ui_captures/…`.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import shutil
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Any, Iterator
|
||||
|
||||
import pytest
|
||||
from meshtastic_mcp import admin as admin_mod
|
||||
from meshtastic_mcp import camera as camera_mod
|
||||
from meshtastic_mcp import ocr as ocr_mod
|
||||
from meshtastic_mcp.input_events import InputEventCode
|
||||
|
||||
from ._screen_log import FrameEvent, get_current_frame, wait_for_frame
|
||||
|
||||
# Roles that carry a screen the UI tier can drive. Only esp32s3 (heltec-v3
|
||||
# SSD1306) qualifies today — nrf52 (rak4631) has no display.
|
||||
UI_CAPABLE_ROLES = ("esp32s3",)
|
||||
|
||||
# Where per-test captures land. One subdirectory per session seed, then per
|
||||
# sanitized test nodeid — identical pattern to other pytest artifacts.
|
||||
CAPTURES_ROOT = Path(__file__).resolve().parent.parent / "ui_captures"
|
||||
|
||||
|
||||
def _sanitize_nodeid(nodeid: str) -> str:
|
||||
return re.sub(r"[^a-zA-Z0-9_.-]+", "_", nodeid)
|
||||
|
||||
|
||||
# ---------- Role gating ----------------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def ui_capable_role(request: pytest.FixtureRequest, hub_devices: dict[str, Any]) -> str:
|
||||
"""Resolve the single role the UI tier drives.
|
||||
|
||||
Today that's `esp32s3`. Skips if the hub doesn't have one. A future
|
||||
multi-screen hub could pick a role per parametrization.
|
||||
"""
|
||||
for role in UI_CAPABLE_ROLES:
|
||||
if role in hub_devices:
|
||||
return role
|
||||
pytest.skip(
|
||||
f"no UI-capable role on hub; need one of {UI_CAPABLE_ROLES} in {sorted(hub_devices)}"
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def ui_port(ui_capable_role: str, hub_devices: dict[str, Any]) -> str:
|
||||
port = (
|
||||
hub_devices[ui_capable_role].get("port")
|
||||
if isinstance(hub_devices[ui_capable_role], dict)
|
||||
else hub_devices[ui_capable_role]
|
||||
)
|
||||
if not port:
|
||||
pytest.skip(f"{ui_capable_role!r} has no usable port")
|
||||
return port
|
||||
|
||||
|
||||
# ---------- Camera + OCR session fixtures ---------------------------------
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def camera(ui_capable_role_session: str | None) -> Iterator[camera_mod.CameraBackend]:
|
||||
"""Session-scoped camera backend. Closed at teardown.
|
||||
|
||||
Backend + device selected by env vars (see `meshtastic_mcp.camera`).
|
||||
Falls through to `NullBackend` when no camera is configured, so the
|
||||
tests run end-to-end on machines without hardware; they just won't
|
||||
have useful images.
|
||||
"""
|
||||
role = ui_capable_role_session or "esp32s3"
|
||||
cam = camera_mod.get_camera(role)
|
||||
try:
|
||||
yield cam
|
||||
finally:
|
||||
cam.close()
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def ui_capable_role_session(hub_devices: dict[str, Any]) -> str | None:
|
||||
"""Session-scoped lookup mirroring `ui_capable_role` but non-skipping.
|
||||
|
||||
Used by the `camera` session fixture so it doesn't depend on a
|
||||
test-scoped skip.
|
||||
"""
|
||||
for role in UI_CAPABLE_ROLES:
|
||||
if role in hub_devices:
|
||||
return role
|
||||
return None
|
||||
|
||||
|
||||
@pytest.fixture(scope="session", autouse=True)
|
||||
def _ocr_warm() -> None:
|
||||
"""Pay easyocr's ~100 MB / cold-start cost ONCE per session.
|
||||
|
||||
Subsequent `ocr_text()` calls hit the cached reader and return quickly.
|
||||
Swallows errors — if OCR isn't installed, warm is a no-op.
|
||||
"""
|
||||
try:
|
||||
ocr_mod.warm()
|
||||
except Exception: # noqa: BLE001 — belt: never block the suite on OCR init
|
||||
pass
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def _ui_screen_kept_on(
|
||||
ui_capable_role_session: str | None, hub_devices: dict[str, Any]
|
||||
) -> Iterator[None]:
|
||||
"""Keep the OLED on throughout the UI tier so input events aren't dropped.
|
||||
|
||||
Why: `InputBroker::handleInputEvent` (src/input/InputBroker.cpp:118-122)
|
||||
silently DROPS any event that arrives while the screen is off — it just
|
||||
wakes the screen and returns. Every first event in each test would
|
||||
disappear. We set `display.screen_on_secs = 86400` at session start
|
||||
(effectively "always on" for the test window) and restore the prior
|
||||
value at teardown.
|
||||
"""
|
||||
if ui_capable_role_session is None:
|
||||
yield
|
||||
return
|
||||
|
||||
hub_entry = hub_devices[ui_capable_role_session]
|
||||
port = hub_entry.get("port") if isinstance(hub_entry, dict) else hub_entry
|
||||
if not port:
|
||||
yield
|
||||
return
|
||||
|
||||
original: int | None = None
|
||||
try:
|
||||
current = admin_mod.get_config(section="display", port=port)
|
||||
original = int(
|
||||
current.get("config", {}).get("display", {}).get("screen_on_secs") or 0
|
||||
)
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
|
||||
try:
|
||||
admin_mod.set_config("display.screen_on_secs", 86400, port=port)
|
||||
# Send one wake event so the screen is actually ON going into the
|
||||
# first test. The event itself gets dropped (screenWasOff), but the
|
||||
# wake side-effect sticks.
|
||||
try:
|
||||
admin_mod.send_input_event(event_code=int(InputEventCode.FN_F1), port=port)
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
time.sleep(1.5) # Let the screen finish its wake transition.
|
||||
except (
|
||||
Exception
|
||||
): # noqa: BLE001 — best-effort; ui_home_state surfaces the real error
|
||||
pass
|
||||
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
if original is not None:
|
||||
try:
|
||||
admin_mod.set_config("display.screen_on_secs", original, port=port)
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
|
||||
|
||||
# ---------- Per-test capture + transcript ----------------------------------
|
||||
|
||||
|
||||
class FrameCapture:
|
||||
"""Per-test capture recorder. Created once per test via the
|
||||
`frame_capture` fixture; call with a label to snapshot the screen.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
cam: camera_mod.CameraBackend,
|
||||
dir_path: Path,
|
||||
lines: list[str],
|
||||
nodeid: str,
|
||||
) -> None:
|
||||
self._cam = cam
|
||||
self._dir = dir_path
|
||||
self._lines = lines
|
||||
self._nodeid = nodeid
|
||||
self._step = 0
|
||||
self.captures: list[dict[str, Any]] = []
|
||||
self._transcript_path = dir_path / "transcript.md"
|
||||
self._dir.mkdir(parents=True, exist_ok=True)
|
||||
self._transcript_path.write_text(
|
||||
f"# {nodeid} — {time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())}\n\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
def __call__(self, label: str) -> dict[str, Any]:
|
||||
self._step += 1
|
||||
stem = f"{self._step:03d}-{re.sub(r'[^a-zA-Z0-9_-]+', '-', label)}"
|
||||
png_path = self._dir / f"{stem}.png"
|
||||
ocr_path = self._dir / f"{stem}.ocr.txt"
|
||||
|
||||
try:
|
||||
png = self._cam.capture()
|
||||
except Exception as exc: # noqa: BLE001
|
||||
png = b""
|
||||
ocr_str = f"[capture error: {exc}]"
|
||||
else:
|
||||
camera_mod.save_capture(png, png_path)
|
||||
try:
|
||||
ocr_str = ocr_mod.ocr_text(png)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
ocr_str = f"[ocr error: {exc}]"
|
||||
ocr_path.write_text(ocr_str or "", encoding="utf-8")
|
||||
|
||||
frame = get_current_frame(self._lines)
|
||||
entry: dict[str, Any] = {
|
||||
"step": self._step,
|
||||
"label": label,
|
||||
"png_path": str(png_path) if png else None,
|
||||
"ocr_text": ocr_str,
|
||||
"frame": (
|
||||
{
|
||||
"idx": frame.idx,
|
||||
"name": frame.name,
|
||||
"reason": frame.reason,
|
||||
}
|
||||
if frame is not None
|
||||
else None
|
||||
),
|
||||
}
|
||||
self.captures.append(entry)
|
||||
|
||||
with self._transcript_path.open("a", encoding="utf-8") as fh:
|
||||
frame_str = (
|
||||
f"frame {frame.idx}/{frame.count} name={frame.name} reason={frame.reason}"
|
||||
if frame is not None
|
||||
else "frame <none>"
|
||||
)
|
||||
ocr_summary = (ocr_str or "").replace("\n", " / ")[:80]
|
||||
fh.write(
|
||||
f"{self._step}. **{label}** — {frame_str} — OCR: `{ocr_summary}`\n"
|
||||
)
|
||||
return entry
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def frame_capture(
|
||||
request: pytest.FixtureRequest,
|
||||
camera: camera_mod.CameraBackend,
|
||||
session_seed: str,
|
||||
) -> Iterator[FrameCapture]:
|
||||
nodeid = _sanitize_nodeid(request.node.nodeid)
|
||||
dir_path = CAPTURES_ROOT / session_seed / nodeid
|
||||
# Fresh directory per test run so reruns don't mix old and new images.
|
||||
if dir_path.exists():
|
||||
shutil.rmtree(dir_path)
|
||||
|
||||
lines = getattr(request.node, "_debug_log_buffer", [])
|
||||
fc = FrameCapture(camera, dir_path, lines, nodeid)
|
||||
# Stash so pytest_runtest_makereport can embed captures in HTML extras.
|
||||
request.node._ui_captures = fc.captures # type: ignore[attr-defined]
|
||||
yield fc
|
||||
|
||||
|
||||
# ---------- Pre-test home-state reset --------------------------------------
|
||||
|
||||
|
||||
def _send_event(port: str, event: InputEventCode) -> None:
|
||||
try:
|
||||
admin_mod.send_input_event(event_code=int(event), port=port)
|
||||
except Exception: # noqa: BLE001
|
||||
# Treat a failed event as soft — the subsequent frame-log assertion
|
||||
# surfaces the real problem with better context.
|
||||
pass
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def ui_home_state(
|
||||
request: pytest.FixtureRequest,
|
||||
hub_devices: dict[str, Any],
|
||||
_ui_screen_kept_on: None,
|
||||
) -> Iterator[None]:
|
||||
"""Before every UI test, jump to frame 0 (usually `home`) via FN_F1 and
|
||||
confirm the device emitted the expected frame log.
|
||||
|
||||
Why FN_F1 (not BACK): FN_F1 maps to `switchToFrame(0)` and ALWAYS
|
||||
produces a `reason=fn_f1` log line, regardless of whatever frame the
|
||||
prior test left us on. BACK is context-sensitive (dismisses overlays
|
||||
on some frames, no-op on others) and can silently fail to transition.
|
||||
|
||||
This fixture doubles as the macro-presence detector: if no `fn_f1`
|
||||
log arrives within 5 s, the firmware almost certainly wasn't baked
|
||||
with `USERPREFS_UI_TEST_LOG`. Skip the tier with an actionable hint
|
||||
instead of letting every test body fail with a confusing assertion.
|
||||
|
||||
Autouse scope is restricted to `tests/ui/` by virtue of this fixture
|
||||
living in that directory's conftest.py — no explicit nodeid guard
|
||||
needed (and earlier attempts at one were wrong, matching `/tests/ui/`
|
||||
against a nodeid that has no leading slash).
|
||||
"""
|
||||
role = next((r for r in UI_CAPABLE_ROLES if r in hub_devices), None)
|
||||
if role is None:
|
||||
yield
|
||||
return
|
||||
|
||||
hub_entry = hub_devices[role]
|
||||
port = hub_entry.get("port") if isinstance(hub_entry, dict) else hub_entry
|
||||
lines: list[str] = getattr(request.node, "_debug_log_buffer", [])
|
||||
start_len = len(lines)
|
||||
|
||||
# First: a wake event. The screen should already be kept on by
|
||||
# `_ui_screen_kept_on`, but belt + suspenders — if it somehow
|
||||
# powered off (sleep after factory_reset, etc.), this first FN_F1
|
||||
# gets dropped by InputBroker's screenWasOff guard. That's fine;
|
||||
# the second FN_F1 below lands cleanly.
|
||||
_send_event(port, InputEventCode.FN_F1)
|
||||
time.sleep(0.4)
|
||||
_send_event(port, InputEventCode.FN_F1)
|
||||
|
||||
# Wait for the fn_f1 transition log. Any new `reason=fn_f1` line
|
||||
# after call-start counts — we don't care about the name (it should
|
||||
# be `home` or `deviceFocused` depending on board-specific frame order).
|
||||
from ._screen_log import wait_for_reason
|
||||
|
||||
try:
|
||||
wait_for_reason(lines, "fn_f1", timeout_s=5.0)
|
||||
except TimeoutError:
|
||||
# One more try — FreeRTOS queue may be draining slowly.
|
||||
_send_event(port, InputEventCode.FN_F1)
|
||||
try:
|
||||
wait_for_reason(lines, "fn_f1", timeout_s=5.0)
|
||||
except TimeoutError:
|
||||
# Look at what the _debug_log_buffer actually contains to
|
||||
# disambiguate "macro off" from "macro on but event lost".
|
||||
frame_lines = [ln for ln in lines[start_len:] if "Screen: frame" in ln]
|
||||
processing_lines = [
|
||||
ln for ln in lines[start_len:] if "Processing input event" in ln
|
||||
]
|
||||
if frame_lines:
|
||||
pytest.skip(
|
||||
f"ui_home_state: events fire but none reach Screen "
|
||||
f"(saw {len(frame_lines)} frame line(s), "
|
||||
f"{len(processing_lines)} admin inject(s)). "
|
||||
f"Device may be in an unusual state — try `--force-bake`."
|
||||
)
|
||||
else:
|
||||
pytest.skip(
|
||||
"ui_home_state: no `Screen: frame` log after FN_F1. "
|
||||
"Firmware not baked with USERPREFS_UI_TEST_LOG — "
|
||||
"run with `--force-bake` to reflash, or verify the "
|
||||
"macro is active in the bake."
|
||||
)
|
||||
yield
|
||||
|
||||
|
||||
# ---------- Small helpers reused by tests ---------------------------------
|
||||
|
||||
|
||||
def send_event(
|
||||
port: str, event: InputEventCode | int | str, **kwargs: Any
|
||||
) -> dict[str, Any]:
|
||||
"""Thin wrapper so tests read `send_event(port, InputEventCode.RIGHT)`."""
|
||||
return admin_mod.send_input_event(event_code=event, port=port, **kwargs)
|
||||
|
||||
|
||||
__all__ = [
|
||||
"FrameCapture",
|
||||
"UI_CAPABLE_ROLES",
|
||||
"send_event",
|
||||
"wait_for_frame",
|
||||
"FrameEvent",
|
||||
]
|
||||
|
||||
|
||||
# Make the helpers discoverable to test modules via `from .conftest import …`.
|
||||
# pytest auto-loads conftest.py, but the symbols above are also re-exported
|
||||
# for readability in the test files.
|
||||
@@ -0,0 +1,61 @@
|
||||
"""FN_F1..F5 directly jumps to frame 0..4 via Screen::handleInputEvent.
|
||||
|
||||
Parametrized over the 5 function keys. Each expects a
|
||||
`Screen: frame <idx>/<count> name=... reason=fn_f<k>` log line, with
|
||||
`idx == k-1`. We don't hardcode the frame *name* because the layout
|
||||
depends on which modules are compiled in for this board.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
|
||||
import pytest
|
||||
from meshtastic_mcp.input_events import InputEventCode
|
||||
|
||||
from ._screen_log import get_current_frame, wait_for_reason
|
||||
from .conftest import FrameCapture, send_event
|
||||
|
||||
|
||||
@pytest.mark.timeout(120)
|
||||
@pytest.mark.parametrize(
|
||||
"event,expected_idx,reason",
|
||||
[
|
||||
(InputEventCode.FN_F1, 0, "fn_f1"),
|
||||
(InputEventCode.FN_F2, 1, "fn_f2"),
|
||||
(InputEventCode.FN_F3, 2, "fn_f3"),
|
||||
(InputEventCode.FN_F4, 3, "fn_f4"),
|
||||
(InputEventCode.FN_F5, 4, "fn_f5"),
|
||||
],
|
||||
ids=["FN_F1", "FN_F2", "FN_F3", "FN_F4", "FN_F5"],
|
||||
)
|
||||
def test_fn_jump_direct_frame(
|
||||
ui_port: str,
|
||||
frame_capture: FrameCapture,
|
||||
request: pytest.FixtureRequest,
|
||||
event: InputEventCode,
|
||||
expected_idx: int,
|
||||
reason: str,
|
||||
) -> None:
|
||||
lines: list[str] = request.node._debug_log_buffer
|
||||
start = get_current_frame(lines)
|
||||
assert start is not None, "no frame log yet — USERPREFS_UI_TEST_LOG not wired?"
|
||||
assert start.name in (
|
||||
"home",
|
||||
"deviceFocused",
|
||||
), f"setup expected frame 0 landing, got {start.name!r}"
|
||||
frame_capture("initial")
|
||||
|
||||
if start.count <= expected_idx:
|
||||
pytest.skip(
|
||||
f"device has {start.count} frames; FN_F{expected_idx + 1} needs > {expected_idx}"
|
||||
)
|
||||
|
||||
send_event(ui_port, event)
|
||||
time.sleep(0.1)
|
||||
evt = wait_for_reason(lines, reason, timeout_s=5.0)
|
||||
assert evt.idx == expected_idx, (
|
||||
f"FN_F{expected_idx + 1} expected idx={expected_idx}, got {evt.idx} "
|
||||
f"(name={evt.name}, count={evt.count})"
|
||||
)
|
||||
frame_capture(f"after-{reason}")
|
||||
@@ -0,0 +1,61 @@
|
||||
"""Out-of-bounds FN_F5 when the device has <5 frames: no crash, idx unchanged.
|
||||
|
||||
`Screen::handleInputEvent` dispatches FN_F5 unconditionally to
|
||||
`ui->switchToFrame(4)`. The OLEDDisplayUi library typically clamps or
|
||||
silently ignores out-of-range indices, but firmware bugs have existed
|
||||
here — this test protects against a regression that would wedge the UI.
|
||||
|
||||
If this test fails, first check: did the device actually crash (Guru
|
||||
Meditation in the log)? Or did switchToFrame accept an OOB index and
|
||||
leave the UI blank?
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
|
||||
import pytest
|
||||
from meshtastic_mcp.input_events import InputEventCode
|
||||
|
||||
from ._screen_log import get_current_frame, wait_for_reason
|
||||
from .conftest import FrameCapture, send_event
|
||||
|
||||
|
||||
@pytest.mark.timeout(90)
|
||||
def test_fn_f5_out_of_bounds(
|
||||
ui_port: str,
|
||||
frame_capture: FrameCapture,
|
||||
request: pytest.FixtureRequest,
|
||||
) -> None:
|
||||
lines: list[str] = request.node._debug_log_buffer
|
||||
start = get_current_frame(lines)
|
||||
assert start is not None
|
||||
|
||||
if start.count > 5:
|
||||
pytest.skip(
|
||||
f"device has {start.count} frames; FN_F5 is in-bounds — not testing OOB here"
|
||||
)
|
||||
|
||||
frame_capture("initial-home")
|
||||
send_event(ui_port, InputEventCode.FN_F5)
|
||||
time.sleep(0.5)
|
||||
|
||||
try:
|
||||
wait_for_reason(lines, "fn_f5", timeout_s=3.0)
|
||||
except TimeoutError:
|
||||
# Firmware may have ignored the event entirely — acceptable.
|
||||
pass
|
||||
|
||||
# Capture whatever is on screen (OCR will tell us if something weird
|
||||
# happened). Device must remain responsive — subsequent events should
|
||||
# still land.
|
||||
frame_capture("after-fn_f5-oob")
|
||||
|
||||
# Send a RIGHT to confirm the UI is still alive. If this times out,
|
||||
# the OOB switchToFrame wedged the UI.
|
||||
send_event(ui_port, InputEventCode.RIGHT)
|
||||
post = wait_for_reason(lines, "next", timeout_s=5.0)
|
||||
assert (
|
||||
post is not None
|
||||
), "UI wedged after OOB FN_F5 — RIGHT no longer produces frame log"
|
||||
frame_capture("after-recovery-right")
|
||||
@@ -0,0 +1,68 @@
|
||||
"""SELECT on the home frame opens the home menu; BACK closes it.
|
||||
|
||||
The home menu is an overlay (menuHandler::homeBaseMenu), not a frame
|
||||
transition — so we verify via OCR difference between before/after
|
||||
captures rather than a `Screen: frame` log line. The underlying
|
||||
mechanism is still InputBroker → Screen::handleInputEvent → menu
|
||||
callback.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
|
||||
import pytest
|
||||
from meshtastic_mcp.input_events import InputEventCode
|
||||
|
||||
from ._screen_log import get_current_frame
|
||||
from .conftest import FrameCapture, send_event
|
||||
|
||||
|
||||
@pytest.mark.timeout(120)
|
||||
def test_select_opens_home_menu(
|
||||
ui_port: str,
|
||||
frame_capture: FrameCapture,
|
||||
request: pytest.FixtureRequest,
|
||||
) -> None:
|
||||
lines: list[str] = request.node._debug_log_buffer
|
||||
start = get_current_frame(lines)
|
||||
assert start is not None
|
||||
if start.name not in ("home", "deviceFocused"):
|
||||
pytest.skip(
|
||||
f"SELECT on {start.name!r} doesn't open homeBaseMenu; "
|
||||
"test is only valid when the landing frame is home/deviceFocused"
|
||||
)
|
||||
|
||||
initial = frame_capture("initial")
|
||||
send_event(ui_port, InputEventCode.SELECT)
|
||||
time.sleep(0.8)
|
||||
opened = frame_capture("after-select")
|
||||
|
||||
# The menu is an overlay (not a frame change). We cannot use log
|
||||
# assertion — instead, OCR should differ because a menu list is now
|
||||
# drawn on top.
|
||||
initial_text = (initial.get("ocr_text") or "").strip()
|
||||
opened_text = (opened.get("ocr_text") or "").strip()
|
||||
if initial_text and opened_text:
|
||||
# When OCR is available, require *some* difference between the two
|
||||
# frames — even a single menu title changes the transcribed text.
|
||||
assert initial_text != opened_text, (
|
||||
f"expected OCR diff after SELECT; both read {initial_text!r}. "
|
||||
"If both are empty, check camera alignment + OCR backend."
|
||||
)
|
||||
|
||||
# Back out — the menu dismisses on BACK.
|
||||
send_event(ui_port, InputEventCode.BACK)
|
||||
time.sleep(0.8)
|
||||
closed = frame_capture("after-back")
|
||||
|
||||
# Soft check: OCR after BACK should look different from the menu
|
||||
# (either back to home or onto a previous frame — BACK's exact
|
||||
# behavior when the menu is up vs. not-up varies). We don't assert
|
||||
# equality because OLED rendering is pixel-stable but camera sampling
|
||||
# introduces noise.
|
||||
if opened_text and closed.get("ocr_text"):
|
||||
close_text = (closed.get("ocr_text") or "").strip()
|
||||
assert (
|
||||
close_text != opened_text
|
||||
), f"after BACK, OCR still looks like the menu: {close_text!r}"
|
||||
@@ -0,0 +1,60 @@
|
||||
"""Once we navigate to the textMessage frame, UP/DOWN exercises the
|
||||
message-scroll path (or opens CannedMessages on empty devices).
|
||||
|
||||
Weaker than a "no frame change" assertion because on a fresh bench
|
||||
device the message store is usually empty, and the firmware's UP
|
||||
handler in that case launches CannedMessage — which DOES rebuild
|
||||
frames. We just verify the path doesn't crash + produce captures for
|
||||
visual inspection.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
|
||||
import pytest
|
||||
from meshtastic_mcp.input_events import InputEventCode
|
||||
|
||||
from ._screen_log import get_current_frame, wait_for_frame
|
||||
from .conftest import FrameCapture, send_event
|
||||
|
||||
|
||||
@pytest.mark.timeout(180)
|
||||
def test_up_down_on_textmessage_survives(
|
||||
ui_port: str,
|
||||
frame_capture: FrameCapture,
|
||||
request: pytest.FixtureRequest,
|
||||
) -> None:
|
||||
lines: list[str] = request.node._debug_log_buffer
|
||||
frame_capture("initial")
|
||||
|
||||
# Walk RIGHT until we land on textMessage — up to 15 hops.
|
||||
for _i in range(15):
|
||||
send_event(ui_port, InputEventCode.RIGHT)
|
||||
time.sleep(0.3)
|
||||
current = get_current_frame(lines)
|
||||
if current is not None and current.name == "textMessage":
|
||||
break
|
||||
else:
|
||||
pytest.skip(
|
||||
"couldn't reach textMessage frame within 15 RIGHTs — not present on this board"
|
||||
)
|
||||
|
||||
wait_for_frame(lines, "textMessage", timeout_s=5.0)
|
||||
frame_capture("on-textMessage")
|
||||
|
||||
# UP and DOWN exercise the message-scroll / canned-message-launch path.
|
||||
# Capture after each so the HTML report shows any visual effect.
|
||||
send_event(ui_port, InputEventCode.UP)
|
||||
time.sleep(0.3)
|
||||
frame_capture("after-up")
|
||||
|
||||
send_event(ui_port, InputEventCode.DOWN)
|
||||
time.sleep(0.3)
|
||||
frame_capture("after-down")
|
||||
|
||||
# Soft check: we should still be in a reachable frame (not wedged).
|
||||
# The next test's `ui_home_state` will error out if the device is
|
||||
# unresponsive, so we don't need a stricter guarantee here.
|
||||
final = get_current_frame(lines)
|
||||
assert final is not None, "no frame log after UP/DOWN — event path broke"
|
||||
@@ -0,0 +1,93 @@
|
||||
"""INPUT_BROKER_RIGHT cycles forward through frames; INPUT_BROKER_LEFT backs.
|
||||
|
||||
The simplest UI test: fire N RIGHT events and assert the frame index
|
||||
moves forward by N (modulo frameCount). Each step captures an image +
|
||||
OCR for the HTML report.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from meshtastic_mcp.input_events import InputEventCode
|
||||
|
||||
from ._screen_log import get_current_frame, wait_for_frame
|
||||
from .conftest import FrameCapture, send_event
|
||||
|
||||
|
||||
@pytest.mark.timeout(120)
|
||||
def test_input_right_cycles_frames(
|
||||
ui_port: str,
|
||||
frame_capture: FrameCapture,
|
||||
request: pytest.FixtureRequest,
|
||||
) -> None:
|
||||
lines: list[str] = request.node._debug_log_buffer
|
||||
start = get_current_frame(lines)
|
||||
assert start is not None, "no frame log yet — USERPREFS_UI_TEST_LOG not wired?"
|
||||
# FN_F1 in ui_home_state lands on frame 0. The name at frame 0 varies
|
||||
# by board (home on heltec-v3, deviceFocused on others) — accept either.
|
||||
assert start.name in (
|
||||
"home",
|
||||
"deviceFocused",
|
||||
), f"setup expected home/deviceFocused at frame 0, got {start.name!r}"
|
||||
|
||||
frame_capture("initial")
|
||||
visited = [start.idx]
|
||||
|
||||
for step in range(4):
|
||||
send_event(ui_port, InputEventCode.RIGHT)
|
||||
# Each RIGHT should bump the frame index by 1. The log fires with
|
||||
# `reason=next` from showFrame(NEXT).
|
||||
before_count = len(list(_frame_events(lines)))
|
||||
deadline = time.monotonic() + 5.0
|
||||
while time.monotonic() < deadline:
|
||||
if len(list(_frame_events(lines))) > before_count:
|
||||
break
|
||||
time.sleep(0.1)
|
||||
evt = get_current_frame(lines)
|
||||
assert evt is not None
|
||||
assert (
|
||||
evt.reason == "next"
|
||||
), f"step {step}: expected reason=next, got {evt.reason!r}"
|
||||
visited.append(evt.idx)
|
||||
frame_capture(f"after-right-{step + 1}")
|
||||
|
||||
# Sanity: each index should differ from its predecessor.
|
||||
diffs = [visited[i + 1] - visited[i] for i in range(len(visited) - 1)]
|
||||
assert all(
|
||||
d in (1, -(start.count - 1)) for d in diffs
|
||||
), f"expected monotonic +1 steps (or a wrap), got visited={visited} diffs={diffs}"
|
||||
|
||||
|
||||
@pytest.mark.timeout(120)
|
||||
def test_input_left_returns_to_home(
|
||||
ui_port: str,
|
||||
frame_capture: FrameCapture,
|
||||
request: pytest.FixtureRequest,
|
||||
) -> None:
|
||||
"""After RIGHT×3 + LEFT×3, we should end up back on the starting frame."""
|
||||
lines: list[str] = request.node._debug_log_buffer
|
||||
start = get_current_frame(lines)
|
||||
assert start is not None
|
||||
start_name = start.name
|
||||
frame_capture("initial")
|
||||
for _ in range(3):
|
||||
send_event(ui_port, InputEventCode.RIGHT)
|
||||
time.sleep(0.3)
|
||||
frame_capture("after-right-3")
|
||||
|
||||
for _ in range(3):
|
||||
send_event(ui_port, InputEventCode.LEFT)
|
||||
time.sleep(0.3)
|
||||
|
||||
# Back to whichever frame we started on (home or deviceFocused).
|
||||
wait_for_frame(lines, start_name, timeout_s=5.0)
|
||||
frame_capture(f"after-left-3-back-{start_name}")
|
||||
|
||||
|
||||
def _frame_events(lines: list[str]) -> Any:
|
||||
from ._screen_log import iter_frame_events
|
||||
|
||||
return iter_frame_events(lines)
|
||||
@@ -0,0 +1,51 @@
|
||||
"""On the nodelist_nodes frame, UP/DOWN scrolls the list via
|
||||
`NodeListRenderer::scrollUp/scrollDown` (src/graphics/Screen.cpp:1779-1788).
|
||||
The firmware returns 0 before notifying observers, so no frame-change
|
||||
log fires. Verify the path doesn't crash and we stay on nodelist_nodes.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
|
||||
import pytest
|
||||
from meshtastic_mcp.input_events import InputEventCode
|
||||
|
||||
from ._screen_log import assert_no_frame_change, get_current_frame, wait_for_frame
|
||||
from .conftest import FrameCapture, send_event
|
||||
|
||||
|
||||
@pytest.mark.timeout(180)
|
||||
def test_up_down_on_nodelist_no_frame_change(
|
||||
ui_port: str,
|
||||
frame_capture: FrameCapture,
|
||||
request: pytest.FixtureRequest,
|
||||
) -> None:
|
||||
lines: list[str] = request.node._debug_log_buffer
|
||||
frame_capture("initial")
|
||||
|
||||
# Walk RIGHT until we land on nodelist_nodes.
|
||||
for _i in range(15):
|
||||
send_event(ui_port, InputEventCode.RIGHT)
|
||||
time.sleep(0.3)
|
||||
current = get_current_frame(lines)
|
||||
if current is not None and current.name == "nodelist_nodes":
|
||||
break
|
||||
else:
|
||||
pytest.skip("couldn't reach nodelist_nodes within 15 RIGHTs")
|
||||
|
||||
wait_for_frame(lines, "nodelist_nodes", timeout_s=5.0)
|
||||
frame_capture("on-nodelist")
|
||||
|
||||
# UP/DOWN on nodelist scroll internally + `return 0` before
|
||||
# notifyObservers — no frame-change log. Verify.
|
||||
send_event(ui_port, InputEventCode.UP)
|
||||
assert_no_frame_change(lines, wait_s=1.5)
|
||||
send_event(ui_port, InputEventCode.DOWN)
|
||||
assert_no_frame_change(lines, wait_s=1.5)
|
||||
|
||||
final = get_current_frame(lines)
|
||||
assert (
|
||||
final is not None and final.name == "nodelist_nodes"
|
||||
), f"UP/DOWN moved us off nodelist_nodes; now on {final!r}"
|
||||
frame_capture("after-up-down")
|
||||
@@ -0,0 +1,90 @@
|
||||
"""Pin `InputEventCode` values to the firmware `input_broker_event` enum.
|
||||
|
||||
If this test fails, someone changed the firmware enum (or this Python
|
||||
mirror) and they must stay in sync — the admin RPC sends these as u8
|
||||
wire values directly.
|
||||
|
||||
Also exercises `coerce_event_code` for the happy + error paths.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from meshtastic_mcp.input_events import InputEventCode, coerce_event_code
|
||||
|
||||
|
||||
class TestInputEventCodeValues:
|
||||
"""These values MUST match src/input/InputBroker.h exactly."""
|
||||
|
||||
def test_navigation_keys(self) -> None:
|
||||
assert int(InputEventCode.UP) == 17
|
||||
assert int(InputEventCode.DOWN) == 18
|
||||
assert int(InputEventCode.LEFT) == 19
|
||||
assert int(InputEventCode.RIGHT) == 20
|
||||
|
||||
def test_action_keys(self) -> None:
|
||||
assert int(InputEventCode.SELECT) == 10
|
||||
assert int(InputEventCode.CANCEL) == 24
|
||||
assert int(InputEventCode.BACK) == 27
|
||||
|
||||
def test_long_press_variants(self) -> None:
|
||||
assert int(InputEventCode.SELECT_LONG) == 11
|
||||
assert int(InputEventCode.UP_LONG) == 12
|
||||
assert int(InputEventCode.DOWN_LONG) == 13
|
||||
|
||||
def test_fn_keys(self) -> None:
|
||||
assert int(InputEventCode.FN_F1) == 0xF1
|
||||
assert int(InputEventCode.FN_F2) == 0xF2
|
||||
assert int(InputEventCode.FN_F3) == 0xF3
|
||||
assert int(InputEventCode.FN_F4) == 0xF4
|
||||
assert int(InputEventCode.FN_F5) == 0xF5
|
||||
|
||||
def test_system_events(self) -> None:
|
||||
assert int(InputEventCode.SHUTDOWN) == 0x9B
|
||||
assert int(InputEventCode.GPS_TOGGLE) == 0x9E
|
||||
assert int(InputEventCode.SEND_PING) == 0xAF
|
||||
|
||||
def test_auto_increment_block(self) -> None:
|
||||
# C enum: `BACK = 27, USER_PRESS, ALT_PRESS, ALT_LONG` → 28, 29, 30.
|
||||
assert int(InputEventCode.USER_PRESS) == 28
|
||||
assert int(InputEventCode.ALT_PRESS) == 29
|
||||
assert int(InputEventCode.ALT_LONG) == 30
|
||||
|
||||
|
||||
class TestCoerceEventCode:
|
||||
def test_int_passthrough(self) -> None:
|
||||
assert coerce_event_code(20) == 20
|
||||
assert coerce_event_code(0) == 0
|
||||
assert coerce_event_code(255) == 255
|
||||
|
||||
def test_enum_passthrough(self) -> None:
|
||||
assert coerce_event_code(InputEventCode.RIGHT) == 20
|
||||
assert coerce_event_code(InputEventCode.FN_F1) == 0xF1
|
||||
|
||||
def test_name_case_insensitive(self) -> None:
|
||||
assert coerce_event_code("right") == 20
|
||||
assert coerce_event_code("RIGHT") == 20
|
||||
assert coerce_event_code("Right") == 20
|
||||
|
||||
def test_input_broker_prefix_stripped(self) -> None:
|
||||
assert coerce_event_code("INPUT_BROKER_FN_F1") == 0xF1
|
||||
assert coerce_event_code("input_broker_select") == 10
|
||||
|
||||
def test_hyphen_and_underscore_equivalence(self) -> None:
|
||||
assert coerce_event_code("fn-f1") == 0xF1
|
||||
|
||||
def test_int_out_of_range_raises(self) -> None:
|
||||
with pytest.raises(ValueError, match="u8"):
|
||||
coerce_event_code(256)
|
||||
with pytest.raises(ValueError, match="u8"):
|
||||
coerce_event_code(-1)
|
||||
|
||||
def test_unknown_name_raises(self) -> None:
|
||||
with pytest.raises(ValueError, match="unknown event code name"):
|
||||
coerce_event_code("NOT_A_KEY")
|
||||
|
||||
def test_wrong_type_raises(self) -> None:
|
||||
with pytest.raises(TypeError):
|
||||
coerce_event_code(1.5) # type: ignore[arg-type]
|
||||
with pytest.raises(TypeError):
|
||||
coerce_event_code(None) # type: ignore[arg-type]
|
||||
@@ -0,0 +1,148 @@
|
||||
"""Pin the `uhubctl` default-output parser against canned real-world samples.
|
||||
|
||||
uhubctl's output format has been stable since v2.x but occasionally adds
|
||||
new hub-descriptor fields (e.g. the `, ppps` marker). The parser uses loose
|
||||
regexes to tolerate additions; this test keeps us honest.
|
||||
|
||||
Samples captured from:
|
||||
- v2.6.0 on macOS (Homebrew) — two USB2 hubs, one populated with an
|
||||
nRF52 and a CP2102, plus chained USB3 hubs.
|
||||
- v2.5.0 on Linux (hypothetical — reconstructed from the project README).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from meshtastic_mcp.uhubctl import (
|
||||
ROLE_VIDS,
|
||||
UhubctlError,
|
||||
parse_list_output,
|
||||
)
|
||||
|
||||
# Actual `uhubctl` stdout on the developer's macOS bench, Apr 2026.
|
||||
_SAMPLE_MACOS_V26 = """\
|
||||
Current status for hub 1-1.3 [2109:2817 VIA Labs, Inc. USB2.0 Hub, USB 2.10, 4 ports, ppps]
|
||||
Port 1: 0100 power
|
||||
Port 2: 0103 power enable connect [239a:8029 RAKwireless WisCore RAK4631 Board 920456B1E6972262]
|
||||
Port 3: 0103 power enable connect [10c4:ea60 Silicon Labs CP2102 USB to UART Bridge Controller 0001]
|
||||
Port 4: 0100 power
|
||||
Current status for hub 1-2.3 [2109:0817 VIA Labs, Inc. USB3.0 Hub, USB 3.10, 4 ports, ppps]
|
||||
Port 1: 02a0 power 5gbps Rx.Detect
|
||||
Port 2: 02a0 power 5gbps Rx.Detect
|
||||
Port 3: 02a0 power 5gbps Rx.Detect
|
||||
Port 4: 02a0 power 5gbps Rx.Detect
|
||||
Current status for hub 1-1 [2109:2817 VIA Labs, Inc. USB2.0 Hub, USB 2.10, 4 ports, ppps]
|
||||
Port 1: 0100 power
|
||||
Port 2: 0100 power
|
||||
Port 3: 0503 power highspeed enable connect [2109:2817 VIA Labs, Inc. USB2.0 Hub, USB 2.10, 4 ports, ppps]
|
||||
Port 4: 0100 power
|
||||
"""
|
||||
|
||||
|
||||
# Minimal Linux-style sample (fewer hubs, shows a non-PPPS hub).
|
||||
_SAMPLE_LINUX_NONPPPS = """\
|
||||
Current status for hub 2-1.4 [05e3:0608 GenesysLogic USB2.1 Hub, USB 2.10, 4 ports]
|
||||
Port 1: 0507 power highspeed suspend enable connect [239a:0029 Adafruit Feather Bootloader]
|
||||
Port 2: 0100 power
|
||||
Port 3: 0100 power
|
||||
Port 4: 0100 power
|
||||
"""
|
||||
|
||||
|
||||
class TestParseListOutput:
|
||||
def test_parses_macos_sample_hub_count(self) -> None:
|
||||
hubs = parse_list_output(_SAMPLE_MACOS_V26)
|
||||
assert len(hubs) == 3
|
||||
|
||||
def test_parses_hub_location_and_vid(self) -> None:
|
||||
hubs = parse_list_output(_SAMPLE_MACOS_V26)
|
||||
via_hub = hubs[0]
|
||||
assert via_hub["location"] == "1-1.3"
|
||||
assert via_hub["vid"] == 0x2109
|
||||
assert via_hub["pid"] == 0x2817
|
||||
assert via_hub["ppps"] is True
|
||||
|
||||
def test_parses_port_with_device(self) -> None:
|
||||
hubs = parse_list_output(_SAMPLE_MACOS_V26)
|
||||
nrf52_hub = hubs[0]
|
||||
port2 = next(p for p in nrf52_hub["ports"] if p["port"] == 2)
|
||||
assert port2["device_vid"] == 0x239A
|
||||
assert port2["device_pid"] == 0x8029
|
||||
assert "RAKwireless" in port2["device_desc"]
|
||||
|
||||
def test_empty_port_has_no_device(self) -> None:
|
||||
hubs = parse_list_output(_SAMPLE_MACOS_V26)
|
||||
nrf52_hub = hubs[0]
|
||||
port1 = next(p for p in nrf52_hub["ports"] if p["port"] == 1)
|
||||
assert port1["device_vid"] is None
|
||||
assert port1["device_pid"] is None
|
||||
assert port1["device_desc"] is None
|
||||
|
||||
def test_ports_count(self) -> None:
|
||||
hubs = parse_list_output(_SAMPLE_MACOS_V26)
|
||||
for hub in hubs:
|
||||
assert len(hub["ports"]) == 4 # each sample hub has 4 ports
|
||||
|
||||
def test_non_ppps_hub_flagged(self) -> None:
|
||||
hubs = parse_list_output(_SAMPLE_LINUX_NONPPPS)
|
||||
assert len(hubs) == 1
|
||||
assert hubs[0]["ppps"] is False
|
||||
|
||||
def test_handles_empty_input(self) -> None:
|
||||
assert parse_list_output("") == []
|
||||
|
||||
def test_handles_malformed_lines_gracefully(self) -> None:
|
||||
# Lines that don't match HUB_RE or PORT_RE are ignored silently.
|
||||
garbage = "uhubctl: warning: something weird\n" + _SAMPLE_LINUX_NONPPPS
|
||||
hubs = parse_list_output(garbage)
|
||||
assert len(hubs) == 1
|
||||
|
||||
|
||||
class TestRoleVids:
|
||||
def test_nrf52_mapped(self) -> None:
|
||||
assert 0x239A in ROLE_VIDS["nrf52"]
|
||||
|
||||
def test_esp32s3_covers_both_vids(self) -> None:
|
||||
# Espressif native USB + CP2102 USB-UART on heltec-v3 boards.
|
||||
assert 0x303A in ROLE_VIDS["esp32s3"]
|
||||
assert 0x10C4 in ROLE_VIDS["esp32s3"]
|
||||
|
||||
|
||||
class TestResolveTargetErrorPaths:
|
||||
def test_unknown_role_raises(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
from meshtastic_mcp.uhubctl import resolve_target
|
||||
|
||||
# Clear any env-var pinning that might make this pass accidentally.
|
||||
for key in (
|
||||
"MESHTASTIC_UHUBCTL_LOCATION_FLUX",
|
||||
"MESHTASTIC_UHUBCTL_PORT_FLUX",
|
||||
):
|
||||
monkeypatch.delenv(key, raising=False)
|
||||
with pytest.raises(UhubctlError, match="unknown role"):
|
||||
resolve_target("flux")
|
||||
|
||||
def test_invalid_env_port_raises(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
from meshtastic_mcp.uhubctl import resolve_target
|
||||
|
||||
monkeypatch.setenv("MESHTASTIC_UHUBCTL_LOCATION_NRF52", "1-1.3")
|
||||
monkeypatch.setenv("MESHTASTIC_UHUBCTL_PORT_NRF52", "not-an-int")
|
||||
with pytest.raises(UhubctlError, match="not a valid integer"):
|
||||
resolve_target("nrf52")
|
||||
|
||||
def test_env_var_pinning_wins(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
from meshtastic_mcp.uhubctl import resolve_target
|
||||
|
||||
# Env-var pinning should NOT require uhubctl to be running / installed.
|
||||
monkeypatch.setenv("MESHTASTIC_UHUBCTL_LOCATION_NRF52", "9-9.9")
|
||||
monkeypatch.setenv("MESHTASTIC_UHUBCTL_PORT_NRF52", "7")
|
||||
assert resolve_target("nrf52") == ("9-9.9", 7)
|
||||
|
||||
def test_normalize_role_strips_alt_suffix(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
from meshtastic_mcp.uhubctl import resolve_target
|
||||
|
||||
# esp32s3_alt collapses to esp32s3 for env-var lookup.
|
||||
monkeypatch.setenv("MESHTASTIC_UHUBCTL_LOCATION_ESP32S3", "2-2")
|
||||
monkeypatch.setenv("MESHTASTIC_UHUBCTL_PORT_ESP32S3", "3")
|
||||
assert resolve_target("esp32s3_alt") == ("2-2", 3)
|
||||
@@ -0,0 +1,80 @@
|
||||
"""Pin the `Screen: frame N/M name=X reason=Y` regex + FrameEvent dataclass.
|
||||
|
||||
The firmware-side format lives in `src/graphics/Screen.cpp::logFrameChange`;
|
||||
if the format string changes, this test — and the parser in
|
||||
`tests/ui/_screen_log.py` — have to be updated together.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from tests.ui._screen_log import FRAME_RE, FrameEvent, iter_frame_events
|
||||
|
||||
|
||||
class TestFrameEventParse:
|
||||
def test_exact_firmware_output(self) -> None:
|
||||
raw = "Screen: frame 2/8 name=home reason=next"
|
||||
evt = FrameEvent.parse(raw)
|
||||
assert evt is not None
|
||||
assert evt.idx == 2
|
||||
assert evt.count == 8
|
||||
assert evt.name == "home"
|
||||
assert evt.reason == "next"
|
||||
assert evt.raw == raw
|
||||
|
||||
def test_with_log_prefix(self) -> None:
|
||||
"""Log lines may be preamble-wrapped by the firmware LOG_INFO macro
|
||||
(timestamp, severity, etc.) — the regex uses .search() not .match()
|
||||
so prefixes are fine."""
|
||||
raw = "[INFO] 00:12:34 567 Screen: frame 4/12 name=nodelist_nodes reason=fn_f3 "
|
||||
evt = FrameEvent.parse(raw)
|
||||
assert evt is not None
|
||||
assert evt.idx == 4
|
||||
assert evt.count == 12
|
||||
assert evt.name == "nodelist_nodes"
|
||||
assert evt.reason == "fn_f3"
|
||||
|
||||
def test_rebuild_reason(self) -> None:
|
||||
evt = FrameEvent.parse("Screen: frame 0/5 name=deviceFocused reason=rebuild")
|
||||
assert evt is not None
|
||||
assert evt.reason == "rebuild"
|
||||
|
||||
def test_all_fn_reasons(self) -> None:
|
||||
for k in range(1, 6):
|
||||
evt = FrameEvent.parse(
|
||||
f"Screen: frame {k - 1}/8 name=settings reason=fn_f{k}"
|
||||
)
|
||||
assert evt is not None and evt.reason == f"fn_f{k}"
|
||||
|
||||
def test_unknown_name_is_preserved(self) -> None:
|
||||
"""If the reverse-map returns 'unknown', that still parses cleanly."""
|
||||
evt = FrameEvent.parse("Screen: frame 99/100 name=unknown reason=prev")
|
||||
assert evt is not None and evt.name == "unknown"
|
||||
|
||||
def test_non_matching_line_returns_none(self) -> None:
|
||||
assert FrameEvent.parse("BOOT Booting firmware 2.7.23") is None
|
||||
assert FrameEvent.parse("") is None
|
||||
assert FrameEvent.parse("Screen: without the right format") is None
|
||||
|
||||
|
||||
class TestIterFrameEvents:
|
||||
def test_filters_non_matching_lines(self) -> None:
|
||||
lines = [
|
||||
"Booting...",
|
||||
"Screen: frame 1/5 name=home reason=rebuild",
|
||||
"Some other log line",
|
||||
"Screen: frame 2/5 name=textMessage reason=next",
|
||||
]
|
||||
evts = list(iter_frame_events(lines))
|
||||
assert len(evts) == 2
|
||||
assert evts[0].reason == "rebuild"
|
||||
assert evts[1].reason == "next"
|
||||
|
||||
|
||||
class TestRegexAnchoring:
|
||||
def test_regex_is_compiled(self) -> None:
|
||||
assert FRAME_RE.search("Screen: frame 0/0 name=home reason=next") is not None
|
||||
|
||||
def test_regex_allows_unusual_names(self) -> None:
|
||||
r"""Name is `\S+`, so compound names with underscores/digits match."""
|
||||
m = FRAME_RE.search("Screen: frame 5/10 name=nodelist_hopsignal reason=fn_f2")
|
||||
assert m is not None and m["name"] == "nodelist_hopsignal"
|
||||
Reference in New Issue
Block a user