From 5ca1d0ae433c153494e8bddcb76b8553b3aff4a8 Mon Sep 17 00:00:00 2001 From: Dmitry Ilyin <6576495+widgetii@users.noreply.github.com> Date: Fri, 2 Oct 2026 18:07:45 +0300 Subject: [PATCH 1/4] power: drive Tasmota plugs and plain HTTP relays MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two new DEFIB_POWER_TYPE backends, so lab benches built from cheap smart plugs no longer need someone at a second shell to flip the power: - tasmota: the /cm?cmnd= HTTP API (DEFIB_TASMOTA_HOST, optional _RELAY, _USER, _PASSWORD). On/off replies are checked against the requested state, and an auth WARNING reply is an error rather than a silent no-op. power_cycle runs on the plug as "Backlog Power OFF; Delay N; Power ON", so the off window does not depend on network latency and a dropped connection mid-cycle still leaves the camera powered; it then polls until the relay reports ON again. - http: any relay with an "on" GET URL and an "off" GET URL (DEFIB_HTTP_POWER_ON_URL / _OFF_URL, optional _TIMEOUT) — e.g. a local bridge in front of a cloud-only Sonoff switch. agent upload accepted --power-cycle but ignored it on the HiSiliconStandard path; only CV6xx used the controller. It now powers off, starts the handshake while the camera is still dark, then powers on. The order matters for relays whose power_on returns seconds after the relay actually closed: the blaster is already on the wire when the bootrom window opens, and nothing a running OS prints can be mistaken for bootrom markers. A handshake that does not complete within 15 s gets one fresh cycle. Verified on hardware: - hi3516ev300 behind a cloud-backed Sonoff Micro USB relay, through a local HTTP bridge: `DEFIB_POWER_TYPE=http ... defib agent upload -c hi3516ev300 --power-cycle` caught the bootrom on the first cycle; DDR step, SPL and agent uploaded, agent READY, `agent info` answered. - GK7205V500 behind a Tasmota 14.3 mains plug: the Backlog cycle reboots the board (vendor U-Boot -> Linux boot log captured read-only), and `defib burn -c gk7205v500 --power-cycle` completed the full upload once the V500 handshake fix in the next commit was in place. --- CLAUDE.md | 3 +- README.md | 45 ++++++- src/defib/cli/app.py | 117 ++++++++++++++++-- src/defib/power/factory.py | 12 +- src/defib/power/http.py | 86 ++++++++++++++ src/defib/power/tasmota.py | 178 +++++++++++++++++++++++++++ tests/test_agent_upload_power.py | 104 ++++++++++++++++ tests/test_power_http.py | 101 ++++++++++++++++ tests/test_power_tasmota.py | 198 +++++++++++++++++++++++++++++++ 9 files changed, 833 insertions(+), 11 deletions(-) create mode 100644 src/defib/power/http.py create mode 100644 src/defib/power/tasmota.py create mode 100644 tests/test_agent_upload_power.py create mode 100644 tests/test_power_http.py create mode 100644 tests/test_power_tasmota.py diff --git a/CLAUDE.md b/CLAUDE.md index fd816f1..7e08c90 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -152,7 +152,8 @@ Backends: `spi_flash.c` (fmc100), `spi_flash_hisfc350.c` (V1-era parts), flash path. Keep installer-transient environment here; persistent board policy belongs to firmware/device profiles. - **Power** (`src/defib/power/`) — `routeros` (MikroTik PoE, default), `vectis`, - `rack`, chosen by `DEFIB_POWER_TYPE`. + `rack`, `tasmota` (smart plug `/cm?cmnd=` API), `http` (relay driven by two GET + URLs), chosen by `DEFIB_POWER_TYPE`. - **TUI** (`src/defib/tui/`) — Textual UI, including the Flash Doctor screen. - **Web** (`web/`) — WebSerial browser UI: standalone HTML/JS, no build step, deployed to GitHub Pages. `src/defib/web/` is an empty placeholder package — diff --git a/README.md b/README.md index 82f4ad2..c1176ab 100644 --- a/README.md +++ b/README.md @@ -206,8 +206,8 @@ Handles both `tftpboot` and `tftp` U-Boot commands transparently. ## Automated Power Cycling Defib can automatically power-cycle devices, eliminating manual intervention -for recovery loops and research workflows. Two backends are supported, -selected via `DEFIB_POWER_TYPE`: +for recovery loops and research workflows. The backend is selected via +`DEFIB_POWER_TYPE`: ### MikroTik RouterOS PoE switch (default) @@ -267,6 +267,47 @@ socat -,raw,echo=0 TCP:172.17.32.17:35240 > window can close before the round-trip completes; running Vectis > on the same host as defib (or close to it on a LAN) is recommended. +### Tasmota smart plug + +Any plug running [Tasmota](https://tasmota.github.io/) (or a firmware that +speaks the same `/cm?cmnd=` HTTP API) in front of the camera's power supply. +Power cycles run on the plug as a `Backlog`, so the off window is timed by +the plug rather than by the network: + +```bash +export DEFIB_POWER_TYPE=tasmota +export DEFIB_TASMOTA_HOST=192.168.1.50 +export DEFIB_TASMOTA_RELAY=1 # optional, multi-relay devices only +export DEFIB_TASMOTA_USER=admin # optional, if a web password is set +export DEFIB_TASMOTA_PASSWORD=secret + +defib burn -c gk7205v500 -f u-boot-gk7205v500-nand.bin -p /dev/ttyUSB0 --power-cycle -t +``` + +On a mains plug, leave the default 3 s off time alone: a camera brick can +hold enough charge to ride through a shorter gap. + +### Generic HTTP relay + +Any relay that turns on and off with a plain GET request — for example a +small local bridge in front of a cloud-only smart switch: + +```bash +export DEFIB_POWER_TYPE=http +export DEFIB_HTTP_POWER_ON_URL=http://127.0.0.1:8090/relay/on +export DEFIB_HTTP_POWER_OFF_URL=http://127.0.0.1:8090/relay/off +export DEFIB_HTTP_POWER_TIMEOUT=10 # optional, seconds per request + +defib agent upload -c hi3516ev300 -p /dev/ttyUSB0 --power-cycle +``` + +The cycle is timed on the host, so it inherits the relay's latency. +`agent upload --power-cycle` copes with that by starting the handshake while +the camera is still off and re-cycling once if the bootrom does not answer. + +Both are single-outlet controllers: `--poe-port` does not apply, and like +Vectis they do not work with `defib restore`. + The `-t` flag auto-detects the post-boot mode: - **Normal U-Boot shell** (e.g. hi3516ev300): a two-way serial terminal — your keystrokes go to the board and its output comes back. Ctrl-C exits the diff --git a/src/defib/cli/app.py b/src/defib/cli/app.py index 5fb6300..63d8726 100644 --- a/src/defib/cli/app.py +++ b/src/defib/cli/app.py @@ -2,12 +2,18 @@ from __future__ import annotations -from typing import Any +from collections.abc import Callable, Coroutine +from typing import TYPE_CHECKING, Any import typer from defib.install import layout as _install_layout +if TYPE_CHECKING: + from defib.power.base import PowerController + from defib.recovery.events import HandshakeResult + from defib.transport.base import Transport + # Compatibility aliases for existing private imports. Install implementation # lives in defib.install.layout; CLI code does not own these helpers. _NAND_LAYOUT = _install_layout.NAND_LAYOUT @@ -29,7 +35,7 @@ def burn( port: str = typer.Option("/dev/ttyUSB0", "-p", "--port", help="Serial device (/dev/ttyUSB0), tcp://host:port, rfc2217://host:port, or socket:///path"), send_break: bool = typer.Option(False, "-b", "--break", help="Send Ctrl-C after upload"), terminal: bool = typer.Option(False, "-t", "--terminal", help="Open serial terminal after upload"), - power_cycle: bool = typer.Option(False, "--power-cycle", help="Auto power-cycle via PoE (needs DEFIB_POE_* env vars)"), + power_cycle: bool = typer.Option(False, "--power-cycle", help="Auto power-cycle via the controller selected by DEFIB_POWER_TYPE (default routeros, needs DEFIB_POE_* env vars)"), poe_port_override: str = typer.Option("", "--poe-port", help="Explicit MikroTik ether port (e.g. ether3) — overrides comment-based auto-discovery. Requires --power-cycle."), output: str = typer.Option("human", "--output", help="Output mode: human, json, quiet"), debug: bool = typer.Option(False, "-d", "--debug", help="Enable debug logging"), @@ -1095,11 +1101,12 @@ async def _agent_upload_async( FlashAgentClient, agent_binary_help, get_agent_binary, ) from defib.firmware import get_cached_path + from defib.power.base import PowerControllerError from defib.profiles.loader import load_profile from defib.protocol.hisilicon_cv6xx import HiSiliconCV6xx from defib.protocol.hisilicon_standard import HiSiliconStandard from defib.protocol.registry import find_protocol - from defib.recovery.events import ProgressEvent + from defib.recovery.events import ProgressEvent, Stage from defib.transport.serial_platform import ( create_transport, normalize_port_name, ) @@ -1164,10 +1171,23 @@ async def _agent_upload_async( spl_data = cached_fw.read_bytes() spl_source = f"full U-Boot ({len(spl_data)} bytes — boundary auto-detected)" + power = None + if power_cycle: + from defib.power.factory import power_controller_from_env + try: + power = power_controller_from_env() + except Exception as e: + if output == "json": + print(json_mod.dumps({"event": "error", "message": str(e)})) + else: + console.print(f"[red]Power controller error:[/red] {e}") + raise typer.Exit(1) + if output == "human": console.print(f"Agent: [cyan]{agent_path.name}[/cyan] ({len(agent_data)} bytes)") console.print(f"SPL: {spl_source}") - console.print("\n[yellow]Power-cycle the camera now![/yellow]\n") + if power is None: + console.print("\n[yellow]Power-cycle the camera now![/yellow]\n") transport = await create_transport(normalize_port_name(port)) protocol = HiSiliconStandard() @@ -1180,7 +1200,32 @@ def on_progress(e: ProgressEvent) -> None: elif output == "json": print(json_mod.dumps({"event": "progress", "message": e.message}), flush=True) - hs = await protocol.handshake(transport, on_progress) + if power is not None: + # The bootrom's catch window is short, so flood 0xAA from the start. + protocol.set_continuous_ack(True) + + def on_power_log(message: str) -> None: + on_progress(ProgressEvent( + stage=Stage.POWER_CYCLE, bytes_sent=0, bytes_total=1, message=message, + )) + + try: + hs = await _power_cycle_into_handshake( + power, transport, + lambda: protocol.handshake(transport, on_progress), + on_power_log, + ) + except PowerControllerError as e: + if output == "json": + print(json_mod.dumps({"event": "error", "message": f"Power cycle failed: {e}"})) + else: + console.print(f"[red]Power cycle failed:[/red] {e}") + await transport.close() + raise typer.Exit(1) + finally: + await power.close() + else: + hs = await protocol.handshake(transport, on_progress) if not hs.success: if output == "json": print(json_mod.dumps({"event": "error", "message": "Handshake failed"})) @@ -1235,6 +1280,64 @@ def on_progress(e: ProgressEvent) -> None: await transport.close() +async def _power_cycle_into_handshake( + power: PowerController, + transport: Transport, + start_handshake: Callable[[], Coroutine[Any, Any, HandshakeResult]], + log: Callable[[str], None], + off_duration: float = 3.0, + handshake_timeout: float = 15.0, + attempts: int = 2, +) -> HandshakeResult: + """Power the device off, start the handshake, then power it back on. + + The handshake starts while the device is still off: the running OS + is dead by then, so nothing on the line can be mistaken for bootrom + markers, and the blaster is already on the wire when the bootrom's + catch window opens. This holds however long the controller takes + to answer ``power_on`` — cloud-backed relays can lag by seconds. + + A handshake that does not complete within ``handshake_timeout`` is + retried with a fresh power cycle, up to ``attempts`` times in all. + Power-controller failures propagate as ``PowerControllerError``. + """ + import asyncio + + from defib.recovery.events import HandshakeResult + + hs = HandshakeResult(success=False, message="no attempt made") + for attempt in range(1, attempts + 1): + suffix = f" (attempt {attempt}/{attempts})" if attempt > 1 else "" + log(f"Powering off via {power.name()}{suffix}...") + await power.power_off("") + await asyncio.sleep(off_duration) + await transport.flush_input() + task = asyncio.create_task(start_handshake()) + # Yield so the first frames reach the wire before power returns. + await asyncio.sleep(0.05) + try: + log("Powering on...") + await power.power_on("") + except BaseException: + task.cancel() + try: + await task + except BaseException: + pass + raise + try: + hs = await asyncio.wait_for(task, timeout=handshake_timeout) + except asyncio.TimeoutError: + hs = HandshakeResult( + success=False, + message=f"no bootrom response within {handshake_timeout:.0f}s of power-on", + ) + if hs.success: + return hs + log(f"Handshake failed: {hs.message}") + return hs + + async def _agent_upload_cv6xx( *, chip: str, @@ -2159,7 +2262,7 @@ def install( help="Explicit U-Boot artifact override", ), port: str = typer.Option("/dev/ttyUSB0", "-p", "--port", help="Serial device (/dev/ttyUSB0), tcp://host:port, rfc2217://host:port, or socket:///path"), - power_cycle: bool = typer.Option(False, "--power-cycle", help="Auto power-cycle via PoE"), + power_cycle: bool = typer.Option(False, "--power-cycle", help="Auto power-cycle via the controller selected by DEFIB_POWER_TYPE"), poe_port_override: str = typer.Option("", "--poe-port", help="Explicit MikroTik ether port (e.g. ether3) — overrides comment-based auto-discovery. Requires --power-cycle."), nic: str = typer.Option("", "--nic", help="Network interface for TFTP (auto-detect if empty)"), host_ip: str = typer.Option("192.168.1.10", "--host-ip", help="IP to assign to host NIC for TFTP"), @@ -2266,7 +2369,7 @@ def restore( host_ip: str = typer.Option("", "--host-ip", help="Host IP for TFTP (auto-detect if empty)"), device_ip: str = typer.Option("", "--device-ip", help="Device IP in U-Boot"), nic: str = typer.Option("", "--nic", help="Network interface for TFTP"), - power_cycle: bool = typer.Option(False, "--power-cycle", help="Auto power-cycle via PoE"), + power_cycle: bool = typer.Option(False, "--power-cycle", help="Auto power-cycle via the controller selected by DEFIB_POWER_TYPE"), poe_port_override: str = typer.Option("", "--poe-port", help="Explicit MikroTik ether port (e.g. ether3) — overrides comment-based auto-discovery. Requires --power-cycle."), tftp_via: str = typer.Option( "auto", "--tftp-via", diff --git a/src/defib/power/factory.py b/src/defib/power/factory.py index 5b1b4b8..12308b1 100644 --- a/src/defib/power/factory.py +++ b/src/defib/power/factory.py @@ -16,6 +16,10 @@ def power_controller_from_env() -> PowerController: configured via ``DEFIB_VECTIS_*``. - ``DEFIB_POWER_TYPE=rack``: rack pod HTTP API (``~/git/rack`` ESP32-S3 spinoff), configured via ``DEFIB_RACK_*``. + - ``DEFIB_POWER_TYPE=tasmota``: Tasmota smart plug ``/cm?cmnd=`` API, + configured via ``DEFIB_TASMOTA_*``. + - ``DEFIB_POWER_TYPE=http``: generic relay driven by two GET URLs, + configured via ``DEFIB_HTTP_POWER_*``. Raises: PowerControllerError: if the type is unknown or required env @@ -31,7 +35,13 @@ def power_controller_from_env() -> PowerController: if kind == "rack": from defib.power.rack import RackController return RackController.from_env() + if kind == "tasmota": + from defib.power.tasmota import TasmotaController + return TasmotaController.from_env() + if kind == "http": + from defib.power.http import HttpRelayController + return HttpRelayController.from_env() raise PowerControllerError( f"Unknown DEFIB_POWER_TYPE: {kind!r} " - "(expected 'routeros', 'vectis', or 'rack')" + "(expected 'routeros', 'vectis', 'rack', 'tasmota', or 'http')" ) diff --git a/src/defib/power/http.py b/src/defib/power/http.py new file mode 100644 index 0000000..4621433 --- /dev/null +++ b/src/defib/power/http.py @@ -0,0 +1,86 @@ +"""Generic HTTP relay power controller. + +For any relay that exposes "turn on" and "turn off" as plain GET URLs — +small bridge services in front of cloud-only smart switches (e.g. a local +eWeLink/Sonoff bridge at ``http://127.0.0.1:8090/relay/on``), ESPHome +web-server buttons, homegrown relay boards. + +Single-port, like :class:`~defib.power.rack.RackController`: the ``port`` +argument is ignored. Power cycling is timed on the host, so it inherits +whatever latency the relay's backend adds. +""" + +from __future__ import annotations + +import asyncio +import logging +import os +import urllib.error +import urllib.request + +from defib.power.base import PowerController, PowerControllerError + +logger = logging.getLogger(__name__) + + +class HttpRelayController(PowerController): + """Drives power by GETting one URL for on and another for off.""" + + def __init__(self, on_url: str, off_url: str, timeout: float = 10.0) -> None: + self._on_url = on_url + self._off_url = off_url + self._timeout = timeout + + @classmethod + def name(cls) -> str: + return "HTTP relay" + + @classmethod + def from_env(cls) -> HttpRelayController: + """Create from ``DEFIB_HTTP_POWER_*`` environment variables. + + Required: + DEFIB_HTTP_POWER_ON_URL: GET this to switch power on. + DEFIB_HTTP_POWER_OFF_URL: GET this to switch power off. + Optional: + DEFIB_HTTP_POWER_TIMEOUT: per-request timeout in seconds + (default 10). + """ + on_url = os.environ.get("DEFIB_HTTP_POWER_ON_URL") + off_url = os.environ.get("DEFIB_HTTP_POWER_OFF_URL") + if not on_url or not off_url: + raise PowerControllerError( + "DEFIB_HTTP_POWER_ON_URL and DEFIB_HTTP_POWER_OFF_URL env vars " + "required for HTTP relay power control" + ) + timeout_env = os.environ.get("DEFIB_HTTP_POWER_TIMEOUT", "10") + try: + timeout = float(timeout_env) + except ValueError: + raise PowerControllerError( + f"DEFIB_HTTP_POWER_TIMEOUT must be a number, got {timeout_env!r}" + ) from None + return cls(on_url=on_url, off_url=off_url, timeout=timeout) + + async def power_off(self, port: str) -> None: + await asyncio.to_thread(self._get_sync, self._off_url) + + async def power_on(self, port: str) -> None: + await asyncio.to_thread(self._get_sync, self._on_url) + + async def close(self) -> None: + # Stateless HTTP — nothing to release. + return None + + def _get_sync(self, url: str) -> None: + logger.info("http relay GET %s", url) + try: + with urllib.request.urlopen(url, timeout=self._timeout) as resp: + resp.read() + except urllib.error.HTTPError as e: + detail = e.read().decode("utf-8", "replace")[:300] + raise PowerControllerError( + f"HTTP relay {e.code} on GET {url}: {detail}" + ) from e + except (urllib.error.URLError, TimeoutError, OSError) as e: + raise PowerControllerError(f"HTTP relay unreachable at {url}: {e}") from e diff --git a/src/defib/power/tasmota.py b/src/defib/power/tasmota.py new file mode 100644 index 0000000..d1fdc83 --- /dev/null +++ b/src/defib/power/tasmota.py @@ -0,0 +1,178 @@ +"""Tasmota smart-plug power controller. + +Drives a single relay on a Tasmota (or API-compatible) smart plug via its +``/cm?cmnd=`` HTTP API. Typical use is a mains plug feeding a camera's +power brick, or an inline USB relay. + +Like :class:`~defib.power.rack.RackController`, each plug owns exactly +one camera, so the ``port`` argument is ignored. Pass ``""`` from the CLI. + +Power cycling runs on the plug itself as a ``Backlog`` (``Power OFF; +Delay N; Power ON``), so the off window does not depend on host or +network latency, and a dropped HTTP connection mid-cycle still leaves the +camera powered. +""" + +from __future__ import annotations + +import asyncio +import json +import logging +import os +import time +import urllib.error +import urllib.parse +import urllib.request + +from defib.power.base import PowerController, PowerControllerError + +logger = logging.getLogger(__name__) + +# Tasmota's Backlog ``Delay`` is in 0.1 s units and accepts 2..3600. +_DELAY_MIN = 2 +_DELAY_MAX = 3600 + + +class TasmotaController(PowerController): + """Drives one relay of a Tasmota plug over HTTP.""" + + def __init__( + self, + host: str, + relay: int | None = None, + user: str | None = None, + password: str | None = None, + timeout: float = 10.0, + ) -> None: + self._host = host + self._relay = relay + self._user = user + self._password = password + self._timeout = timeout + + @classmethod + def name(cls) -> str: + return "Tasmota plug" + + @classmethod + def from_env(cls) -> TasmotaController: + """Create from ``DEFIB_TASMOTA_*`` environment variables. + + Required: + DEFIB_TASMOTA_HOST: plug IP or hostname (e.g. ``10.216.128.80``). + Optional: + DEFIB_TASMOTA_RELAY: relay index on multi-relay devices + (``Power``); omit for single-relay plugs. + DEFIB_TASMOTA_USER / DEFIB_TASMOTA_PASSWORD: web credentials, + when the plug has a web password set. + """ + host = os.environ.get("DEFIB_TASMOTA_HOST") + if not host: + raise PowerControllerError( + "DEFIB_TASMOTA_HOST env var required for Tasmota power control" + ) + relay_env = os.environ.get("DEFIB_TASMOTA_RELAY") + try: + relay = int(relay_env) if relay_env else None + except ValueError: + raise PowerControllerError( + f"DEFIB_TASMOTA_RELAY must be an integer, got {relay_env!r}" + ) from None + return cls( + host=host, + relay=relay, + user=os.environ.get("DEFIB_TASMOTA_USER") or None, + password=os.environ.get("DEFIB_TASMOTA_PASSWORD") or None, + ) + + @property + def _power_cmd(self) -> str: + return f"Power{self._relay}" if self._relay is not None else "Power" + + async def power_off(self, port: str) -> None: + await self._set("OFF") + + async def power_on(self, port: str) -> None: + await self._set("ON") + + async def power_cycle(self, port: str, off_duration: float = 3.0) -> None: + delay = max(_DELAY_MIN, min(_DELAY_MAX, round(off_duration * 10))) + cmd = self._power_cmd + await self._cmnd(f"Backlog {cmd} OFF; Delay {delay}; {cmd} ON") + # The backlog runs asynchronously on the plug. Wait for the relay + # to come back so callers can rely on "power is on" afterwards. + deadline = time.monotonic() + delay / 10 + self._timeout + await asyncio.sleep(delay / 10) + while True: + if await self._state() == "ON": + return + if time.monotonic() > deadline: + raise PowerControllerError( + f"Tasmota {self._host}: relay did not come back ON after cycle" + ) + await asyncio.sleep(0.2) + + async def close(self) -> None: + # Stateless HTTP — nothing to release. + return None + + async def _set(self, want: str) -> None: + reply = await self._cmnd(f"{self._power_cmd} {want}") + got = self._extract_state(reply) + if got != want: + raise PowerControllerError( + f"Tasmota {self._host}: asked for {want}, plug reports {got!r}" + ) + + async def _state(self) -> str | None: + return self._extract_state(await self._cmnd(self._power_cmd)) + + def _extract_state(self, reply: dict[str, object]) -> str | None: + # Single-relay plugs answer {"POWER": "ON"}; multi-relay ones + # answer {"POWER2": "ON"}. "Power1" may come back as "POWER". + keys = [self._power_cmd.upper()] + if self._relay in (None, 1): + keys += ["POWER", "POWER1"] + for key in keys: + value = reply.get(key) + if isinstance(value, str): + return value.upper() + return None + + async def _cmnd(self, command: str) -> dict[str, object]: + params = {"cmnd": command} + if self._user is not None: + params["user"] = self._user + if self._password is not None: + params["password"] = self._password + url = f"http://{self._host}/cm?{urllib.parse.urlencode(params)}" + logger.info("tasmota GET %s cmnd=%r", self._host, command) + return await asyncio.to_thread(self._get_sync, url, self._timeout) + + def _get_sync(self, url: str, timeout: float) -> dict[str, object]: + try: + with urllib.request.urlopen(url, timeout=timeout) as resp: + payload = resp.read() + except urllib.error.HTTPError as e: + raise PowerControllerError( + f"Tasmota {self._host}: HTTP {e.code}" + ) from e + except (urllib.error.URLError, TimeoutError, OSError) as e: + raise PowerControllerError( + f"Tasmota unreachable at {self._host}: {e}" + ) from e + try: + result = json.loads(payload) + except json.JSONDecodeError as e: + raise PowerControllerError( + f"Tasmota {self._host}: non-JSON reply {payload[:100]!r}" + ) from e + if not isinstance(result, dict): + raise PowerControllerError( + f"Tasmota {self._host}: unexpected reply {result!r}" + ) + if "WARNING" in result: + # Tasmota answers {"WARNING": "Need user=&password=..."} + # with HTTP 200 when the web password is wrong or missing. + raise PowerControllerError(f"Tasmota {self._host}: {result['WARNING']}") + return result diff --git a/tests/test_agent_upload_power.py b/tests/test_agent_upload_power.py new file mode 100644 index 0000000..3a3483b --- /dev/null +++ b/tests/test_agent_upload_power.py @@ -0,0 +1,104 @@ +"""Tests for agent upload's power-cycle-into-handshake sequencing.""" + +from __future__ import annotations + +import asyncio + +import pytest + +from defib.cli.app import _power_cycle_into_handshake +from defib.power.base import PowerController, PowerControllerError +from defib.recovery.events import HandshakeResult +from defib.transport.mock import MockTransport + + +class RecordingPower(PowerController): + def __init__(self, events: list[str], fail_on: str | None = None) -> None: + self.events = events + self.fail_on = fail_on + + @classmethod + def name(cls) -> str: + return "recording" + + async def power_off(self, port: str) -> None: + self.events.append("off") + if self.fail_on == "off": + raise PowerControllerError("relay offline") + + async def power_on(self, port: str) -> None: + self.events.append("on") + if self.fail_on == "on": + raise PowerControllerError("relay offline") + + async def close(self) -> None: + return None + + +async def test_handshake_starts_before_power_on() -> None: + events: list[str] = [] + + async def handshake() -> HandshakeResult: + events.append("handshake") + await asyncio.sleep(0) + return HandshakeResult(success=True) + + hs = await _power_cycle_into_handshake( + RecordingPower(events), MockTransport(), handshake, lambda _m: None, + off_duration=0.0, + ) + assert hs.success + assert events == ["off", "handshake", "on"] + + +async def test_retries_with_fresh_cycle_on_timeout() -> None: + events: list[str] = [] + results = iter([None, HandshakeResult(success=True)]) + + async def handshake() -> HandshakeResult: + events.append("handshake") + result = next(results) + if result is None: + await asyncio.sleep(10) # never answers; times out + assert result is not None + return result + + logs: list[str] = [] + hs = await _power_cycle_into_handshake( + RecordingPower(events), MockTransport(), handshake, logs.append, + off_duration=0.0, handshake_timeout=0.05, + ) + assert hs.success + assert events == ["off", "handshake", "on", "off", "handshake", "on"] + assert any("attempt 2/2" in m for m in logs) + + +async def test_gives_up_after_attempts() -> None: + async def handshake() -> HandshakeResult: + return HandshakeResult(success=False, message="bad marker") + + hs = await _power_cycle_into_handshake( + RecordingPower([]), MockTransport(), handshake, lambda _m: None, + off_duration=0.0, attempts=2, + ) + assert not hs.success + assert hs.message == "bad marker" + + +async def test_power_on_failure_cancels_handshake() -> None: + cancelled = asyncio.Event() + + async def handshake() -> HandshakeResult: + try: + await asyncio.sleep(10) + except asyncio.CancelledError: + cancelled.set() + raise + return HandshakeResult(success=True) + + with pytest.raises(PowerControllerError): + await _power_cycle_into_handshake( + RecordingPower([], fail_on="on"), MockTransport(), handshake, + lambda _m: None, off_duration=0.0, + ) + assert cancelled.is_set() diff --git a/tests/test_power_http.py b/tests/test_power_http.py new file mode 100644 index 0000000..786ff71 --- /dev/null +++ b/tests/test_power_http.py @@ -0,0 +1,101 @@ +"""Tests for the generic HTTP relay power controller.""" + +from __future__ import annotations + +import io +import urllib.error +from typing import Any + +import pytest + +from defib.power import http as http_mod +from defib.power.base import PowerControllerError +from defib.power.factory import power_controller_from_env +from defib.power.http import HttpRelayController + +ON = "http://127.0.0.1:8090/relay/on" +OFF = "http://127.0.0.1:8090/relay/off" + + +class FakeResponse(io.BytesIO): + def __enter__(self) -> "FakeResponse": + return self + + def __exit__(self, *exc_info: object) -> None: + self.close() + + +@pytest.fixture +def calls(monkeypatch: pytest.MonkeyPatch) -> list[str]: + seen: list[str] = [] + + def fake(url: str, timeout: float | None = None) -> FakeResponse: + seen.append(url) + return FakeResponse(b'{"state":"on"}') + + monkeypatch.setattr(http_mod.urllib.request, "urlopen", fake) + return seen + + +class TestFromEnv: + @pytest.mark.parametrize("missing", ["DEFIB_HTTP_POWER_ON_URL", "DEFIB_HTTP_POWER_OFF_URL"]) + def test_missing_url(self, monkeypatch: pytest.MonkeyPatch, missing: str) -> None: + monkeypatch.setenv("DEFIB_HTTP_POWER_ON_URL", ON) + monkeypatch.setenv("DEFIB_HTTP_POWER_OFF_URL", OFF) + monkeypatch.delenv(missing) + with pytest.raises(PowerControllerError, match="DEFIB_HTTP_POWER_ON_URL"): + HttpRelayController.from_env() + + def test_timeout(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("DEFIB_HTTP_POWER_ON_URL", ON) + monkeypatch.setenv("DEFIB_HTTP_POWER_OFF_URL", OFF) + monkeypatch.setenv("DEFIB_HTTP_POWER_TIMEOUT", "2.5") + assert HttpRelayController.from_env()._timeout == 2.5 + monkeypatch.setenv("DEFIB_HTTP_POWER_TIMEOUT", "soon") + with pytest.raises(PowerControllerError, match="TIMEOUT"): + HttpRelayController.from_env() + + def test_factory_dispatches_http(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("DEFIB_POWER_TYPE", "http") + monkeypatch.setenv("DEFIB_HTTP_POWER_ON_URL", ON) + monkeypatch.setenv("DEFIB_HTTP_POWER_OFF_URL", OFF) + monkeypatch.delenv("DEFIB_HTTP_POWER_TIMEOUT", raising=False) + assert isinstance(power_controller_from_env(), HttpRelayController) + + +class TestPowerOps: + async def test_on_off(self, calls: list[str]) -> None: + ctrl = HttpRelayController(on_url=ON, off_url=OFF) + await ctrl.power_off("ignored") + await ctrl.power_on("ignored") + assert calls == [OFF, ON] + + async def test_cycle_is_host_timed( + self, calls: list[str], monkeypatch: pytest.MonkeyPatch, + ) -> None: + sleeps: list[float] = [] + + async def fake_sleep(seconds: float) -> None: + sleeps.append(seconds) + + monkeypatch.setattr("defib.power.base.asyncio.sleep", fake_sleep) + ctrl = HttpRelayController(on_url=ON, off_url=OFF) + await ctrl.power_cycle("", off_duration=2.0) + assert calls == [OFF, ON] + assert sleeps == [2.0] + + async def test_http_error(self, monkeypatch: pytest.MonkeyPatch) -> None: + def fail(url: str, timeout: float | None = None) -> Any: # noqa: ANN401 + raise urllib.error.HTTPError(url, 502, "Bad Gateway", {}, io.BytesIO(b"cloud down")) # type: ignore[arg-type] + + monkeypatch.setattr(http_mod.urllib.request, "urlopen", fail) + with pytest.raises(PowerControllerError, match="502.*cloud down"): + await HttpRelayController(on_url=ON, off_url=OFF).power_on("") + + async def test_unreachable(self, monkeypatch: pytest.MonkeyPatch) -> None: + def fail(url: str, timeout: float | None = None) -> Any: # noqa: ANN401 + raise urllib.error.URLError("connection refused") + + monkeypatch.setattr(http_mod.urllib.request, "urlopen", fail) + with pytest.raises(PowerControllerError, match="unreachable"): + await HttpRelayController(on_url=ON, off_url=OFF).power_off("") diff --git a/tests/test_power_tasmota.py b/tests/test_power_tasmota.py new file mode 100644 index 0000000..a830916 --- /dev/null +++ b/tests/test_power_tasmota.py @@ -0,0 +1,198 @@ +"""Tests for the Tasmota smart-plug power controller.""" + +from __future__ import annotations + +import io +import json +import urllib.error +import urllib.parse +from typing import Any + +import pytest + +from defib.power import tasmota as tasmota_mod +from defib.power.base import PowerControllerError +from defib.power.factory import power_controller_from_env +from defib.power.tasmota import TasmotaController + + +class FakeResponse(io.BytesIO): + def __enter__(self) -> "FakeResponse": + return self + + def __exit__(self, *exc_info: object) -> None: + self.close() + + +class FakePlug: + """Answers /cm?cmnd= like a single-relay Tasmota plug.""" + + def __init__(self, relay_key: str = "POWER") -> None: + self.cmnds: list[str] = [] + self.queries: list[dict[str, list[str]]] = [] + self.state = "ON" + self.relay_key = relay_key + self.reply_override: bytes | None = None + + def __call__(self, url: str, timeout: float | None = None) -> FakeResponse: + parsed = urllib.parse.urlparse(url) + assert parsed.path == "/cm" + query = urllib.parse.parse_qs(parsed.query) + self.queries.append(query) + cmnd = query["cmnd"][0] + self.cmnds.append(cmnd) + if self.reply_override is not None: + return FakeResponse(self.reply_override) + parts = cmnd.split() + if parts[0] == "Backlog": + self.state = "ON" + return FakeResponse(b'{"POWER":"OFF"}') + if len(parts) == 2: + self.state = parts[1] + return FakeResponse(json.dumps({self.relay_key: self.state}).encode()) + + +@pytest.fixture +def plug(monkeypatch: pytest.MonkeyPatch) -> FakePlug: + fake = FakePlug() + monkeypatch.setattr(tasmota_mod.urllib.request, "urlopen", fake) + return fake + + +@pytest.fixture +def no_sleep(monkeypatch: pytest.MonkeyPatch) -> list[float]: + sleeps: list[float] = [] + + async def fake_sleep(seconds: float) -> None: + sleeps.append(seconds) + + monkeypatch.setattr(tasmota_mod.asyncio, "sleep", fake_sleep) + return sleeps + + +class TestFromEnv: + def test_missing_host(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("DEFIB_TASMOTA_HOST", raising=False) + with pytest.raises(PowerControllerError, match="DEFIB_TASMOTA_HOST"): + TasmotaController.from_env() + + def test_defaults(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("DEFIB_TASMOTA_HOST", "10.216.128.80") + for var in ("DEFIB_TASMOTA_RELAY", "DEFIB_TASMOTA_USER", "DEFIB_TASMOTA_PASSWORD"): + monkeypatch.delenv(var, raising=False) + ctrl = TasmotaController.from_env() + assert ctrl._host == "10.216.128.80" + assert ctrl._relay is None + assert ctrl._user is None + + def test_bad_relay(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("DEFIB_TASMOTA_HOST", "x") + monkeypatch.setenv("DEFIB_TASMOTA_RELAY", "two") + with pytest.raises(PowerControllerError, match="DEFIB_TASMOTA_RELAY"): + TasmotaController.from_env() + + def test_factory_dispatches_tasmota(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("DEFIB_POWER_TYPE", "tasmota") + monkeypatch.setenv("DEFIB_TASMOTA_HOST", "10.0.0.5") + assert isinstance(power_controller_from_env(), TasmotaController) + + +class TestPowerOps: + async def test_power_on_off(self, plug: FakePlug) -> None: + ctrl = TasmotaController(host="10.0.0.5") + await ctrl.power_off("ignored") + await ctrl.power_on("ignored") + assert plug.cmnds == ["Power OFF", "Power ON"] + + async def test_relay_index(self, monkeypatch: pytest.MonkeyPatch) -> None: + fake = FakePlug(relay_key="POWER2") + monkeypatch.setattr(tasmota_mod.urllib.request, "urlopen", fake) + ctrl = TasmotaController(host="x", relay=2) + await ctrl.power_off("") + assert fake.cmnds == ["Power2 OFF"] + + async def test_credentials_in_query(self, plug: FakePlug) -> None: + ctrl = TasmotaController(host="x", user="admin", password="s3cret") + await ctrl.power_on("") + assert plug.queries[0]["user"] == ["admin"] + assert plug.queries[0]["password"] == ["s3cret"] + + async def test_state_mismatch_raises(self, plug: FakePlug) -> None: + plug.reply_override = b'{"POWER":"ON"}' + ctrl = TasmotaController(host="x") + with pytest.raises(PowerControllerError, match="asked for OFF"): + await ctrl.power_off("") + + async def test_auth_warning_raises(self, plug: FakePlug) -> None: + plug.reply_override = b'{"WARNING":"Need user=&password="}' + ctrl = TasmotaController(host="x") + with pytest.raises(PowerControllerError, match="Need user"): + await ctrl.power_on("") + + async def test_non_json_raises(self, plug: FakePlug) -> None: + plug.reply_override = b"" + ctrl = TasmotaController(host="x") + with pytest.raises(PowerControllerError, match="non-JSON"): + await ctrl.power_on("") + + +class TestPowerCycle: + async def test_cycle_runs_on_plug(self, plug: FakePlug, no_sleep: list[float]) -> None: + ctrl = TasmotaController(host="x") + await ctrl.power_cycle("", off_duration=3.0) + assert plug.cmnds == ["Backlog Power OFF; Delay 30; Power ON", "Power"] + assert no_sleep == [3.0] + + @pytest.mark.parametrize(("off", "delay"), [(0.0, 2), (0.24, 2), (1.26, 13), (9999.0, 3600)]) + async def test_delay_clamped_and_rounded( + self, plug: FakePlug, no_sleep: list[float], off: float, delay: int, + ) -> None: + ctrl = TasmotaController(host="x") + await ctrl.power_cycle("", off_duration=off) + assert plug.cmnds[0] == f"Backlog Power OFF; Delay {delay}; Power ON" + + async def test_cycle_polls_until_on( + self, monkeypatch: pytest.MonkeyPatch, no_sleep: list[float], + ) -> None: + replies = iter([b'{"POWER":"OFF"}', b'{"POWER":"OFF"}', b'{"POWER":"OFF"}', b'{"POWER":"ON"}']) + calls: list[str] = [] + + def fake(url: str, timeout: float | None = None) -> FakeResponse: + calls.append(url) + return FakeResponse(next(replies)) + + monkeypatch.setattr(tasmota_mod.urllib.request, "urlopen", fake) + ctrl = TasmotaController(host="x") + await ctrl.power_cycle("", off_duration=3.0) + assert len(calls) == 4 + + async def test_cycle_gives_up( + self, monkeypatch: pytest.MonkeyPatch, no_sleep: list[float], + ) -> None: + def fake(url: str, timeout: float | None = None) -> FakeResponse: + return FakeResponse(b'{"POWER":"OFF"}') + + clock = iter(range(0, 10_000, 5)) + monkeypatch.setattr(tasmota_mod.urllib.request, "urlopen", fake) + monkeypatch.setattr(tasmota_mod.time, "monotonic", lambda: float(next(clock))) + ctrl = TasmotaController(host="x", timeout=10.0) + with pytest.raises(PowerControllerError, match="did not come back ON"): + await ctrl.power_cycle("", off_duration=3.0) + + +class TestErrorMapping: + async def test_url_error(self, monkeypatch: pytest.MonkeyPatch) -> None: + def fail(url: str, timeout: float | None = None) -> Any: # noqa: ANN401 + raise urllib.error.URLError("unreachable") + + monkeypatch.setattr(tasmota_mod.urllib.request, "urlopen", fail) + with pytest.raises(PowerControllerError, match="unreachable"): + await TasmotaController(host="x").power_on("") + + async def test_http_error(self, monkeypatch: pytest.MonkeyPatch) -> None: + def fail(url: str, timeout: float | None = None) -> Any: # noqa: ANN401 + raise urllib.error.HTTPError(url, 401, "Unauthorized", {}, io.BytesIO()) # type: ignore[arg-type] + + monkeypatch.setattr(tasmota_mod.urllib.request, "urlopen", fail) + with pytest.raises(PowerControllerError, match="HTTP 401"): + await TasmotaController(host="x").power_on("") From ae7aecc17006f7d0369662929dfe315d6ca9e6b6 Mon Sep 17 00:00:00 2001 From: Dmitry Ilyin <6576495+widgetii@users.noreply.github.com> Date: Fri, 2 Oct 2026 18:07:45 +0300 Subject: [PATCH 2/4] protocol/v500: keep the handshake on the wire continuously MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The V500 handshake wrote one 14-byte frame (~1.2 ms at 115200) and then blocked up to 100 ms waiting for a 14-byte reply that had to start exactly at the read boundary. The line sat idle ~99% of the time, and the bootrom only listens for a few tens of ms after reset before it falls through to flash boot. On a board with healthy NAND that was a lottery, lost most of the time — it is why catching a GK7205V510 on a known-good flash "worked once, then failed three times in a row". Now each iteration writes an 8-frame burst (~10 ms), drains whatever has arrived without blocking, and searches the accumulated stream for the BD 00 reply wherever it lands — after boot noise or split across reads. Once it is found, the replies to the rest of the burst are allowed to settle and are flushed, so they cannot be read as ACKs in the HEAD stage. Verified on a GK7205V500 with GD5F1GM7 NAND and working vendor firmware, power-cycled by a Tasmota plug: `defib burn -c gk7205v500 -f u-boot-gk7205v500-nand.bin --power-cycle` caught the bootrom on the first cycle and uploaded HEAD, AUX and the 292 KB boot image (34 s). Before the change the same command never caught it in a 120 s run. --- src/defib/protocol/hisilicon_v500.py | 54 ++++++++++++++++++++-------- tests/test_protocol_v500.py | 48 +++++++++++++++++++++++++ 2 files changed, 87 insertions(+), 15 deletions(-) diff --git a/src/defib/protocol/hisilicon_v500.py b/src/defib/protocol/hisilicon_v500.py index 48d3944..c05745c 100644 --- a/src/defib/protocol/hisilicon_v500.py +++ b/src/defib/protocol/hisilicon_v500.py @@ -35,6 +35,8 @@ ]) HANDSHAKE_TIMEOUT = 20.0 # seconds +HANDSHAKE_BURST_FRAMES = 8 # 112 B, ~10 ms at 115200 baud +HANDSHAKE_REPLY_LEN = 14 CHUNK_ACK_TIMEOUT = 4.0 # seconds MAX_NAK_RETRIES = 10 BOOT_LOAD_ADDR = 0x41000000 @@ -74,25 +76,47 @@ async def handshake( handshake_frame = append_crc( V500_HANDSHAKE_MAGIC + b"\x00\x00\x00\x00\x00\x00\x00\x00" ) + # The bootrom only listens for a few tens of ms after reset before + # it falls through to flash boot, so the line must never go idle: + # one 14-byte frame is ~1.2 ms on the wire, and waiting 100 ms for + # a reply after each one left the window uncovered ~99% of the time. + burst = handshake_frame * HANDSHAKE_BURST_FRAMES + buffer = bytearray() while True: - await transport.write(handshake_frame) + await transport.write(burst) try: - response = await transport.read(14, timeout=0.1) - if response.startswith(b"\xbd\x00") and len(response) >= 12: - chip_id = struct.unpack(">I", response[8:12])[0] - self._chip_id = chip_id - _emit(on_progress, ProgressEvent( - stage=Stage.HANDSHAKE, bytes_sent=1, bytes_total=1, - message=f"Detected SoC: {hex(chip_id)}", - )) - return HandshakeResult( - success=True, - chip_id=chip_id, - message=f"Detected SoC: {hex(chip_id)}", - ) + waiting = await transport.bytes_waiting() + if waiting > 0: + buffer += await transport.read(waiting, timeout=0.01) except TransportTimeout: - continue + pass + + # The reply can land anywhere in the stream — after boot noise + # from a still-running OS, or mid-way through a burst. + idx = buffer.find(b"\xbd\x00") + if idx != -1 and len(buffer) - idx >= HANDSHAKE_REPLY_LEN: + chip_id = struct.unpack(">I", buffer[idx + 8:idx + 12])[0] + self._chip_id = chip_id + # The rest of the burst is still in flight and the bootrom + # answers each frame; let that settle and drop it so the + # replies are not mistaken for ACKs during the HEAD stage. + await asyncio.sleep(0.1) + await transport.flush_input() + _emit(on_progress, ProgressEvent( + stage=Stage.HANDSHAKE, bytes_sent=1, bytes_total=1, + message=f"Detected SoC: {hex(chip_id)}", + )) + return HandshakeResult( + success=True, + chip_id=chip_id, + message=f"Detected SoC: {hex(chip_id)}", + ) + if idx == -1: + # Keep a trailing 0xBD: it may be the start of a reply. + del buffer[:max(0, len(buffer) - 1)] + else: + del buffer[:idx] async def _send_frame_wait_ack( self, diff --git a/tests/test_protocol_v500.py b/tests/test_protocol_v500.py index 35eef4b..ef168db 100644 --- a/tests/test_protocol_v500.py +++ b/tests/test_protocol_v500.py @@ -52,6 +52,54 @@ async def test_handshake_reports_chip_id(self): assert result.chip_id == 0xAABBCCDD +class _ScriptedRx(MockTransport): + """Delivers one scripted RX chunk after each write, like a live UART.""" + + def __init__(self, *chunks: bytes) -> None: + super().__init__(flush_clears_buffer=True) + self._script = list(chunks) + + async def write(self, data: bytes) -> None: + await super().write(data) + if self._script: + self.enqueue_rx(self._script.pop(0)) + + +class TestV500HandshakeCatch: + REPLY = b"\xbd\x00" + b"\x00" * 6 + struct.pack(">I", 0x72050500) + b"\x00\x00" + + @pytest.mark.asyncio + async def test_reply_after_boot_noise(self): + transport = _ScriptedRx(b"", b"Starting kernel ...\r\n" + self.REPLY) + result = await HiSiliconV500().handshake(transport) + assert result.success + assert result.chip_id == 0x72050500 + + @pytest.mark.asyncio + async def test_reply_split_across_reads(self): + transport = _ScriptedRx(b"\x00\x01", self.REPLY[:1], self.REPLY[1:9], self.REPLY[9:]) + result = await HiSiliconV500().handshake(transport) + assert result.chip_id == 0x72050500 + + @pytest.mark.asyncio + async def test_line_never_idles_between_polls(self): + """Every write is a multi-frame burst, so the bootrom's short listen + window cannot fall into a gap between frames.""" + transport = _ScriptedRx(b"", b"", self.REPLY) + await HiSiliconV500().handshake(transport) + assert len(transport.tx_log) == 3 + for burst in transport.tx_log: + assert len(burst) >= 8 * 14 + assert burst[:4] == b"\xbd\x00\xff\x01" + + @pytest.mark.asyncio + async def test_stale_replies_flushed(self): + """Replies to the rest of the burst must not linger as fake ACKs.""" + transport = _ScriptedRx(self.REPLY * 3) + await HiSiliconV500().handshake(transport) + assert await transport.bytes_waiting() == 0 + + class TestV500FirmwareTransfer: @pytest.mark.asyncio async def test_send_firmware_with_acks(self): From 2ae4eeaa6015ba1aa9a32586ba11df4d943aabf1 Mon Sep 17 00:00:00 2001 From: Dmitry Ilyin <6576495+widgetii@users.noreply.github.com> Date: Fri, 2 Oct 2026 18:53:20 +0300 Subject: [PATCH 3/4] transport/socket: make bytes_waiting() see unread data MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SocketTransport.read() goes straight to the socket, so _buf only ever held bytes pushed back by unread(), and bytes_waiting() reported 0 while data sat in the kernel. Anything that polls before reading was blind on socket://, tcp:// and rack:// links — Transport.drain_input(), the CV6xx handshake, and the V500 handshake from the previous commit, which would blast forever without ever seeing the bootrom's reply. bytes_waiting() now pulls whatever the non-blocking socket already has into _buf. The V500 handshake also drains its TX queue once the reply arrives, before the settle delay and input flush: bursts still queued on the host would otherwise keep reaching the bootrom, and their replies could be read as ACK/NAK bytes in the HEAD stage. (A timed read every pass instead of the bytes_waiting() gate was tried; it reprograms the serial timeout with a tcsetattr per pass and was dropped.) Both found in review (Qodo). A socketpair-backed test fails on the old bytes_waiting() and passes now. Re-verified on the GK7205V510: three consecutive `defib burn --power-cycle` catches, 3/3. --- src/defib/protocol/hisilicon_v500.py | 16 +++++++++++--- src/defib/transport/socket.py | 12 ++++++++++ tests/test_protocol_v500.py | 33 ++++++++++++++++++++++++++++ tests/test_transport_socket.py | 23 +++++++++++++++++++ 4 files changed, 81 insertions(+), 3 deletions(-) diff --git a/src/defib/protocol/hisilicon_v500.py b/src/defib/protocol/hisilicon_v500.py index c05745c..7e6c1b4 100644 --- a/src/defib/protocol/hisilicon_v500.py +++ b/src/defib/protocol/hisilicon_v500.py @@ -25,7 +25,7 @@ RecoveryResult, Stage, ) -from defib.transport.base import Transport, TransportTimeout +from defib.transport.base import Transport, TransportError, TransportTimeout logger = logging.getLogger(__name__) @@ -85,6 +85,11 @@ async def handshake( buffer = bytearray() while True: await transport.write(burst) + # Read only when something has arrived (bytes_waiting() is + # non-blocking on every transport). An unconditional timed read + # would reprogram the serial port's timeout — a tcsetattr — on + # every pass. The host TX queue paces the writes; whatever is + # still queued when the reply lands is drained below. try: waiting = await transport.bytes_waiting() if waiting > 0: @@ -99,8 +104,13 @@ async def handshake( chip_id = struct.unpack(">I", buffer[idx + 8:idx + 12])[0] self._chip_id = chip_id # The rest of the burst is still in flight and the bootrom - # answers each frame; let that settle and drop it so the - # replies are not mistaken for ACKs during the HEAD stage. + # answers each frame: drain our TX queue, let the replies + # settle and drop them so they are not mistaken for ACKs + # during the HEAD stage. + try: + await transport.flush_output() + except TransportError: + pass # best effort: the settle delay still covers it await asyncio.sleep(0.1) await transport.flush_input() _emit(on_progress, ProgressEvent( diff --git a/src/defib/transport/socket.py b/src/defib/transport/socket.py index 86bc7b3..a669b1a 100644 --- a/src/defib/transport/socket.py +++ b/src/defib/transport/socket.py @@ -116,6 +116,18 @@ async def flush_output(self) -> None: pass # sendall already ensures data is sent async def bytes_waiting(self) -> int: + # read() goes straight to the socket, so _buf alone only ever holds + # bytes pulled in here or pushed back by unread(). Pull whatever the + # kernel already has (the socket is non-blocking) so callers that + # poll before reading see data that has actually arrived. + try: + data = self._sock.recv(65536) + except (BlockingIOError, InterruptedError): + data = None + except OSError: + data = None # let the next read() report the error + if data: + self._buf += data return len(self._buf) async def close(self) -> None: diff --git a/tests/test_protocol_v500.py b/tests/test_protocol_v500.py index ef168db..6930e43 100644 --- a/tests/test_protocol_v500.py +++ b/tests/test_protocol_v500.py @@ -100,6 +100,39 @@ async def test_stale_replies_flushed(self): assert await transport.bytes_waiting() == 0 +class TestV500HandshakeOverSocket: + @pytest.mark.asyncio + async def test_reply_reaches_handshake_over_socket(self): + """tcp:// and socket:// transports only count already-received bytes + in bytes_waiting(), so the handshake must read, not poll.""" + import asyncio + import socket + + from defib.transport.socket import SocketTransport + + ours, bootrom = socket.socketpair() + bootrom.setblocking(False) + reply = b"\xbd\x00" + b"\x00" * 6 + struct.pack(">I", 0x72050510) + b"\x00\x00" + loop = asyncio.get_running_loop() + + async def fake_bootrom() -> None: + await loop.sock_recv(bootrom, 4096) # first burst arrives + await loop.sock_sendall(bootrom, reply) + while True: # keep swallowing the rest of the flood + if not await loop.sock_recv(bootrom, 4096): + return + + peer = asyncio.create_task(fake_bootrom()) + transport = SocketTransport(ours) + try: + result = await asyncio.wait_for(HiSiliconV500().handshake(transport), 5) + finally: + await transport.close() + bootrom.close() + peer.cancel() + assert result.chip_id == 0x72050510 + + class TestV500FirmwareTransfer: @pytest.mark.asyncio async def test_send_firmware_with_acks(self): diff --git a/tests/test_transport_socket.py b/tests/test_transport_socket.py index 51a34b2..cb1709d 100644 --- a/tests/test_transport_socket.py +++ b/tests/test_transport_socket.py @@ -95,3 +95,26 @@ async def test_flush_on_empty_buffer_is_noop(self) -> None: await transport.close() server.close() listener.close() + + +async def test_bytes_waiting_sees_data_not_yet_read() -> None: + """Callers poll bytes_waiting() before read(); it must count bytes that + are sitting in the kernel, not only ones read() already buffered.""" + import socket + + from defib.transport.socket import SocketTransport + + ours, peer = socket.socketpair() + transport = SocketTransport(ours) + try: + assert await transport.bytes_waiting() == 0 + peer.sendall(b"\xbd\x00hello") + for _ in range(100): + if await transport.bytes_waiting(): + break + assert await transport.bytes_waiting() == 7 + assert await transport.read(7, timeout=1.0) == b"\xbd\x00hello" + assert await transport.bytes_waiting() == 0 + finally: + await transport.close() + peer.close() From 1faa8a05fd577fb9b9b6339dc0676c3d860bf942 Mon Sep 17 00:00:00 2001 From: Dmitry Ilyin <6576495+widgetii@users.noreply.github.com> Date: Fri, 2 Oct 2026 18:53:20 +0300 Subject: [PATCH 4/4] power: verify Tasmota cycles; port, pulse and secrets fixes from review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review (Qodo) of the new power backends: - tasmota: power_cycle sent "Backlog Power OFF; Delay N; Power ON" and never checked it happened; Tasmota answers any Backlog with {}. Polling to verify made it worse: on Tasmota 14.3 a command sent during the Delay cuts the delay short (relay back ON at ~2 s instead of ~4.5 s), and the mains-powered GK7205V510 then never booted at all — measured by timing poll replies against the board's first UART byte. The plug now uses the base off / wait / on, each step confirmed by its reply. - agent upload --power-cycle passed "" as the port, which RouterOS cannot resolve. It now takes --poe-port and falls back to comment discovery from the serial device name, like burn. - Vectis can only pulse, and rejects power_off/power_on. Controllers now declare supports_independent_power; for pulse-only ones the handshake starts before the pulse on proactive protocols and after it on the reactive HiSilicon-standard one (whose 0x20 markers a still-running OS could fake), matching RecoverySession. A live RFC 2217 transport is attached so Vectis pulses over the UART connection defib holds. - http: configured URLs were logged and echoed in errors verbatim. Logs and errors now show scheme://host/path with user info and query redacted (also inside echoed response bodies), and user:pass@ in a URL is sent as Basic auth — urllib would otherwise try to resolve a host literally named "user:pass@host". Re-verified on hardware: GK7205V510 behind the Tasmota plug, `defib burn --power-cycle` 3/3; hi3516ev300 behind the HTTP-bridged Sonoff, `agent upload --power-cycle` to READY. --- README.md | 4 +- src/defib/cli/app.py | 70 +++++++++++++++++++++++++++--- src/defib/power/base.py | 5 +++ src/defib/power/http.py | 57 ++++++++++++++++++++++--- src/defib/power/tasmota.py | 34 +++------------ src/defib/power/vectis.py | 2 + tests/test_agent_upload_power.py | 73 ++++++++++++++++++++++++++++---- tests/test_power_http.py | 65 +++++++++++++++++++++++++++- tests/test_power_tasmota.py | 67 ++++++++--------------------- tests/test_power_vectis.py | 9 ++++ 10 files changed, 286 insertions(+), 100 deletions(-) diff --git a/README.md b/README.md index c1176ab..82dcb4d 100644 --- a/README.md +++ b/README.md @@ -271,8 +271,8 @@ socat -,raw,echo=0 TCP:172.17.32.17:35240 Any plug running [Tasmota](https://tasmota.github.io/) (or a firmware that speaks the same `/cm?cmnd=` HTTP API) in front of the camera's power supply. -Power cycles run on the plug as a `Backlog`, so the off window is timed by -the plug rather than by the network: +Each switch is confirmed from the plug's reply, so a cycle that did not +happen is an error rather than a silent no-op: ```bash export DEFIB_POWER_TYPE=tasmota diff --git a/src/defib/cli/app.py b/src/defib/cli/app.py index 63d8726..837779d 100644 --- a/src/defib/cli/app.py +++ b/src/defib/cli/app.py @@ -1082,16 +1082,18 @@ def agent_upload( output: str = typer.Option("human", "--output", help="Output mode: human, json"), file: str | None = typer.Option(None, "-f", "--file", help="CV6xx composite boot file (GSL+DDR+U-Boot); required for CV6xx, ignored for other protocols"), power_cycle: bool = typer.Option(False, "--power-cycle", help="Auto power-cycle via configured controller (DEFIB_POWER_TYPE)"), + poe_port_override: str = typer.Option("", "--poe-port", help="Explicit MikroTik ether port (e.g. ether3) — overrides comment-based auto-discovery. Requires --power-cycle."), ) -> None: """Upload flash agent to device via boot protocol (requires power-cycle).""" import asyncio - asyncio.run(_agent_upload_async(chip, port, output, file, power_cycle)) + asyncio.run(_agent_upload_async(chip, port, output, file, power_cycle, poe_port_override)) async def _agent_upload_async( chip: str, port: str, output: str, composite_path: str | None = None, power_cycle: bool = False, + poe_port_override: str = "", ) -> None: import json as json_mod @@ -1172,11 +1174,15 @@ async def _agent_upload_async( spl_source = f"full U-Boot ({len(spl_data)} bytes — boundary auto-detected)" power = None + power_port = "" if power_cycle: from defib.power.factory import power_controller_from_env try: power = power_controller_from_env() + power_port = await _resolve_power_port(power, port, poe_port_override) except Exception as e: + if power is not None: + await power.close() if output == "json": print(json_mod.dumps({"event": "error", "message": str(e)})) else: @@ -1190,6 +1196,8 @@ async def _agent_upload_async( console.print("\n[yellow]Power-cycle the camera now![/yellow]\n") transport = await create_transport(normalize_port_name(port)) + if power is not None: + _attach_power_transport(power, transport) protocol = HiSiliconStandard() protocol.set_profile(profile) @@ -1211,7 +1219,7 @@ def on_power_log(message: str) -> None: try: hs = await _power_cycle_into_handshake( - power, transport, + power, power_port, transport, lambda: protocol.handshake(transport, on_progress), on_power_log, ) @@ -1280,14 +1288,46 @@ def on_power_log(message: str) -> None: await transport.close() +async def _resolve_power_port( + power: PowerController, serial_port: str, poe_port_override: str, +) -> str: + """The controller port that powers the camera on ``serial_port``. + + Only RouterOS addresses ports by name: ``--poe-port`` wins, otherwise + the interface whose comment names the device (``/dev/uart-`` -> + ````). Single-outlet controllers ignore the port, so ``""``. + """ + from pathlib import Path + + from defib.power.routeros import RouterOSController + + if not isinstance(power, RouterOSController): + return "" + if poe_port_override: + return poe_port_override + label = Path(serial_port).name.removeprefix("uart-") + return await power.find_port_by_comment(label) + + +def _attach_power_transport(power: PowerController, transport: Transport) -> None: + """Let Vectis pulse reset over the UART connection defib already holds.""" + from defib.power.vectis import VectisController + from defib.transport.rfc2217 import Rfc2217Transport + + if isinstance(power, VectisController) and isinstance(transport, Rfc2217Transport): + power.attach_transport(transport) + + async def _power_cycle_into_handshake( power: PowerController, + power_port: str, transport: Transport, start_handshake: Callable[[], Coroutine[Any, Any, HandshakeResult]], log: Callable[[str], None], off_duration: float = 3.0, handshake_timeout: float = 15.0, attempts: int = 2, + proactive: bool = False, ) -> HandshakeResult: """Power the device off, start the handshake, then power it back on. @@ -1297,6 +1337,13 @@ async def _power_cycle_into_handshake( catch window opens. This holds however long the controller takes to answer ``power_on`` — cloud-backed relays can lag by seconds. + Pulse-only controllers (``supports_independent_power`` False) cannot + hold the device off, so the order follows RecoverySession: a + ``proactive`` handshake (V500, CV6xx: the magic must already be on the + wire when the bootrom wakes) starts before the pulse; a reactive one + (HiSilicon standard, which counts 0x20 markers that a running OS could + also print) starts after it. + A handshake that does not complete within ``handshake_timeout`` is retried with a fresh power cycle, up to ``attempts`` times in all. Power-controller failures propagate as ``PowerControllerError``. @@ -1308,16 +1355,25 @@ async def _power_cycle_into_handshake( hs = HandshakeResult(success=False, message="no attempt made") for attempt in range(1, attempts + 1): suffix = f" (attempt {attempt}/{attempts})" if attempt > 1 else "" - log(f"Powering off via {power.name()}{suffix}...") - await power.power_off("") - await asyncio.sleep(off_duration) + if power.supports_independent_power: + log(f"Powering off via {power.name()}{suffix}...") + await power.power_off(power_port) + await asyncio.sleep(off_duration) + pulse_first = not power.supports_independent_power and not proactive + if pulse_first: + log(f"Power-cycling via {power.name()}{suffix}...") + await power.power_cycle(power_port) await transport.flush_input() task = asyncio.create_task(start_handshake()) # Yield so the first frames reach the wire before power returns. await asyncio.sleep(0.05) try: - log("Powering on...") - await power.power_on("") + if power.supports_independent_power: + log("Powering on...") + await power.power_on(power_port) + elif not pulse_first: + log(f"Power-cycling via {power.name()}{suffix}...") + await power.power_cycle(power_port) except BaseException: task.cancel() try: diff --git a/src/defib/power/base.py b/src/defib/power/base.py index fb1638f..3831b63 100644 --- a/src/defib/power/base.py +++ b/src/defib/power/base.py @@ -17,6 +17,11 @@ class PowerController(ABC): smart PDU outlet, relay board). """ + #: Whether ``power_off`` and ``power_on`` work on their own. Controllers + #: that can only pulse (Vectis) set this False and support only + #: ``power_cycle``. + supports_independent_power: bool = True + @classmethod @abstractmethod def name(cls) -> str: diff --git a/src/defib/power/http.py b/src/defib/power/http.py index 4621433..1d6fc6b 100644 --- a/src/defib/power/http.py +++ b/src/defib/power/http.py @@ -13,9 +13,11 @@ from __future__ import annotations import asyncio +import base64 import logging import os import urllib.error +import urllib.parse import urllib.request from defib.power.base import PowerController, PowerControllerError @@ -73,14 +75,59 @@ async def close(self) -> None: return None def _get_sync(self, url: str) -> None: - logger.info("http relay GET %s", url) + shown = _redact(url) + logger.info("http relay GET %s", shown) try: - with urllib.request.urlopen(url, timeout=self._timeout) as resp: + with urllib.request.urlopen(_request(url), timeout=self._timeout) as resp: resp.read() except urllib.error.HTTPError as e: - detail = e.read().decode("utf-8", "replace")[:300] + detail = _redact_text(e.read().decode("utf-8", "replace")[:300], url) raise PowerControllerError( - f"HTTP relay {e.code} on GET {url}: {detail}" + f"HTTP relay {e.code} on GET {shown}: {detail}" ) from e except (urllib.error.URLError, TimeoutError, OSError) as e: - raise PowerControllerError(f"HTTP relay unreachable at {url}: {e}") from e + raise PowerControllerError( + f"HTTP relay unreachable at {shown}: {_redact_text(str(e), url)}" + ) from e + + +def _request(url: str) -> urllib.request.Request: + """A GET for ``url``, sending any ``user:pass@`` as HTTP Basic auth. + + urllib does not do that itself: it would try to connect to a host + literally named ``user:pass@host``. + """ + parts = urllib.parse.urlsplit(url) + if parts.username is None: + return urllib.request.Request(url) + netloc = parts.hostname or "" + if parts.port is not None: + netloc = f"{netloc}:{parts.port}" + bare = urllib.parse.urlunsplit(parts._replace(netloc=netloc)) + creds = f"{urllib.parse.unquote(parts.username)}:{urllib.parse.unquote(parts.password or '')}" + token = base64.b64encode(creds.encode()).decode() + return urllib.request.Request(bare, headers={"Authorization": f"Basic {token}"}) + + +def _redact(url: str) -> str: + """The URL without user info or query, which may carry credentials.""" + parts = urllib.parse.urlsplit(url) + host = parts.hostname or "" + if parts.port is not None: + host = f"{host}:{parts.port}" + query = "?..." if parts.query else "" + return f"{parts.scheme}://{host}{parts.path}{query}" + + +def _redact_text(text: str, url: str) -> str: + """Strip the URL's secrets from a message that may echo them.""" + parts = urllib.parse.urlsplit(url) + secrets = [parts.password, parts.query] + secrets += [value for _, value in urllib.parse.parse_qsl(parts.query)] + # Longest first, so a query value never leaves part of the full query. + for secret in sorted(filter(None, secrets), key=len, reverse=True): + # Skip trivially short values ("1", "on"): blanking them would + # mangle the message without hiding anything worth hiding. + if len(secret) >= 3: + text = text.replace(secret, "...") + return text diff --git a/src/defib/power/tasmota.py b/src/defib/power/tasmota.py index d1fdc83..6a8f41f 100644 --- a/src/defib/power/tasmota.py +++ b/src/defib/power/tasmota.py @@ -7,10 +7,13 @@ Like :class:`~defib.power.rack.RackController`, each plug owns exactly one camera, so the ``port`` argument is ignored. Pass ``""`` from the CLI. -Power cycling runs on the plug itself as a ``Backlog`` (``Power OFF; -Delay N; Power ON``), so the off window does not depend on host or -network latency, and a dropped HTTP connection mid-cycle still leaves the -camera powered. +Power cycling is the base class's off / wait / on, with each switch +confirmed by the plug's reply. A ``Backlog Power OFF; Delay N; Power ON`` +looks tempting (timed on the plug), but its reply is ``{}`` whatever the +commands do, so it cannot be verified without polling — and on Tasmota +14.3 any command sent during the ``Delay`` cuts the delay short: the off +window shrank from ~4.5 s to ~2 s and the camera behind a mains brick +never booted. """ from __future__ import annotations @@ -19,7 +22,6 @@ import json import logging import os -import time import urllib.error import urllib.parse import urllib.request @@ -28,11 +30,6 @@ logger = logging.getLogger(__name__) -# Tasmota's Backlog ``Delay`` is in 0.1 s units and accepts 2..3600. -_DELAY_MIN = 2 -_DELAY_MAX = 3600 - - class TasmotaController(PowerController): """Drives one relay of a Tasmota plug over HTTP.""" @@ -95,23 +92,6 @@ async def power_off(self, port: str) -> None: async def power_on(self, port: str) -> None: await self._set("ON") - async def power_cycle(self, port: str, off_duration: float = 3.0) -> None: - delay = max(_DELAY_MIN, min(_DELAY_MAX, round(off_duration * 10))) - cmd = self._power_cmd - await self._cmnd(f"Backlog {cmd} OFF; Delay {delay}; {cmd} ON") - # The backlog runs asynchronously on the plug. Wait for the relay - # to come back so callers can rely on "power is on" afterwards. - deadline = time.monotonic() + delay / 10 + self._timeout - await asyncio.sleep(delay / 10) - while True: - if await self._state() == "ON": - return - if time.monotonic() > deadline: - raise PowerControllerError( - f"Tasmota {self._host}: relay did not come back ON after cycle" - ) - await asyncio.sleep(0.2) - async def close(self) -> None: # Stateless HTTP — nothing to release. return None diff --git a/src/defib/power/vectis.py b/src/defib/power/vectis.py index a186ee6..c8b63f9 100644 --- a/src/defib/power/vectis.py +++ b/src/defib/power/vectis.py @@ -52,6 +52,8 @@ class VectisController(PowerController): suitable for driving a recovery flow. """ + supports_independent_power = False + def __init__( self, host: str, diff --git a/tests/test_agent_upload_power.py b/tests/test_agent_upload_power.py index 3a3483b..13473f4 100644 --- a/tests/test_agent_upload_power.py +++ b/tests/test_agent_upload_power.py @@ -22,12 +22,12 @@ def name(cls) -> str: return "recording" async def power_off(self, port: str) -> None: - self.events.append("off") + self.events.append(f"off{port}") if self.fail_on == "off": raise PowerControllerError("relay offline") async def power_on(self, port: str) -> None: - self.events.append("on") + self.events.append(f"on{port}") if self.fail_on == "on": raise PowerControllerError("relay offline") @@ -44,11 +44,11 @@ async def handshake() -> HandshakeResult: return HandshakeResult(success=True) hs = await _power_cycle_into_handshake( - RecordingPower(events), MockTransport(), handshake, lambda _m: None, + RecordingPower(events), "ether3", MockTransport(), handshake, lambda _m: None, off_duration=0.0, ) assert hs.success - assert events == ["off", "handshake", "on"] + assert events == ["offether3", "handshake", "onether3"] async def test_retries_with_fresh_cycle_on_timeout() -> None: @@ -65,11 +65,11 @@ async def handshake() -> HandshakeResult: logs: list[str] = [] hs = await _power_cycle_into_handshake( - RecordingPower(events), MockTransport(), handshake, logs.append, + RecordingPower(events), "ether3", MockTransport(), handshake, logs.append, off_duration=0.0, handshake_timeout=0.05, ) assert hs.success - assert events == ["off", "handshake", "on", "off", "handshake", "on"] + assert events == ["offether3", "handshake", "onether3"] * 2 assert any("attempt 2/2" in m for m in logs) @@ -78,7 +78,7 @@ async def handshake() -> HandshakeResult: return HandshakeResult(success=False, message="bad marker") hs = await _power_cycle_into_handshake( - RecordingPower([]), MockTransport(), handshake, lambda _m: None, + RecordingPower([]), "", MockTransport(), handshake, lambda _m: None, off_duration=0.0, attempts=2, ) assert not hs.success @@ -98,7 +98,64 @@ async def handshake() -> HandshakeResult: with pytest.raises(PowerControllerError): await _power_cycle_into_handshake( - RecordingPower([], fail_on="on"), MockTransport(), handshake, + RecordingPower([], fail_on="on"), "", MockTransport(), handshake, lambda _m: None, off_duration=0.0, ) assert cancelled.is_set() + + +class PulseOnlyPower(RecordingPower): + supports_independent_power = False + + async def power_off(self, port: str) -> None: + raise PowerControllerError("pulse only") + + async def power_on(self, port: str) -> None: + raise PowerControllerError("pulse only") + + async def power_cycle(self, port: str, off_duration: float = 3.0) -> None: + self.events.append("pulse") + + +@pytest.mark.parametrize(("proactive", "order"), [ + (False, ["pulse", "handshake"]), # reactive: let the OS die first + (True, ["handshake", "pulse"]), # proactive: magic on the wire first +]) +async def test_pulse_only_controller(proactive: bool, order: list[str]) -> None: + events: list[str] = [] + + async def handshake() -> HandshakeResult: + events.append("handshake") + await asyncio.sleep(0) + return HandshakeResult(success=True) + + hs = await _power_cycle_into_handshake( + PulseOnlyPower(events), "", MockTransport(), handshake, lambda _m: None, + off_duration=0.0, proactive=proactive, + ) + assert hs.success + assert events == order + + +class TestResolvePowerPort: + async def test_single_outlet_controllers_take_no_port(self) -> None: + from defib.cli.app import _resolve_power_port + + assert await _resolve_power_port(RecordingPower([]), "/dev/uart-CAM", "ether9") == "" + + async def test_routeros_override_and_discovery(self, monkeypatch: pytest.MonkeyPatch) -> None: + from defib.cli.app import _resolve_power_port + from defib.power.routeros import RouterOSController + + searched: list[str] = [] + + async def find(self: RouterOSController, search: str) -> str: + searched.append(search) + return "ether4" + + monkeypatch.setattr(RouterOSController, "find_port_by_comment", find) + ctrl = RouterOSController.__new__(RouterOSController) + assert await _resolve_power_port(ctrl, "/dev/uart-IVG85HG50PYA-S", "ether3") == "ether3" + assert searched == [] + assert await _resolve_power_port(ctrl, "/dev/uart-IVG85HG50PYA-S", "") == "ether4" + assert searched == ["IVG85HG50PYA-S"] diff --git a/tests/test_power_http.py b/tests/test_power_http.py index 786ff71..604dd42 100644 --- a/tests/test_power_http.py +++ b/tests/test_power_http.py @@ -29,8 +29,8 @@ def __exit__(self, *exc_info: object) -> None: def calls(monkeypatch: pytest.MonkeyPatch) -> list[str]: seen: list[str] = [] - def fake(url: str, timeout: float | None = None) -> FakeResponse: - seen.append(url) + def fake(req: Any, timeout: float | None = None) -> FakeResponse: + seen.append(req.full_url) return FakeResponse(b'{"state":"on"}') monkeypatch.setattr(http_mod.urllib.request, "urlopen", fake) @@ -99,3 +99,64 @@ def fail(url: str, timeout: float | None = None) -> Any: # noqa: ANN401 monkeypatch.setattr(http_mod.urllib.request, "urlopen", fail) with pytest.raises(PowerControllerError, match="unreachable"): await HttpRelayController(on_url=ON, off_url=OFF).power_off("") + + +class TestRedaction: + SECRET_ON = "http://admin:hunter2@relay.local:8080/relay/on?token=s3cret" + + async def test_logs_hide_credentials( + self, calls: list[str], caplog: pytest.LogCaptureFixture, + ) -> None: + import logging + + caplog.set_level(logging.INFO, logger="defib.power.http") + await HttpRelayController(on_url=self.SECRET_ON, off_url=OFF).power_on("") + # The query still goes out; the user info moves into a header. + assert calls == ["http://relay.local:8080/relay/on?token=s3cret"] + assert "hunter2" not in caplog.text + assert "s3cret" not in caplog.text + assert "http://relay.local:8080/relay/on?..." in caplog.text + + async def test_errors_hide_credentials(self, monkeypatch: pytest.MonkeyPatch) -> None: + def fail(url: str, timeout: float | None = None) -> Any: # noqa: ANN401 + raise urllib.error.HTTPError( + url.full_url, 401, "Unauthorized", {}, io.BytesIO(b"bad token s3cret"), # type: ignore[attr-defined, arg-type] + ) + + monkeypatch.setattr(http_mod.urllib.request, "urlopen", fail) + with pytest.raises(PowerControllerError) as exc: + await HttpRelayController(on_url=self.SECRET_ON, off_url=OFF).power_on("") + assert "hunter2" not in str(exc.value) + assert "s3cret" not in str(exc.value) + assert "401" in str(exc.value) + + async def test_unreachable_message_hides_credentials( + self, monkeypatch: pytest.MonkeyPatch, + ) -> None: + def fail(url: str, timeout: float | None = None) -> Any: # noqa: ANN401 + raise urllib.error.URLError(f"cannot reach {url}") + + monkeypatch.setattr(http_mod.urllib.request, "urlopen", fail) + with pytest.raises(PowerControllerError) as exc: + await HttpRelayController(on_url=self.SECRET_ON, off_url=OFF).power_on("") + assert "hunter2" not in str(exc.value) + assert "s3cret" not in str(exc.value) + + +class TestBasicAuth: + async def test_userinfo_becomes_basic_auth(self, monkeypatch: pytest.MonkeyPatch) -> None: + import base64 + + seen: list[Any] = [] + + def fake(req: Any, timeout: float | None = None) -> FakeResponse: + seen.append(req) + return FakeResponse(b"ok") + + monkeypatch.setattr(http_mod.urllib.request, "urlopen", fake) + url = "http://admin:p%40ss@relay.local/relay/off" + await HttpRelayController(on_url=ON, off_url=url).power_off("") + req = seen[0] + assert req.full_url == "http://relay.local/relay/off" + expected = base64.b64encode(b"admin:p@ss").decode() + assert req.get_header("Authorization") == f"Basic {expected}" diff --git a/tests/test_power_tasmota.py b/tests/test_power_tasmota.py index a830916..1a45f66 100644 --- a/tests/test_power_tasmota.py +++ b/tests/test_power_tasmota.py @@ -44,9 +44,6 @@ def __call__(self, url: str, timeout: float | None = None) -> FakeResponse: if self.reply_override is not None: return FakeResponse(self.reply_override) parts = cmnd.split() - if parts[0] == "Backlog": - self.state = "ON" - return FakeResponse(b'{"POWER":"OFF"}') if len(parts) == 2: self.state = parts[1] return FakeResponse(json.dumps({self.relay_key: self.state}).encode()) @@ -59,17 +56,6 @@ def plug(monkeypatch: pytest.MonkeyPatch) -> FakePlug: return fake -@pytest.fixture -def no_sleep(monkeypatch: pytest.MonkeyPatch) -> list[float]: - sleeps: list[float] = [] - - async def fake_sleep(seconds: float) -> None: - sleeps.append(seconds) - - monkeypatch.setattr(tasmota_mod.asyncio, "sleep", fake_sleep) - return sleeps - - class TestFromEnv: def test_missing_host(self, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.delenv("DEFIB_TASMOTA_HOST", raising=False) @@ -137,47 +123,30 @@ async def test_non_json_raises(self, plug: FakePlug) -> None: class TestPowerCycle: - async def test_cycle_runs_on_plug(self, plug: FakePlug, no_sleep: list[float]) -> None: - ctrl = TasmotaController(host="x") - await ctrl.power_cycle("", off_duration=3.0) - assert plug.cmnds == ["Backlog Power OFF; Delay 30; Power ON", "Power"] - assert no_sleep == [3.0] - - @pytest.mark.parametrize(("off", "delay"), [(0.0, 2), (0.24, 2), (1.26, 13), (9999.0, 3600)]) - async def test_delay_clamped_and_rounded( - self, plug: FakePlug, no_sleep: list[float], off: float, delay: int, - ) -> None: - ctrl = TasmotaController(host="x") - await ctrl.power_cycle("", off_duration=off) - assert plug.cmnds[0] == f"Backlog Power OFF; Delay {delay}; Power ON" - - async def test_cycle_polls_until_on( - self, monkeypatch: pytest.MonkeyPatch, no_sleep: list[float], + async def test_cycle_is_verified_off_then_on( + self, plug: FakePlug, monkeypatch: pytest.MonkeyPatch, ) -> None: - replies = iter([b'{"POWER":"OFF"}', b'{"POWER":"OFF"}', b'{"POWER":"OFF"}', b'{"POWER":"ON"}']) - calls: list[str] = [] + sleeps: list[float] = [] - def fake(url: str, timeout: float | None = None) -> FakeResponse: - calls.append(url) - return FakeResponse(next(replies)) + async def fake_sleep(seconds: float) -> None: + sleeps.append(seconds) - monkeypatch.setattr(tasmota_mod.urllib.request, "urlopen", fake) - ctrl = TasmotaController(host="x") - await ctrl.power_cycle("", off_duration=3.0) - assert len(calls) == 4 + monkeypatch.setattr("defib.power.base.asyncio.sleep", fake_sleep) + await TasmotaController(host="x").power_cycle("", off_duration=3.0) + # No Backlog: commands sent during its Delay cut the delay short. + assert plug.cmnds == ["Power OFF", "Power ON"] + assert sleeps == [3.0] - async def test_cycle_gives_up( - self, monkeypatch: pytest.MonkeyPatch, no_sleep: list[float], + async def test_cycle_fails_if_plug_does_not_switch_off( + self, plug: FakePlug, monkeypatch: pytest.MonkeyPatch, ) -> None: - def fake(url: str, timeout: float | None = None) -> FakeResponse: - return FakeResponse(b'{"POWER":"OFF"}') + async def fake_sleep(seconds: float) -> None: + return None - clock = iter(range(0, 10_000, 5)) - monkeypatch.setattr(tasmota_mod.urllib.request, "urlopen", fake) - monkeypatch.setattr(tasmota_mod.time, "monotonic", lambda: float(next(clock))) - ctrl = TasmotaController(host="x", timeout=10.0) - with pytest.raises(PowerControllerError, match="did not come back ON"): - await ctrl.power_cycle("", off_duration=3.0) + monkeypatch.setattr("defib.power.base.asyncio.sleep", fake_sleep) + plug.reply_override = b'{"POWER":"ON"}' + with pytest.raises(PowerControllerError, match="asked for OFF"): + await TasmotaController(host="x").power_cycle("", off_duration=3.0) class TestErrorMapping: diff --git a/tests/test_power_vectis.py b/tests/test_power_vectis.py index c86d757..447c5ca 100644 --- a/tests/test_power_vectis.py +++ b/tests/test_power_vectis.py @@ -147,3 +147,12 @@ async def test_close_is_idempotent(self) -> None: # should be a safe no-op. await ctrl.close() await ctrl.close() + + +def test_vectis_is_pulse_only() -> None: + """Callers must not sequence Vectis with separate off/on calls.""" + from defib.power.base import PowerController + from defib.power.vectis import VectisController + + assert VectisController.supports_independent_power is False + assert PowerController.supports_independent_power is True