setup.focus: the test reads the LASER_ON count and the tube current from sysfs itself at every poll of the burn

The focus burn's emission was judged from the burn's result alone: the
daemon's own 25 Hz reads of cnc/laser_on_sampled and pic/hv_current,
reported as laser_on_samples and hv_max. The one live catalog test of
the wizard's burn took the daemon under test at its word for the
emission it reported, where the laser.* tests judge from the test's own
/status samples.

run_check takes on_poll(doc), called once per poll of /wiz/dark with
the status document, a caller's own witness read beside the check's;
the default keeps every positional caller as it is. Witnesses, in
setup_burns, reads the two attributes from sysfs at every poll and
keeps each at its peak and the count of polls that read both. burn()
passes it to run_check, and emission_ok(ctx, last, seen) keeps the
result's three checks and adds the test's own: at least one poll read
both witnesses, the LASER_ON count read nonzero from here, and the tube
current read over HV_ON (30, the daemon's own rule for a lit burn,
jobstream.h JOBSTREAM_HV_ON) from here. Once the result's three checks
pass, the evidence carries the two peaks and the read count as
emission_seen; the two readings are not compared with each other. The beam sensor's handling is unchanged, and
the description names both readings.

Proof: tests/test_setup_burns.py, the Unit tests step of
forgetest-ci.yml (python -B -m unittest discover -s tests), 20 tests
OK on Windows and on Linux, with tests/test_setup_dark.py (20 OK) for
the positional callers of run_check. BurnRunTests plants
cnc/laser_on_sampled and pic/hv_current under GF_SYSFS_ROOT (the shape
of a bench record), the lit burn's evidence carries the two peaks with
reads over zero, and a burn whose result reports a lit burn while the
two attributes read 0 from here fails on the LASER_ON count, with the
setting put back and the controller restarted on it. pyflakes reports
no undefined name. Negative control on a scratch copy: with
setup_burns.py and setup_dark.py as they were before this change, that
same case passes (Failed not raised) and the evidence has no
emission_seen; the two new assertions catch both.

Acceptance: setup.focus's fingerprint moves (its own text, and the
shared text of setup_burns.py and setup_dark.py), and every test of
setup_dark.py moves through run_check: setup.check-switches,
setup.check-airflow, setup.check-motion, setup.check-lens and
setup.cloud-header-capture. motion.release does not move (its run_check
import is inside the function body). The next campaign runs setup.focus
with both readings on the bench.

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 "

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"])

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.