From 3e6b2e82634b4f69adb78608de0abf399960817e Mon Sep 17 00:00:00 2001 From: Edward Viaene Date: Fri, 9 Oct 2026 15:18:36 -0500 Subject: [PATCH] One wiring profile and one saved connection across web, CLI, SDK and MCP - A LAN/USB BenchPod now uses the wiring profile stored on embeddedci.com for the device the pod reports in cloud_status, when BENCHPOD_API_KEY, a caller's user token or the `benchpod login` session is available. wiring=/BENCHPOD_WIRING still win; every failure logs and falls back to the defaults within ~3 s. New wiring_source= / BENCHPOD_WIRING_SOURCE=local opt-out. - BenchPod() falls back to the connection benchpod-cli saved (~/.config/benchpod-cli/config.json, read-only): argument > BENCHPOD_CONNECTION > saved. The "saved" connection string names it; the pytest plugin only uses it when asked (--benchpod-connection=saved). - MCP connect: argument > --connection > BENCHPOD_CONNECTION > saved > mDNS; the login is handed to the SDK on LAN/USB too; connect/status report wiring_source. - New SDK modules cli_config and cli_login; READMEs, CHANGELOGs (behavior changes), tests. Co-Authored-By: Claude Opus 5.5 --- packages/embeddedci-mcp/CHANGELOG.md | 15 + packages/embeddedci-mcp/README.md | 36 +- .../src/embeddedci_mcp/guide.py | 4 +- .../src/embeddedci_mcp/models.py | 7 +- .../src/embeddedci_mcp/server.py | 12 +- .../src/embeddedci_mcp/session.py | 29 +- packages/embeddedci-mcp/tests/conftest.py | 13 + .../tests/test_saved_connection_and_wiring.py | 136 +++++++ .../embeddedci-mcp/tests/tools_surface.json | 2 +- packages/embeddedci-openhtf/tests/conftest.py | 14 + packages/embeddedci/CHANGELOG.md | 28 ++ packages/embeddedci/README.md | 52 ++- .../src/embeddedci/benchpod/cli_config.py | 59 +++ .../src/embeddedci/benchpod/cli_login.py | 151 ++++++++ .../src/embeddedci/benchpod/client.py | 139 ++++++- .../src/embeddedci/benchpod/connection.py | 26 +- .../src/embeddedci/benchpod/pytest_plugin.py | 31 +- packages/embeddedci/tests/api_surface.json | 2 +- packages/embeddedci/tests/conftest.py | 12 + packages/embeddedci/tests/test_one_source.py | 357 ++++++++++++++++++ 20 files changed, 1073 insertions(+), 52 deletions(-) create mode 100644 packages/embeddedci-mcp/tests/test_saved_connection_and_wiring.py create mode 100644 packages/embeddedci/src/embeddedci/benchpod/cli_config.py create mode 100644 packages/embeddedci/src/embeddedci/benchpod/cli_login.py create mode 100644 packages/embeddedci/tests/test_one_source.py diff --git a/packages/embeddedci-mcp/CHANGELOG.md b/packages/embeddedci-mcp/CHANGELOG.md index d636ae1..53869ea 100644 --- a/packages/embeddedci-mcp/CHANGELOG.md +++ b/packages/embeddedci-mcp/CHANGELOG.md @@ -2,6 +2,21 @@ ## Unreleased +- `connect` without a connection now uses, in order: `--connection`, `BENCHPOD_CONNECTION`, the + default connection `benchpod-cli` saved (`benchpod set-connection`), then mDNS discovery. The + saved connection needs embeddedci 2.10 or newer; with an older SDK the step is skipped. + **Behavior change:** on a machine with a saved CLI connection, `connect()` goes there instead of + discovering a pod over mDNS. +- One wiring profile everywhere: on a LAN or USB connection the `wiring` tool (and every tool that + falls back to the profile) now uses the profile stored on embeddedci.com for the pod, as the web + UI's Wiring tab shows it, when `BENCHPOD_API_KEY` or the `benchpod login` session is available + and the pod is registered on that account (embeddedci 2.10 or newer). `BENCHPOD_WIRING` still + wins, `BENCHPOD_WIRING_SOURCE=local` skips the lookup, and any failure falls back to the + defaults within about 3 seconds. **Behavior change:** a signed-in user's registered LAN pod + reports `source: "server"` instead of `"defaults"` when a profile is stored. +- `connect` and `status` report `wiring_source` (`server`, `file`, `dict` or `defaults`). +- The `benchpod login` session is also handed to the SDK for a LAN or USB connection when a login + exists, so the waveform library works there without an API key too. - `connect` without a connection, `--connection` or `BENCHPOD_CONNECTION` now looks for the pod on the LAN over mDNS (as `discover` does) instead of failing with "no connection given", so `BENCHPOD_CONNECTION` is optional for a single pod on your network. When discovery finds no pod diff --git a/packages/embeddedci-mcp/README.md b/packages/embeddedci-mcp/README.md index 188916a..5883981 100644 --- a/packages/embeddedci-mcp/README.md +++ b/packages/embeddedci-mcp/README.md @@ -32,9 +32,18 @@ claude mcp add benchpod \ `BENCHPOD_LA_VOLTAGE` is the board's I/O voltage, configured once here — use `1.8` for a 1V8 board. Without it the agent is told to call `set_la_voltage` before touching the LA bank. -Both variables are optional. With no connection configured (no `connection` argument, no -`--connection`, no `BENCHPOD_CONNECTION`), `connect` looks for a BenchPod on the LAN over mDNS -(like `discover`) and uses it when it finds exactly one. When it finds none or several, the error +Both variables are optional. `connect` picks the pod, first match wins: + +1. its `connection` argument; +2. the server default, `--connection`; +3. `BENCHPOD_CONNECTION`; +4. the default connection `benchpod-cli` saved with `benchpod set-connection ` + (read-only, from `~/.config/benchpod-cli/config.json`, or under `$XDG_CONFIG_HOME`); +5. mDNS discovery of the one pod on the LAN. + +So the CLI, the SDK and the MCP server share one saved connection. With none of the first four, +`connect` looks for a BenchPod on the LAN over mDNS (like `discover`) and uses it when it finds +exactly one. When it finds none or several, the error lists the pods it heard (name and address) and what to pass instead: `host[:port]`, `usb`, or `embeddedci:` for a cloud pod (`cloud_list_devices` lists those). So `claude mcp add benchpod -- uvx embeddedci-mcp` is enough for a single pod on your network: just @@ -93,8 +102,9 @@ transport is included: ``` Or skip the API key: after `benchpod login`, the server reuses that session -(`~/.config/benchpod-cli/token.json`, refreshed and written back when it has expired) for both -`cloud_list_devices` and `connect("embeddedci:")`. `BENCHPOD_API_KEY` wins when set, and +(`~/.config/benchpod-cli/token.json`, refreshed and written back when it has expired) for +`cloud_list_devices`, `connect("embeddedci:")` and, on a LAN or USB connection, the pod's +stored wiring profile. `BENCHPOD_API_KEY` wins when set, and inside GitHub Actions the SDK uses OIDC instead. `cloud_list_devices` lists the pods on your account and which are online, without connecting to @@ -110,10 +120,12 @@ chat never blocks CI on that pod. | Flag | Environment | Default | | | --- | --- | --- | --- | -| `--connection` | `BENCHPOD_CONNECTION` | mDNS | host[:port], serial device, `usb`, `discover`, or `embeddedci:`; unset = find the one pod on the LAN | +| `--connection` | `BENCHPOD_CONNECTION` | the CLI's saved connection, else mDNS | host[:port], serial device, `usb`, `discover`, `saved`, or `embeddedci:`; unset = `benchpod set-connection`'s saved one, else find the one pod on the LAN | | `--la-voltage` | `BENCHPOD_LA_VOLTAGE` | — | LA I/O voltage (1.8 or 3.3) applied on connect | | — | `BENCHPOD_API_KEY` | — | cloud pods and the waveform library (without it, cloud tools use the `benchpod login` session) | | — | `BENCHPOD_API_BASE` | `https://www.embeddedci.com` | another embeddedci server | +| — | `BENCHPOD_WIRING` | — | wiring profile file (`.json`/`.toml`) that wins over the stored profile | +| — | `BENCHPOD_WIRING_SOURCE` | `auto` | `local` never asks embeddedci.com for the pod's stored wiring profile | | `--lease-wait` | — | `30` | cloud: seconds to wait for a busy pod | | `--idle-timeout` | — | `600` | cloud: release the lease after this many idle seconds (0 = never) | | `--timeout` | — | `30` | per-command device timeout | @@ -152,7 +164,7 @@ description, ranges and enums, so an agent sees more than this table. | Tool | Purpose | Parameters | | --- | --- | --- | -| `connect` | Open a connection (closing any previous one) and report status and capabilities. Without a `connection` or a configured default, finds the one pod on the LAN over mDNS. | `connection`, `la_voltage`, `lease_wait=30` | +| `connect` | Open a connection (closing any previous one) and report status, capabilities and where the wiring profile came from. Without a `connection` or a configured default (`--connection`, `BENCHPOD_CONNECTION`, the CLI's saved connection), finds the one pod on the LAN over mDNS. | `connection`, `la_voltage`, `lease_wait=30` | | `disconnect` | Close UART/CAN sessions and the connection, releasing a cloud lease. | none | | `status` | Firmware, capabilities, LA voltage, open sessions and warnings (works when not connected). | none | | `set_la_voltage` | Select the LA I/O voltage to match the DUT (1.8 V needs a rev3 pod). | **`voltage`** | @@ -356,6 +368,16 @@ resistors and analog paths) and `benchpod://help` (the server instructions). agent that read `wiring` once can call `flash()`, `uart_open()` or `power_on()` with no pin numbers at all. `set_wiring` replaces it for the connection, or stores it on embeddedci.com with `save: true`. + + It is the same profile the web UI's Wiring tab edits. The SDK resolves it, first match wins: + `BENCHPOD_WIRING` (a file); the profile stored on embeddedci.com for this pod (a cloud pod by + name; a LAN or USB pod by the device it reports it is registered as, when `BENCHPOD_API_KEY` or + the `benchpod login` session is available and that device is on the same account); the + defaults. `wiring` and `connect`/`status` (`wiring_source`) report where it came from: `server`, + `file`, `dict` (from `set_wiring`) or `defaults`. When the lookup fails (not signed in, pod not + registered, server unreachable, another account) the defaults apply and nothing fails; over the + LAN or USB it gives up after about 3 seconds. The STM32 pod's USB text console cannot report its + registration, so over USB the defaults apply. `BENCHPOD_WIRING_SOURCE=local` skips the lookup. - **Agent-sized captures.** `capture_adc` returns calibrated statistics, the dominant frequency and a min/max envelope; `capture_la` returns per-channel levels, edges and frequencies. The last captures stay in the session, so `decode_la`, `replay(from_last_capture=true)` and diff --git a/packages/embeddedci-mcp/src/embeddedci_mcp/guide.py b/packages/embeddedci-mcp/src/embeddedci_mcp/guide.py index e11414b..a4c4df8 100644 --- a/packages/embeddedci-mcp/src/embeddedci_mcp/guide.py +++ b/packages/embeddedci-mcp/src/embeddedci_mcp/guide.py @@ -8,13 +8,13 @@ Drive an EmbeddedCI BenchPod: a hardware-in-the-loop tester wired to a real target board (the DUT). Start every session with: -1. connect — a host[:port] or 'embeddedci:'. Omit it to use the server default (--connection or BENCHPOD_CONNECTION); with none configured, connect finds the one pod on the LAN over mDNS, and if it finds none or several its error lists them and what to pass instead (a host, 'usb', or a cloud pod from cloud_list_devices). A USB connection ('usb' or a serial device path) only supports status, LA voltage and power on an STM32 pod; everything else needs the network or cloud. +1. connect — a host[:port] or 'embeddedci:'. Omit it to use the server default (--connection, else BENCHPOD_CONNECTION, else the connection saved with `benchpod set-connection`); with none configured, connect finds the one pod on the LAN over mDNS, and if it finds none or several its error lists them and what to pass instead (a host, 'usb', or a cloud pod from cloud_list_devices). A USB connection ('usb' or a serial device path) only supports status, LA voltage and power on an STM32 pod; everything else needs the network or cloud. 2. set_la_voltage (1.8 or 3.3, the DUT's I/O voltage) unless status already reports one. The pod refuses flash, UART, LA capture, pull resistors and I2C-sensor emulation until it is set. 3. status shows firmware, capabilities, the LA voltage and open sessions. Cloud pods: cloud_list_devices lists the pods registered on embeddedci.com (name, online) without connecting to any; connect to one with its `connection` ('embeddedci:'). Tools named cloud_* talk to embeddedci.com, not to a pod. Cloud access (cloud_* and connect to 'embeddedci:') authenticates with BENCHPOD_API_KEY, else the user's `benchpod login` session; if neither exists, ask the user to run `benchpod login`. -Wiring: the pod has 14 generic LA channels (LA1-LA14) and nothing is fixed. Call `wiring` first: it is the bench's profile: which DUT signal is on which channel, the target-power rail, the UART baud, the SWD target. Every channel, baud, rail or SWD argument you omit comes from it, and channel arguments also accept its names ("READY", "uart_rx"). Confirm it with the user if it looks wrong; `set_wiring` replaces it (save=true stores it on embeddedci.com). Target power is internal 5 V (efuse 1) or external (efuse 2). The DUT's reset line goes to the pod's own reset pin: flash(nreset=true), reset_target. +Wiring: the pod has 14 generic LA channels (LA1-LA14) and nothing is fixed. Call `wiring` first: it is the bench's profile (the one stored on embeddedci.com and shown in the web UI's Wiring tab when the pod is registered and the user is signed in; `source` says where it came from): which DUT signal is on which channel, the target-power rail, the UART baud, the SWD target. Every channel, baud, rail or SWD argument you omit comes from it, and channel arguments also accept its names ("READY", "uart_rx"). Confirm it with the user if it looks wrong; `set_wiring` replaces it (save=true stores it on embeddedci.com). Target power is internal 5 V (efuse 1) or external (efuse 2). The DUT's reset line goes to the pod's own reset pin: flash(nreset=true), reset_target. GPIO: each LA channel has one owner at a time. gpio_mode claims channels (input/output/open_drain), gpio_write drives them, gpio_read reads any channel's live level, gpio_wait polls for one, gpio_pulse emits FPGA-timed pulses, gpio_release frees them. Release a GPIO channel before uart_open, flash or enable_i2c_sensor uses it, otherwise the pod refuses with "pin conflict: LA4 is in use by gpio; release it with gpio_release(la=[4])": the error names the tool that frees the channel (gpio_release, uart_close, disable_i2c_sensor, disable_gps); la_pins shows every channel's owner. Captures observe all 14 channels whatever owns them. diff --git a/packages/embeddedci-mcp/src/embeddedci_mcp/models.py b/packages/embeddedci-mcp/src/embeddedci_mcp/models.py index 97b7797..ae11a3d 100644 --- a/packages/embeddedci-mcp/src/embeddedci_mcp/models.py +++ b/packages/embeddedci-mcp/src/embeddedci_mcp/models.py @@ -105,6 +105,9 @@ class StatusResult(BaseModel): session: SessionInfo = Field(default_factory=SessionInfo) warnings: List[str] = Field(default_factory=list) firmware: Dict[str, Any] = Field(default_factory=dict, description="The pod's raw status report.") + wiring_source: Optional[str] = Field(None, description=( + "Where this connection's wiring profile came from: server (stored on embeddedci.com, as the " + "web UI's Wiring tab shows it), file (BENCHPOD_WIRING), dict (set_wiring) or defaults.")) class LaVoltageResult(BaseModel): @@ -134,7 +137,9 @@ class WiringPin(BaseModel): class WiringResult(BaseModel): """The bench's effective wiring profile: which DUT signal is on which LA channel.""" - source: str = Field(description="Where the profile came from: defaults, file, server or dict.") + source: str = Field(description=( + "Where the profile came from: server (stored on embeddedci.com for this pod, over the cloud " + "or, signed in, over the LAN/USB), file (BENCHPOD_WIRING), dict (set_wiring) or defaults.")) version: int = 1 la_voltage: float = Field(description="LA I/O-bank voltage the profile asks for, in volts.") efuse: int = Field(description=( diff --git a/packages/embeddedci-mcp/src/embeddedci_mcp/server.py b/packages/embeddedci-mcp/src/embeddedci_mcp/server.py index 7ccbf3a..e278fef 100644 --- a/packages/embeddedci-mcp/src/embeddedci_mcp/server.py +++ b/packages/embeddedci-mcp/src/embeddedci_mcp/server.py @@ -292,7 +292,8 @@ def _status() -> m.StatusResult: if not SESSION.connected and not SESSION.reconnectable: default = SESSION.default_connection hint = (f"call connect (the server default is {default!r})" if default - else "call connect: without a connection it finds the one pod on the LAN over mDNS; " + else "call connect: without a connection it uses the one saved with `benchpod " + "set-connection`, else finds the one pod on the LAN over mDNS; " "else pass a host, 'usb' or 'embeddedci:'") return m.StatusResult(connected=False, warnings=[f"not connected — {hint}"]) pod = SESSION.require() @@ -313,13 +314,17 @@ def _status() -> m.StatusResult: if isinstance(firmware, dict) and firmware.get("lan_policy") == "locked" and SESSION.kind == "tcp": warnings.append("the pod's LAN policy is locked: configuration and firmware commands need " "the cloud ('embeddedci:') or the USB console") + try: + wiring_source: Optional[str] = pod.wiring.source + except (BenchPodError, ValueError, OSError): + wiring_source = None if la is None: warnings.append("LA I/O voltage not selected — call set_la_voltage (1.8 or 3.3, matching " "the DUT) before flash, UART, LA capture, pulls or I2C-sensor emulation") return m.StatusResult( connected=True, connection=SESSION.connection, kind=SESSION.kind, leased=pod.leased, la_voltage=la, capabilities=m.CapabilitiesInfo.from_caps(pod.capabilities), - session=SESSION.info(), warnings=warnings, firmware=firmware, + session=SESSION.info(), warnings=warnings, firmware=firmware, wiring_source=wiring_source, ) @@ -328,7 +333,8 @@ async def connect( connection: Annotated[Optional[str], Field(description=( "host[:port] (TCP, default port 8080), a serial device path, 'usb' (auto-detect), " "'discover' (mDNS) or 'embeddedci:' (cloud). Omit to use the server default " - "(--connection or BENCHPOD_CONNECTION), else to find the one pod on the LAN over mDNS."))] = None, + "(--connection, else BENCHPOD_CONNECTION, else the connection saved with `benchpod " + "set-connection`), else to find the one pod on the LAN over mDNS."))] = None, la_voltage: Annotated[Optional[Literal[1.8, 3.3]], Field(description=( "Select the LA I/O-bank voltage right after connecting (the DUT's I/O voltage)."))] = None, lease_wait: Annotated[float, Field(ge=0, le=3600, description=( diff --git a/packages/embeddedci-mcp/src/embeddedci_mcp/session.py b/packages/embeddedci-mcp/src/embeddedci_mcp/session.py index c5fefda..27ed563 100644 --- a/packages/embeddedci-mcp/src/embeddedci_mcp/session.py +++ b/packages/embeddedci-mcp/src/embeddedci_mcp/session.py @@ -59,21 +59,34 @@ def connection_kind(connection: str) -> str: def cloud_user_token(connection: str) -> Optional[Callable[[], str]]: - """For a cloud connection without BENCHPOD_API_KEY, outside GitHub Actions (which uses OIDC): - the `benchpod login` session, exchanged by the SDK for a device session token.""" - if connection_kind(connection) != "embeddedci" or os.environ.get(API_KEY_ENV): + """The `benchpod login` session, without BENCHPOD_API_KEY and outside GitHub Actions (which uses + OIDC). For a cloud connection the SDK exchanges it for a device session token; for a LAN or USB + pod (only when a login exists) the SDK reads the pod's stored wiring profile with it, so the + `wiring` tool shows what the web UI's Wiring tab shows.""" + if os.environ.get(API_KEY_ENV) or os.environ.get("GITHUB_ACTIONS") == "true": return None - if os.environ.get("GITHUB_ACTIONS") == "true": + if connection_kind(connection) != "embeddedci" and not cli_login.token_path().is_file(): return None base = os.environ.get(API_BASE_ENV) or DEFAULT_API_BASE return lambda: cli_login.access_token(base) +def saved_connection() -> Optional[str]: + """The default connection `benchpod-cli` saved (`benchpod set-connection`), or None. Needs an + SDK that reads it (embeddedci 2.10); older ones have none.""" + try: + from embeddedci.benchpod.cli_config import saved_connection as _saved + except ImportError: # pragma: no cover - embeddedci < 2.10 + return None + return _saved() + + #: What to pass instead when mDNS finds no single pod. CONNECTION_HELP = ( "pass connection= one of: host[:port] (a pod on the LAN, default port 8080), 'usb' (a pod on " "this machine's USB), or 'embeddedci:' for a cloud pod (cloud_list_devices lists " - "them). The user can also make one the default with " + ENV_VAR + " or --connection." + "them). The user can also make one the default with `benchpod set-connection `, " + + ENV_VAR + " or --connection." ) @@ -101,7 +114,8 @@ def __init__(self) -> None: self.lock = threading.RLock() self._pod: Optional[BenchPod] = None #: Connection used by ``connect`` when called without one (``--connection``); falls back to - #: ``BENCHPOD_CONNECTION``, then to mDNS discovery of a single pod on the LAN. + #: ``BENCHPOD_CONNECTION``, then to the connection ``benchpod-cli`` saved, then to mDNS + #: discovery of a single pod on the LAN. self.default_connection: Optional[str] = None #: LA voltage applied on connect when the tool call does not pass one (``--la-voltage``); #: falls back to ``BENCHPOD_LA_VOLTAGE`` inside the SDK. @@ -145,7 +159,8 @@ def connect(self, connection: Optional[str] = None, *, la_voltage: Optional[floa """Open (or re-open) the device, closing any prior connection and sessions first.""" with self.lock: self.disconnect() - conn = connection or self.default_connection or os.environ.get(ENV_VAR) + conn = (connection or self.default_connection or os.environ.get(ENV_VAR) + or saved_connection()) if not conn: conn = discover_connection() la = self.default_la_voltage if la_voltage is None else la_voltage diff --git a/packages/embeddedci-mcp/tests/conftest.py b/packages/embeddedci-mcp/tests/conftest.py index 11330c9..b05f0ab 100644 --- a/packages/embeddedci-mcp/tests/conftest.py +++ b/packages/embeddedci-mcp/tests/conftest.py @@ -4,6 +4,7 @@ from __future__ import annotations import math +import os from typing import Any, Dict, Iterator, List, Optional import anyio @@ -431,3 +432,15 @@ def connected(fake_open, fake_transport) -> FakeTransport: """A session connected (via the connect tool) to the fake pod, LA bank at 3.3 V.""" call("connect", connection="192.168.1.50", la_voltage=3.3) return fake_transport + + +@pytest.fixture(autouse=True) +def _isolated_benchpod_cli_config(request, tmp_path_factory, monkeypatch): + """Unit tests never see this machine's benchpod-cli config (its saved connection) or its + `benchpod login` session: both change what BenchPod() connects to and where its wiring comes + from. Hardware tests (e2e, examples) keep them.""" + path = str(request.node.fspath) + if "e2e" in os.path.basename(path) or os.sep + "e2e" + os.sep in path \ + or os.sep + "examples" + os.sep in path: + return + monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path_factory.mktemp("xdg"))) diff --git a/packages/embeddedci-mcp/tests/test_saved_connection_and_wiring.py b/packages/embeddedci-mcp/tests/test_saved_connection_and_wiring.py new file mode 100644 index 0000000..8fa5aad --- /dev/null +++ b/packages/embeddedci-mcp/tests/test_saved_connection_and_wiring.py @@ -0,0 +1,136 @@ +"""One saved connection and one wiring profile: `connect` falls back to the connection benchpod-cli +saved, and a LAN pod's `wiring` is the profile stored on embeddedci.com when the user is signed in. + +Both need an SDK with ``embeddedci.benchpod.cli_config`` (2.10); CI also runs these tests against +the oldest SDK the server allows, where they skip.""" + +from __future__ import annotations + +import io +import json +import urllib.request +from datetime import datetime, timedelta, timezone + +import pytest + +from embeddedci.benchpod import BenchPod + +import embeddedci_mcp.session as session_mod +from embeddedci_mcp.session import SESSION, cloud_user_token + +from conftest import call + +try: + import embeddedci.benchpod.cli_config # noqa: F401 + + NEW_SDK = True +except ImportError: # pragma: no cover - the minimum-SDK CI job + NEW_SDK = False + +needs_new_sdk = pytest.mark.skipif(not NEW_SDK, reason="needs embeddedci with cli_config (2.10)") + +DEVICE_ID = "dev-1234" + + +@pytest.fixture +def xdg(monkeypatch, tmp_path): + monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path)) + monkeypatch.delenv("GITHUB_ACTIONS", raising=False) + for var in ("BENCHPOD_API_BASE", "BENCHPOD_WIRING", "BENCHPOD_WIRING_SOURCE"): + monkeypatch.delenv(var, raising=False) + d = tmp_path / "benchpod-cli" + d.mkdir() + return d + + +def _save(xdg, connection): + (xdg / "config.json").write_text(json.dumps({"connection": connection})) + + +def _login(xdg): + now = datetime.now(timezone.utc) + (xdg / "token.json").write_text(json.dumps({ + "access_token": "login-access", "refresh_token": "r", + "access_expires_at": (now + timedelta(hours=1)).isoformat(), + "refresh_expires_at": (now + timedelta(days=7)).isoformat()})) + + +def _no_mdns(monkeypatch): + from embeddedci.benchpod import discovery + + def no_browse(timeout=None): + raise AssertionError("mDNS must not run when a connection is saved") + + monkeypatch.setattr(discovery, "discover", no_browse) + + +@needs_new_sdk +def test_connect_uses_the_cli_saved_connection(fake_open, monkeypatch, xdg): + _save(xdg, "192.168.1.220") + _no_mdns(monkeypatch) + result = call("connect") + assert fake_open["connection"] == "192.168.1.220" and result["connected"] is True + + +@needs_new_sdk +@pytest.mark.parametrize("configured", ["argument", "server", "env"]) +def test_the_saved_connection_comes_last(fake_open, monkeypatch, xdg, configured): + _save(xdg, "192.168.1.220") + if configured == "argument": + call("connect", connection="10.0.0.7") + else: + if configured == "server": + SESSION.default_connection = "10.0.0.7" + else: + monkeypatch.setenv("BENCHPOD_CONNECTION", "10.0.0.7") + call("connect") + assert fake_open["connection"] == "10.0.0.7" + + +def test_the_login_reaches_a_lan_pod_only_when_there_is_one(monkeypatch, xdg): + assert cloud_user_token("192.168.1.220") is None # not signed in + _login(xdg) + provider = cloud_user_token("192.168.1.220") + assert provider is not None and provider() == "login-access" + assert cloud_user_token("usb")() == "login-access" + monkeypatch.setenv("BENCHPOD_API_KEY", "eci_x") # the SDK uses the key itself + assert cloud_user_token("192.168.1.220") is None + + +class _Resp(io.BytesIO): + status = 200 + + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + +@needs_new_sdk +def test_wiring_of_a_lan_pod_is_the_stored_profile(monkeypatch, xdg, fake_transport): + _login(xdg) + fake_transport._cmd_cloud_status = lambda req: {"configured": True, "device_id": DEVICE_ID} + requests = [] + + def urlopen(req, timeout=None): + requests.append((req.full_url, req.get_header("Authorization"))) + return _Resp(json.dumps({"profile": {"uart_rx": 3, "uart_tx": 6}, "stored": True}).encode()) + + monkeypatch.setattr(urllib.request, "urlopen", urlopen) + + def factory(connection, **kwargs): + return BenchPod(transport=fake_transport, lease=False, la_voltage=kwargs.get("la_voltage"), + cloud_user_token=kwargs.get("cloud_user_token")) + + monkeypatch.setattr(session_mod, "BenchPod", factory) + status = call("connect", connection="192.168.1.220", la_voltage=3.3) + assert status["wiring_source"] == "server" + w = call("wiring") + assert w["source"] == "server" and w["profile"]["uart_rx"] == 3 and w["profile"]["uart_tx"] == 6 + assert requests == [(f"https://www.embeddedci.com/api/benchpod/devices/{DEVICE_ID}/wiring/profile", + "Bearer login-access")] + + +def test_status_reports_the_defaults_when_nothing_is_stored(connected): + assert call("status")["wiring_source"] == "defaults" diff --git a/packages/embeddedci-mcp/tests/tools_surface.json b/packages/embeddedci-mcp/tests/tools_surface.json index 73c0461..f7d9d52 100644 --- a/packages/embeddedci-mcp/tests/tools_surface.json +++ b/packages/embeddedci-mcp/tests/tools_surface.json @@ -808,7 +808,7 @@ } ], "default": null, - "description": "host[:port] (TCP, default port 8080), a serial device path, 'usb' (auto-detect), 'discover' (mDNS) or 'embeddedci:' (cloud). Omit to use the server default (--connection or BENCHPOD_CONNECTION), else to find the one pod on the LAN over mDNS.", + "description": "host[:port] (TCP, default port 8080), a serial device path, 'usb' (auto-detect), 'discover' (mDNS) or 'embeddedci:' (cloud). Omit to use the server default (--connection, else BENCHPOD_CONNECTION, else the connection saved with `benchpod set-connection`), else to find the one pod on the LAN over mDNS.", "title": "Connection" }, "la_voltage": { diff --git a/packages/embeddedci-openhtf/tests/conftest.py b/packages/embeddedci-openhtf/tests/conftest.py index c88dc5c..888f55d 100644 --- a/packages/embeddedci-openhtf/tests/conftest.py +++ b/packages/embeddedci-openhtf/tests/conftest.py @@ -6,6 +6,20 @@ import os import sys +import pytest + _SRC = os.path.join(os.path.dirname(os.path.dirname(__file__)), "src") if os.environ.get("EMBEDDEDCI_TEST_INSTALLED") != "1" and _SRC not in sys.path: sys.path.insert(0, _SRC) + + +@pytest.fixture(autouse=True) +def _isolated_benchpod_cli_config(request, tmp_path_factory, monkeypatch): + """Unit tests never see this machine's benchpod-cli config (its saved connection) or its + `benchpod login` session: both change what BenchPod() connects to and where its wiring comes + from. Hardware tests (e2e, examples) keep them.""" + path = str(request.node.fspath) + if "e2e" in os.path.basename(path) or os.sep + "e2e" + os.sep in path \ + or os.sep + "examples" + os.sep in path: + return + monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path_factory.mktemp("xdg"))) diff --git a/packages/embeddedci/CHANGELOG.md b/packages/embeddedci/CHANGELOG.md index 6b04738..efff8e3 100644 --- a/packages/embeddedci/CHANGELOG.md +++ b/packages/embeddedci/CHANGELOG.md @@ -2,6 +2,34 @@ ## Unreleased +- One wiring profile everywhere. A LAN or USB `BenchPod` now uses the wiring profile stored on + embeddedci.com for the pod (the web UI's Wiring tab), as a cloud connection already did, when + credentials are available (`api_key=` / `BENCHPOD_API_KEY`, `cloud_user_token=`, or the + `benchpod login` session) and the pod reports (`cloud_status`) a device id on that account; + `bp.wiring.source` is then `"server"`. The `wiring=` argument, `--benchpod-wiring`, the + `benchpod_wiring` fixture and `BENCHPOD_WIRING` still win. Any failure (no credentials, pod not + registered, server unreachable, HTTP 404, another account) logs why and uses the defaults; it + never raises and gives up after about 3 seconds. (The STM32 pod's USB text console has no + `cloud_status`, so over USB the defaults still apply there.) New `wiring_source=` argument and + `BENCHPOD_WIRING_SOURCE` variable: `"local"` skips the embeddedci.com lookup on every + connection kind, `"auto"` is the default. + **Behavior change:** a signed-in developer (or a run with `BENCHPOD_API_KEY`) whose LAN or USB + pod is registered and has a stored profile now gets that profile instead of the defaults, and + the first `bp.wiring` access on such a connection makes one `cloud_status` call to the pod and + one request to embeddedci.com. Set `BENCHPOD_WIRING_SOURCE=local` for the old behavior. +- One saved connection. `BenchPod()` with no `connection` argument and no `BENCHPOD_CONNECTION` + now uses the default connection `benchpod-cli` saved (`benchpod set-connection`, + `benchpod discover --save`) in `~/.config/benchpod-cli/config.json` (under `$XDG_CONFIG_HOME` + when set). Precedence: argument, then `BENCHPOD_CONNECTION`, then the saved connection. The file + is only read; a missing or malformed one counts as nothing saved. The connection string `saved` + names it explicitly. **Behavior change:** `BenchPod()` used to raise `ConnectionConfigError` + here; it now connects to the saved pod when there is one. +- The pytest plugin keeps its behavior: without `--benchpod-connection`, the ini option or + `BENCHPOD_CONNECTION` the hardware fixtures skip, and the CLI's saved connection is never used + implicitly, so a suite does not start driving hardware on a developer's machine. Opt in with + `--benchpod-connection=saved`. +- New modules `embeddedci.benchpod.cli_config` (`saved_connection()`, `config_path()`) and + `embeddedci.benchpod.cli_login` (the `benchpod login` session, as the MCP server reads it). - Friendlier pod refusals: a `FirmwareError` whose fix the firmware spells as a raw protocol command now names the SDK call instead. "la voltage not set" says to set the board's I/O voltage with `BenchPod(conn, la_voltage=3.3)`, the `benchpod_la_voltage` fixture or diff --git a/packages/embeddedci/README.md b/packages/embeddedci/README.md index 0b49043..3fb1575 100644 --- a/packages/embeddedci/README.md +++ b/packages/embeddedci/README.md @@ -163,8 +163,24 @@ fixture in conftest.py, or bp.set_la_voltage(3.3)` (prefixed with the command, a `FirmwareError` is). Other constructor arguments: `timeout` (seconds, default 30), `api_key`, `api_base`, -`cloud_token`, `cloud_audience`, `cloud_user_token`, `lease`, `lease_wait`, `lease_ttl` (see [Cloud](#cloud-embeddedcidevice-name)) -and `transport` (inject a custom backend). +`cloud_token`, `cloud_audience`, `cloud_user_token`, `lease`, `lease_wait`, `lease_ttl` (see [Cloud](#cloud-embeddedcidevice-name)), +`wiring` and `wiring_source` (see [Wiring profile](#wiring-profile)) and `transport` (inject a +custom backend). + +### Which pod `BenchPod()` connects to + +The CLI, the SDK and the MCP server share one saved connection. `BenchPod()` picks, first match +wins: + +1. the `connection` argument; +2. the `BENCHPOD_CONNECTION` environment variable; +3. the default connection `benchpod-cli` saved with `benchpod set-connection ` + (or `benchpod discover --save`), read from `~/.config/benchpod-cli/config.json` + (`$XDG_CONFIG_HOME/benchpod-cli/config.json` when that is set). + +With none of them it raises `ConnectionConfigError`. The SDK only reads that file, and a missing or +malformed file counts as "nothing saved". The connection string `saved` names the saved connection +explicitly. The pytest plugin does **not** fall back to it (see [pytest plugin](#pytest-plugin)). ### Connection strings @@ -175,6 +191,7 @@ and `transport` (inject a custom backend). | `usb` | USB console, auto-detected by probing the serial ports (`serial` also accepted) | | `discover` (or `mdns`, `auto`) | find exactly one pod on the LAN via mDNS (needs `[discovery]`); errors on zero or several | | `embeddedci:` | a named device through embeddedci.com (needs `[cloud]`; an API key or a user token anywhere, or GitHub Actions OIDC) | +| `saved` | the default connection `benchpod-cli` saved (`benchpod set-connection`); errors when none is saved | **Use the network (or the cloud) for testing.** The STM32 pod's USB console is a text shell for setup and diagnostics with no JSON mode, so over USB the SDK can only read `status()`, `ping()`, @@ -194,12 +211,13 @@ automatically.) | Variable | Used when | Purpose | |---|---|---| -| `BENCHPOD_CONNECTION` | no `connection` argument / `--benchpod-connection` | connection string | +| `BENCHPOD_CONNECTION` | no `connection` argument / `--benchpod-connection` | connection string (`BenchPod()` then falls back to the CLI's saved connection; pytest does not) | | `BENCHPOD_LA_VOLTAGE` | no `la_voltage` argument / `--benchpod-la-voltage` | LA bank voltage to select on connect (`1.8` or `3.3`) | | `BENCHPOD_API_KEY` | no `api_key` argument / `--benchpod-api-key` | EmbeddedCI API key (`eci_…`) for the cloud destination and server features | | `BENCHPOD_API_BASE` | no `api_base` argument / `--benchpod-api-base` | server base URL (default `https://www.embeddedci.com`) | | `BENCHPOD_BUILD_TARGET` | no `--benchpod-build-target` | platform id recorded by the `build_report` fixture | | `BENCHPOD_WIRING` | no `wiring` argument / `--benchpod-wiring` / `benchpod_wiring` fixture | wiring profile file (`.json`/`.toml`), see [Wiring profile](#wiring-profile) | +| `BENCHPOD_WIRING_SOURCE` | no `wiring_source` argument | `local` never asks embeddedci.com for the pod's stored wiring profile; `auto` (default) does, see [Wiring profile](#wiring-profile) | | `BENCHPOD_LIFT_DAC_LIMITS` | pytest sessions | `1` clears the pod's DAC output limits for the session and restores the same limits at the end, even when tests fail (only with the output stage switched off) | | `BENCHPOD_STALL_TIMEOUT` | flashing | seconds with no SWD traffic before a flash attempt is aborted and retried (default 60) | @@ -339,6 +357,11 @@ The `benchpod` fixture is a `BenchPod` instance, not the module — import const Connection resolution order: `--benchpod-connection` → `benchpod_connection` ini option → `BENCHPOD_CONNECTION` → `--benchpod-discover`. +The plugin never falls back to the connection `benchpod-cli` saved on this machine, unlike +`BenchPod()`: a suite must not start driving hardware on a developer's machine just because they +once ran `benchpod set-connection`. Opt in with `--benchpod-connection=saved` (or `saved` in the ini +option or `BENCHPOD_CONNECTION`); with nothing saved the hardware tests skip and say why. + ```ini # pytest.ini [pytest] @@ -350,7 +373,7 @@ benchpod_connection = 192.168.1.213 | Fixture | Scope | Provides | |---|---|---| | `benchpod_la_voltage` | session | the board's I/O voltage selected on connect — **override it in `conftest.py`** (default `None` → `BENCHPOD_LA_VOLTAGE`) | -| `benchpod_wiring` | session | the bench's wiring profile — **override it in `conftest.py`** (a `Wiring`, dict or file path; default `None` → `BENCHPOD_WIRING`, a cloud device's stored profile, the defaults). An explicit profile also supplies the LA voltage when nothing else sets one | +| `benchpod_wiring` | session | the bench's wiring profile — **override it in `conftest.py`** (a `Wiring`, dict or file path; default `None` → `BENCHPOD_WIRING`, the pod's profile stored on embeddedci.com, the defaults; see [Wiring profile](#wiring-profile)). An explicit profile also supplies the LA voltage when nothing else sets one | | `benchpod` | session | a connected `BenchPod` with the options above applied; LA channels left in GPIO mode are released when the session starts and ends; closed at session end | | `benchpod_connection` | session | the resolved connection string (skips when none) | | `benchpod_target` | function | `benchpod` with the target rail (`--benchpod-efuse`, else the wiring profile's) powered on for the test, off at teardown | @@ -418,9 +441,24 @@ A role set to `None` is not wired. Construction validates the whole profile and listing every problem — including two roles or signals on one LA channel; `wiring.warnings()` lists wiring that works but is risky, such as an I2C bus on a channel without a pull-up. -**Where the profile comes from** (`bp.wiring`, resolved once): the `wiring=` argument (a `Wiring`, a -dict, or a `.json`/`.toml` file) → the `BENCHPOD_WIRING` file → for a cloud device -(`embeddedci:`), the profile stored on embeddedci.com, which the web UI edits → the defaults. +**Where the profile comes from.** The web UI, the CLI, the SDK and the MCP server use one profile +per pod: the one stored on embeddedci.com. `bp.wiring` is resolved once, first match wins: + +| # | Source | `bp.wiring.source` | +|---|---|---| +| 1 | the `wiring=` argument (a `Wiring`, a dict, or a `.json`/`.toml` file); in pytest `--benchpod-wiring`, then the `benchpod_wiring` fixture | `dict`, `file` (a `Wiring` keeps its own) | +| 2 | the `BENCHPOD_WIRING` file | `file` | +| 3 | the profile stored on embeddedci.com for this pod, as the web UI's Wiring tab shows it: for `embeddedci:` that device; over the LAN (or a USB console that answers JSON commands), the device the pod reports it is registered as, when an API key (`api_key=` / `BENCHPOD_API_KEY`) or the `benchpod login` session is available and that device is on the same account | `server` | +| 4 | the defaults | `defaults` | + +Step 3 never fails the connection. Without credentials, with the pod not registered, the server +unreachable, or the pod on another account, the SDK logs why (logger `embeddedci.benchpod`) and uses +the defaults; over the LAN or USB it gives up after about 3 seconds. The STM32 pod's USB text +console cannot say which device it is registered as, so over USB the defaults apply unless +`wiring=` or `BENCHPOD_WIRING` gives a profile. `wiring_source="local"` (or +`BENCHPOD_WIRING_SOURCE=local`) skips step 3 entirely, for offline benches or a bench that should +ignore the stored profile. + `bp.wiring = ...` swaps it for this connection; `bp.save_wiring(profile)` stores it for a cloud device. A file holds the same JSON object the server stores: diff --git a/packages/embeddedci/src/embeddedci/benchpod/cli_config.py b/packages/embeddedci/src/embeddedci/benchpod/cli_config.py new file mode 100644 index 0000000..ec85339 --- /dev/null +++ b/packages/embeddedci/src/embeddedci/benchpod/cli_config.py @@ -0,0 +1,59 @@ +"""Read-only access to ``benchpod-cli``'s saved settings, so the SDK, the MCP server and the CLI +share one default connection. + +``benchpod-cli`` keeps its configuration in ``~/.config/benchpod-cli/config.json`` (or under +``$XDG_CONFIG_HOME``; see its ``internal/benchpodconfig``). ``benchpod set-connection `` +and ``benchpod discover --save`` store the default connection there as ``connection``; older CLI +versions wrote a TCP address as ``bench_pod_addr``, which is read the same way. The SDK never +writes this file. A missing, unreadable or malformed file means "nothing saved". +""" + +from __future__ import annotations + +import json +import logging +import os +from pathlib import Path +from typing import Optional + +#: The CLI's settings directory name under the config root. +CLI_DIR = "benchpod-cli" +#: The connection keyword that stands for the CLI's saved connection. +SAVED_KEYWORD = "saved" + +_log = logging.getLogger("embeddedci.benchpod") + + +def config_dir() -> Path: + """``$XDG_CONFIG_HOME/benchpod-cli`` when set, else ``~/.config/benchpod-cli`` (as the CLI).""" + base = os.environ.get("XDG_CONFIG_HOME") or os.path.join(os.path.expanduser("~"), ".config") + return Path(base) / CLI_DIR + + +def config_path() -> Path: + """The CLI's ``config.json``.""" + return config_dir() / "config.json" + + +def saved_connection(path: Optional[Path] = None) -> Optional[str]: + """The default connection ``benchpod-cli`` saved (``connection``, else the legacy + ``bench_pod_addr``), or None when there is none or the file cannot be read.""" + path = path or config_path() + try: + data = json.loads(path.read_text(encoding="utf-8")) + except FileNotFoundError: + return None + except (OSError, ValueError) as exc: + _log.warning("ignoring the benchpod-cli config %s: %s", path, exc) + return None + if not isinstance(data, dict): + _log.warning("ignoring the benchpod-cli config %s: not a JSON object", path) + return None + for key in ("connection", "bench_pod_addr"): + value = data.get(key) + if isinstance(value, str) and value.strip(): + conn = value.strip() + if conn.lower() == SAVED_KEYWORD: # would point at itself + return None + return conn + return None diff --git a/packages/embeddedci/src/embeddedci/benchpod/cli_login.py b/packages/embeddedci/src/embeddedci/benchpod/cli_login.py new file mode 100644 index 0000000..47af9e4 --- /dev/null +++ b/packages/embeddedci/src/embeddedci/benchpod/cli_login.py @@ -0,0 +1,151 @@ +"""The ``benchpod login`` session, reused by the SDK for embeddedci.com access. + +``benchpod-cli`` keeps a logged-in user's tokens in ``~/.config/benchpod-cli/token.json`` (see its +``internal/authstore``). A developer who has run ``benchpod login`` therefore needs no API key for +the SDK to read the pod's wiring profile from embeddedci.com. This module reads that file, +refreshes the access token through ``POST /api/auth/guest/refresh`` when it has expired, and writes +the rotated tokens back in the same shape, so the CLI keeps working afterwards. + +The ``embeddedci-mcp`` server carries the same logic (``embeddedci_mcp.cli_login``) because it +still supports SDK releases without this module. +""" + +from __future__ import annotations + +import json +import os +import re +import tempfile +import threading +import urllib.error +import urllib.request +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any, Dict, Optional + +from .cli_config import config_dir +from .cloud_auth import USER_AGENT +from .errors import CloudAuthError + +#: Same margin as the CLI: a token this close to expiry counts as expired. +EXPIRY_SKEW = timedelta(seconds=60) +_HTTP_TIMEOUT = 15.0 +_LOGIN_HINT = "run `benchpod login` (or set BENCHPOD_API_KEY)" + +# One refresh at a time: the server rotates the refresh token, so two concurrent refreshes would +# leave one of them holding a revoked token. +_lock = threading.Lock() + + +def token_path() -> Path: + """Where ``benchpod-cli`` stores its tokens: ``$XDG_CONFIG_HOME`` first, else ``~/.config``.""" + return config_dir() / "token.json" + + +def has_session(path: Optional[Path] = None) -> bool: + """Whether a ``benchpod login`` token file exists (it may still have expired).""" + return (path or token_path()).is_file() + + +def _parse_time(value: Any) -> Optional[datetime]: + """An RFC 3339 time as Go writes it (up to 9 fractional digits); None for Go's zero time.""" + if not value or not isinstance(value, str): + return None + s = value.strip().replace("Z", "+00:00") + s = re.sub(r"(\.\d{6})\d+", r"\1", s) # Python 3.10 accepts at most microseconds + try: + t = datetime.fromisoformat(s) + except ValueError: + return None + if t.year <= 1: + return None + return t if t.tzinfo else t.replace(tzinfo=timezone.utc) + + +def _expired(token: Any, expires_at: Any, now: datetime) -> bool: + """Mirrors authstore's AccessExpired/RefreshExpired: no expiry recorded means still valid.""" + if not token: + return True + t = _parse_time(expires_at) + return t is not None and now >= t - EXPIRY_SKEW + + +def _load(path: Path) -> Dict[str, Any]: + try: + data = json.loads(path.read_text()) + except FileNotFoundError: + raise CloudAuthError(f"not logged in to embeddedci.com ({path} not found): {_LOGIN_HINT}") from None + except (OSError, ValueError) as exc: + raise CloudAuthError(f"cannot read {path}: {exc}") from exc + if not isinstance(data, dict): + raise CloudAuthError(f"cannot read {path}: not a JSON object") + return data + + +def _save(path: Path, tokens: Dict[str, Any]) -> None: + """Replace the file atomically with owner-only permissions, as authstore.Save does.""" + path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + fd, tmp = tempfile.mkstemp(dir=str(path.parent), prefix="token-", suffix=".json") + try: + with os.fdopen(fd, "w") as fh: + json.dump(tokens, fh, indent=2) + os.chmod(tmp, 0o600) + os.replace(tmp, path) + except BaseException: + try: + os.unlink(tmp) + except OSError: + pass + raise + + +def _refresh(api_base: str, refresh_token: str, timeout: float = _HTTP_TIMEOUT) -> Dict[str, Any]: + request = urllib.request.Request( + api_base.rstrip("/") + "/api/auth/guest/refresh", + data=json.dumps({"refresh_token": refresh_token}).encode("utf-8"), + method="POST", + headers={"Content-Type": "application/json", "Accept": "application/json", + "User-Agent": USER_AGENT}, + ) + try: + with urllib.request.urlopen(request, timeout=timeout) as resp: + body = json.loads(resp.read().decode("utf-8")) + except urllib.error.HTTPError as exc: + msg = exc.read().decode("utf-8", "replace") + try: + msg = json.loads(msg).get("error", msg) + except Exception: + pass + raise CloudAuthError( + f"refreshing the benchpod login failed (HTTP {exc.code}): {msg}; {_LOGIN_HINT}") from exc + except Exception as exc: + raise CloudAuthError(f"refreshing the benchpod login failed: {exc}") from exc + if not body.get("access_token") or not body.get("refresh_token"): + raise CloudAuthError("refreshing the benchpod login returned no tokens; " + _LOGIN_HINT) + return body + + +def access_token(api_base: str, path: Optional[Path] = None, *, timeout: float = _HTTP_TIMEOUT) -> str: + """A valid access token from the ``benchpod login`` session, refreshing (and saving) it first + when it has expired (waiting at most ``timeout`` seconds for the server). Raises + :class:`CloudAuthError` when there is no usable session.""" + path = path or token_path() + with _lock: + tokens = _load(path) + now = datetime.now(timezone.utc) + if not _expired(tokens.get("access_token"), tokens.get("access_expires_at"), now): + return str(tokens["access_token"]) + if _expired(tokens.get("refresh_token"), tokens.get("refresh_expires_at"), now): + raise CloudAuthError(f"the benchpod login has expired: {_LOGIN_HINT}") + resp = _refresh(api_base, str(tokens["refresh_token"]), timeout) + local_now = datetime.now().astimezone() + tokens.update( + access_token=resp["access_token"], + refresh_token=resp["refresh_token"], + access_expires_at=(local_now + timedelta(seconds=int(resp.get("access_expires_in") or 0))).isoformat(), + refresh_expires_at=(local_now + timedelta(seconds=int(resp.get("refresh_expires_in") or 0))).isoformat(), + session_id=resp.get("session_id") or tokens.get("session_id", ""), + user_id=resp.get("user_id") or tokens.get("user_id", ""), + ) + _save(path, tokens) + return str(resp["access_token"]) diff --git a/packages/embeddedci/src/embeddedci/benchpod/client.py b/packages/embeddedci/src/embeddedci/benchpod/client.py index 669dd05..1d5dd85 100644 --- a/packages/embeddedci/src/embeddedci/benchpod/client.py +++ b/packages/embeddedci/src/embeddedci/benchpod/client.py @@ -120,6 +120,12 @@ API_BASE_ENV = "BENCHPOD_API_BASE" API_KEY_ENV = "BENCHPOD_API_KEY" WIRING_ENV = "BENCHPOD_WIRING" +#: ``local`` keeps the wiring lookup off embeddedci.com (see :attr:`BenchPod.wiring`). +WIRING_SOURCE_ENV = "BENCHPOD_WIRING_SOURCE" +#: Accepted ``wiring_source`` values: ``auto`` (the default) and ``local``. +WIRING_SOURCES = ("auto", "local") +#: Budget, in seconds, for finding a LAN/USB pod's stored wiring profile on embeddedci.com. +_WIRING_LOOKUP_TIMEOUT = 3.0 #: Levels of the firmware's parametric generator (it builds each waveform from 8-bit codes). _GENERATOR_MAX_CODE = 255 @@ -199,13 +205,16 @@ def __init__( lease_wait: float = DEFAULT_LEASE_WAIT, lease_ttl: int = DEFAULT_LEASE_TTL, wiring: Union[Wiring, Mapping[str, Any], str, "os.PathLike[str]", None] = None, + wiring_source: Optional[str] = None, ) -> None: """Open a BenchPod. ``connection`` is a host[:port], a serial device path, ``"usb"`` (auto-detect the USB console), ``"discover"`` (find one pod on the LAN via mDNS), or ``"embeddedci:"`` to drive a named device through embeddedci.com. When omitted ``BENCHPOD_CONNECTION`` is - used. Pass ``transport`` to inject a custom backend instead. + used, else the default connection ``benchpod-cli`` saved (``benchpod set-connection``); + ``"saved"`` names that one explicitly. Pass ``transport`` to inject a custom backend + instead. ``la_voltage`` (1.8 or 3.3 volts, falling back to ``BENCHPOD_LA_VOLTAGE``) selects the LA I/O-bank voltage right after connecting. The pod refuses every LA-bank operation — @@ -225,8 +234,16 @@ def __init__( TCP/serial connections never lease. ``wiring`` is the bench's :class:`~embeddedci.benchpod.wiring.Wiring` profile (a ``Wiring``, a - dict, or a ``.json``/``.toml`` path) — see :attr:`wiring`. + dict, or a ``.json``/``.toml`` path) — see :attr:`wiring`. ``wiring_source`` (or + ``BENCHPOD_WIRING_SOURCE``) is ``"auto"`` (default) or ``"local"``, which never asks + embeddedci.com for the stored profile. """ + source = (wiring_source if wiring_source is not None + else os.environ.get(WIRING_SOURCE_ENV, "")).strip().lower() or "auto" + if source not in WIRING_SOURCES: + what = "wiring_source" if wiring_source is not None else WIRING_SOURCE_ENV + raise ValueError(f"{what} must be 'auto' or 'local', got {source!r}") + self._wiring_source = source self._wiring: Optional[Wiring] = Wiring.coerce(wiring) if wiring is not None else None self.timeout = timeout self._lease: Optional[DeviceLease] = None @@ -421,8 +438,21 @@ def wiring(self) -> Wiring: """The bench's :class:`~embeddedci.benchpod.wiring.Wiring` profile: which DUT signal is on which LA channel, the target-power rail and the LA I/O voltage. - Resolved once: the ``wiring`` argument; else the ``BENCHPOD_WIRING`` file; else, for a cloud - device, the profile stored on embeddedci.com (edited in the web UI); else the defaults. + Resolved once, first match wins: + + 1. the ``wiring`` argument (or an assigned profile); + 2. the ``BENCHPOD_WIRING`` file; + 3. the profile stored on embeddedci.com for this pod (edited in the web UI's Wiring tab): + for a cloud connection, the named device; for a LAN or USB connection, the device the + pod reports it is registered as (``cloud_status``), when credentials are available + (``api_key``/``BENCHPOD_API_KEY``, or the ``benchpod login`` session) and the device is + on that account. ``source`` is then ``"server"``. Skipped with + ``wiring_source="local"`` / ``BENCHPOD_WIRING_SOURCE=local``; + 4. the defaults. + + The lookup in step 3 never raises: without credentials, with the pod not registered, the + server unreachable or the device on another account, it logs why and uses the defaults, + and on a LAN/USB connection it gives up after about 3 seconds. Channel arguments fall back to it — ``open_uart()``, ``capture_uart()``, ``flash()``, ``enable_i2c_sensor()``, the power rail — and :meth:`signal` / GPIO names come from it. Assign a profile to use another one for this connection (not saved; see :meth:`save_wiring`). @@ -439,17 +469,100 @@ def _resolve_wiring(self) -> Wiring: env = os.environ.get(WIRING_ENV, "").strip() if env: return Wiring.load(env) - if self._device_name: - api = self._try_server_api() - if api is not None: - try: - data = api.wiring_profile(api.resolve_device_id(self._device_name)) - return Wiring.from_dict(data.get("profile") or {}, source="server", strict=False) - except (BenchPodError, ValueError) as exc: - _log.warning("could not load the wiring profile of %s from embeddedci.com (%s); " - "using the defaults", self._device_name, exc) + if self._wiring_source == "local": + return Wiring.defaults() + if not self._device_name: # a LAN or USB pod + return self._local_pod_server_wiring() or Wiring.defaults() + api = self._try_server_api() + if api is not None: + try: + data = api.wiring_profile(api.resolve_device_id(self._device_name)) + return Wiring.from_dict(data.get("profile") or {}, source="server", strict=False) + except (BenchPodError, ValueError) as exc: + _log.warning("could not load the wiring profile of %s from embeddedci.com (%s); " + "using the defaults", self._device_name, exc) return Wiring.defaults() + def _wiring_credential(self, deadline: float) -> Optional[Tuple[Optional[str], Optional[Callable[[], str]]]]: + """``(api_key, token_provider)`` for the wiring lookup of a LAN/USB pod, or None without + any: the API key, else the caller's user token, else the ``benchpod login`` session.""" + if self._api_key: + return self._api_key.strip(), None + if self._cloud_user_token is not None: + return None, self._cloud_user_token + from . import cli_login + + if not cli_login.has_session(): + return None + from .cloud_auth import DEFAULT_API_BASE + + base = self._api_base or DEFAULT_API_BASE + return None, lambda: cli_login.access_token( + base, timeout=max(0.5, deadline - time.monotonic())) + + def _pod_cloud_device_id(self, deadline: float) -> str: + """The embeddedci.com device id the pod says it is registered as (``cloud_status``), or + "" when it is not registered. Asks with the lookup's short timeout.""" + cmd = getattr(self._transport, "command", None) + if cmd is None: + return "" + budget = max(0.5, deadline - time.monotonic()) + saved = {} + for attr in ("timeout", "dial_timeout"): + value = getattr(self._transport, attr, None) + if isinstance(value, (int, float)) and not isinstance(value, bool) and value > budget: + saved[attr] = value + setattr(self._transport, attr, budget) + try: + data = _dict(cmd({"cmd": "cloud_status"})) + finally: + for attr, value in saved.items(): + setattr(self._transport, attr, value) + if data.get("configured") is False: + return "" + return str(data.get("device_id") or "").strip() + + def _local_pod_server_wiring(self) -> Optional[Wiring]: + """The profile stored on embeddedci.com for the device a LAN/USB pod is registered as, or + None (logged) when there is no credential, no registration, or the lookup fails.""" + deadline = time.monotonic() + _WIRING_LOOKUP_TIMEOUT + try: + cred = self._wiring_credential(deadline) + if cred is None: + _log.info("wiring: no embeddedci.com credentials (BENCHPOD_API_KEY or `benchpod " + "login`), so the stored profile is not used; using the defaults") + return None + try: + device_id = self._pod_cloud_device_id(deadline) + except BenchPodError as exc: + # The USB console of current STM32 firmware has no cloud_status: expected, so quieter. + level = logging.INFO if type(self._transport).__name__ == "SerialTransport" \ + else logging.WARNING + _log.log(level, "wiring: could not ask the pod which embeddedci.com device it is " + "(%s); using the defaults", exc) + return None + if not device_id: + _log.info("wiring: the pod is not registered on embeddedci.com; using the defaults") + return None + from .server_api import ServerApi, ServerApiError + + api_key, token_provider = cred + api = ServerApi(api_base=self._api_base, api_key=api_key, token_provider=token_provider, + timeout=max(0.5, deadline - time.monotonic())) + try: + data = api.wiring_profile(device_id) + except ServerApiError as exc: + if exc.status in (401, 403, 404): + _log.warning("wiring: the pod's embeddedci.com device %s is not on this account " + "(HTTP %s); using the defaults", device_id, exc.status) + return None + raise + return Wiring.from_dict(data.get("profile") or {}, source="server", strict=False) + except Exception as exc: # the lookup is a convenience: never fail the caller over it + _log.warning("wiring: could not load the pod's profile from embeddedci.com (%s); " + "using the defaults", exc) + return None + def save_wiring(self, wiring: Union[Wiring, Mapping[str, Any], str, None] = None) -> Wiring: """Store a wiring profile for this cloud device on embeddedci.com (default: :attr:`wiring`) and use it. The server validates it too; a LAN pod keeps its profile in a file instead.""" diff --git a/packages/embeddedci/src/embeddedci/benchpod/connection.py b/packages/embeddedci/src/embeddedci/benchpod/connection.py index a9b4433..fefc9f5 100644 --- a/packages/embeddedci/src/embeddedci/benchpod/connection.py +++ b/packages/embeddedci/src/embeddedci/benchpod/connection.py @@ -9,9 +9,12 @@ * ``discover`` / ``mdns`` / ``auto`` -> find a single pod on the LAN via mDNS * ``embeddedci:`` -> drive a named device through embeddedci.com (an API key anywhere, or GitHub Actions OIDC) +* ``saved`` -> the default connection ``benchpod-cli`` saved + (``benchpod set-connection``, ``benchpod discover --save``) -Precedence is handled by the caller: an explicit argument wins over the -``BENCHPOD_CONNECTION`` environment variable. +:func:`resolve_connection` applies the precedence: an explicit argument, then the +``BENCHPOD_CONNECTION`` environment variable, then the connection ``benchpod-cli`` saved in +``~/.config/benchpod-cli/config.json``. """ from __future__ import annotations @@ -20,6 +23,7 @@ import re from dataclasses import dataclass +from .cli_config import SAVED_KEYWORD, config_path, saved_connection from .errors import ConnectionConfigError DEFAULT_PORT = 8080 @@ -84,6 +88,14 @@ def parse_connection(raw: str) -> ConnSpec: return ConnSpec(kind="embeddedci", device_name=name) # "usb" is the documented spelling; "serial" is the older one, still accepted # so connection strings written before the rename keep working. + if s.lower() == SAVED_KEYWORD: + saved = saved_connection() + if not saved: + raise ConnectionConfigError( + f"connection 'saved' asks for benchpod-cli's saved connection, but {config_path()} " + "holds none; save one with `benchpod set-connection `" + ) + return parse_connection(saved) if s.lower() in ("usb", "serial"): return ConnSpec(kind="serial", device="") if s.lower() in DISCOVER_KEYWORDS: @@ -119,12 +131,16 @@ def _discover_one() -> ConnSpec: def resolve_connection(connection: "str | None" = None) -> ConnSpec: - """Resolve a connection from an explicit value or the environment.""" + """Resolve a connection: the ``connection`` argument, else ``BENCHPOD_CONNECTION``, else the + default connection ``benchpod-cli`` saved (see :mod:`~embeddedci.benchpod.cli_config`).""" raw = connection if connection is not None else os.environ.get(ENV_VAR) + if not raw or not str(raw).strip(): + raw = saved_connection() if not raw or not str(raw).strip(): raise ConnectionConfigError( - "no BenchPod connection configured; pass connection=... or set " + "no BenchPod connection configured; pass connection=..., set " f"the {ENV_VAR} environment variable " - "(e.g. '192.168.1.213', '/dev/ttyACM0', 'usb', or 'embeddedci:')" + "(e.g. '192.168.1.213', '/dev/ttyACM0', 'usb', or 'embeddedci:'), " + "or save a default with `benchpod set-connection `" ) return parse_connection(str(raw)) diff --git a/packages/embeddedci/src/embeddedci/benchpod/pytest_plugin.py b/packages/embeddedci/src/embeddedci/benchpod/pytest_plugin.py index 2473198..37dbe0e 100644 --- a/packages/embeddedci/src/embeddedci/benchpod/pytest_plugin.py +++ b/packages/embeddedci/src/embeddedci/benchpod/pytest_plugin.py @@ -6,6 +6,11 @@ option > ``BENCHPOD_CONNECTION`` environment variable. With no connection configured the fixtures ``skip`` rather than fail, so the suite stays green without hardware. + +Unlike ``BenchPod()``, the plugin never falls back to the connection ``benchpod-cli`` saved on +this machine: a suite must not start driving hardware just because a developer once ran +``benchpod set-connection``. Opt in with ``--benchpod-connection=saved`` (or ``saved`` in the +ini option or ``BENCHPOD_CONNECTION``). """ from __future__ import annotations @@ -16,6 +21,7 @@ import pytest from .client import BenchPod +from .cli_config import SAVED_KEYWORD, config_path, saved_connection from .connection import ENV_VAR from .wiring import Wiring from .constants import PULL_OHMS as _PULL_OHMS @@ -116,8 +122,9 @@ def pytest_addoption(parser: "pytest.Parser") -> None: default=None, dest="benchpod_connection", help="BenchPod connection: host[:port], a serial device path, 'usb' (auto-detect), " - "'discover' (mDNS), or 'embeddedci:' to drive a named device through " - f"embeddedci.com. Falls back to the {ENV_VAR} env var.", + "'discover' (mDNS), 'embeddedci:' to drive a named device through " + "embeddedci.com, or 'saved' for the default connection benchpod-cli saved " + f"(never used unless asked for). Falls back to the {ENV_VAR} env var.", ) group.addoption( "--benchpod-api-base", @@ -160,7 +167,8 @@ def pytest_addoption(parser: "pytest.Parser") -> None: "--benchpod-wiring", action="store", default=None, dest="benchpod_wiring", help="Wiring profile file (.json or .toml): which DUT signal is on which LA channel. Normally " "set once by overriding the benchpod_wiring fixture in conftest.py; this flag wins over it. " - "Without either: BENCHPOD_WIRING, then a cloud device's stored profile, then the defaults.", + "Without either: BENCHPOD_WIRING, then the profile stored on embeddedci.com for the pod " + "(BENCHPOD_WIRING_SOURCE=local skips it), then the defaults.", ) group.addoption( "--benchpod-discover", @@ -260,6 +268,8 @@ def _resolve_connection(config: "pytest.Config") -> Optional[str]: or os.environ.get(ENV_VAR) ) if explicit: + if str(explicit).strip().lower() == SAVED_KEYWORD: + return saved_connection() # None (nothing saved) skips like no connection return explicit # No explicit target: opt into LAN auto-discovery. "discover" is turned into # an mDNS lookup by connection.parse_connection(). @@ -286,6 +296,9 @@ def benchpod_connection(pytestconfig: "pytest.Config") -> str: """The configured connection string, or skip the test if none is set.""" conn = _resolve_connection(pytestconfig) if not conn: + if _asked_for_saved(pytestconfig): + pytest.skip(f"--benchpod-connection=saved, but benchpod-cli saved no connection in " + f"{config_path()} (`benchpod set-connection `)") pytest.skip( "no BenchPod connection configured; pass --benchpod-connection=... " f"or set {ENV_VAR}" @@ -293,6 +306,12 @@ def benchpod_connection(pytestconfig: "pytest.Config") -> str: return conn +def _asked_for_saved(config: "pytest.Config") -> bool: + explicit = (config.getoption("benchpod_connection") or config.getini("benchpod_connection") + or os.environ.get(ENV_VAR) or "") + return str(explicit).strip().lower() == SAVED_KEYWORD + + LIFT_DAC_LIMITS_ENV = "BENCHPOD_LIFT_DAC_LIMITS" @@ -361,8 +380,10 @@ def benchpod_wiring(): return Wiring(uart_rx=3, uart_tx=4, signals=[Signal("TRIGGER", 9, "output")]) Tests then use names and defaults: ``benchpod.open_uart()``, ``benchpod.signal("TRIGGER")``. - The default returns ``None``: ``BENCHPOD_WIRING``, then a cloud device's profile stored on - embeddedci.com (edited in the web UI), then the defaults. ``--benchpod-wiring`` overrides it. + The default returns ``None``: ``BENCHPOD_WIRING``, then the pod's profile stored on + embeddedci.com (edited in the web UI; for a LAN or USB pod when an API key or the + ``benchpod login`` session is available; ``BENCHPOD_WIRING_SOURCE=local`` skips it), then the + defaults. ``--benchpod-wiring`` overrides it. """ return None diff --git a/packages/embeddedci/tests/api_surface.json b/packages/embeddedci/tests/api_surface.json index 4afa8d6..439c223 100644 --- a/packages/embeddedci/tests/api_surface.json +++ b/packages/embeddedci/tests/api_surface.json @@ -38,7 +38,7 @@ ] }, "BenchPod": { - "__init__": "(self, connection: 'Optional[str]' = None, *, la_voltage: 'Optional[float]' = None, timeout: 'float' = 30.0, transport: 'Optional[Transport]' = None, api_base: 'Optional[str]' = None, api_key: 'Optional[str]' = None, cloud_token: 'Optional[str]' = None, cloud_audience: 'Optional[str]' = None, cloud_user_token: 'Optional[Callable[[], str]]' = None, lease: 'bool' = True, lease_wait: 'float' = 600.0, lease_ttl: 'int' = 120, wiring: \"Union[Wiring, Mapping[str, Any], str, 'os.PathLike[str]', None]\" = None) -> 'None'", + "__init__": "(self, connection: 'Optional[str]' = None, *, la_voltage: 'Optional[float]' = None, timeout: 'float' = 30.0, transport: 'Optional[Transport]' = None, api_base: 'Optional[str]' = None, api_key: 'Optional[str]' = None, cloud_token: 'Optional[str]' = None, cloud_audience: 'Optional[str]' = None, cloud_user_token: 'Optional[Callable[[], str]]' = None, lease: 'bool' = True, lease_wait: 'float' = 600.0, lease_ttl: 'int' = 120, wiring: \"Union[Wiring, Mapping[str, Any], str, 'os.PathLike[str]', None]\" = None, wiring_source: 'Optional[str]' = None) -> 'None'", "adc_read": "(self, source: 'AdcSource' = 'ext') -> 'AdcReading'", "analog_path": "(self, path: 'AnalogPath') -> 'AnalogPathState'", "calibrate": "(self, source: 'CalibrateSource' = 'current_in') -> 'Calibration'", diff --git a/packages/embeddedci/tests/conftest.py b/packages/embeddedci/tests/conftest.py index 9e58830..210a600 100644 --- a/packages/embeddedci/tests/conftest.py +++ b/packages/embeddedci/tests/conftest.py @@ -22,3 +22,15 @@ def benchpod_la_voltage(): Change this to 1.8 when testing a 1V8 board (or pass --benchpod-la-voltage for one run). """ return 3.3 + + +@pytest.fixture(autouse=True) +def _isolated_benchpod_cli_config(request, tmp_path_factory, monkeypatch): + """Unit tests never see this machine's benchpod-cli config (its saved connection) or its + `benchpod login` session: both change what BenchPod() connects to and where its wiring comes + from. Hardware tests (e2e, examples) keep them.""" + path = str(request.node.fspath) + if "e2e" in os.path.basename(path) or os.sep + "e2e" + os.sep in path \ + or os.sep + "examples" + os.sep in path: + return + monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path_factory.mktemp("xdg"))) diff --git a/packages/embeddedci/tests/test_one_source.py b/packages/embeddedci/tests/test_one_source.py new file mode 100644 index 0000000..9616830 --- /dev/null +++ b/packages/embeddedci/tests/test_one_source.py @@ -0,0 +1,357 @@ +"""One wiring profile and one saved connection across the web UI, the CLI, the SDK and MCP. + +* A LAN/USB BenchPod reads the profile stored on embeddedci.com for the device the pod says it is + registered as, when credentials exist; every failure falls back to the defaults. +* ``BenchPod()`` without a connection uses the connection ``benchpod-cli`` saved; the pytest + plugin only does so when asked with ``saved``. +""" + +from __future__ import annotations + +import io +import json +import logging +import urllib.error +import urllib.request +from datetime import datetime, timedelta, timezone +from typing import Any, Dict, List + +import pytest + +from embeddedci.benchpod import BenchPod, Wiring +from embeddedci.benchpod import cli_config +from embeddedci.benchpod.connection import resolve_connection +from embeddedci.benchpod.errors import ConnectionConfigError, FirmwareError + +DEVICE_ID = "8f1c0d2e-0000-4000-8000-000000000001" +PROFILE = {"uart_rx": 3, "uart_tx": 6, "efuse": 2, "signals": [{"name": "READY", "la": 10}]} + + +@pytest.fixture(autouse=True) +def _clean_env(monkeypatch, tmp_path): + for var in ("BENCHPOD_CONNECTION", "BENCHPOD_WIRING", "BENCHPOD_WIRING_SOURCE", + "BENCHPOD_API_KEY", "BENCHPOD_API_BASE"): + monkeypatch.delenv(var, raising=False) + monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "xdg")) + + +def _cli_dir(tmp_path): + d = tmp_path / "xdg" / "benchpod-cli" + d.mkdir(parents=True, exist_ok=True) + return d + + +def _save_config(tmp_path, data: Any) -> None: + text = data if isinstance(data, str) else json.dumps(data) + (_cli_dir(tmp_path) / "config.json").write_text(text) + + +def _login(tmp_path, access: str = "login-access") -> None: + now = datetime.now(timezone.utc) + (_cli_dir(tmp_path) / "token.json").write_text(json.dumps({ + "access_token": access, "refresh_token": "r", + "access_expires_at": (now + timedelta(hours=1)).isoformat(), + "refresh_expires_at": (now + timedelta(days=7)).isoformat(), + })) + + +# -- the saved connection ------------------------------------------------------------------- + +def test_saved_connection_reads_the_cli_config(tmp_path): + assert cli_config.saved_connection() is None # no file + _save_config(tmp_path, {"connection": " 192.168.1.220 ", "last_serial": "/dev/x"}) + assert cli_config.saved_connection() == "192.168.1.220" + assert cli_config.config_path() == tmp_path / "xdg" / "benchpod-cli" / "config.json" + _save_config(tmp_path, {"bench_pod_addr": "10.0.0.5:8080"}) # the legacy field + assert cli_config.saved_connection() == "10.0.0.5:8080" + _save_config(tmp_path, {"connection": "usb", "bench_pod_addr": "10.0.0.5"}) + assert cli_config.saved_connection() == "usb" + + +@pytest.mark.parametrize("content", ["{not json", "[1, 2]", '{"connection": 5}', "{}", + '{"connection": "saved"}']) +def test_a_bad_or_empty_cli_config_means_nothing_saved(tmp_path, content): + _save_config(tmp_path, content) + assert cli_config.saved_connection() is None + + +def test_config_path_without_xdg_is_under_home(monkeypatch, tmp_path): + monkeypatch.delenv("XDG_CONFIG_HOME") + monkeypatch.setenv("HOME", str(tmp_path)) + assert cli_config.config_path() == tmp_path / ".config" / "benchpod-cli" / "config.json" + + +def test_connection_precedence_argument_env_saved(monkeypatch, tmp_path): + _save_config(tmp_path, {"connection": "10.0.0.3"}) + assert resolve_connection().addr == "10.0.0.3:8080" # saved + monkeypatch.setenv("BENCHPOD_CONNECTION", "10.0.0.2") + assert resolve_connection().addr == "10.0.0.2:8080" # env beats saved + assert resolve_connection("10.0.0.1").addr == "10.0.0.1:8080" # argument beats both + assert resolve_connection("saved").addr == "10.0.0.3:8080" # explicit keyword + + +def test_no_connection_anywhere_says_how_to_save_one(): + with pytest.raises(ConnectionConfigError, match="benchpod set-connection"): + resolve_connection() + with pytest.raises(ConnectionConfigError, match="holds none"): + resolve_connection("saved") + + +def test_benchpod_without_a_connection_opens_the_saved_one(monkeypatch, tmp_path): + import embeddedci.benchpod.client as client_mod + + _save_config(tmp_path, {"connection": "usb"}) + opened: List[Any] = [] + + def fake_open(spec, **kwargs): + opened.append(spec) + return Pod() + + monkeypatch.setattr(client_mod, "open_transport", fake_open) + BenchPod(lease=False).close() + assert opened[0].kind == "serial" and opened[0].device == "" + + +# -- the wiring profile of a LAN/USB pod ---------------------------------------------------- + +class Pod: + """A LAN pod that answers cloud_status.""" + + def __init__(self, cloud: Any = None) -> None: + self.cloud = {"configured": True, "state": "connected", "device_id": DEVICE_ID} \ + if cloud is None else cloud + self.commands: List[dict] = [] + self.timeout = 30.0 + self.dial_timeout = 10.0 + self.seen_timeouts: List[tuple] = [] + + def status(self) -> Dict[str, Any]: + return {"board": "stm32h563"} + + def ping(self) -> Any: + return "pong" + + def command(self, req: dict) -> Any: + self.commands.append(req) + if req["cmd"] == "cloud_status": + self.seen_timeouts.append((self.timeout, self.dial_timeout)) + if isinstance(self.cloud, Exception): + raise self.cloud + return self.cloud + return {} + + def close(self) -> None: + pass + + +class Http: + """Stands in for urllib: answers GET .../wiring/profile and records each request.""" + + def __init__(self, status: int = 200, body: Any = None, error: Exception = None) -> None: + self.status = status + self.body = {"profile": PROFILE, "stored": True} if body is None else body + self.error = error + self.requests: List[tuple] = [] + + def urlopen(self, req, timeout=None): + self.requests.append((req.get_method(), req.full_url, req.get_header("Authorization"), timeout)) + if self.error is not None: + raise self.error + if self.status >= 400: + raise urllib.error.HTTPError(req.full_url, self.status, "x", {}, + io.BytesIO(json.dumps({"error": "nope"}).encode())) + return _Resp(json.dumps(self.body).encode(), self.status) + + +class _Resp: + def __init__(self, body: bytes, status: int) -> None: + self._body, self.status = body, status + + def read(self) -> bytes: + return self._body + + def __enter__(self) -> "_Resp": + return self + + def __exit__(self, *exc: object) -> None: + pass + + +@pytest.fixture +def http(monkeypatch): + h = Http() + monkeypatch.setattr(urllib.request, "urlopen", h.urlopen) + return h + + +def test_lan_pod_uses_the_stored_profile_with_an_api_key(monkeypatch, http): + monkeypatch.setenv("BENCHPOD_API_KEY", "eci_test") + monkeypatch.setenv("BENCHPOD_API_BASE", "https://staging.example") + pod = Pod() + bp = BenchPod(transport=pod, lease=False) + assert bp.wiring.source == "server" + assert bp.wiring.uart_rx == 3 and bp.wiring.efuse == 2 and bp.signal("READY").la == 10 + method, url, auth, timeout = http.requests[0] + assert (method, url, auth) == ( + "GET", f"https://staging.example/api/benchpod/devices/{DEVICE_ID}/wiring/profile", "ApiKey eci_test") + assert 0 < timeout <= 3.0 + # cloud_status went out with the short budget, and the transport's timeouts came back. + assert pod.seen_timeouts and all(t <= 3.0 and d <= 3.0 for t, d in pod.seen_timeouts) + assert (pod.timeout, pod.dial_timeout) == (30.0, 10.0) + bp.wiring # resolved once + assert len(http.requests) == 1 + + +def test_lan_pod_uses_the_benchpod_login_session(tmp_path, http): + _login(tmp_path, access="tok-123") + bp = BenchPod(transport=Pod(), lease=False) + assert bp.wiring.source == "server" + assert http.requests[0][1].startswith("https://www.embeddedci.com/api/benchpod/devices/") + assert http.requests[0][2] == "Bearer tok-123" + + +def test_a_callers_user_token_is_used_for_the_lookup(http): + bp = BenchPod(transport=Pod(), lease=False, cloud_user_token=lambda: "mcp-token") + assert bp.wiring.source == "server" and http.requests[0][2] == "Bearer mcp-token" + + +def test_no_credentials_means_defaults_without_asking_anyone(http, caplog): + pod = Pod() + with caplog.at_level(logging.INFO, logger="embeddedci.benchpod"): + w = BenchPod(transport=pod, lease=False).wiring + assert w == Wiring() and w.source == "defaults" + assert pod.commands == [] and http.requests == [] + assert "no embeddedci.com credentials" in caplog.text + + +@pytest.mark.parametrize("cloud", [{"configured": False, "state": "unconfigured"}, + {"configured": True, "device_id": ""}, {}]) +def test_an_unregistered_pod_gets_the_defaults(monkeypatch, http, cloud): + monkeypatch.setenv("BENCHPOD_API_KEY", "eci_test") + assert BenchPod(transport=Pod(cloud), lease=False).wiring.source == "defaults" + assert http.requests == [] + + +def test_a_pod_that_refuses_cloud_status_gets_the_defaults(monkeypatch, http, caplog): + monkeypatch.setenv("BENCHPOD_API_KEY", "eci_test") + pod = Pod(FirmwareError("cloud_status: unknown command")) + with caplog.at_level(logging.WARNING, logger="embeddedci.benchpod"): + assert BenchPod(transport=pod, lease=False).wiring.source == "defaults" + assert http.requests == [] and "could not ask the pod" in caplog.text + assert (pod.timeout, pod.dial_timeout) == (30.0, 10.0) + + +@pytest.mark.parametrize("status, phrase", [(404, "not on this account"), (403, "not on this account"), + (500, "HTTP 500")]) +def test_server_errors_fall_back_to_the_defaults(monkeypatch, http, caplog, status, phrase): + monkeypatch.setenv("BENCHPOD_API_KEY", "eci_test") + http.status = status + with caplog.at_level(logging.WARNING, logger="embeddedci.benchpod"): + assert BenchPod(transport=Pod(), lease=False).wiring.source == "defaults" + assert phrase in caplog.text + + +@pytest.mark.parametrize("error", [urllib.error.URLError("offline"), TimeoutError("timed out"), + OSError("network is unreachable")]) +def test_an_unreachable_server_falls_back_to_the_defaults(monkeypatch, http, error): + monkeypatch.setenv("BENCHPOD_API_KEY", "eci_test") + http.error = error + assert BenchPod(transport=Pod(), lease=False).wiring.source == "defaults" + + +def test_an_expired_login_falls_back_to_the_defaults(tmp_path, http): + past = datetime.now(timezone.utc) - timedelta(days=1) + (_cli_dir(tmp_path) / "token.json").write_text(json.dumps({ + "access_token": "a", "refresh_token": "r", + "access_expires_at": past.isoformat(), "refresh_expires_at": past.isoformat()})) + assert BenchPod(transport=Pod(), lease=False).wiring.source == "defaults" + assert http.requests == [] + + +def test_a_malformed_stored_profile_falls_back_to_the_defaults(monkeypatch, http): + monkeypatch.setenv("BENCHPOD_API_KEY", "eci_test") + http.body = {"profile": {"uart_rx": 99}} + assert BenchPod(transport=Pod(), lease=False).wiring.source == "defaults" + + +def test_explicit_wiring_and_the_wiring_file_still_win(monkeypatch, http, tmp_path): + monkeypatch.setenv("BENCHPOD_API_KEY", "eci_test") + pod = Pod() + assert BenchPod(transport=pod, lease=False, wiring={"uart_rx": 7}).wiring.uart_rx == 7 + p = tmp_path / "bench.json" + p.write_text(json.dumps({"uart_rx": 8})) + monkeypatch.setenv("BENCHPOD_WIRING", str(p)) + w = BenchPod(transport=pod, lease=False).wiring + assert w.uart_rx == 8 and w.source == "file" + assert pod.commands == [] and http.requests == [] + + +def test_wiring_source_local_skips_the_lookup(monkeypatch, http): + monkeypatch.setenv("BENCHPOD_API_KEY", "eci_test") + pod = Pod() + assert BenchPod(transport=pod, lease=False, wiring_source="local").wiring.source == "defaults" + monkeypatch.setenv("BENCHPOD_WIRING_SOURCE", "LOCAL") + assert BenchPod(transport=pod, lease=False).wiring.source == "defaults" + assert pod.commands == [] and http.requests == [] + # The argument wins over the variable. + assert BenchPod(transport=pod, lease=False, wiring_source="auto").wiring.source == "server" + + +def test_wiring_source_local_also_skips_a_cloud_devices_profile(monkeypatch): + bp = BenchPod(transport=Pod(), lease=False, wiring_source="local") + bp._device_name = "bench-1" + monkeypatch.setattr(bp, "_try_server_api", lambda: pytest.fail("asked the server")) + assert bp.wiring.source == "defaults" + + +def test_a_bad_wiring_source_is_rejected(monkeypatch): + with pytest.raises(ValueError, match="wiring_source"): + BenchPod(transport=Pod(), lease=False, wiring_source="cloud") + monkeypatch.setenv("BENCHPOD_WIRING_SOURCE", "server-only") + with pytest.raises(ValueError, match="BENCHPOD_WIRING_SOURCE"): + BenchPod(transport=Pod(), lease=False) + + +# -- the pytest plugin never picks up the saved connection by itself ---------------------- + +PLUGIN_CONFTEST = """ +import pytest +from embeddedci.benchpod import pytest_plugin + +made = {} + +class FakePod: + def __init__(self, connection, **kwargs): + made["connection"] = connection + self.capabilities = type("C", (), {"la_pins": False})() + def close(self): + pass + +pytest_plugin.BenchPod = FakePod +""" + +PLUGIN_TEST = """ +import conftest + +def test_session(benchpod): + assert conftest.made["connection"] == "10.9.8.7" +""" + + +def test_pytest_plugin_ignores_the_saved_connection_unless_asked(pytester, tmp_path): + _save_config(tmp_path, {"connection": "10.9.8.7"}) + pytester.makeconftest(PLUGIN_CONFTEST) + pytester.makepyfile(PLUGIN_TEST) + result = pytester.runpytest("-p", "no:cacheprovider", "-rs") + result.assert_outcomes(skipped=1) + result.stdout.fnmatch_lines(["*no BenchPod connection configured*"]) + result = pytester.runpytest("-p", "no:cacheprovider", "--benchpod-connection=saved") + result.assert_outcomes(passed=1) + + +def test_pytest_plugin_saved_without_a_saved_connection_skips(pytester): + pytester.makeconftest(PLUGIN_CONFTEST) + pytester.makepyfile(PLUGIN_TEST) + result = pytester.runpytest("-p", "no:cacheprovider", "-rs", "--benchpod-connection=saved") + result.assert_outcomes(skipped=1) + result.stdout.fnmatch_lines(["*benchpod-cli saved no connection*"])