diff --git a/forgetest/forgetest/suite/setup_burns.py b/forgetest/forgetest/suite/setup_burns.py index 7266d3b..74425ef 100644 --- a/forgetest/forgetest/suite/setup_burns.py +++ b/forgetest/forgetest/suite/setup_burns.py @@ -48,6 +48,36 @@ HOME = "The head is home" EDGE_KEY = "lens_hall_edge_z_mm" FOCUS_KEYS = ["thickness_mm", "z_passes", "ladder_z", "window", "emission", "pick", "picked_lines", "pick_half_steps", "edge_z_mm", "steps_per_mm", "max_height_mm", "focus_range_mm", "summary"] +# pic/hv_current above this: the discharge is present (the daemon's own +# rule for a lit burn, jobstream.h JOBSTREAM_HV_ON). +HV_ON = 30 + + +class Witnesses: + """The two emission witnesses read from sysfs by the test itself at + every poll of the burn: the kernel's LASER_ON sample count + (cnc/laser_on_sampled) and the tube current (pic/hv_current), each + at its peak, and how many polls read both. The burn's result reports + the daemon's own reading of the same two; this one is taken past the + daemon.""" + + def __init__(self): + self.laser_on_max = None + self.hv_max = None + self.reads = 0 + + def __call__(self, _doc): + lon = hw.sysfs_int("cnc/laser_on_sampled") + hv = hw.sysfs_int("pic/hv_current") + if lon is not None: + self.laser_on_max = lon if self.laser_on_max is None else max(self.laser_on_max, lon) + if hv is not None: + self.hv_max = hv if self.hv_max is None else max(self.hv_max, hv) + if lon is not None and hv is not None: + self.reads += 1 + + def record(self): + return {"laser_on_max": self.laser_on_max, "hv_max": self.hv_max, "reads": self.reads} def preview_ok(ctx, wid): @@ -91,17 +121,29 @@ def preview_ok(ctx, wid): ctx.check(tail == want, "the %s program ends %s, not the tail %s", wid, tail, want) -def emission_ok(ctx, last): - """The two witnesses of the burn, from the wizard's result. The head's - beam sensor is recorded, never required: the result reports its rise - and whether it read as the beam seen.""" +def emission_ok(ctx, last, seen): + """The two witnesses of the burn, from the wizard's result and from + the test's own reads of them at every poll (`seen`, a Witnesses). + The head's beam sensor is recorded, never required: the result + reports its rise and whether it read as the beam seen.""" e = (last.get("result") or {}).get("emission") or {} ctx.evidence["emission"] = e - ctx.check(e.get("hv_max", 0) > 30, "the tube current never rose (%s)", e.get("hv_max")) + ctx.check(e.get("hv_max", 0) > HV_ON, "the tube current never rose (%s)", e.get("hv_max")) ctx.check(e.get("laser_on_samples", 0) > 0, "the kernel never sampled LASER_ON") ctx.check(isinstance(e.get("thermopile_delta"), int) and isinstance(e.get("beam_sensor_seen"), bool), "the result does not record the beam sensor: %s", e) ctx.log("the beam sensor rose %s (seen: %s)", e.get("thermopile_delta"), e.get("beam_sensor_seen")) + w = seen.record() + ctx.evidence["emission_seen"] = w + ctx.log("seen from here: LASER_ON count peak %s, tube current peak %s over %d polls", + w["laser_on_max"], w["hv_max"], w["reads"]) + ctx.check(w["reads"] > 0, "the test could not read the witnesses from sysfs during the burn") + ctx.check((w["laser_on_max"] or 0) > 0, + "the kernel's LASER_ON count never read nonzero from here during the burn (%d reads); the " + "burn reports %s", w["reads"], e.get("laser_on_samples")) + ctx.check((w["hv_max"] or 0) > HV_ON, + "the tube current never read over %d from here during the burn (peak %s over %d reads); the " + "burn reports %s", HV_ON, w["hv_max"], w["reads"], e.get("hv_max")) def thickness_answer(ctx): @@ -116,7 +158,8 @@ def burn(ctx, wid, arm, pick, want): """One burn: the preview, then the burn after the press with every prompt answered (the gantry stops installed, the head at home, the scrap in the corner, its thickness in the machine's units, the arm, - and the pick in `pick`), the witnesses, and the keys of the result in + and the pick in `pick`), the witnesses (the burn's own, and the two + read from sysfs here at every poll), and the keys of the result in `want`. run_check makes the arm press when the burn asks for it. A prompt the test does not answer, or one asked again because its answer was refused, ends the burn. Returns the result.""" @@ -127,6 +170,7 @@ def burn(ctx, wid, arm, pick, want): answers.update(pick) preview_ok(ctx, wid) asked = set() + seen = Witnesses() def on_prompt(p): pid = p.get("id") @@ -142,10 +186,10 @@ def burn(ctx, wid, arm, pick, want): return answers[pid] try: - last = run_check(ctx, wid, on_prompt, 900) + last = run_check(ctx, wid, on_prompt, 900, on_poll=seen) finally: ctx.clear_notice() - emission_ok(ctx, last) + emission_ok(ctx, last, seen) r = last.get("result") or {} missing = [k for k in want if k not in r] ctx.check(not missing, "the result lacks %s: %s", missing, sorted(r)) @@ -236,7 +280,8 @@ def focus_window_ok(ctx, r, before): "inside the reach /status reports, and the program ending in that tail at the " "edge /status reports, it burns twelve lines after the press, from the top of " "the window to the bottom, seen by the two witnesses (the tube current and the " - "kernel's LASER_ON samples), with the beam sensor's rise recorded and not " + "kernel's LASER_ON samples) as the burn reports them and as the test reads them " + "from sysfs itself at every poll, with the beam sensor's rise recorded and not " "required. The pick is line 11. The " "result carries its keys; the ladder ran in the window the machine held, which " "the burn leaves as found (it writes the edge alone, the one its result " diff --git a/forgetest/forgetest/suite/setup_dark.py b/forgetest/forgetest/suite/setup_dark.py index 9c2089a..42c6f3d 100644 --- a/forgetest/forgetest/suite/setup_dark.py +++ b/forgetest/forgetest/suite/setup_dark.py @@ -82,11 +82,13 @@ def answer(fc, wid, prompt, value): return fc.post("/wiz/%s/answer" % wid, data={"seq": str(prompt["seq"]), "value": value}) -def run_check(ctx, wid, on_prompt, timeout_s): +def run_check(ctx, wid, on_prompt, timeout_s, on_poll=None): """Start `wid` and follow it to its end. on_prompt(prompt) is called once per prompt (by sequence number) and returns the value to answer with, or None when it handled the prompt another way (a fixture - action). Returns the final status document.""" + action). on_poll(doc), when given, is called once per poll of + /wiz/dark with the status document: a caller's own witness read + beside the check's. Returns the final status document.""" fc = ctx.forgectrl ev = ctx.evidence start(ctx, fc, wid) @@ -100,6 +102,8 @@ def run_check(ctx, wid, on_prompt, timeout_s): d = dark(fc) if d.get("id") != wid: raise hw.HwError("another check took the slot: %s" % d.get("id")) + if on_poll is not None: + on_poll(d) if d.get("phase") and (not phases or phases[-1] != d["phase"]): phases.append(d["phase"]) ctx.log("%s: %s", wid, d["phase"]) diff --git a/forgetest/tests/test_setup_burns.py b/forgetest/tests/test_setup_burns.py index dbd2932..be4bfb6 100644 --- a/forgetest/tests/test_setup_burns.py +++ b/forgetest/tests/test_setup_burns.py @@ -9,11 +9,14 @@ sender, the page, and the driver's laser module; and the burn against a scripted daemon: the preview and the program read, the start, the prompts answered in order (the gantry stops, the head at home, the scrap, its thickness in the machine's units, the arm, the machine's own press -prompt, the pick), the witnesses judged, the record read at version 1, -the setting the burn wrote put back as found, and the controller started +prompt, the pick), the witnesses judged (the burn's own, and the two the +test reads from sysfs at every poll), the record read at version 1, the +setting the burn wrote put back as found, and the controller started again on the edge and the window put back.""" import os +import shutil import sys +import tempfile import unittest sys.path.insert(0, os.path.dirname(__file__)) @@ -279,6 +282,13 @@ class ScriptedDaemon: class BurnRunTests(unittest.TestCase): def setUp(self): + # the two witnesses as sysfs serves them through a lit burn (the + # shape of a bench record: the LASER_ON count and the tube current) + self.sysfs = tempfile.mkdtemp(prefix="forgetest-burn-") + self.plant("cnc/laser_on_sampled", "118") + self.plant("pic/hv_current", "431") + self.saved_root = os.environ.get("GF_SYSFS_ROOT") + os.environ["GF_SYSFS_ROOT"] = self.sysfs + "/" self.fake = helpers.FakeForgectrl().start() self.daemon = ScriptedDaemon(self.fake) self.poll = setup_dark.POLL_S @@ -287,6 +297,17 @@ class BurnRunTests(unittest.TestCase): def tearDown(self): setup_dark.POLL_S = self.poll self.fake.stop() + if self.saved_root is None: + os.environ.pop("GF_SYSFS_ROOT", None) + else: + os.environ["GF_SYSFS_ROOT"] = self.saved_root + shutil.rmtree(self.sysfs, ignore_errors=True) + + def plant(self, attr, value): + path = os.path.join(self.sysfs, *attr.split("/")) + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w") as f: + f.write(value + "\n") def run_test(self, tid): t = catalog.load_suite()[tid] @@ -312,6 +333,10 @@ class BurnRunTests(unittest.TestCase): self.assertNotIn("lens_hall_edge_z_mm", start) self.assertEqual((start["lens_stop_below_steps"], start["lens_stop_above_steps"]), ("14", "20")) self.assertEqual(run.evidence["result"]["emission"]["hv_max"], 610) + # the witnesses read from here at every poll: each at its peak + seen = run.evidence["emission_seen"] + self.assertEqual((seen["laser_on_max"], seen["hv_max"]), (118, 431)) + self.assertGreater(seen["reads"], 0) fw = run.evidence["focus_window"] self.assertEqual(fw["window"]["below"], 14) self.assertEqual(fw["window"]["above"], 20) @@ -404,6 +429,18 @@ class BurnRunTests(unittest.TestCase): self.assertIn("the tube current never rose", str(cm.exception)) self.assert_found() + def test_a_burn_the_test_itself_never_saw_fails_and_still_restores(self): + # The burn's result reports a lit burn, and the test's own reads + # of the two witnesses never leave zero: the burn is judged on + # what was read from here, not on the daemon's word alone. + self.plant("cnc/laser_on_sampled", "0") + self.plant("pic/hv_current", "0") + with self.assertRaises(Failed) as cm: + self.run_test("setup.focus") + self.assertIn("LASER_ON count never read nonzero", str(cm.exception)) + self.assert_found() + self.assert_restarted_on_found() + def test_a_burn_with_a_silent_beam_sensor_passes(self): # A head whose beam sensor gives no output: the tube current and # LASER_ON saw the beam, so the burn counts.