From b54b94e8ae49f02b9493e8d6d4fc5fbf138a02bf Mon Sep 17 00:00:00 2001 From: ScottW514 Date: Sun, 23 Aug 2026 14:15:53 -0400 Subject: [PATCH] forgetest: the bench actuator plugs into the action seam fixture.py: the bench's /data/forgetest/fixture.json (hostname, key, optional ip, the channels wired, arm_press), a resolver for .local asked of the network directly (the image has no mDNS resolver), and the client. The runner probes it before every run and at most every 30 s otherwise; ctx.act asks it for a channel it covers and still waits for the machine's own reading, falling back to the operator's notice when the box fails. A test declares with hands=(...) what it asks of a person beyond its typed actions; an operator test with none, whose actions the fixture covers, is routed into the unattended queue, its Ready gates pass, and a prompt it raises anyway is a FAIL naming the undeclared step. Live tests never move; their arm press stays a person's unless the bench opted in, in which case the fixture presses when the button lights. Whatever the box still holds after a run is released before the baseline's post pass and recorded. The page shows what the fixture covers. Contract in ACCEPTANCE.md; the wiring facts, with the interlock connector left to the bench to settle (SAFETY.md and the sister map differ), in BRINGUP. Catalog unchanged in its definitions; the cloud and laser suites' shared code moved, so their implementation hashes move with it. --- docs/ACCEPTANCE.md | 42 ++- docs/BRINGUP.md | 26 +- forgetest/forgetest/catalog.py | 32 +- forgetest/forgetest/fixture.py | 324 ++++++++++++++++++++ forgetest/forgetest/page.py | 13 +- forgetest/forgetest/runner.py | 128 +++++++- forgetest/forgetest/suite/cloud.py | 2 +- forgetest/forgetest/suite/laser.py | 4 +- forgetest/tests/helpers.py | 4 +- forgetest/tests/test_fixture.py | 474 +++++++++++++++++++++++++++++ 10 files changed, 1026 insertions(+), 23 deletions(-) create mode 100644 forgetest/forgetest/fixture.py create mode 100644 forgetest/tests/test_fixture.py diff --git a/docs/ACCEPTANCE.md b/docs/ACCEPTANCE.md index a4c30b1..0ecf1d3 100644 --- a/docs/ACCEPTANCE.md +++ b/docs/ACCEPTANCE.md @@ -253,9 +253,9 @@ during the run is those steps, taken in turn, in one of four forms: manages: the wording is the action's own, the test adds its context, the machine's reading proves it done, and the result's `evidence.actions` records each one with who performed it. This is the - seam a bench actuator plugs into: a runner `fixture` covering a channel - performs the action instead of the notice, and a test reads the same - either way; + seam the bench actuator plugs into (below): a `fixture` covering a + channel performs the action instead of the notice, and a test reads + the same either way; - a **confirm** is a yes/no the evidence cannot answer. One is left in the catalog: the mark `laser.emission-witness` leaves on the scrap, the once-per-campaign calibration of the sensor witnesses (the head's beam @@ -264,6 +264,42 @@ during the run is those steps, taken in turn, in one of four forms: for "did the gantry move", the button LEDs for "is the button dark", the lid lamp toggled between two snapshots for "is the camera live". +### The bench actuator + +`forgefixture` (`fixture/`, its own README) is an ESP32-S3 on the bench +network driving three relays at the machine's connectors: a normally +closed contact in the lid-switch loop, another in the interlock loop, a +normally open contact across the button input that is only ever pulsed. +Unpowered, unplugged or rebooting, it leaves a stock machine. The tool +finds it through `/data/forgetest/fixture.json` (bench-local, mode +0600: the hostname, the API key, an optional `ip` override, the +`channels` wired, `arm_press`), resolves `.local` itself (the +image carries no mDNS resolver), and probes it before every run and at +most every 30 s otherwise. What holds: + +- **Every action is still proven by the machine.** The fixture does not + read the switches; `ctx.act` asks it and then waits for forgectrl's + reading exactly as it waits for an operator's hand. An action the box + fails to perform falls back to the operator's notice, and the record + says so (`evidence.actions[].by`, `fixture_error`). +- **An operator test the fixture can run alone runs unattended.** A test + declares its actions and, with `hands=(...)`, whatever else it asks of + a person ("app" for a job in the Glowforge app). An `operator` test + whose actions the fixture covers and whose `hands` are empty is routed + into the unattended queue and out of the attended one; its Ready gates + pass (the fixture performs the timed step), and a prompt it raises + anyway is a FAIL naming the undeclared step, never a wait for nobody. + `live` never moves: the fire watch and the acknowledgment are a + person's. +- **The button channel needs the jumper.** With the fixture's enable + jumper out the button is not covered, and tests that press it stay + attended. The arm press of a live test stays a person's unless the + bench config says `arm_press: true`: then, and only with the jumper in, + the fixture presses when the button lights, recorded as its own. +- **What the box still holds after a run is released** and recorded + (`evidence.fixture.released`), before the baseline's post pass, so a + lid left open by a failed test never reaches the next one. + ### Every run starts from, and leaves, the fresh-boot idle state The runner brackets every test and bench tool with a **baseline** pass diff --git a/docs/BRINGUP.md b/docs/BRINGUP.md index 0f093e1..3ca6087 100644 --- a/docs/BRINGUP.md +++ b/docs/BRINGUP.md @@ -641,6 +641,20 @@ until `releases/v/acceptance.json` is committed. ## Hardware facts bank (measured) +- **The bench actuator's wiring** (`fixture/README.md` for the box itself). + Three 3.3 V optocoupler relay modules, high-level trigger, coils from the + machine's 3.3 V (about 100 mA each), inputs from the ESP32-S3 DevKitC-1's + GPIO 4 (lid), 5 (interlock), 6 (button), the button's enable jumper on + GPIO 7 to GND. The lid contact (NC) goes in series with the lid-switch + loop at J4.12/13; the button contact (NO) across the front button input + at J5 (BTN and its 12 V); the interlock contact (NC) in the remote + interlock loop, which SAFETY.md places on J8 and the sister project's + connector map on J6 (J8 being Y1's coils there): **to be settled at the + bench before the harness is made.** The DevKit and the machine share a + ground through the modules, so the DevKit is powered from a USB wall + adapter. The interposer harness itself is bench-local and is not + described in any repository. + - **The factory's envelope, decoded** (firmware 2.6.0-2228, the 23 captured headers, this board's own factory logs). The pulse header is the job's operating envelope and the factory refuses to cut without it: 29 tags are @@ -1169,8 +1183,16 @@ Open items only. Anything closed is in `CAMPAIGN-LOG.md`. `/run/gfcloud-nohunt` (`gfcloud --no-hunt`, the first settings report in the reconnect form), while the two homing tests and the one real print get theirs, the print by never reusing a session that has not hunted - the machine itself (the contract's cloud split in ACCEPTANCE.md). A bench - actuator for the lid, interlock and button is planned, not started. Catalog + the machine itself (the contract's cloud split in ACCEPTANCE.md). The + coverage maps follow the split (a sign-in change re-requires the + protocol test and the print, a feeder change the offline tests and the + print, a doc edit nothing), and the bench actuator `forgefixture` + (`fixture/`: ESP32-S3, three relays, the `ctx.act` seam, an operator + test it covers routed into the unattended queue) is written and + host-proven (the firmware builds in the pinned ESP-IDF container, the + policy and the tool's client have host tests) and **owed its bench + proof**: the harness at the machine's connectors, then a campaign with + it up. Catalog gaps left from the tool's own plan: `cooling.confirm-escalate` and `cooling.fire-gate-blocks-arm` are not ported (both need the pump switched by hand mid-run, so they are bench-tab material first), and whether diff --git a/forgetest/forgetest/catalog.py b/forgetest/forgetest/catalog.py index 62e1a43..7374e7e 100644 --- a/forgetest/forgetest/catalog.py +++ b/forgetest/forgetest/catalog.py @@ -40,7 +40,7 @@ REGISTRY = {} class Test: def __init__(self, id, title, subsystem, kind, hardware, covers, requires, always, est_min, steps, description, fn, mode=None, actions=(), - precheck=None): + precheck=None, hands=()): self.id = id self.title = title self.subsystem = subsystem @@ -53,6 +53,7 @@ class Test: self.est_min = est_min self.steps = tuple(steps) self.actions = tuple(actions) + self.hands = tuple(hands) self.precheck = precheck self.description = description or (fn.__doc__ or "").strip() self.fn = fn @@ -102,9 +103,20 @@ class Test: d = self.definition() d.update({"title": self.title, "est_min": self.est_min, "mode": self.mode, "steps": list(self.steps), "actions": list(self.actions), - "precheck": bool(self.precheck), "description": self.description}) + "hands": list(self.hands), "precheck": bool(self.precheck), + "description": self.description}) return d + def fixture_runnable(self, channels): + """Whether a bench actuator covering `channels` lets this test + run with nobody in the room: an operator test whose every action + is a covered channel and which asks a person for nothing else. A + live test never qualifies (the arm press and the fire watch are + a person's); an auto test needs no fixture.""" + if self.kind != "operator" or self.hands or not self.actions: + return False + return all(a in channels for a in self.actions) + def cannot_start(self): """The reason the test cannot start on the machine as it is, or None. Evaluated right before a start; never a result.""" @@ -174,7 +186,7 @@ def implementation_sha(path, test_id): def test(id, *, title, subsystem, kind="auto", hardware="api", mode=None, covers=(), requires=(), always=False, est_min=1, steps=(), description="", actions=(), - precheck=None): + precheck=None, hands=()): """`mode` names the controller mode the test needs live when it starts ("grbl" or "cloud"); the runner switches the machine there before the test and leaves it there, so a queue crosses modes only where a test @@ -183,9 +195,13 @@ def test(id, *, title, subsystem, kind="auto", hardware="api", mode=None, covers which also waits for the service session). `actions` names the machine actions (ACTIONS) the test performs - through Context.act; an `auto` test declares none. `precheck` is a - callable returning a reason string when the machine cannot run the - test as it is (None when it can).""" + through Context.act; an `auto` test declares none. `hands` names what + the test asks of a person beyond those actions ("app" for a job in + the Glowforge app, "scrap", "mark" for an eye on the result...): an + operator test with none, whose actions a bench actuator covers, runs + unattended when one is up. `precheck` is a callable returning a + reason string when the machine cannot run the test as it is (None + when it can).""" if not _ID_RX.match(id): raise ValueError("test id %r must look like subsystem.name" % id) if kind not in KINDS: @@ -202,6 +218,8 @@ def test(id, *, title, subsystem, kind="auto", hardware="api", mode=None, covers raise ValueError("test %s: action %r" % (id, a)) if actions and kind == "auto": raise ValueError("test %s: an auto test asks for no machine actions" % id) + if hands and kind == "auto": + raise ValueError("test %s: an auto test asks nothing of a person" % id) if precheck is not None and not callable(precheck): raise ValueError("test %s: precheck must be callable" % id) @@ -210,7 +228,7 @@ def test(id, *, title, subsystem, kind="auto", hardware="api", mode=None, covers raise ValueError("duplicate test id %r" % id) REGISTRY[id] = Test(id, title, subsystem, kind, hardware, covers, requires, always, est_min, steps, description, fn, mode=mode, - actions=actions, precheck=precheck) + actions=actions, precheck=precheck, hands=hands) return fn return deco diff --git a/forgetest/forgetest/fixture.py b/forgetest/forgetest/fixture.py new file mode 100644 index 0000000..3e1e86b --- /dev/null +++ b/forgetest/forgetest/fixture.py @@ -0,0 +1,324 @@ +"""The bench actuator (forgefixture): the box on the bench network that +opens the lid loop, pulls the interlock loop and presses the button on +request, so an operator test can run with nobody in the room. + +Configured by a bench-local file, FIXTURE_CONFIG (default +/data/forgetest/fixture.json, mode 0600, never in the repo): + + {"hostname": "forgefixture", "key": "", + "ip": null, "channels": ["lid", "interlock", "button"], + "arm_press": false} + +`hostname` is resolved as .local by the resolver here (the +image carries no mDNS resolver; the fixture answers the query itself); +`ip` overrides that. `channels` names what is wired. `arm_press` is the +one policy switch: whether the fixture may press the button to arm the +laser for a live test (a person's press by default). + +The fixture does not read the machine: every action the runner asks of +it is verified through the machine's own readings afterward, exactly as +an operator's hand is (Context.act). Contract: fixture/README.md. +""" +import json +import os +import socket +import struct +import threading +import time +import urllib.error +import urllib.request + +FIXTURE_CONFIG = os.environ.get("FORGETEST_FIXTURE_CONFIG", "/data/forgetest/fixture.json") +CHANNELS = ("lid", "interlock", "button") +MDNS_GROUP = "224.0.0.251" +MDNS_PORT = 5353 +RESOLVE_TTL_S = 300.0 # a resolved address is trusted this long + + +class FixtureError(Exception): + pass + + +# ---------------------------------------------------------------- mDNS + +def mdns_query(name): + """One mDNS A query for `name` (e.g. "forgefixture.local"), as the + bytes on the wire: a standard DNS header with id 0, one question, + class IN with the unicast-response bit set so the answer comes back + to this socket rather than the group.""" + labels = b"".join(struct.pack("B", len(p)) + p.encode("ascii") for p in name.rstrip(".").split(".")) + header = struct.pack("!HHHHHH", 0, 0, 1, 0, 0, 0) + return header + labels + b"\x00" + struct.pack("!HH", 1, 0x8001) + + +def _read_name(data, off): + """A DNS name at `off`, compression pointers followed; (name, next + offset).""" + parts = [] + jumped = False + end = off + guard = 0 + while True: + if off >= len(data): + raise ValueError("truncated name") + n = data[off] + if n == 0: + off += 1 + break + if n & 0xC0 == 0xC0: + ptr = struct.unpack("!H", data[off:off + 2])[0] & 0x3FFF + if not jumped: + end = off + 2 + jumped = True + off = ptr + guard += 1 + if guard > 32: + raise ValueError("pointer loop") + continue + parts.append(data[off + 1:off + 1 + n].decode("ascii", "replace")) + off += 1 + n + if not jumped: + end = off + return ".".join(parts).lower(), end + + +def mdns_answers(data, name): + """The IPv4 addresses in an mDNS response's A records for `name` + (answers and additionals; a response is anything with QR set).""" + name = name.rstrip(".").lower() + out = [] + try: + _id, flags, qd, an, ns, ar = struct.unpack("!HHHHHH", data[:12]) + if not flags & 0x8000: + return out + off = 12 + for _ in range(qd): + _n, off = _read_name(data, off) + off += 4 + for _ in range(an + ns + ar): + rname, off = _read_name(data, off) + rtype, rclass, _ttl, rdlen = struct.unpack("!HHIH", data[off:off + 10]) + off += 10 + rdata = data[off:off + rdlen] + off += rdlen + if rtype == 1 and rdlen == 4 and rname == name: + out.append(socket.inet_ntoa(rdata)) + except (struct.error, ValueError): + pass + return out + + +def resolve_mdns(hostname, timeout=2.0, tries=3): + """The IPv4 address of .local, asked of the network + directly; None when nothing answers.""" + name = hostname if hostname.endswith(".local") else hostname + ".local" + query = mdns_query(name) + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP) + try: + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 1) + # On the mDNS port and in the group, so a responder that answers + # the group rather than the asker is heard too; a port already + # taken (a resolver daemon) means that daemon would have served + # the name, so the ephemeral port and unicast replies will do. + try: + sock.bind(("", MDNS_PORT)) + sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, + socket.inet_aton(MDNS_GROUP) + socket.inet_aton("0.0.0.0")) + except OSError: + sock.bind(("", 0)) + for _ in range(tries): + sock.sendto(query, (MDNS_GROUP, MDNS_PORT)) + deadline = time.time() + timeout + while True: + left = deadline - time.time() + if left <= 0: + break + sock.settimeout(left) + try: + data, _peer = sock.recvfrom(2048) + except socket.timeout: + break + except OSError: + break + for ip in mdns_answers(data, name): + return ip + return None + finally: + sock.close() + + +# -------------------------------------------------------------- client + +def load_config(path=None): + """The bench's fixture config, or None when there is none (a bench + without a fixture: the runner behaves exactly as before).""" + path = path or FIXTURE_CONFIG + try: + with open(path, "r", encoding="utf-8") as f: + cfg = json.load(f) + except OSError: + return None + except ValueError as e: + raise FixtureError("%s: not JSON: %s" % (path, e)) + if not isinstance(cfg, dict) or not cfg.get("key"): + raise FixtureError("%s: needs at least a key" % path) + cfg.setdefault("hostname", "forgefixture") + cfg.setdefault("ip", None) + cfg.setdefault("port", 80) + chans = cfg.get("channels") + if chans is None: + chans = list(CHANNELS) + bad = [c for c in chans if c not in CHANNELS] + if bad: + raise FixtureError("%s: unknown channel(s) %s" % (path, ", ".join(bad))) + cfg["channels"] = list(chans) + cfg["arm_press"] = bool(cfg.get("arm_press", False)) + return cfg + + +class Fixture: + """The actuator as the runner sees it: which channels it covers, an + action per channel, a release, its state. Every request carries the + key; the address comes from the config's ip or the mDNS lookup, + re-resolved when a request fails to connect.""" + + def __init__(self, cfg, timeout=3.0, resolver=resolve_mdns): + self.hostname = cfg["hostname"] + self.ip_override = cfg.get("ip") or None + self.port = int(cfg.get("port") or 80) + self.key = cfg["key"] + self.channels = tuple(cfg["channels"]) + self.arm_press = bool(cfg.get("arm_press", False)) + self.timeout = timeout + self._resolver = resolver + self._ip = self.ip_override + self._resolved_at = time.time() if self.ip_override else 0.0 + self._lock = threading.Lock() + self.last_state = None + + # -- address ------------------------------------------------------- + def address(self, refresh=False): + with self._lock: + if self.ip_override: + return self.ip_override + stale = time.time() - self._resolved_at > RESOLVE_TTL_S + if self._ip is None or refresh or stale: + ip = self._resolver(self.hostname) + if ip is None: + raise FixtureError("%s.local did not answer the mDNS lookup (set ip in the config " + "to skip it)" % self.hostname) + self._ip = ip + self._resolved_at = time.time() + return self._ip + + def where(self): + return "%s (%s)" % (self.hostname, self._ip or "unresolved") + + # -- requests ------------------------------------------------------ + def _request(self, method, path, body=None, retry=True): + ip = self.address() + data = json.dumps(body).encode() if body is not None else None + host = ip if self.port == 80 else "%s:%d" % (ip, self.port) + req = urllib.request.Request("http://%s%s" % (host, path), data=data, method=method, + headers={"X-Fixture-Key": self.key, "Host": host, + "Content-Type": "application/json"}) + try: + with urllib.request.urlopen(req, timeout=self.timeout) as resp: + return resp.status, json.loads(resp.read().decode("utf-8", "replace") or "{}") + except urllib.error.HTTPError as e: + try: + body = json.loads(e.read().decode("utf-8", "replace") or "{}") + except ValueError: + body = {} + return e.code, body + except (urllib.error.URLError, socket.timeout, OSError) as e: + if retry and not self.ip_override: + # the address may have moved: one fresh lookup, one retry + self.address(refresh=True) + return self._request(method, path, body, retry=False) + raise FixtureError("fixture %s: %s %s: %s" % (self.where(), method, path, e)) + + def status(self): + st, body = self._request("GET", "/") + if st == 401: + raise FixtureError("fixture %s refused the key" % self.where()) + if st != 200 or body.get("device") != "forgefixture": + raise FixtureError("fixture %s: unexpected answer %s %s" % (self.where(), st, body)) + self.last_state = body + return body + + def covers(self, channel): + """The runner asks this per action. The button needs the jumper + in, which the fixture reports; a channel it does not cover falls + back to the operator.""" + if channel not in self.channels: + return False + if channel == "button": + st = self.last_state + return bool(st and st.get("button_enabled")) + return True + + def act(self, channel, state): + if channel == "button": + if state != "press": + raise FixtureError("the button is only ever pressed") + st, body = self._request("POST", "/button", {}) + else: + if state not in ("open", "close"): + raise FixtureError("%s: unknown state %r" % (channel, state)) + st, body = self._request("POST", "/" + channel, {"state": state}) + if st != 200: + raise FixtureError("fixture %s: %s %s -> %s %s" % (self.where(), channel, state, st, + body.get("error") or body)) + self.last_state = body + return body + + def release(self): + st, body = self._request("POST", "/release", {}) + if st != 200: + raise FixtureError("fixture %s: release -> %s %s" % (self.where(), st, body.get("error") or body)) + self.last_state = body + return body + + @staticmethod + def energized(state): + """The channels a state report shows energized (a loop open, the + button pressed).""" + ch = (state or {}).get("channels") or {} + return [c for c, s in ch.items() if s in ("open", "pressed")] + + def summary(self): + st = self.last_state or {} + return {"hostname": self.hostname, "ip": self._ip, "channels": list(self.channels), + "button_enabled": bool(st.get("button_enabled")), "arm_press": self.arm_press, + "version": st.get("version"), "uptime_s": st.get("uptime_s")} + + +def probe(log, path=None, resolver=resolve_mdns): + """The bench's fixture, up and answering, or None: no config means + no fixture; a config whose fixture does not answer is logged and + treated the same, so the run goes to the operator.""" + try: + cfg = load_config(path) + except FixtureError as e: + log("fixture: %s - running without it" % e) + return None + if cfg is None: + return None + fx = Fixture(cfg, resolver=resolver) + try: + st = fx.status() + except FixtureError as e: + log("fixture: %s - running without it" % e) + return None + chans = ", ".join(fx.channels) + log("fixture up: %s at %s, v%s, covers %s%s" % ( + fx.hostname, fx._ip, st.get("version"), chans, + "" if st.get("button_enabled") or "button" not in fx.channels + else " (button disabled: the enable jumper is out)")) + return fx + + +__all__ = ["Fixture", "FixtureError", "CHANNELS", "FIXTURE_CONFIG", "load_config", "probe", + "resolve_mdns", "mdns_query", "mdns_answers"] diff --git a/forgetest/forgetest/page.py b/forgetest/forgetest/page.py index a03bbed..0ac749a 100644 --- a/forgetest/forgetest/page.py +++ b/forgetest/forgetest/page.py @@ -134,10 +134,11 @@ pre#log{background:#1d1e26;color:#d7dae0;font-family:ui-monospace,Consolas,monos -
+

