From ba0bf41749cb5252f0e4e10b2b222f8665502ea1 Mon Sep 17 00:00:00 2001 From: ScottW514 Date: Thu, 3 Sep 2026 16:55:34 -0400 Subject: [PATCH] forgetest: the operator proves presence at the machine, the bench presses A live test asked a person to click Ready on a page and then make timing-critical presses in the middle of a burning cut. That is how tonight's pause test became unanswerable: the actuator had dropped off the network, the harness fell back to the operator without saying so, and afterwards nobody could tell a second press from the machine resuming on its own. Where an actuator is up and wired to the button, the ready gate now takes a press on the machine's own button as the presence check, and the actuator performs every press in that test. The button does nothing at Idle, so the press is only a presence check, and the gate waits for the release so it is never read as the arm press. With no actuator the operator does the presses and answers on the page, as before. An actuator lost after that takeover is now said out loud, in the log and in the evidence, instead of quietly becoming a person's press. The live-fire cue was four lines of machine-shaped prose. It is now what a person needs: protection, exhaust, extinguisher, scrap, lid. --- forgetest/forgetest/runner.py | 70 +- forgetest/forgetest/suite/laser.py | 6 +- forgetest/tests/test_fixture.py | 1166 +++++++++++++++------------- 3 files changed, 686 insertions(+), 556 deletions(-) diff --git a/forgetest/forgetest/runner.py b/forgetest/forgetest/runner.py index c3f1248..bd866d9 100644 --- a/forgetest/forgetest/runner.py +++ b/forgetest/forgetest/runner.py @@ -116,6 +116,10 @@ ACTION_TEXT = { ("button", "press"): "Press the button once.", } ACT_TIMEOUT_S = 180 +# How long a live test waits for the operator to prove they are at the +# machine by pressing its button. Generous: they may be setting up the +# scrap, and nothing fires until it lands. +PRESENCE_TIMEOUT_S = 600 class Run: @@ -131,6 +135,10 @@ class Run: self.notice = None # {"id","text"}: a standing instruction, no button self.answers = [] self.unattended = False # a fixture-run test: no prompt can be answered + # The operator proved they are at the machine by pressing its + # button, so the bench actuator performs every press from there + # on. Set by Ctx.ready() when an actuator is up to take over. + self.fixture_takeover = False self.evidence = {} self.baseline_captured = None # preserved state the post pass hands back self.aborted = threading.Event() @@ -266,14 +274,45 @@ class Context: raise Failed("operator could not: %s" % text) 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. With the - fixture performing the step (the run is unattended) there is - nobody to announce it to: the gate passes at once, logged.""" + """Pre-announce a timed step and wait for the operator to say the + machine is set up. + + Where an actuator is up and wired to the button, the operator + proves they are at the machine by pressing the machine's own + button, and the actuator then performs every press in the test. + A person doing timing-critical presses in the middle of a live + cut is how a press count becomes unknowable: a press that lands + late, or twice, cannot be told apart afterwards from the machine + misbehaving. The button does nothing at Idle, so this press is + only a presence check. + + With no actuator the operator does the presses and answers on the + page, as before. With the run unattended there is nobody to + announce it to and the gate passes at once, logged.""" if self.run.unattended: self.log("READY (fixture performs the step): %s", text) return + + fixture = getattr(self.runner, "fixture", None) if self.runner is not None else None + if fixture is not None and fixture.covers("button"): + self.notice(text + " Then press the button on the machine to start.") + self.log("READY: waiting for the operator's press on the machine " + "(the bench then does every press in this test)") + pressed = self.wait_for(lambda: self.switch("button") is True, PRESENCE_TIMEOUT_S) + self.clear_notice() + if pressed is None: + raise Failed("no press on the machine within %d s: %s" + % (PRESENCE_TIMEOUT_S, text)) + # Let the button come back up, so this press is never read as + # the arm press or a pause toggle. + self.wait_for(lambda: self.switch("button") is False, 10) + self.run.fixture_takeover = True + self.evidence.setdefault("actions", []).append( + {"channel": "button", "state": "presence", "by": "operator", + "took_s": round(pressed, 2), "ts": now_ts()}) + self.log("READY: operator present (press after %.1f s); the bench takes the presses", pressed) + return + ans = self.prompt(text, ("Ready", "Cannot")) if ans != "Ready": raise Failed("operator could not: %s" % text) @@ -332,6 +371,15 @@ class Context: fixture = getattr(self.runner, "fixture", None) if self.runner is not None else None rec = {"channel": channel, "state": state, "by": "operator", "ts": now_ts()} self.evidence.setdefault("actions", []).append(rec) + # The test began with the bench taking the presses. If it is gone + # now, say so loudly: falling back to the operator silently is + # what turns one dropped actuator into a press nobody can account + # for afterwards. + if self.run.fixture_takeover and not (fixture is not None and fixture.covers(channel)): + rec["fixture_lost"] = True + self.log("ACT %s %s: WARNING the bench actuator took this test's presses and is now " + "gone - asking the operator instead; a press from here on is a person's", + channel, state) if fixture is not None and fixture.covers(channel): self.log("ACT %s %s (fixture)", channel, state) rec["by"] = "fixture" @@ -377,7 +425,17 @@ class Context: 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")): + # A presence press at the ready gate hands this test's presses to + # the bench, whatever the config's standing opt-in says: the + # operator has already proved they are at the machine, which is + # the thing the opt-in exists to establish. + opted_in = fixture is not None and fixture.covers("button") and \ + (fixture.arm_press or self.run.fixture_takeover) + if not opted_in: + if self.run.fixture_takeover: + rec["fixture_lost"] = True + self.log("ARM: WARNING the bench actuator took this test's presses and is now " + "gone - asking the operator for the arm press") self.notice(text) return False self.log("ARM: the fixture presses when the button lights (the bench's arm_press opt-in)") diff --git a/forgetest/forgetest/suite/laser.py b/forgetest/forgetest/suite/laser.py index 1545e92..8256ec2 100644 --- a/forgetest/forgetest/suite/laser.py +++ b/forgetest/forgetest/suite/laser.py @@ -60,10 +60,8 @@ def _grbl_settings(lines): pass return vals -ARM_CUE = ("LIVE FIRE. Eye protection on, exhaust running, fire watch and extinguisher in reach, " - "scrap under the head with room to move (%s), lid closed. On Ready the job starts: " - "the white button lights and the stream waits for your press of the physical arm " - "button; the machine fires only after that press.") +ARM_CUE = ("LIVE FIRE. Eye protection on. Exhaust on. Extinguisher in reach. " + "Put scrap under the head with %s clear, and close the lid.") # The beam detector's emission signature: this much over its pre-fire # level while the beam is on (bench: ~1834 idle, 2600-2890 at S300/S400). diff --git a/forgetest/tests/test_fixture.py b/forgetest/tests/test_fixture.py index e7bcffb..f3a8a4a 100644 --- a/forgetest/tests/test_fixture.py +++ b/forgetest/tests/test_fixture.py @@ -1,546 +1,620 @@ -"""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 = [] - self.presses = 0 # presses the device performed - 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": srv.state["button"] == "pressed", - "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"}) - if srv.state["button"] == "pressed": - return self._json(409, {"error": "a button pulse is in progress"}) - srv.state["button"] = "pressed" - srv.presses += 1 - 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_two_presses_are_spaced_so_the_controller_sees_the_release(self): - f = self.client() - f.status() - t0 = time.time() - f.act("button", "press") - f.act("button", "press") - took = time.time() - t0 - posts = [c for c in self.dev.calls if c[:2] == ("POST", "/button")] - self.assertEqual(len(posts), 2) # no 409 round trip was needed - self.assertEqual(self.dev.presses, 2) - # the second waited for the first pulse (200 ms as reported) and the gap - self.assertGreaterEqual(took, 0.2 + fx.BUTTON_GAP_S - 0.05) - - def test_a_pulse_in_progress_is_waited_out_then_retried(self): - f = self.client() - f.status() - self.dev.state["button"] = "pressed" # a press this client did not time - threading.Timer(0.3, lambda: self.dev.state.__setitem__("button", "idle")).start() - f.act("button", "press") - posts = [c for c in self.dev.calls if c[:2] == ("POST", "/button")] - self.assertEqual(len(posts), 2) # the 409, then the press - self.assertEqual(self.dev.presses, 1) - - def test_a_pulse_that_never_ends_is_the_fixtures_error(self): - f = self.client() - f.status() - self.dev.state["button"] = "pressed" - with self.assertRaises(fx.FixtureError) as cm: - f.act("button", "press") - self.assertIn("in progress", 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 test_a_queue_started_during_a_probe_waits_for_the_fixture(self): - # The daemon just came up: nothing probed yet, a page poll starts - # the first probe (slow: an mDNS answer), and the queue start - # lands while it is in flight. The start must see the fixture. - stub = self.stub - - def slow_probe(log, path=None, resolver=None): - time.sleep(0.5) - return stub - fx.probe = slow_probe - self.runner._fixture_probed = 0.0 - self.runner.fixture = None - poll = threading.Thread(target=self.runner.probe_fixture) - poll.start() - time.sleep(0.1) - order = self.runner.batch_selection("unattended") - poll.join() - self.assertIn("r.lid", order) - self.assertIn("r.btn", order) - - 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_a_failing_box_in_an_unattended_run_ends_the_test_as_an_error(self): - # nobody is in the room: the run must not wait ACT_TIMEOUT_S for a - # hand that is not there; it ends at once, as the harness's error - self.stub.fail = True - t0 = time.time() - b, order = self.run_queue("unattended") - results = {x["test"]: x["result"] for x in b["done"]} - self.assertEqual(results["r.lid"], "ERROR") - self.assertLess(time.time() - t0, 20) - rec = self.last_result("r.lid") - self.assertIn("the fixture could not perform a step", rec["message"]) - self.assertIn("the box is off", rec["message"]) - act = rec["evidence"]["actions"][0] - self.assertEqual((act["by"], act["fixture_error"]), ("fixture", "the box is off")) - self.assertFalse(any("asking the operator" in l for l in rec["log"]), rec["log"]) - - 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() +"""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 = [] + self.presses = 0 # presses the device performed + 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": srv.state["button"] == "pressed", + "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"}) + if srv.state["button"] == "pressed": + return self._json(409, {"error": "a button pulse is in progress"}) + srv.state["button"] = "pressed" + srv.presses += 1 + 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_two_presses_are_spaced_so_the_controller_sees_the_release(self): + f = self.client() + f.status() + t0 = time.time() + f.act("button", "press") + f.act("button", "press") + took = time.time() - t0 + posts = [c for c in self.dev.calls if c[:2] == ("POST", "/button")] + self.assertEqual(len(posts), 2) # no 409 round trip was needed + self.assertEqual(self.dev.presses, 2) + # the second waited for the first pulse (200 ms as reported) and the gap + self.assertGreaterEqual(took, 0.2 + fx.BUTTON_GAP_S - 0.05) + + def test_a_pulse_in_progress_is_waited_out_then_retried(self): + f = self.client() + f.status() + self.dev.state["button"] = "pressed" # a press this client did not time + threading.Timer(0.3, lambda: self.dev.state.__setitem__("button", "idle")).start() + f.act("button", "press") + posts = [c for c in self.dev.calls if c[:2] == ("POST", "/button")] + self.assertEqual(len(posts), 2) # the 409, then the press + self.assertEqual(self.dev.presses, 1) + + def test_a_pulse_that_never_ends_is_the_fixtures_error(self): + f = self.client() + f.status() + self.dev.state["button"] = "pressed" + with self.assertRaises(fx.FixtureError) as cm: + f.act("button", "press") + self.assertIn("in progress", 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 test_a_queue_started_during_a_probe_waits_for_the_fixture(self): + # The daemon just came up: nothing probed yet, a page poll starts + # the first probe (slow: an mDNS answer), and the queue start + # lands while it is in flight. The start must see the fixture. + stub = self.stub + + def slow_probe(log, path=None, resolver=None): + time.sleep(0.5) + return stub + fx.probe = slow_probe + self.runner._fixture_probed = 0.0 + self.runner.fixture = None + poll = threading.Thread(target=self.runner.probe_fixture) + poll.start() + time.sleep(0.1) + order = self.runner.batch_selection("unattended") + poll.join() + self.assertIn("r.lid", order) + self.assertIn("r.btn", order) + + 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_a_failing_box_in_an_unattended_run_ends_the_test_as_an_error(self): + # nobody is in the room: the run must not wait ACT_TIMEOUT_S for a + # hand that is not there; it ends at once, as the harness's error + self.stub.fail = True + t0 = time.time() + b, order = self.run_queue("unattended") + results = {x["test"]: x["result"] for x in b["done"]} + self.assertEqual(results["r.lid"], "ERROR") + self.assertLess(time.time() - t0, 20) + rec = self.last_result("r.lid") + self.assertIn("the fixture could not perform a step", rec["message"]) + self.assertIn("the box is off", rec["message"]) + act = rec["evidence"]["actions"][0] + self.assertEqual((act["by"], act["fixture_error"]), ("fixture", "the box is off")) + self.assertFalse(any("asking the operator" in l for l in rec["log"]), rec["log"]) + + 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 press_button(self, down): + self.fc.state["status"]["switches"]["button"] = bool(down) + + def test_ready_takes_a_press_on_the_machine_as_the_presence_check(self): + """With an actuator wired to the button, the operator proves they + are at the machine by pressing it, not by clicking the page. The + actuator then owns every press in the test, so no press in a live + cut is a person's and none can go uncounted.""" + run = Run("test", "r.live", "r.live") + ctx = Context(run, self.runner, self.reg["r.live"]) + self.runner.probe_fixture(force=True) + asked = [] + run.ask = lambda q, o: asked.append(q) + self.press_button(False) + threading.Timer(0.3, self.press_button, args=(True,)).start() + threading.Timer(0.8, self.press_button, args=(False,)).start() + ctx.ready("LIVE FIRE. Scrap under the head.") + self.assertEqual(asked, []) # no page click asked for + self.assertTrue(run.fixture_takeover) + rec = [r for r in run.evidence["actions"] if r["state"] == "presence"] + self.assertEqual(len(rec), 1) + self.assertEqual(rec[0]["by"], "operator") + self.assertTrue(any("press the button on the machine" in ln.lower() for ln in run.lines)) + + def test_the_presence_press_hands_the_arm_press_to_the_actuator(self): + """The bench's standing opt-in is not needed once the operator has + proved presence: that press is what the opt-in existed to + establish.""" + run = Run("test", "r.live", "r.live") + ctx = Context(run, self.runner, self.reg["r.live"]) + self.runner.probe_fixture(force=True) + self.stub.arm_press = False + run.fixture_takeover = True + saved = runner_mod.hw.button_lit + runner_mod.hw.button_lit = lambda: True + try: + self.assertTrue(ctx.arm_press()) + deadline = time.time() + 5 + while ("button", "press") not in self.stub.acts and time.time() < deadline: + time.sleep(0.05) + self.assertIn(("button", "press"), self.stub.acts) + finally: + runner_mod.hw.button_lit = saved + + def test_an_actuator_lost_after_the_takeover_is_said_out_loud(self): + """Falling back to the operator without a word is how one dropped + actuator becomes a press nobody can account for afterwards.""" + run = Run("test", "r.live", "r.live") + ctx = Context(run, self.runner, self.reg["r.live"]) + self.runner.probe_fixture(force=True) + run.fixture_takeover = True + self.runner.fixture = None # the box drops off mid-test + self.fc.state["status"]["switches"]["lid"] = True + threading.Timer(0.3, lambda: self.fc.state["status"]["switches"].__setitem__("lid", False)).start() + ctx.act("lid", "open", timeout=5) + rec = run.evidence["actions"][-1] + self.assertTrue(rec.get("fixture_lost")) + self.assertEqual(rec["by"], "operator") + self.assertTrue(any("WARNING" in ln and "now gone" in ln for ln in run.lines)) + + def test_ready_still_asks_the_page_with_no_actuator(self): + run = Run("test", "r.live", "r.live") + ctx = Context(run, self.runner, self.reg["r.live"]) + self.runner.fixture = None + asked = [] + + def ask(q, o): + asked.append((q, list(o))) + return "Ready" + run.ask = ask + ctx.ready("LIVE FIRE. Scrap under the head.") + self.assertEqual(asked, [("LIVE FIRE. Scrap under the head.", ["Ready", "Cannot"])]) + self.assertFalse(run.fixture_takeover) + + 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()