diff --git a/Makefile b/Makefile index 9313219..16c65f3 100644 --- a/Makefile +++ b/Makefile @@ -1,7 +1,7 @@ # Developer shortcuts. Unit tests need no hardware; the e2e targets drive a real BenchPod. # # make test -# make e2e POD=192.168.1.215 [FIRMWARE=path/to/scenario-sensors.elf] [USB=/dev/cu.usbmodem…] [REV=v3] +# make e2e POD=192.168.1.215 [FIRMWARE=path/to/scenario-sensors.elf] [USB=/dev/cu.usbmodem…] [REV=v3] [BOARD=digital] # BENCHPOD_API_KEY=eci_… make e2e-cloud CLOUD_DEVICE=benchpod-v2.0.0 # # The board's LA voltage is set once in packages/embeddedci/tests/conftest.py; the bench wiring in @@ -13,13 +13,17 @@ FIRMWARE ?= USB ?= CLOUD_DEVICE ?= REV ?= +BOARD ?= # 1 (default): clear the pod's DAC output limits for the run and restore them afterwards # (pytest_plugin._benchpod_lift_dac_limits). LIFT_DAC_LIMITS=0 when an output stage is live. LIFT_DAC_LIMITS ?= 1 FIRMWARE_OPT = $(if $(FIRMWARE),--benchpod-firmware=$(abspath $(FIRMWARE)),) # REV=v3 makes a pod that reports another board revision fail instead of taking the v2 branches. -REV_ENV = $(if $(REV),BENCHPOD_E2E_BOARD_REV=$(REV),) BENCHPOD_LIFT_DAC_LIMITS=$(LIFT_DAC_LIMITS) +# BOARD=analog|digital fails the run on the other board: a digital pod skips every analog test, so +# an analog pod whose front end went undetected would otherwise pass green. +REV_ENV = $(if $(REV),BENCHPOD_E2E_BOARD_REV=$(REV),) $(if $(BOARD),BENCHPOD_EXPECT_BOARD=$(BOARD),) \ + BENCHPOD_LIFT_DAC_LIMITS=$(LIFT_DAC_LIMITS) .PHONY: test e2e e2e-cloud check-sdk @@ -36,7 +40,7 @@ check-sdk: @$(PYTHON) scripts/check_sdk.py e2e: check-sdk - @test -n "$(POD)" || { echo "usage: make e2e POD= [FIRMWARE=app.elf] [USB=/dev/…] [REV=v3]"; exit 2; } + @test -n "$(POD)" || { echo "usage: make e2e POD= [FIRMWARE=app.elf] [USB=/dev/…] [REV=v3] [BOARD=digital]"; exit 2; } # tests/examples too: it is the first thing a new user runs, so it must not rot. $(REV_ENV) BENCHPOD_E2E_USB="$(USB)" $(PYTHON) -m pytest -v -rs \ packages/embeddedci/tests/e2e packages/embeddedci/tests/examples \ @@ -48,6 +52,6 @@ e2e: check-sdk e2e-cloud: @test -n "$(CLOUD_DEVICE)" || { echo "usage: BENCHPOD_API_KEY=eci_… make e2e-cloud CLOUD_DEVICE="; exit 2; } - BENCHPOD_E2E_CLOUD_DEVICE="$(CLOUD_DEVICE)" BENCHPOD_LIFT_DAC_LIMITS=$(LIFT_DAC_LIMITS) \ + BENCHPOD_E2E_CLOUD_DEVICE="$(CLOUD_DEVICE)" $(REV_ENV) \ $(PYTHON) -m pytest -v -rs packages/embeddedci/tests/e2e/test_e2e_cloud.py \ --benchpod-connection=embeddedci:$(CLOUD_DEVICE) diff --git a/packages/embeddedci-mcp/tests/test_e2e_mcp.py b/packages/embeddedci-mcp/tests/test_e2e_mcp.py index abfdb5d..c3fe204 100644 --- a/packages/embeddedci-mcp/tests/test_e2e_mcp.py +++ b/packages/embeddedci-mcp/tests/test_e2e_mcp.py @@ -4,7 +4,9 @@ [--benchpod-firmware=] Skips without a connection. The DUT part expects the examples/scenario-sensors-stm32 app with its -UART on BENCHPOD_E2E_UART_RX/TX (default LA3/LA4) and SWD on LA11/LA12. +UART on BENCHPOD_E2E_UART_RX/TX (default LA3/LA4) and SWD on LA11/LA12. On the digital-only board +the analog tests skip and test_digital_board_refuses_the_analog_tools runs instead; +BENCHPOD_EXPECT_BOARD=analog|digital fails the run on the other board. """ from __future__ import annotations @@ -30,15 +32,26 @@ def _pin(name: str, default: int) -> int: def connected(benchpod_connection): status = call("connect", connection=benchpod_connection, la_voltage=LA_VOLTAGE) assert status["connected"] and status["la_voltage"] == LA_VOLTAGE, status + board = "digital" if status["capabilities"].get("analog") is False else "analog" + want = os.environ.get("BENCHPOD_EXPECT_BOARD", "").strip().lower() + if want and want != board: + pytest.fail(f"the pod reports the {board} board, the bench expects {want} (BENCHPOD_EXPECT_BOARD)") yield status - for tool, args in (("dac_stop", {}), ("analog_path", {"path": "off"}), ("can_close", {}), - ("disable_i2c_sensor", {}), ("disconnect", {})): + cleanup = [("dac_stop", {}), ("can_close", {}), ("disable_i2c_sensor", {}), ("disconnect", {})] + if board == "analog": + cleanup.insert(1, ("analog_path", {"path": "off"})) + for tool, args in cleanup: try: call(tool, **args) except ToolError: pass +def _require_analog(connected) -> None: + if connected["capabilities"].get("analog") is False: + pytest.skip("digital-only board: no analog front end (DAC, ADC, 4-20 mA)") + + def test_status_and_typed_state(connected): assert connected["capabilities"]["board"] # A bench pod on a computer's USB port is told about its 500 mA port; nothing else may warn. @@ -49,6 +62,7 @@ def test_status_and_typed_state(connected): def test_analog_round_trip(connected): + _require_analog(connected) out = call("dac_output", path="5v", volts=2.5) assert out["path"] == "5v" and abs(out["voltage"] - 2.5) < 0.05 reading = call("adc_read", source="cal1") @@ -59,6 +73,7 @@ def test_current_loop_terminals(connected): """The 4-20 mA tools with nothing wired: a held current and a waveform in amps reach the DAC (read back on the 5 V loopback), and the open input is captured in amps with the logic channels. Loop current itself needs the bench fixture.""" + _require_analog(connected) if not connected["capabilities"].get("current_out"): pytest.skip("this pod's firmware has no current_out command") if os.environ.get("BENCHPOD_E2E_J8_WIRED") == "1": @@ -93,6 +108,7 @@ def test_current_loop_terminals(connected): def test_generate_capture_summary_and_replay(connected): + _require_analog(connected) pytest.importorskip("numpy") call("analog_path", path="cal1") call("generate", waveform="sine", freq_hz=200, amplitude=0.8, offset=1.5, route=False) @@ -104,6 +120,18 @@ def test_generate_capture_summary_and_replay(connected): assert replay["samples"] == 2000 and not replay["deep"] +def test_digital_board_refuses_the_analog_tools(connected): + if connected["capabilities"].get("analog") is not False: + pytest.skip("analog board: the digital-only checks do not apply") + assert not connected["capabilities"].get("scope") + for tool, args in (("dac_output", {"path": "5v", "volts": 1.0}), ("adc_read", {"source": "cal1"}), + ("current_out", {"current": 0.012}), ("capture_adc", {"samples": 1024})): + with pytest.raises(ToolError, match=r"UnsupportedFeatureError: .*digital-only board"): + call(tool, **args) + call("dac_stop") # the cleanup every agent runs still works + assert "internal" in call("power_status") + + def test_logic_capture_and_decode(connected): la = call("capture_la", samples=8192, sample_rate_hz=1_000_000) assert la["samples"] == 8192 and len(la["channels"]) == 14 diff --git a/packages/embeddedci-openhtf/tests/test_e2e_openhtf.py b/packages/embeddedci-openhtf/tests/test_e2e_openhtf.py index 1821260..96781fd 100644 --- a/packages/embeddedci-openhtf/tests/test_e2e_openhtf.py +++ b/packages/embeddedci-openhtf/tests/test_e2e_openhtf.py @@ -45,7 +45,16 @@ def _run(*phases): return record, failed +def _require_analog(connection: str) -> None: + from embeddedci.benchpod import BenchPod + + with BenchPod(connection, lease=False) as probe: + if not probe.can_analog: + pytest.skip("digital-only board: no analog front end (DAC, ADC, 4-20 mA)") + + def test_analog_phases_against_the_pod(benchpod_connection): + _require_analog(benchpod_connection) bench = benchpod_plug(benchpod_connection, la_voltage=LA_VOLTAGE) record, failed = _run( dac_output_phase(bench, path="5v", volts=2.5), @@ -62,6 +71,8 @@ def test_current_loop_phases_against_the_pod(benchpod_connection): from embeddedci.benchpod import BenchPod with BenchPod(benchpod_connection, lease=False) as probe: + if not probe.can_analog: + pytest.skip("digital-only board: no analog front end (DAC, ADC, 4-20 mA)") if not probe.capabilities.current_out: pytest.skip("this pod's firmware has no current_out command") if os.environ.get("BENCHPOD_E2E_J8_WIRED") == "1": diff --git a/packages/embeddedci/CHANGELOG.md b/packages/embeddedci/CHANGELOG.md index edf9afc..3fb145f 100644 --- a/packages/embeddedci/CHANGELOG.md +++ b/packages/embeddedci/CHANGELOG.md @@ -1,5 +1,13 @@ # Changelog — `embeddedci` +## Unreleased + +- `@pytest.mark.benchpod_capability("analog")` skips a test on the digital-only board and runs it + on firmware that does not announce the flag, as `BenchPod.can_analog` does (it used to skip + there too). +- On the digital-only board, `current_out` and `calibrate` raise `UnsupportedFeatureError` that + names the missing analog front end (`feature="analog"`) instead of asking for newer firmware. + ## 2.12.0 - Firmware 3.9.0 keeps ST's USB id `0483:5740` (pre-release builds used `1209:0001`); auto-detect diff --git a/packages/embeddedci/README.md b/packages/embeddedci/README.md index 1218e7f..78e4d68 100644 --- a/packages/embeddedci/README.md +++ b/packages/embeddedci/README.md @@ -451,6 +451,9 @@ nothing changes. `dac_deep_replay`, `dac_control_loop`, `dac_loop_sources`, `dac_cotrig`, …). For the image-bound `dac_control_loop` and `dac_deep_replay` it switches the pod's gateware image instead of skipping, when the pod carries both images (see [Gateware images](#gateware-images)). + `benchpod_capability("analog")` skips only on the digital-only board (no DAC, ADC or 4-20 mA); + firmware that does not announce the flag ran on analog boards only and counts as analog, like + `BenchPod.can_analog`. ## Wiring profile diff --git a/packages/embeddedci/src/embeddedci/benchpod/client.py b/packages/embeddedci/src/embeddedci/benchpod/client.py index 12e276a..ec83563 100644 --- a/packages/embeddedci/src/embeddedci/benchpod/client.py +++ b/packages/embeddedci/src/embeddedci/benchpod/client.py @@ -138,6 +138,11 @@ _log = logging.getLogger("embeddedci.benchpod") +#: Capability flags only the analog board announces: the digital-only board lacks them whatever its +#: firmware (cloud_caps_hw.c). +_ANALOG_FLAGS = frozenset({"scope", "dac_replay", "dac_deep_replay", "dac_control_loop", "dac_cotrig", + "dac_loop_sources", "dac_loop_input_map", "calibrate", "current_out"}) + #: The capability flag and FPGA_FEATURES bit each gateware image carries. _IMAGE_FLAG = {FpgaImage.LOOP: "dac_control_loop", FpgaImage.DEEP_REPLAY: "dac_deep_replay"} _IMAGE_FEATURE = {FpgaImage.LOOP: 0x01, FpgaImage.DEEP_REPLAY: 0x02} @@ -426,6 +431,12 @@ def can_analog(self) -> bool: def _require_capability(self, flag: str, what: str) -> None: if not self.supports(flag): fw = self.capabilities.firmware_version or "unknown" + if flag in _ANALOG_FLAGS and not self.can_analog: + # The digital board drops these flags: newer firmware would not bring them back. + raise UnsupportedFeatureError( + f"{what} need the analog front end: this is the digital-only board (no DAC, " + f"ADC or 4-20 mA)", feature="analog", + firmware_version=self.capabilities.firmware_version) raise UnsupportedFeatureError( f"{what} need newer pod firmware (the {flag!r} capability is missing; " f"this pod runs firmware {fw})", diff --git a/packages/embeddedci/src/embeddedci/benchpod/pytest_plugin.py b/packages/embeddedci/src/embeddedci/benchpod/pytest_plugin.py index 5c6673d..7aab49d 100644 --- a/packages/embeddedci/src/embeddedci/benchpod/pytest_plugin.py +++ b/packages/embeddedci/src/embeddedci/benchpod/pytest_plugin.py @@ -222,7 +222,8 @@ def pytest_configure(config: "pytest.Config") -> None: "markers", "benchpod_capability(name): skip the test unless the connected device advertises the " "given capability (e.g. 'scope', 'analyzer', 'dac_replay'); 'dac_control_loop' and " - "'dac_deep_replay' switch the gateware image when the pod carries both.", + "'dac_deep_replay' switch the gateware image when the pod carries both; 'analog' skips " + "on the digital-only board (firmware that does not say counts as analog).", ) @@ -239,6 +240,7 @@ def _benchpod_capability_gate(request: "pytest.FixtureRequest") -> None: ``benchpod`` device and skips on any capability it lacks — except an image-bound one (``dac_control_loop`` / ``dac_deep_replay``) on a pod that carries both gateware images, which is switched to the right image instead, as the SDK does for ``control_loop`` / ``replay``. + ``analog`` follows :attr:`BenchPod.can_analog`: only the digital-only board skips it. """ if request.node.get_closest_marker("hardware") and not _resolve_connection(request.config): pytest.skip(f"hardware test: no BenchPod connection configured (--benchpod-connection / {ENV_VAR})") @@ -251,6 +253,10 @@ def _benchpod_capability_gate(request: "pytest.FixtureRequest") -> None: caps = pod.capabilities for mark in marks: for name in map(str, mark.args): + if name == "analog": + if not pod.can_analog: + pytest.skip("digital-only board: no analog front end (DAC, ADC, 4-20 mA)") + continue if getattr(caps, name, False): continue image = _IMAGE_CAPABILITIES.get(name) @@ -332,7 +338,8 @@ def _benchpod_lift_dac_limits(pytestconfig: "pytest.Config") -> Iterator[None]: kwargs = _cloud_kwargs(pytestconfig) try: with BenchPod(conn, **kwargs) as pod: - limits = pod.command({"cmd": "dac_limits"}) + # The digital-only board has no DAC (and refuses dac_limits): nothing to lift. + limits = pod.command({"cmd": "dac_limits"}) if pod.can_analog else None if isinstance(limits, dict) and limits.get("enabled"): pod.command({"cmd": "dac_limits", "enabled": False}) saved = limits diff --git a/packages/embeddedci/tests/e2e/README.md b/packages/embeddedci/tests/e2e/README.md index 6bb2496..746b8cd 100644 --- a/packages/embeddedci/tests/e2e/README.md +++ b/packages/embeddedci/tests/e2e/README.md @@ -11,6 +11,10 @@ make e2e POD=192.168.1.215 FIRMWARE=../examples/scenario-sensors-stm32/build/sce # a rev3 (v3) pod: REV=v3 makes a misdetected board fail instead of passing as v2 make e2e POD=192.168.1.215 FIRMWARE=… REV=v3 +# a digital-only pod (no analog board): BOARD=digital fails the run on an analog pod and the +# reverse, so a front end that went undetected cannot skip the analog tests and pass green +make e2e POD=192.168.1.215 FIRMWARE=… BOARD=digital + # add the USB console and the cloud tiers make e2e POD=192.168.1.215 FIRMWARE=… USB=/dev/cu.usbmodem357D377F31331 BENCHPOD_API_KEY=eci_… make e2e-cloud CLOUD_DEVICE=benchpod-v2.0.0 @@ -22,6 +26,7 @@ BENCHPOD_API_KEY=eci_… make e2e-cloud CLOUD_DEVICE=benchpod-v2.0.0 | gpio | `test_e2e_gpio.py` | `--benchpod-connection`; firmware with `la_pins`, gateware v35+ for triggers | GPIO output/open-drain read-back, pin ownership (UART refused on a GPIO pin and the reverse, release), bias-resistor conflicts both ways, step trains on GPIO pins, LA-voltage lock (rev3), wiring signal names with active-low, rising/level triggers, trigger timeout, triggered ADC capture, FPGA pulse widths measured from the trigger | | rev3 | `test_e2e_rev3.py` | `--benchpod-connection` on a rev3 pod; `BENCHPOD_E2E_NRST_LA` for the wire checks | board strap (v3, ~3.0 V), 1.8 V bank read back from the mux status pin, pull-ups released and refused at 1.8 V, GPIO + capture on the 1.8 V bank, reset pin hold/release/pulse, pulses capped at 1 s, and with the jumper: the reset line open-drain to the bank rail at 3.3 V and 1.8 V and the pulse width on the wire; USB-C CC report consistency (and a cable seen with `BENCHPOD_E2E_USB_CC=1`) | | dut | `test_e2e_dut.py` | + `--benchpod-firmware` and the wired board | SWD flash, boot banner via power cycle, power monitor + eFuse see the board, interactive console (help/status/reset), firmware detects the emulated BMP280, firmware reports a missing sensor, `i2c_sensor_capture` sees the boot probe, a 4 s logic capture of the boot decodes the chip-id read (and every transaction the pod served) and the UART banner; power profiles of a boot, of a delayed power-on and of an unpowered rail; with `BENCHPOD_E2E_NRESET=1` (rev3): flash under reset, a reset pulse reboots the DUT with the rail up, a held reset keeps it quiet until released | +| digital | `test_e2e_digital.py` | `--benchpod-connection` on a digital-only pod (skips on the analog board) | status says `analog: false` and drops the analog capabilities; every analog call (DAC, ADC, 4-20 mA, ADC capture, correlated capture with ADC samples, generate, replay) fails fast with `UnsupportedFeatureError` (`feature="analog"`), raw commands carry the pod's own refusal, `dac_stop` still works and the pod still answers and captures LA data afterwards | | usb | `test_e2e_usb.py` | `BENCHPOD_E2E_USB` | the text console: status/ping/LA voltage/power, fail-fast hints for everything else | | cloud | `test_e2e_cloud.py` | `BENCHPOD_E2E_CLOUD_DEVICE` + `BENCHPOD_API_KEY` | lease, command channel, tunnel captures, commands during a UART session, waveform library save → replay → delete | @@ -29,6 +34,17 @@ The MCP server and the OpenHTF plug have their own e2e files, run by `make e2e` `packages/embeddedci-mcp/tests/test_e2e_mcp.py` and `packages/embeddedci-openhtf/tests/test_e2e_openhtf.py`. +## Digital-only board + +The digital-only BenchPod (digital-v3, digital-v3r2) has no analog front end: its firmware reports +`analog: false` and refuses the analog commands. Tests that need the DAC, ADC or 4-20 mA carry +`ANALOG` (`@pytest.mark.benchpod_capability("analog")`, from `e2e_helpers.py`) or call +`require_analog(pod)` when the pod is not the `benchpod` fixture (the cloud tier); they skip there. +Tests that mix analog and digital checks are split, so the digital half still runs. The MCP and +OpenHTF e2e files skip their analog tests the same way. Set `BENCHPOD_EXPECT_BOARD=analog` or +`digital` (`make e2e BOARD=…`) on a bench whose board you know: a pod reporting the other one fails +the run. + ## Bench configuration The LA voltage (3.3 V) is set once in `tests/conftest.py`. The wiring defaults in `conftest.py` @@ -41,6 +57,7 @@ match the EmbeddedCI bench — a NUCLEO-F446RE running `examples/scenario-sensor | I2C SDA / SCL | LA2 / LA1 | `BENCHPOD_E2E_I2C_SDA` / `_I2C_SCL` | | SWCLK / SWDIO | LA11 / LA12 | `BENCHPOD_E2E_SWCLK` / `_SWDIO` | | PCB revision the pod must report | any | `BENCHPOD_E2E_BOARD_REV=v3` (`make e2e REV=v3`) | +| Board the pod must report (analog or digital-only) | any | `BENCHPOD_EXPECT_BOARD=digital` (`make e2e BOARD=digital`) | | DUT reset on the pod's reset pin (J1 pin 22) | no | `BENCHPOD_E2E_NRESET=1` | | LA channel jumpered to the reset pin | none | `BENCHPOD_E2E_NRST_LA` (e.g. 8) | | Pod's USB-C port plugged into a host | no | `BENCHPOD_E2E_USB_CC=1` | diff --git a/packages/embeddedci/tests/e2e/conftest.py b/packages/embeddedci/tests/e2e/conftest.py index 4dd9694..c1ee3c1 100644 --- a/packages/embeddedci/tests/e2e/conftest.py +++ b/packages/embeddedci/tests/e2e/conftest.py @@ -3,6 +3,12 @@ The LA voltage comes from ``tests/conftest.py`` (3.3 V). The bench wiring below matches the EmbeddedCI bench (NUCLEO-F446RE running examples/scenario-sensors-stm32); override any of it with ``BENCHPOD_E2E_*`` environment variables for another bench. + +On the digital-only board (no analog front end) the analog tests skip: they carry +``ANALOG`` (``@pytest.mark.benchpod_capability("analog")``) or call ``require_analog`` (both in +e2e_helpers.py). +``BENCHPOD_EXPECT_BOARD=analog|digital`` makes the run fail when the pod reports the other board, +so a pod whose analog section went undetected cannot skip half the suite and pass green. """ from __future__ import annotations @@ -15,6 +21,8 @@ from embeddedci.benchpod import BenchPod, BenchPodError, FpgaImage +from e2e_helpers import check_expected_board, require_analog + def pytest_report_header(config): import embeddedci @@ -81,7 +89,8 @@ def bench() -> Bench: def quiesce(pod: BenchPod) -> None: """Stop every output a test may have left running (never raises).""" - for step in (pod.dac_stop, lambda: pod.analog_path("off"), pod.disable_i2c_sensor, + for step in (pod.dac_stop, lambda: pod.analog_path("off") if pod.can_analog else None, + pod.disable_i2c_sensor, pod.can_respond_clear, pod.can_disable, lambda: pod.release_gpio() if pod.capabilities.la_pins else None): try: @@ -94,6 +103,7 @@ def quiesce(pod: BenchPod) -> None: def pod(benchpod: BenchPod): """The session's pod; the gateware image it started on is restored at the end.""" started_on_loop = benchpod.refresh_capabilities().dac_control_loop + check_expected_board(benchpod) yield benchpod quiesce(benchpod) try: @@ -111,6 +121,7 @@ def _leave_the_pod_quiet(request): def _on_image(pod: BenchPod, image: FpgaImage, flag: str) -> BenchPod: + require_analog(pod) # both images are for the DAC; the digital board has neither feature if not getattr(pod.refresh_capabilities(), flag): try: pod.fpga_image(image) diff --git a/packages/embeddedci/tests/e2e/e2e_helpers.py b/packages/embeddedci/tests/e2e/e2e_helpers.py index c18ee3c..0a8dd5b 100644 --- a/packages/embeddedci/tests/e2e/e2e_helpers.py +++ b/packages/embeddedci/tests/e2e/e2e_helpers.py @@ -2,12 +2,39 @@ from __future__ import annotations +import os import threading import time from typing import Any, Callable, List, Optional, Sequence, Tuple +import pytest + from embeddedci.benchpod import BenchPod, FirmwareError +#: Marks a test that needs the analog front end (DAC, ADC, 4-20 mA): it skips on the digital board. +ANALOG = pytest.mark.benchpod_capability("analog") + + +def require_analog(pod: BenchPod) -> None: + """Skip on the digital-only board, for tests on a pod the ``ANALOG`` marker does not see (the + cloud tier's own connection).""" + if not pod.can_analog: + pytest.skip("digital-only board: no analog front end (DAC, ADC, 4-20 mA)") + + +def check_expected_board(pod: BenchPod) -> None: + """Fail when ``BENCHPOD_EXPECT_BOARD`` (analog or digital) names the other board.""" + want = os.environ.get("BENCHPOD_EXPECT_BOARD", "").strip().lower() + if not want: + return + if want not in ("analog", "digital"): + pytest.fail(f"BENCHPOD_EXPECT_BOARD={want!r}: use analog or digital") + got = "analog" if pod.can_analog else "digital" + if got != want: + pytest.fail(f"the pod reports the {got} board (status analog={pod.capabilities.analog!r}), " + f"the bench expects {want} (BENCHPOD_EXPECT_BOARD)") + + #: Seconds for relays and a freshly started DAC output to settle before a capture. SETTLE = 0.3 diff --git a/packages/embeddedci/tests/e2e/test_e2e_cloud.py b/packages/embeddedci/tests/e2e/test_e2e_cloud.py index 92f5277..a938f47 100644 --- a/packages/embeddedci/tests/e2e/test_e2e_cloud.py +++ b/packages/embeddedci/tests/e2e/test_e2e_cloud.py @@ -15,6 +15,8 @@ from embeddedci.benchpod import BenchPod +from e2e_helpers import check_expected_board, require_analog + # Not marked `hardware`: that gate wants --benchpod-connection, and this tier opens its own cloud # connection (the `cloud` fixture skips without BENCHPOD_E2E_CLOUD_DEVICE + BENCHPOD_API_KEY). @@ -25,6 +27,7 @@ def cloud(benchpod_la_voltage): if not name or not os.environ.get("BENCHPOD_API_KEY"): pytest.skip("set BENCHPOD_E2E_CLOUD_DEVICE and BENCHPOD_API_KEY to run the cloud tier") with BenchPod(f"embeddedci:{name}", la_voltage=benchpod_la_voltage, lease_wait=120) as pod: + check_expected_board(pod) yield pod @@ -36,12 +39,24 @@ def test_lease_and_status(cloud): def test_commands_over_the_command_channel(cloud): assert cloud.power_status().internal.ok + + +def test_analog_commands_over_the_command_channel(cloud): + require_analog(cloud) assert cloud.analog_path("cal1").path == "cal1" cloud.analog_path("off") -def test_captures_over_the_tunnel(cloud): +def test_la_capture_over_the_tunnel(cloud): + la = cloud.capture_la(8192, sample_rate_hz=1_000_000) + assert len(la) == 8192 + assert la.sample_rate_hz == pytest.approx(1_000_000, rel=0.05), la.sample_rate_hz + assert all(0 <= w < 1 << la.channels for w in la.words), f"LA words must fit {la.channels} bits" + + +def test_adc_capture_over_the_tunnel(cloud): """Length alone would pass on 2048 zeros — check the payload survived the tunnel intact.""" + require_analog(cloud) adc = cloud.capture_adc(2048, sample_rate_hz=100_000) assert len(adc) == 2048 assert adc.sample_rate_hz == pytest.approx(100_000, rel=0.05), adc.sample_rate_hz @@ -49,11 +64,6 @@ def test_captures_over_the_tunnel(cloud): assert all(0 <= c <= 65535 for c in adc.counts) assert len(set(adc.counts)) > 1, "every ADC sample identical: the tunnel delivered filler" - la = cloud.capture_la(8192, sample_rate_hz=1_000_000) - assert len(la) == 8192 - assert la.sample_rate_hz == pytest.approx(1_000_000, rel=0.05), la.sample_rate_hz - assert all(0 <= w < 1 << la.channels for w in la.words), f"LA words must fit {la.channels} bits" - def test_commands_work_while_a_uart_session_holds_the_tunnel(cloud): """The point is that the command channel still answers while the tunnel is busy.""" @@ -66,6 +76,7 @@ def test_commands_work_while_a_uart_session_holds_the_tunnel(cloud): def test_waveform_library_save_and_replay(cloud): + require_analog(cloud) cloud.analog_path("cal1") cloud.dac_output("5v", volts=1.5) cap = cloud.capture_adc(2048, sample_rate_hz=100_000) diff --git a/packages/embeddedci/tests/e2e/test_e2e_digital.py b/packages/embeddedci/tests/e2e/test_e2e_digital.py new file mode 100644 index 0000000..be48a5e --- /dev/null +++ b/packages/embeddedci/tests/e2e/test_e2e_digital.py @@ -0,0 +1,86 @@ +"""Tier "digital": the digital-only board (digital-v3, digital-v3r2), the pod without the analog +front end. + + BENCHPOD_EXPECT_BOARD=digital pytest packages/embeddedci/tests/e2e --benchpod-connection= + +These tests run only on a pod that reports ``analog: false`` and skip on the analog board; the +analog tests elsewhere skip here. The firmware finds the analog section at boot by its two I2C +expanders and refuses every analog command on a board without them (cmd_gate.c): the SDK raises +:class:`UnsupportedFeatureError` for those, and everything digital keeps working. +""" + +from __future__ import annotations + +import time + +import pytest + +from embeddedci.benchpod import UnsupportedFeatureError + +pytestmark = pytest.mark.hardware + +#: The pod's refusal on the digital board (cmd_gate.c). +NO_ANALOG = "no analog front end" + +#: Capabilities only the analog board announces (cloud_caps_hw.c, cloud_caps.c). +ANALOG_ONLY_CAPS = ("scope", "dac_replay", "dac_deep_replay", "dac_control_loop", "dac_cotrig", + "dac_loop_sources", "dac_loop_input_map", "calibrate", "current_out") + + +@pytest.fixture +def digital(pod): + if pod.can_analog: + pytest.skip("analog board: the digital-only checks do not apply") + return pod + + +def test_status_says_digital_and_drops_the_analog_capabilities(digital): + assert digital.status().get("analog") is False + caps = digital.refresh_capabilities() + assert caps.analog is False and not digital.can_analog + assert [c for c in ANALOG_ONLY_CAPS if getattr(caps, c, False)] == [] + assert caps.analyzer # the logic analyzer is on the digital board + + +@pytest.mark.parametrize("call", [ + lambda p: p.dac_output("5v", volts=1.0), + lambda p: p.adc_read("cal1"), + lambda p: p.analog_path("cal1"), + lambda p: p.current_out(0.012), + lambda p: p.capture_adc(1024, sample_rate_hz=100_000), + lambda p: p.capture_correlated(adc_samples=1024, adc_sample_rate_hz=100_000, + la_samples=1024, la_sample_rate_hz=1_000_000), + lambda p: p.generate("sine", freq_hz=1000, amplitude=0.5, dac_path="5v"), + lambda p: p.replay([0.5, 1.0, 1.5], dac_path="5v", sample_rate_hz=1000), +], ids=["dac_output", "adc_read", "analog_path", "current_out", "capture_adc", + "capture_correlated", "generate", "replay"]) +def test_analog_calls_are_refused_fast(digital, call): + t0 = time.monotonic() + with pytest.raises(UnsupportedFeatureError) as err: + call(digital) + assert time.monotonic() - t0 < 5.0, "the refusal must not wait for a timeout" + # From the pod (cmd_gate.c) or, for a capability-gated call, from the SDK before sending: + # either way it names the missing front end, never a firmware update. + assert err.value.feature == "analog", err.value + assert "newer" not in str(err.value) + + +def test_refused_raw_commands_name_the_missing_front_end(digital): + for cmd in ({"cmd": "dac_out", "path": "5v", "mv": 1000}, {"cmd": "adc_read", "source": "cal1"}): + with pytest.raises(UnsupportedFeatureError, match=NO_ANALOG): + digital.command(cmd) + + +def test_dac_stop_still_works(digital): + """dac_stop is the cleanup every test and tool runs: it must not fail on the digital board.""" + digital.dac_stop() + + +def test_pod_answers_after_the_refusals(digital): + for call in (lambda: digital.adc_read("cal1"), lambda: digital.capture_adc(1024)): + with pytest.raises(UnsupportedFeatureError): + call() + assert digital.status().get("board") + assert digital.power_status().internal.ok + la = digital.capture_la(8192, sample_rate_hz=1_000_000) + assert len(la) == 8192 and la.sample_rate_hz == pytest.approx(1e6, rel=0.05) diff --git a/packages/embeddedci/tests/e2e/test_e2e_gpio.py b/packages/embeddedci/tests/e2e/test_e2e_gpio.py index 89777eb..31445a0 100644 --- a/packages/embeddedci/tests/e2e/test_e2e_gpio.py +++ b/packages/embeddedci/tests/e2e/test_e2e_gpio.py @@ -23,6 +23,8 @@ Wiring, ) +from e2e_helpers import ANALOG + pytestmark = pytest.mark.hardware @@ -194,6 +196,7 @@ def test_trigger_timeout_aborts_and_later_captures_are_untriggered(trigger_pod, assert time.monotonic() - start < 5.0 +@ANALOG def test_triggered_adc_capture(trigger_pod, bench): ch = bench.free_la[0] pin = trigger_pod.gpio(ch, "output", level=0) @@ -211,6 +214,7 @@ def run(): assert len(result["capture"]) == 8192 and result["capture"].trigger.la == ch +@ANALOG def test_triggered_capture_of_the_current_input(trigger_pod, bench): """An LA edge starts a capture of the 4-20 mA input (J8) like any other: amps, t = 0 on the edge, alone and together with the logic channels.""" diff --git a/packages/embeddedci/tests/e2e/test_e2e_pod.py b/packages/embeddedci/tests/e2e/test_e2e_pod.py index 06fe366..f2f9194 100644 --- a/packages/embeddedci/tests/e2e/test_e2e_pod.py +++ b/packages/embeddedci/tests/e2e/test_e2e_pod.py @@ -36,7 +36,7 @@ from embeddedci.benchpod.control_loop import CURVE_POINTS from embeddedci.benchpod.decode import decode_uart -from e2e_helpers import SETTLE, adc_levels, events_during, host_clock_rate, p2p, split_levels +from e2e_helpers import ANALOG, SETTLE, adc_levels, events_during, host_clock_rate, p2p, split_levels pytestmark = pytest.mark.hardware @@ -49,7 +49,9 @@ def test_status_and_capabilities(pod): caps = pod.refresh_capabilities() assert caps.board == status["board"] if caps.board == "stm32h563": - assert caps.adc_bits == 16 and caps.scope and caps.analyzer + assert caps.analyzer + if pod.can_analog: + assert caps.adc_bits == 16 and caps.scope def test_la_voltage_is_selected_from_conftest(pod): @@ -118,6 +120,7 @@ def test_firmware_errors_surface_as_firmware_error(pod): # -- analog ---------------------------------------------------------------------------- +@ANALOG @pytest.mark.parametrize("path", [p for p in ANALOG_PATHS if p != "dac_12v"]) def test_analog_path_applies(pod, path): state = pod.analog_path(path) @@ -129,6 +132,7 @@ def test_analog_path_applies(pod, path): (pod.analog_path(path).dac_mux_register, pod.analog_path(path).cal_relay_register) +@ANALOG def test_analog_paths_are_not_all_the_same_switch_setting(pod): """Distinct paths must drive distinct hardware, not just return distinct names.""" settings = {p: (s.dac_mux_register, s.cal_relay_register) @@ -137,6 +141,7 @@ def test_analog_paths_are_not_all_the_same_switch_setting(pod): assert len(set(settings.values())) > 1, f"every path set the same registers: {settings}" +@ANALOG def test_dac_output_codes_match_the_sdk_volts_mapping(pod, bench): rows = [] points = [("3v3", 0.5), ("3v3", 1.5), ("5v", 1.0), ("5v", min(2.5, bench.dac_max_v))] @@ -152,6 +157,7 @@ def test_dac_output_codes_match_the_sdk_volts_mapping(pod, bench): assert all(abs(achieved - volts) < 0.06 for _, volts, achieved, _, _ in rows), rows +@ANALOG def test_dac_output_reads_back_through_the_adc(pod): for volts in (1.0, 2.5): pod.dac_output("5v", volts=volts) @@ -160,6 +166,7 @@ def test_dac_output_reads_back_through_the_adc(pod): assert abs(reading.voltage - volts) < 0.1, (volts, reading) +@ANALOG def test_current_out_holds_a_current_and_the_dac_follows(pod): """The 4-20 mA output (J9) with nothing wired: the replies, and the DAC behind them. @@ -205,6 +212,7 @@ def test_current_out_holds_a_current_and_the_dac_follows(pod): pod.analog_path("off") +@ANALOG def test_waveforms_on_the_current_output_reach_the_dac(pod): """generate and replay on ``current_out`` with nothing wired: the levels asked for in amps are what the DAC holds, read back through the internal 5 V loopback (up to 2.5 V on the 5 V @@ -239,6 +247,7 @@ def check(readings) -> None: pod.analog_path("off") +@ANALOG def test_capture_of_the_current_input_is_in_amps(pod): """A capture of J8 with nothing wired: amps, about zero, from the pod's own fit.""" if os.environ.get("BENCHPOD_E2E_J8_WIRED") == "1": @@ -267,6 +276,7 @@ def _mean(values) -> float: return sum(values) / len(values) +@ANALOG def test_current_input_is_captured_with_the_logic_channels(pod): """ADC + LA off one trigger, with the analog half on J8: amps next to the logic words.""" if os.environ.get("BENCHPOD_E2E_J8_WIRED") == "1": @@ -283,6 +293,7 @@ def test_current_input_is_captured_with_the_logic_channels(pod): assert ext.source == "ext" and ext.currents == [] +@ANALOG def test_slow_captures_of_the_open_input_stay_within_the_documented_shift(pod): """Below 400 kS/s the input stage reads a little higher. The docs say 10 to 20 uA; this pins it under 60 uA at the slowest rate the Capture tab offers.""" @@ -297,6 +308,7 @@ def test_slow_captures_of_the_open_input_stay_within_the_documented_shift(pod): assert -10e-6 < means[80_000] - means[400_000] < 40e-6, means +@ANALOG def test_a_current_waveform_starts_on_the_capture_t0(pod): """The DAC co-trigger on the 4-20 mA output: armed, the waveform's first sample lands on the capture's t0, so the capture opens on the first level and steps to the second.""" @@ -320,6 +332,7 @@ def test_a_current_waveform_starts_on_the_capture_t0(pod): pod.analog_path("off") +@ANALOG def test_stop_dac_after_freezes_the_loop_current(pod): """The capture-timed DAC cut on the 4-20 mA output: the waveform runs, then the DAC freezes, so the loop HOLDS a current from the waveform. It does not fall back to 4 mA by itself.""" @@ -345,6 +358,7 @@ def test_stop_dac_after_freezes_the_loop_current(pod): pod.analog_path("off") +@ANALOG def test_dac_stop_alone_leaves_the_loop_where_it_was(pod): """`dac_stop` stops the DAC engine and nothing else: a held current stays. Going back to 4 mA takes `current_out(0.004)`, which is what the handles and the web Stop do.""" @@ -363,6 +377,7 @@ def test_dac_stop_alone_leaves_the_loop_where_it_was(pod): pod.analog_path("off") +@ANALOG def test_a_current_switches_the_voltage_outputs_off(pod): """One DAC: the voltage outputs would follow the current, so setting one switches them off, and it leaves the ADC relays alone (J8 stays routed).""" @@ -380,6 +395,7 @@ def test_a_current_switches_the_voltage_outputs_off(pod): pod.analog_path("off") +@ANALOG def test_current_in_reads_zero_with_j8_open(pod): """The 4-20 mA input (J8) with nothing wired: about 0 mA, and the current matches the voltage.""" if os.environ.get("BENCHPOD_E2E_J8_WIRED") == "1": @@ -393,6 +409,7 @@ def test_current_in_reads_zero_with_j8_open(pod): assert abs(reading.current) < limit, f"J8 open reads {reading.current * 1e6:.0f} uA: is something connected?" +@ANALOG def test_generate_levels_are_volts(pod): pod.analog_path("cal1") with pod.generate("square", freq_hz=2, amplitude=0.8, offset=1.5, dac_path="5v", route=False): @@ -403,6 +420,7 @@ def test_generate_levels_are_volts(pod): assert abs(hi - 2.3) < 0.15 and abs(lo - 0.7) < 0.15, (hi, lo, readings) +@ANALOG def test_generate_on_the_bipolar_12v_path_centres_on_zero(pod, bench): if not bench.allow_12v: pytest.skip("set BENCHPOD_E2E_ALLOW_12V=1 when nothing is wired to the 12v output") @@ -414,6 +432,7 @@ def test_generate_on_the_bipolar_12v_path_centres_on_zero(pod, bench): assert abs(hi - 1.04) < 0.2 and abs(lo + 1.0) < 0.2, (hi, lo) +@ANALOG def test_route_false_keeps_a_loopback_and_route_true_reroutes(pod): pytest.importorskip("numpy") pod.analog_path("cal1") @@ -427,6 +446,7 @@ def test_route_false_keeps_a_loopback_and_route_true_reroutes(pod): assert pod.lowlevel.cal_switch_status()["cal1"] == 0 # dac_out re-routed the output +@ANALOG def test_capture_adc_source_routes_the_input(pod): pod.analog_path("off") cap = pod.capture_adc(256, source="cal1") @@ -434,6 +454,7 @@ def test_capture_adc_source_routes_the_input(pod): assert pod.lowlevel.cal_switch_status()["cal1"] == 1 +@ANALOG def test_replay_volts_on_the_loopback(pod): pytest.importorskip("numpy") pod.analog_path("cal1") @@ -444,6 +465,7 @@ def test_replay_volts_on_the_loopback(pod): assert abs(cap.dominant_frequency() - 100) < 5 and p2p(cap) > 800 +@ANALOG def test_co_triggered_replay_arms_on_the_capture(pod): if not pod.capabilities.dac_cotrig: pytest.skip("the running gateware has no DAC co-trigger") @@ -454,6 +476,7 @@ def test_co_triggered_replay_arms_on_the_capture(pod): assert p2p(cc.adc) > 800 +@ANALOG def test_stop_dac_after_cuts_the_output_mid_capture(pod): pod.analog_path("cal1") pod.generate("square", freq_hz=200, amplitude=0.8, offset=1.5, route=False) @@ -465,6 +488,7 @@ def test_stop_dac_after_cuts_the_output_mid_capture(pod): assert max(late) - min(late) < 300, "the DAC was still toggling after stop_dac_after" +@ANALOG def test_a_restarted_dac_plays_the_new_waveform_from_its_first_sample(pod): # Gateware <= v32 ran the first pass of a DAC start at the PREVIOUS waveform's length, so a # 5 kHz square (181 samples) after a 200 Hz one (2034) played ~2 ms of stale 200 Hz levels. @@ -485,6 +509,7 @@ def test_a_restarted_dac_plays_the_new_waveform_from_its_first_sample(pod): assert flat.size == 0, f"the square stalled in blocks {flat.tolist()} (spreads {spread.tolist()})" +@ANALOG def test_generated_frequency_matches_the_adc_clock(pod): # The firmware picks the DAC divider and period from its model of what one DAC sample costs # in the gateware (max(divider, 3) + 51 clk48). If the gateware's frame sequencing drifts @@ -517,6 +542,7 @@ def _measure(pod, waveform: str, samples: int): return np.array([pod.capabilities.counts_to_volts(c) for c in counts]) +@ANALOG def test_measure_captures_one_period_of_the_played_waveform(pod): # measure ties the DAC period to the capture count (gateware MEASURE), so the capture holds # exactly one period. It does not route the analog path: without cal1 it reads the idle SMA. @@ -564,6 +590,7 @@ def test_measure_captures_one_period_of_the_played_waveform(pod): # -- captures ---------------------------------------------------------------------------- +@ANALOG def test_capture_adc_shallow_and_deep(pod): shallow = pod.capture_adc(4096, sample_rate_hz=100_000) deep = pod.capture_adc(100_000, sample_rate_hz=400_000) # above 32768: streams from PSRAM @@ -571,9 +598,13 @@ def test_capture_adc_shallow_and_deep(pod): assert len(deep) == 100_000 and deep.duration == pytest.approx(100_000 / deep.sample_rate_hz) -def test_capture_la_and_correlated(pod): +def test_capture_la(pod): la = pod.capture_la(65536, sample_rate_hz=1_000_000) assert len(la) == 65536 and la.sample_rate_hz == pytest.approx(1e6, rel=0.05) + + +@ANALOG +def test_capture_correlated(pod): cc = pod.capture_correlated(adc_samples=4096, adc_sample_rate_hz=100_000, la_samples=4096, la_sample_rate_hz=1_000_000) assert len(cc.adc) == 4096 and len(cc.la) == 4096 @@ -648,6 +679,7 @@ def test_la_sample_rate_matches_the_host_clock(pod, bench): f"{la.sample_rate_hz:.0f} Hz)") +@ANALOG def test_adc_sample_rate_matches_the_host_clock(pod): np = pytest.importorskip("numpy") pod.analog_path("cal1") @@ -676,12 +708,14 @@ def toggle_dac(k): # a 5 kHz burst on even events, flat on odd ones # -- DAC replay depth vs gateware image ---------------------------------------------------- +@ANALOG def test_deep_replay_is_refused_on_the_loop_image_without_switching(loop_image): with pytest.raises(BenchPodError, match="DEEP_REPLAY"): loop_image.replay([1.0] * 4096, dac_path="5v", route=False, switch_image=False) assert loop_image.refresh_capabilities().dac_control_loop # still on the loop image +@ANALOG def test_deep_replay_switches_to_the_deep_replay_image(loop_image): np = pytest.importorskip("numpy") pod = loop_image @@ -696,6 +730,7 @@ def test_deep_replay_switches_to_the_deep_replay_image(loop_image): assert abs(adc.dominant_frequency() - 100) < 5 and int(np.ptp(adc.counts)) > 800 +@ANALOG def test_control_loop_switches_to_the_loop_image(deep_image): pod = deep_image if not pod.capabilities.dac_loop_sources: @@ -708,12 +743,14 @@ def test_control_loop_switches_to_the_loop_image(deep_image): assert pod.refresh_capabilities().dac_control_loop +@ANALOG def test_control_loop_without_switching_is_refused_on_the_deep_replay_image(deep_image): with pytest.raises(BenchPodError, match="switch_image"): deep_image.control_loop(curve=build_linear_curve(30000), switch_image=False) assert deep_image.refresh_capabilities().dac_deep_replay # still on the deep-replay image +@ANALOG def test_control_loop_fixed_input_indexes_the_curve(loop_image): pod = loop_image if not pod.capabilities.dac_loop_sources: @@ -729,6 +766,7 @@ def test_control_loop_fixed_input_indexes_the_curve(loop_image): assert pt.loop_input == code and abs(pt.v - curve_output_at(curve, code)) < 1500, (code, pt) +@ANALOG def test_control_loop_sweep_advances(loop_image): pod = loop_image if not pod.capabilities.dac_loop_sources: @@ -740,6 +778,7 @@ def test_control_loop_sweep_advances(loop_image): assert loop.probe().loop_input != first +@ANALOG def test_control_loop_input_map(loop_image): pod = loop_image if not pod.capabilities.dac_loop_input_map: @@ -757,6 +796,7 @@ def _regulator_curve(setpoint_v, lo, hi, n=CURVE_POINTS): for i in range(n)] +@ANALOG def test_control_loop_regulates_through_the_external_wire(loop_image, bench): # The loop closed through real wiring: DAC 0-5 V output -> ADC front SMA. The input map puts # the curve on a 0-5 V axis, so the loop must hold the SMA at the setpoint; then again at a @@ -790,6 +830,7 @@ def test_control_loop_regulates_through_the_external_wire(loop_image, bench): assert pod.status().get("loop_tripped") is True +@ANALOG def test_deep_replay_plays_while_capturing(deep_image): pytest.importorskip("numpy") pod = deep_image @@ -802,6 +843,7 @@ def test_deep_replay_plays_while_capturing(deep_image): assert abs(adc.dominant_frequency() - 100) < 5 and p2p(adc) > 800 +@ANALOG def test_fpga_image_round_trip(pod): started_on_loop = pod.refresh_capabilities().dac_control_loop other = FpgaImage.DEEP_REPLAY if started_on_loop else FpgaImage.LOOP diff --git a/packages/embeddedci/tests/test_adc_fft_hw.py b/packages/embeddedci/tests/test_adc_fft_hw.py index 9fdcc33..a81ff4a 100644 --- a/packages/embeddedci/tests/test_adc_fft_hw.py +++ b/packages/embeddedci/tests/test_adc_fft_hw.py @@ -25,6 +25,16 @@ np = pytest.importorskip("numpy") # the FFT path needs numpy +@pytest.fixture(scope="module", autouse=True) +def _analog_board(): + """The loopback needs the DAC and the ADC: skip on the digital-only board.""" + from embeddedci.benchpod import BenchPod + + with BenchPod(HW, lease=False) as pod: + if not pod.can_analog: + pytest.skip("digital-only board: no analog front end (DAC, ADC, 4-20 mA)") + + def _load_tool(): path = pathlib.Path(__file__).resolve().parents[1] / "scripts" / "adc_fft_capture.py" spec = importlib.util.spec_from_file_location("adc_fft_capture", path) diff --git a/packages/embeddedci/tests/test_capture_replay_hw.py b/packages/embeddedci/tests/test_capture_replay_hw.py index aa4c156..2bca4a2 100644 --- a/packages/embeddedci/tests/test_capture_replay_hw.py +++ b/packages/embeddedci/tests/test_capture_replay_hw.py @@ -19,6 +19,7 @@ pytestmark = pytest.mark.hardware +@pytest.mark.benchpod_capability("analog") def test_capture_adc_returns_volts(benchpod): cap = benchpod.capture_adc(256, sample_rate_hz=1e6) assert len(cap) == 256, f"asked for 256 samples, got {len(cap)}" @@ -38,6 +39,7 @@ def test_raw_la_capture(benchpod): assert all(0 <= w < (1 << 12) for w in la.words), "LA words must be 12-bit" +@pytest.mark.benchpod_capability("analog") def test_correlated_adc_la_capture(benchpod): ac = benchpod.capture_correlated(adc_samples=256, adc_sample_rate_hz=400_000, la_samples=256, la_sample_rate_hz=1e6) @@ -46,12 +48,14 @@ def test_correlated_adc_la_capture(benchpod): assert ac.la.sample_rate_hz == pytest.approx(1e6, rel=0.05), ac.la.sample_rate_hz +@pytest.mark.benchpod_capability("analog") def test_deep_adc_capture_streams_from_psram(benchpod): """Above the 32768-sample single-shot buffer, capture_adc switches to the PSRAM stream.""" cap = benchpod.capture_adc(65536, sample_rate_hz=400_000) assert len(cap) == 65536 +@pytest.mark.benchpod_capability("analog") @pytest.mark.benchpod_capability("dac_replay") def test_save_recording_and_replay(benchpod, benchpod_waveforms): """Capture → save to library → replay from the library → stop (the headline flow).""" @@ -70,6 +74,7 @@ def test_save_recording_and_replay(benchpod, benchpod_waveforms): handle.stop() +@pytest.mark.benchpod_capability("analog") @pytest.mark.benchpod_capability("dac_deep_replay") def test_replay_while_capturing_concurrently(benchpod_dac): """gateware v18: a looping deep replay runs WHILE an LA capture streams (shared PSRAM bus).""" diff --git a/packages/embeddedci/tests/test_control_loop_hw.py b/packages/embeddedci/tests/test_control_loop_hw.py index e10f1c1..81eb2d4 100644 --- a/packages/embeddedci/tests/test_control_loop_hw.py +++ b/packages/embeddedci/tests/test_control_loop_hw.py @@ -18,7 +18,8 @@ from embeddedci import benchpod as bp -pytestmark = pytest.mark.hardware +# Every test here drives the DAC (and reads the ADC): none runs on the digital-only board. +pytestmark = [pytest.mark.hardware, pytest.mark.benchpod_capability("analog")] def _ensure_loop_image(device) -> bool: diff --git a/packages/embeddedci/tests/test_dac_limits_fixture.py b/packages/embeddedci/tests/test_dac_limits_fixture.py index 91fdb2a..000caa9 100644 --- a/packages/embeddedci/tests/test_dac_limits_fixture.py +++ b/packages/embeddedci/tests/test_dac_limits_fixture.py @@ -16,7 +16,8 @@ @pytest.fixture def pod(monkeypatch) -> Dict[str, Any]: """A fake BenchPod recording every dac_limits command across all the connections it opens.""" - state: Dict[str, Any] = {"limits": dict(LIMITS), "sent": [], "opened": 0, "refuse": False} + state: Dict[str, Any] = {"limits": dict(LIMITS), "sent": [], "opened": 0, "refuse": False, + "analog": True} class FakePod: def __init__(self, connection, **kwargs): @@ -33,6 +34,10 @@ def __exit__(self, *exc): def close(self): pass + @property + def can_analog(self): + return state["analog"] + def command(self, req): sent: List[dict] = state["sent"] sent.append(dict(req)) @@ -113,6 +118,16 @@ def test_firmware_without_dac_limits_does_not_stop_the_run(pytester, pod, monkey assert pod["sent"] == [{"cmd": "dac_limits"}] # nothing to restore +def test_the_digital_board_is_not_asked(pytester, pod, monkeypatch): + monkeypatch.setenv(pytest_plugin.LIFT_DAC_LIMITS_ENV, "1") + pod["analog"] = False + pytester.makepyfile("def test_x():\n pass\n") + result = pytester.runpytest("-s", "--benchpod-connection=10.0.0.1") + result.assert_outcomes(passed=1) + assert pod["sent"] == [] # no DAC: nothing to lift, and no refusal in the output + assert "could not read/clear" not in result.stdout.str() + + def test_the_api_key_option_reaches_its_connection(pytester, pod, monkeypatch): """--benchpod-api-key (and the lease options) reach the fixture's own connection, so a cloud run authenticated only by the flag can lift and restore the limits too.""" diff --git a/packages/embeddedci/tests/test_unsupported_feature.py b/packages/embeddedci/tests/test_unsupported_feature.py index d8544cc..fd99ea0 100644 --- a/packages/embeddedci/tests/test_unsupported_feature.py +++ b/packages/embeddedci/tests/test_unsupported_feature.py @@ -5,6 +5,7 @@ from __future__ import annotations +import os from typing import Any, Dict, List, Optional import pytest @@ -91,3 +92,48 @@ def test_can_analog_is_false_only_on_the_digital_board(): assert BenchPod(transport=CapsPod([], analog=True), lease=False).can_analog assert BenchPod(transport=CapsPod([]), lease=False).can_analog # older firmware: analog boards assert not BenchPod(transport=CapsPod([], analog=False), lease=False).can_analog + + +@pytest.mark.parametrize("call", [lambda p: p.current_out(0.012), lambda p: p.calibrate()], + ids=["current_out", "calibrate"]) +def test_digital_board_names_the_missing_front_end_not_old_firmware(call): + pod = BenchPod(transport=CapsPod([], analog=False), lease=False) + with pytest.raises(UnsupportedFeatureError, match="digital-only board") as err: + call(pod) + assert err.value.feature == "analog" and "newer" not in str(err.value) + assert pod._transport.commands == [] + # Older firmware on an analog board still gets the update hint. + with pytest.raises(UnsupportedFeatureError, match="newer pod firmware"): + call(BenchPod(transport=CapsPod([]), lease=False)) + + +@pytest.mark.parametrize("analog,outcome", [(True, "passed"), (None, "passed"), (False, "skipped")], + ids=["analog board", "older firmware", "digital board"]) +def test_analog_capability_marker_skips_only_on_the_digital_board(pytester, analog, outcome): + tests_dir = os.path.dirname(__file__) + pytester.makeconftest(f""" +import sys + +import pytest + +sys.path.insert(0, {tests_dir!r}) +from test_unsupported_feature import CapsPod +from embeddedci.benchpod import BenchPod + + +@pytest.fixture(scope="session") +def benchpod(): + return BenchPod(transport=CapsPod([], analog={analog!r}), lease=False) +""") + pytester.makepyfile(""" +import pytest + + +@pytest.mark.benchpod_capability("analog") +def test_needs_the_analog_front_end(benchpod): + pass +""") + result = pytester.runpytest("-p", "no:cacheprovider", "-rs") + result.assert_outcomes(**{outcome: 1}) + if outcome == "skipped": + result.stdout.fnmatch_lines(["*digital-only board*"])