Each queue takes every test of its kind that the campaign does not already count as satisfied, runs them one at a time in prerequisite order, and stops - on the first result that is not a PASS. The unattended queue needs nobody in the room. + on the first result that is not a PASS. The unattended queue needs nobody in the room; + with the bench fixture up, the operator tests it can perform by itself run there too. The other one does: it prompts, and it fires the laser. Stop-the-queue cancels what is still waiting and lets the run in progress finish; Abort ends that one too.

@@ -259,7 +260,15 @@ function render(){if(!state||!catalog)return; getting on. Built from the state, so a reload picks the queue back up exactly where it is - the queue lives in the runner, not in this tab. */ var QUEUES=[['unattended','Unattended'],['attended','Operator and live']]; +function renderFixture(){var f=state.fixture,e=$('fixture');if(!e)return; + if(!f){setHtml(e,'');return} + var ch=(f.channels||[]).filter(function(c){return c!=='button'||f.button_enabled}); + setHtml(e,'Bench fixture '+esc(f.hostname)+' at '+esc(f.ip||'?')+' (v'+esc(f.version||'?')+') covers '+ + (ch.length?esc(ch.join(', ')):'nothing')+''+ + ((f.channels||[]).indexOf('button')>=0&&!f.button_enabled?' (button disabled: enable jumper out)':'')+ + '; arm press: '+(f.arm_press?'the fixture':'the operator'))} function renderQueue(){var av=state.batch_available||{},b=state.batch,busy=isBusy(); + renderFixture(); QUEUES.forEach(function(p){var e=$('q-'+p[0]);if(!e)return; var ids=av[p[0]]||[]; setText(e,ids.length?(p[1]+' ('+ids.length+')'):(p[1]+' (none left)')); diff --git a/forgetest/forgetest/runner.py b/forgetest/forgetest/runner.py index 8d5a1a6..a4e5cff 100644 --- a/forgetest/forgetest/runner.py +++ b/forgetest/forgetest/runner.py @@ -49,6 +49,7 @@ import traceback from . import artifact as _artifact from . import baseline as _baseline from . import campaign as _campaign +from . import fixture as _fixture from . import catalog as _catalog from . import hw from .log import now_ts, data_dir @@ -65,6 +66,11 @@ BATCH_GROUPS = { "attended": ("operator", "live"), } +# A bench actuator (fixture.py) moves an operator test whose actions it +# covers into the unattended queue. The probe that decides is one GET; +# it is repeated at most this often, and before every run. +FIXTURE_PROBE_S = 30.0 + class Aborted(Exception): pass @@ -124,6 +130,7 @@ class Run: self.prompt = None # {"id","question","options"} self.notice = None # {"id","text"}: a standing instruction, no button self.answers = [] + self.unattended = False # a fixture-run test: no prompt can be answered self.evidence = {} self.baseline_captured = None # preserved state the post pass hands back self.aborted = threading.Event() @@ -156,6 +163,9 @@ class Run: # -- prompt channel ----------------------------------------------- def ask(self, question, options): + if self.unattended: + raise Failed("the test asked a person (%r) while running unattended with the fixture: " + "declare the step in hands=... so the test stays in the attended queue" % question) with self._cv: self._prompt_seq += 1 pid = "p%d" % self._prompt_seq @@ -258,7 +268,12 @@ class Context: def ready(self, text): """Pre-announce a timed step: what happens when the operator clicks Ready and what they do during it. Returns on the click; - the test then starts the thing and watches the machine.""" + the test then starts the thing and watches the machine. With the + fixture performing the step (the run is unattended) there is + nobody to announce it to: the gate passes at once, logged.""" + if self.run.unattended: + self.log("READY (fixture performs the step): %s", text) + return ans = self.prompt(text, ("Ready", "Cannot")) if ans != "Ready": raise Failed("operator could not: %s" % text) @@ -320,7 +335,15 @@ class Context: if fixture is not None and fixture.covers(channel): self.log("ACT %s %s (fixture)", channel, state) rec["by"] = "fixture" - fixture.act(channel, state) + try: + fixture.act(channel, state) + except _fixture.FixtureError as e: + # the box did not do it: the operator is asked instead, + # and the record says so + self.log("ACT %s %s: fixture failed (%s) - asking the operator", channel, state, e) + rec["by"] = "operator" + rec["fixture_error"] = str(e) + self.notice(wording) else: self.notice(wording) try: @@ -337,6 +360,43 @@ class Context: self.log("ACT %s %s: done after %.1f s", channel, state, dt) return dt + def arm_press(self, text="The button lights white: press it to arm. The machine fires after your press."): + """The arm cue of a live test. A person's press by default: a + standing notice until the caller clears it. The fixture presses + only where the bench opted in (arm_press in its config) and its + button channel is enabled: a thread waits for the button to light + (the job may still be on its way to the arm wait) and presses + once, recorded as the fixture's; if the button never lights or + the press fails, the notice goes up for a person. Returns True + when the fixture has been asked.""" + fixture = getattr(self.runner, "fixture", None) if self.runner is not None else None + rec = {"channel": "button", "state": "arm", "by": "operator", "ts": now_ts()} + self.evidence.setdefault("actions", []).append(rec) + if not (fixture is not None and fixture.arm_press and fixture.covers("button")): + self.notice(text) + return False + self.log("ARM: the fixture presses when the button lights (the bench's arm_press opt-in)") + + def press(): + lit = self.wait_for(hw.button_lit, 60) + if lit is None: + self.log("ARM: the button never lit within 60 s - asking the operator") + self.notice(text) + return + try: + fixture.act("button", "press") + except _fixture.FixtureError as e: + self.log("ARM press: fixture failed (%s) - asking the operator", e) + rec["fixture_error"] = str(e) + self.notice(text) + return + rec["by"] = "fixture" + rec["took_s"] = round(lit, 2) + self.log("ARM press by the fixture, button lit after %.1f s", lit) + + threading.Thread(target=press, daemon=True, name="forgetest-arm-press").start() + return True + # -- hardware ------------------------------------------------------ @property def forgectrl(self): @@ -473,7 +533,8 @@ class Runner: self.current = None self.last = None self.batch = None - self.fixture = None # the bench actuator, when one is configured (none yet) + self.fixture = None # the bench actuator, when one is up (fixture.py) + self._fixture_probed = 0.0 self.boot_ref = None self.recover() threading.Thread(target=self._take_boot_reference, daemon=True, @@ -492,6 +553,47 @@ class Runner: if r is not None and not r.finished: r.log(msg) + # -- the bench actuator ------------------------------------------------ + def probe_fixture(self, force=False): + """The fixture, up and answering, or None; re-probed at most every + FIXTURE_PROBE_S unless forced (before a run). A probe that finds + it gone, or a config that appeared, changes the queues' routing + from then on.""" + now = time.time() + if not force and now - self._fixture_probed < FIXTURE_PROBE_S: + return self.fixture + self._fixture_probed = now + had = self.fixture + fx = _fixture.probe(self._note if had is None else (lambda m: None)) + if fx is None and had is not None: + self._note("fixture: %s no longer answers - running without it" % had.hostname) + self.fixture = fx + return fx + + def fixture_channels(self): + """The channels the fixture covers right now (the button only + with its enable jumper in).""" + fx = self.probe_fixture() + if fx is None: + return () + return tuple(c for c in fx.channels if fx.covers(c)) + + def fixture_release(self, run): + """After a run: any channel the fixture still holds energized is + released and recorded, like any other leftover.""" + fx = self.fixture + if fx is None: + return + try: + held = fx.energized(fx.status()) + if held: + fx.release() + run.log("fixture: released %s left energized" % ", ".join(held)) + run.evidence.setdefault("fixture", {})["released"] = held + except _fixture.FixtureError as e: + run.log("fixture: release check failed: %s" % e) + run.evidence.setdefault("fixture", {})["release_error"] = str(e) + # -- startup recovery ------------------------------------------------ def recover(self): m = marker_path() @@ -532,6 +634,8 @@ class Runner: # what each queue would run if started now, so the page can label # its buttons with the work rather than a bare verb st["batch_available"] = {g: self.batch_selection(g, st) for g in BATCH_GROUPS} + fx = self.fixture + st["fixture"] = fx.summary() if fx is not None else None return st, records def busy(self): @@ -605,6 +709,16 @@ class Runner: self.last = self.current self.current = run run.log("start %s (%s, %s) in campaign %s" % (t.id, t.kind, t.hardware, campaign["id"])) + fx = self.probe_fixture(force=True) + if fx is not None: + run.log("fixture: %s at %s covers %s" % (fx.hostname, fx._ip, ", ".join(self.fixture_channels()))) + run.evidence["fixture"] = {"hostname": fx.hostname, "ip": fx._ip, + "channels": list(self.fixture_channels())} + # nobody is expected in the room for a test the fixture runs + # in the unattended queue: a prompt there is a defect, not a wait + if batch is not None and batch.get("group") == "unattended" and t.kind != "auto": + run.unattended = True + run.log("fixture: running unattended (no prompt can be answered)") if missing: run.log("prerequisites overridden by the operator - not satisfied: %s" % ", ".join(missing)) run.evidence["prerequisites"] = {"overridden": True, "missing": missing, "ts": now_ts()} @@ -633,8 +747,13 @@ class Runner: if state is None: state, _ = self.state() tests = self.tests() + # an operator test the fixture can run alone goes to the + # unattended queue and leaves the attended one + channels = self.fixture_channels() + routed = set(t.id for t in tests if channels and t.fixture_runnable(channels)) want = [t.id for t in tests - if t.kind in kinds and not state["tests"][t.id]["satisfied"]] + if ((t.kind in kinds and t.id not in routed) or (group == "unattended" and t.id in routed)) + and not state["tests"][t.id]["satisfied"]] return _catalog.order_by_requires(tests, want) def start_batch(self, group, ack_live=False, ignore_requires=False): @@ -775,6 +894,7 @@ class Runner: except Exception as e: # noqa: BLE001 - an erroring test is a failed test result, message = _campaign.ERROR, "%s: %s" % (type(e).__name__, e) run.log(traceback.format_exc().rstrip()) + self.fixture_release(run) try: self._baseline_post(run, captured) except Exception as e: # noqa: BLE001 - never lose the result over the cleanup diff --git a/forgetest/forgetest/suite/cloud.py b/forgetest/forgetest/suite/cloud.py index 0126621..37381b7 100644 --- a/forgetest/forgetest/suite/cloud.py +++ b/forgetest/forgetest/suite/cloud.py @@ -385,7 +385,7 @@ def mode_switch(ctx): "machine's identity", subsystem="cloud", kind="operator", est_min=5, covers=_SERVICE_LAYER, - requires=["forgectrl.auth"], + requires=["forgectrl.auth"], hands=["app"], steps=["Cloud credentials configured; the machine on the network; the app open in a browser " "(anyone can drive it, at the machine or not: nothing here moves, arms, or fires).", "When told, set up any small job in the app and press Print; the emulator runs it at " diff --git a/forgetest/forgetest/suite/laser.py b/forgetest/forgetest/suite/laser.py index 84ad63c..828669a 100644 --- a/forgetest/forgetest/suite/laser.py +++ b/forgetest/forgetest/suite/laser.py @@ -153,7 +153,7 @@ def arm_and_fire(ctx, g, room="40 mm +X and +Y", job=None, timeout=240): button was never pressed.""" ctx.ready(ARM_CUE % room) stream(g, job or MARK_JOB) - ctx.notice("The button lights white: press it to arm. The machine fires after your press.") + ctx.arm_press() t0 = time.time() while time.time() - t0 < timeout: ctx.checkpoint() @@ -355,7 +355,7 @@ def emission_witness(ctx): job = ["G91", "G21", "M4", "S400", "G1 X20 F600", "G1 Y20 F600", "G1 X-20 F600", "G1 Y-20 F600", "M5", "G90", "M2"] - ctx.notice("The button lights white: press it to arm. The machine fires after your press.") + ctx.arm_press() try: samples = run_and_sample(ctx, g, job) finally: diff --git a/forgetest/tests/helpers.py b/forgetest/tests/helpers.py index 357090f..619de4f 100644 --- a/forgetest/tests/helpers.py +++ b/forgetest/tests/helpers.py @@ -70,10 +70,10 @@ def _noop(ctx): def make_test(id, covers, always=False, requires=(), kind="auto", fn=None, subsystem=None, mode=None, - actions=(), precheck=None, steps=()): + actions=(), precheck=None, steps=(), hands=()): return catalog_mod.Test(id, "Title " + id, subsystem or id.split(".")[0], kind, "api", covers, requires, always, 1, steps, "desc", fn or _noop, mode=mode, - actions=actions, precheck=precheck) + actions=actions, precheck=precheck, hands=hands) def registry(*tests): diff --git a/forgetest/tests/test_fixture.py b/forgetest/tests/test_fixture.py new file mode 100644 index 0000000..4aa729d --- /dev/null +++ b/forgetest/tests/test_fixture.py @@ -0,0 +1,474 @@ +"""The bench actuator as the tool sees it. + +The fixture is a box on the network that opens the lid loop, pulls the +interlock loop and presses the button on request. What has to hold: the +tool finds it by name without a resolver on the image, speaks its API +under the key, falls back to the operator when it cannot act, moves an +operator test it can perform alone into the unattended queue (never a +live one, never one that needs a person for anything else), refuses a +prompt during such a run rather than hanging on it, releases what the +box still holds after every run, and presses the arm button only where +the bench opted in. +""" +import json +import os +import shutil +import socket +import struct +import tempfile +import threading +import time +import unittest +from http.server import BaseHTTPRequestHandler, HTTPServer + +import helpers +from forgetest import fixture as fx +from forgetest import runner as runner_mod +from forgetest.log import Log +from forgetest.runner import Context, Failed, Run, Runner + +KEY = "0123456789abcdef0123456789abcdef" + + +class FakeFixture: + """The device's API, as fixture/README.md describes it.""" + + def __init__(self, button_enabled=True): + self.state = {"lid": "closed", "interlock": "closed", "button": "idle"} + self.button_enabled = button_enabled + self.calls = [] + srv = self + + class H(BaseHTTPRequestHandler): + def log_message(self, *a): + pass + + def _json(self, code, obj): + body = json.dumps(obj).encode() + self.send_response(code) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def _state(self): + return {"device": "forgefixture", "hostname": "forgefixture", "version": "1.0.0", + "idf": "v5.5.5", "uptime_s": 12, "channels": dict(srv.state), + "button_enabled": srv.button_enabled, "button_pulsing": False, + "wifi": {"connected": True, "ip": "127.0.0.1", "rssi": -50}} + + def _auth(self): + if self.headers.get("X-Fixture-Key") != KEY: + self._json(401, {"error": "X-Fixture-Key missing or wrong"}) + return False + return True + + def do_GET(self): + srv.calls.append(("GET", self.path)) + if not self._auth(): + return + self._json(200, self._state()) + + def do_POST(self): + n = int(self.headers.get("Content-Length") or 0) + body = json.loads(self.rfile.read(n).decode() or "{}") + srv.calls.append(("POST", self.path, body)) + if not self._auth(): + return + if self.path in ("/lid", "/interlock"): + st = body.get("state") + if st not in ("open", "close", "closed"): + return self._json(400, {"error": "state must be \"open\" or \"close\""}) + srv.state[self.path[1:]] = "open" if st == "open" else "closed" + return self._json(200, self._state()) + if self.path == "/button": + if not srv.button_enabled: + return self._json(409, {"error": "button disabled: the enable jumper is out"}) + srv.state["button"] = "pressed" + threading.Timer(0.05, lambda: srv.state.__setitem__("button", "idle")).start() + s = self._state() + s["pulse_ms"] = 200 + return self._json(200, s) + if self.path == "/release": + srv.state = {"lid": "closed", "interlock": "closed", "button": "idle"} + return self._json(200, self._state()) + self._json(404, {"error": "no such path"}) + + self.httpd = HTTPServer(("127.0.0.1", 0), H) + self.port = self.httpd.server_address[1] + threading.Thread(target=self.httpd.serve_forever, daemon=True).start() + + def stop(self): + self.httpd.shutdown() + self.httpd.server_close() + + +class MdnsTests(unittest.TestCase): + def test_query_packet(self): + q = fx.mdns_query("forgefixture.local") + self.assertEqual(q[:12], struct.pack("!HHHHHH", 0, 0, 1, 0, 0, 0)) + self.assertEqual(q[12:], b"\x0cforgefixture\x05local\x00" + struct.pack("!HH", 1, 0x8001)) + + def _response(self, name, ip, extra_name=None): + labels = b"".join(struct.pack("B", len(p)) + p.encode() for p in name.split(".")) + b"\x00" + hdr = struct.pack("!HHHHHH", 0, 0x8400, 0, 2 if extra_name else 1, 0, 0) + rr = labels + struct.pack("!HHIH", 1, 0x8001, 120, 4) + socket.inet_aton(ip) + if extra_name: + # a second answer naming the first through a compression pointer + # to offset 12 (the first name), plus its own first label + other = struct.pack("B", len(extra_name)) + extra_name.encode() + struct.pack("!H", 0xC000 | 12) + rr += other + struct.pack("!HHIH", 1, 0x8001, 120, 4) + socket.inet_aton("10.0.0.9") + return hdr + rr + + def test_answers_are_the_named_a_records(self): + data = self._response("forgefixture.local", "192.0.2.50") + self.assertEqual(fx.mdns_answers(data, "forgefixture.local"), ["192.0.2.50"]) + self.assertEqual(fx.mdns_answers(data, "FORGEFIXTURE.local."), ["192.0.2.50"]) + self.assertEqual(fx.mdns_answers(data, "other.local"), []) + # a query (QR clear) is never an answer + self.assertEqual(fx.mdns_answers(fx.mdns_query("forgefixture.local"), "forgefixture.local"), []) + # compression pointers are followed; the other record is not ours + data2 = self._response("forgefixture.local", "192.0.2.50", extra_name="printer") + self.assertEqual(fx.mdns_answers(data2, "forgefixture.local"), ["192.0.2.50"]) + self.assertEqual(fx.mdns_answers(data2, "printer.forgefixture.local"), ["10.0.0.9"]) + # garbage does not raise + self.assertEqual(fx.mdns_answers(b"\x00\x01", "forgefixture.local"), []) + + +class ConfigTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="forgetest-fixture-") + + def tearDown(self): + shutil.rmtree(self.tmp, ignore_errors=True) + + def write(self, obj): + p = os.path.join(self.tmp, "fixture.json") + with open(p, "w") as f: + json.dump(obj, f) + return p + + def test_no_file_is_no_fixture(self): + self.assertIsNone(fx.load_config(os.path.join(self.tmp, "none.json"))) + + def test_defaults_and_checks(self): + cfg = fx.load_config(self.write({"key": KEY})) + self.assertEqual(cfg["hostname"], "forgefixture") + self.assertEqual(cfg["channels"], ["lid", "interlock", "button"]) + self.assertFalse(cfg["arm_press"]) + self.assertIsNone(cfg["ip"]) + with self.assertRaises(fx.FixtureError): + fx.load_config(self.write({"hostname": "x"})) # no key + with self.assertRaises(fx.FixtureError): + fx.load_config(self.write({"key": KEY, "channels": ["lid", "laser"]})) + p = self.write({"key": KEY}) + with open(p, "w") as f: + f.write("{not json") + with self.assertRaises(fx.FixtureError): + fx.load_config(p) + + +class ClientTests(unittest.TestCase): + def setUp(self): + self.dev = FakeFixture() + self.resolved = [] + + def tearDown(self): + self.dev.stop() + + def resolver(self, hostname): + self.resolved.append(hostname) + return "127.0.0.1" + + def client(self, **over): + cfg = {"hostname": "forgefixture", "key": KEY, "channels": ["lid", "interlock", "button"], + "port": self.dev.port} + cfg.update(over) + return fx.Fixture(cfg, resolver=self.resolver) + + def test_status_covers_act_release(self): + f = self.client() + st = f.status() + self.assertEqual(st["device"], "forgefixture") + self.assertEqual(self.resolved, ["forgefixture"]) # resolved once + self.assertTrue(f.covers("lid") and f.covers("interlock") and f.covers("button")) + f.act("lid", "open") + self.assertEqual(self.dev.state["lid"], "open") + f.act("interlock", "open") + f.act("button", "press") + self.assertEqual(self.dev.calls[-1], ("POST", "/button", {})) + time.sleep(0.15) # the fake's pulse ends + self.assertEqual(sorted(fx.Fixture.energized(f.status())), ["interlock", "lid"]) + f.release() + self.assertEqual(fx.Fixture.energized(f.status()), []) + self.assertEqual(self.resolved, ["forgefixture"]) # the address is cached + with self.assertRaises(fx.FixtureError): + f.act("button", "hold") + with self.assertRaises(fx.FixtureError): + f.act("lid", "ajar") + + def test_the_button_is_covered_only_with_the_jumper_in(self): + self.dev.button_enabled = False + f = self.client() + f.status() + self.assertTrue(f.covers("lid")) + self.assertFalse(f.covers("button")) + with self.assertRaises(fx.FixtureError) as cm: + f.act("button", "press") + self.assertIn("jumper", str(cm.exception)) + + def test_a_wrong_key_is_refused(self): + f = self.client(key="wrong") + with self.assertRaises(fx.FixtureError) as cm: + f.status() + self.assertIn("refused the key", str(cm.exception)) + + def test_an_ip_override_skips_the_lookup(self): + f = self.client(ip="127.0.0.1") + f.status() + self.assertEqual(self.resolved, []) + + def test_channels_not_wired_are_not_covered(self): + f = self.client(channels=["lid"]) + f.status() + self.assertTrue(f.covers("lid")) + self.assertFalse(f.covers("interlock")) + + def test_probe_without_a_config_is_none_and_a_silent_box_is_logged(self): + lines = [] + self.assertIsNone(fx.probe(lines.append, path=os.path.join(tempfile.gettempdir(), "no-such-fixture.json"))) + self.assertEqual(lines, []) + tmp = tempfile.mkdtemp(prefix="forgetest-fixture-") + try: + p = os.path.join(tmp, "fixture.json") + with open(p, "w") as f: + json.dump({"key": KEY, "ip": "127.0.0.1", "port": 1}, f) + # port 1 on localhost: nothing listens there + f2 = fx.probe(lines.append, path=p, resolver=lambda h: None) + self.assertIsNone(f2) + self.assertTrue(lines and "running without it" in lines[-1], lines) + finally: + shutil.rmtree(tmp, ignore_errors=True) + + +class CatalogTests(unittest.TestCase): + def test_fixture_runnable(self): + op = helpers.make_test("m.op", [], kind="operator", actions=("lid", "button")) + self.assertTrue(op.fixture_runnable(("lid", "interlock", "button"))) + self.assertFalse(op.fixture_runnable(("lid",))) # the button is not covered + self.assertFalse(helpers.make_test("m.live", [], kind="live", actions=("lid",)) + .fixture_runnable(("lid",))) # live never downgrades + self.assertFalse(helpers.make_test("m.app", [], kind="operator", hands=("app",)) + .fixture_runnable(("lid", "interlock", "button"))) + self.assertFalse(helpers.make_test("m.none", [], kind="operator") + .fixture_runnable(("lid",))) # no actions: a person does something + self.assertFalse(helpers.make_test("m.auto", []).fixture_runnable(("lid",))) + + def test_an_auto_test_asks_nothing_of_a_person(self): + from forgetest import catalog + with self.assertRaises(ValueError): + catalog.test("z.auto", title="t", subsystem="z", hands=("app",))(lambda ctx: None) + + +class StubFixture: + """What the runner needs of a fixture, scripted.""" + + def __init__(self, channels=("lid", "interlock", "button"), button_enabled=True, arm_press=False, + fail=False, fc=None): + self.fc = fc # the fake forgectrl whose switches follow the actions + self.hostname = "forgefixture" + self._ip = "127.0.0.1" + self.channels = tuple(channels) + self.button_enabled = button_enabled + self.arm_press = arm_press + self.fail = fail + self.acts = [] + self.held = [] + self.released = 0 + + def covers(self, channel): + return channel in self.channels and (channel != "button" or self.button_enabled) + + def act(self, channel, state): + if self.fail: + raise fx.FixtureError("the box is off") + self.acts.append((channel, state)) + if state == "open": + self.held.append(channel) + if self.fc is not None and channel in ("lid", "interlock"): + sw = self.fc.state["status"]["switches"] + sw["lid" if channel == "lid" else "interlock_ok"] = state != "open" + + def status(self): + return {"channels": {c: ("open" if c in self.held else "closed") for c in ("lid", "interlock")}} + + @staticmethod + def energized(state): + return fx.Fixture.energized(state) + + def release(self): + self.released += 1 + self.held = [] + + def summary(self): + return {"hostname": self.hostname, "ip": self._ip, "channels": list(self.channels), + "button_enabled": self.button_enabled, "arm_press": self.arm_press} + + +def t_lid(ctx): + ctx.ready("On Ready the lid opens") + ctx.act("lid", "open") + ctx.act("lid", "close") + + +def t_asks(ctx): + ctx.act("lid", "open") + ctx.confirm("Did it?") + + +class RoutingTests(unittest.TestCase): + """The queues with a fixture up: the runner's probe is scripted.""" + + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="forgetest-fixture-run-") + os.environ["FORGETEST_DATA"] = self.tmp + os.environ["FORGETEST_MARKER"] = os.path.join(self.tmp, "marker") + self.fc = helpers.FakeForgectrl().start() + self.man = helpers.make_manifest() + self.reg = helpers.registry( + helpers.make_test("r.auto", [("forgectrl", "src/ui.c")]), + helpers.make_test("r.lid", [("forgectrl", "src/auth.c")], kind="operator", actions=("lid",), fn=t_lid), + helpers.make_test("r.btn", [("forgectrl", "src/cool.c")], kind="operator", actions=("button",), fn=t_lid), + helpers.make_test("r.app", [("forgectrl", "src/main.c")], kind="operator", actions=("lid",), + hands=("app",), fn=t_lid), + helpers.make_test("r.live", [("grblhal-glowforge", "src/**")], kind="live", actions=("lid",)), + helpers.make_test("r.asks", [("linux-fslc", "**")], kind="operator", actions=("lid",), fn=t_asks), + ) + self.log = Log(os.path.join(self.tmp, "results.jsonl")) + self.runner = Runner(self.log, self.man, self.reg) + self.saved_probe = fx.probe + self.stub = StubFixture(fc=self.fc) + fx.probe = lambda log, path=None, resolver=None: self.stub + + def tearDown(self): + fx.probe = self.saved_probe + self.fc.stop() + for k in ("FORGETEST_DATA", "FORGETEST_MARKER"): + os.environ.pop(k, None) + shutil.rmtree(self.tmp, ignore_errors=True) + + def selection(self): + st, _ = self.runner.state() + return st["batch_available"], st["fixture"] + + def test_operator_tests_the_fixture_covers_move_to_the_unattended_queue(self): + av, summary = self.selection() + self.assertEqual(av["unattended"], ["r.auto", "r.lid", "r.btn", "r.asks"]) + self.assertEqual(av["attended"], ["r.app", "r.live"]) # needs the app; live never moves + self.assertEqual(summary["hostname"], "forgefixture") + + def test_without_the_jumper_the_button_tests_stay_attended(self): + self.stub.button_enabled = False + av, _ = self.selection() + self.assertIn("r.btn", av["attended"]) + self.assertIn("r.lid", av["unattended"]) + + def test_no_fixture_is_the_old_routing(self): + fx.probe = lambda log, path=None, resolver=None: None + self.runner._fixture_probed = 0.0 + av, summary = self.selection() + self.assertEqual(av["unattended"], ["r.auto"]) + self.assertIsNone(summary) + + def run_queue(self, group, ack_live=False): + ok, msg, order = self.runner.start_batch(group, ack_live=ack_live) + self.assertTrue(ok, msg) + deadline = time.time() + 30 + while not self.runner.batch["finished"] and time.time() < deadline: + time.sleep(0.05) + return self.runner.batch_snapshot(), order + + def last_result(self, tid): + for rec in reversed(self.log.read()): + if rec.get("t") == "result" and rec["test"] == tid: + return rec + return None + + def test_the_fixture_performs_the_actions_and_the_ready_gate_passes(self): + # r.asks confirms after its action: a person is asked -> a FAIL that + # says so, and the queue stops there, after r.lid and r.btn passed + b, order = self.run_queue("unattended") + self.assertEqual(order, ["r.auto", "r.lid", "r.btn", "r.asks"]) + results = {x["test"]: x["result"] for x in b["done"]} + self.assertEqual(results["r.lid"], "PASS") + self.assertEqual(results["r.btn"], "PASS") + self.assertEqual(results["r.asks"], "FAIL") + rec = self.last_result("r.lid") + acts = rec["evidence"]["actions"] + self.assertEqual([(a["channel"], a["state"], a["by"]) for a in acts], + [("lid", "open", "fixture"), ("lid", "close", "fixture")]) + self.assertTrue(any("READY (fixture performs the step)" in l for l in rec["log"]), rec["log"]) + self.assertEqual(rec["evidence"]["fixture"]["channels"], ["lid", "interlock", "button"]) + asks = self.last_result("r.asks") + self.assertIn("declare the step in hands=", asks["message"]) + + def test_a_failing_box_hands_the_action_to_the_operator(self): + self.stub.fail = True + run = Run("test", "r.lid", "r.lid") + ctx = Context(run, self.runner, self.reg["r.lid"]) + self.runner.probe_fixture(force=True) + seen = [] + ctx.wait_for = lambda cond, timeout, poll=0.25: 1.0 # the machine shows the state + run.set_notice = lambda text: seen.append(text) + ctx.act("lid", "open") + rec = run.evidence["actions"][0] + self.assertEqual(rec["by"], "operator") + self.assertIn("the box is off", rec["fixture_error"]) + self.assertTrue(seen and "lid" in seen[0].lower()) + + def test_what_the_box_still_holds_is_released_after_a_run(self): + def holds(ctx): + ctx.act("lid", "open") # and never closes it + self.reg["r.hold"] = helpers.make_test("r.hold", [("forgectrl", "src/ui.c")], kind="operator", + actions=("lid",), fn=holds) + self.runner.registry = self.reg + ok, msg, run = self.runner._start_test("r.hold") + self.assertTrue(ok, msg) + while run.finished is None: + time.sleep(0.05) + self.assertEqual(self.stub.released, 1) + rec = self.last_result("r.hold") + self.assertEqual(rec["evidence"]["fixture"]["released"], ["lid"]) + + def test_the_arm_press_is_the_operators_unless_the_bench_opted_in(self): + run = Run("test", "r.live", "r.live") + ctx = Context(run, self.runner, self.reg["r.live"]) + self.runner.probe_fixture(force=True) + seen = [] + run.set_notice = lambda text: seen.append(text) + self.assertFalse(ctx.arm_press()) + self.assertEqual(run.evidence["actions"][0]["by"], "operator") + self.assertTrue(seen) + # opted in: the press waits for the button to light + self.stub.arm_press = True + lit = {"v": False} + saved = runner_mod.hw.button_lit + runner_mod.hw.button_lit = lambda: lit["v"] + try: + seen[:] = [] + self.assertTrue(ctx.arm_press()) + time.sleep(0.3) + self.assertEqual(self.stub.acts, []) # not yet: dark button + lit["v"] = True + deadline = time.time() + 5 + while not self.stub.acts and time.time() < deadline: + time.sleep(0.05) + self.assertEqual(self.stub.acts, [("button", "press")]) + self.assertEqual(run.evidence["actions"][1]["by"], "fixture") + self.assertEqual(seen, []) # no notice went up + finally: + runner_mod.hw.button_lit = saved + + +if __name__ == "__main__": + unittest.main()