From 4c5dde25494763105572be8c7bfe2e1dcf472b99 Mon Sep 17 00:00:00 2001 From: ScottW514 Date: Mon, 12 Oct 2026 19:53:20 -0400 Subject: [PATCH] forgetest: the burns, the lens and the intake checks follow forgectrl's review fixes - baseline: the lid lamp fades to a new level (the kernel's smooth trigger), so the hand-back reads it until it holds the wanted level twice in a row, for up to 3 s, before it is a leftover. motion.deadman failed its hand-back reading 232 against 236 0.05 s after the supervisor re-applied 236 to a respawned controller. - setup.check-lens and setup.focus: grblHAL reads the lens window and the edge only at its start, and their restore came after the wizard had restarted the controller on the test's values. Each test now waits for the step to end, puts the settings back, and restarts the controller (stop, start, running and verified with a new pid) on every exit path; cloud mode restarts nothing. - setup.focus and setup.floor: the served program ends with G0 Z, G0 X0 Y0, G92.1, M2; setup.floor requires setup.focus and checks the scrap's thickness against the lens reach; the burns cover lenshome. - setup.check-intakes: the two assertions that could not fail are gone; an intake-count prompt fails the test unanswered (the bench reference's intakes both turn). The precheck's refusal names no setting. - setup.check-airflow: the machine's count rules (intake 2 turning writes 2; silent on a one-fan machine, the count is not written). Shared text: setup_dark.py's precheck string (all setup_dark tests, setup_burns, setup.check-envelope), setup_burns.py's imports, covers, docstring and preview check, cooling_intake.py's docstring; every test they move covers paths forgectrl changed in the same work. Proof: the forgetest host suite passes (548), with new host tests for the lamp's fade, the restarts on pass, fail and abort, the floor's Focus refusal and reach, the tail without G92.1, the intakes prompt, and the airflow count rules; pyflakes is clean on the changed modules. Acceptance: these are the tests. --- forgetest/forgetest/baseline.py | 40 +++- forgetest/forgetest/suite/cooling_intake.py | 3 +- forgetest/forgetest/suite/setup_burns.py | 150 ++++++++++--- forgetest/forgetest/suite/setup_dark.py | 180 ++++++++++----- forgetest/tests/test_baseline.py | 54 +++++ forgetest/tests/test_setup_burns.py | 151 ++++++++++++- forgetest/tests/test_setup_dark.py | 234 ++++++++++++++++---- 7 files changed, 666 insertions(+), 146 deletions(-) diff --git a/forgetest/forgetest/baseline.py b/forgetest/forgetest/baseline.py index f6e1e10..4711297 100644 --- a/forgetest/forgetest/baseline.py +++ b/forgetest/forgetest/baseline.py @@ -19,7 +19,9 @@ restores it after (on every exit path), so a test cannot hand the next one restored where the interface allows. The lid lamp is fixed too, at forgectrl's `lid_lamp_idle` setting (unset = -236): the daemon asserts it at start and at every controller spawn. +236): the daemon asserts it at start and at every controller spawn. The +lamp fades toward a level written to it, so a level written a moment +before the check is read again until it holds before it is judged. The controller mode in force decides what the baseline owns. In GRBL mode everything above applies. In cloud mode the cloud client owns what it @@ -140,6 +142,15 @@ PRESERVED_SYSFS = [] # sysfs attrs captured before, restored afte LID_LAMP_ATTR = "pic/lid_led" LID_LAMP_DEFAULT = "236" # forgectrl's lid_lamp_idle default +# The lid LED runs on the kernel's "smooth" trigger: a level written to it +# is reached by a fade over a fraction of a second, so a read right after +# a write (the supervisor re-applies the idle level at every controller +# spawn) finds the lamp part-way there: on the bench reference, 232 read +# 0.05 s after the supervisor re-applies 236. The lamp is read again every +# LID_LAMP_POLL_S until it holds the level, for at most LID_LAMP_SETTLE_S, +# before it is called a leftover. +LID_LAMP_SETTLE_S = 3.0 +LID_LAMP_POLL_S = 0.1 BOOT_MAX_AGE_S = 600 # a boot reference is taken only this soon after boot SETTLE_S = 150 # the supervisor's probe + rail-off ladder @@ -743,13 +754,34 @@ class Baseline: got = hw.sysfs_read(LID_LAMP_ATTR) if got is None or got == want: return + held, now = self._lamp_holds(want) + if held: + self.log("lid lamp: read %s, then held %s; the fade toward a level just written, " + "not a leftover" % (got, want)) + return + found = now if now is not None else got try: hw.sysfs_write(LID_LAMP_ATTR, want) - back = hw.sysfs_read(LID_LAMP_ATTR) - act = "restored" if back == want else "failed: reads %s" % back + held, back = self._lamp_holds(want) + act = "restored" if held else "failed: reads %s" % back except OSError as e: act = "failed: %s" % e - left.append(Leftover(LID_LAMP_ATTR, got, want, act)) + left.append(Leftover(LID_LAMP_ATTR, found, want, act)) + + def _lamp_holds(self, want): + """Read the lid lamp every LID_LAMP_POLL_S, for at most + LID_LAMP_SETTLE_S, until it holds `want`: two reads in a row at + it, so a fade that only passes through the level does not count. + Returns (held, the last read).""" + last = prev = None + for _ in range(max(2, int(round(LID_LAMP_SETTLE_S / LID_LAMP_POLL_S)))): + if self.abort(): + break + time.sleep(LID_LAMP_POLL_S) + prev, last = last, hw.sysfs_read(LID_LAMP_ATTR) + if last == want and prev == want: + return True, last + return False, last def _kernel_side(self, left): if hw.sysfs_read("cnc/state") is None: diff --git a/forgetest/forgetest/suite/cooling_intake.py b/forgetest/forgetest/suite/cooling_intake.py index f7d5c3c..9f75ae6 100644 --- a/forgetest/forgetest/suite/cooling_intake.py +++ b/forgetest/forgetest/suite/cooling_intake.py @@ -4,7 +4,8 @@ # SPDX-License-Identifier: MIT """cooling.intake-one-fan - a machine built with one intake fan -(cool_intake_fans 1, which the setup's airflow check writes): the second +(cool_intake_fans 1, which the setup's Intake fans step writes when the +operator counts one): the second intake tach is published as not fitted and never judged, and intake 1 still holds the intake floor. On a machine with two intake fans the setting alone proves both halves: intake 2 turns and is not judged, and diff --git a/forgetest/forgetest/suite/setup_burns.py b/forgetest/forgetest/suite/setup_burns.py index 4526183..c778e78 100644 --- a/forgetest/forgetest/suite/setup_burns.py +++ b/forgetest/forgetest/suite/setup_burns.py @@ -10,6 +10,8 @@ result judged, and the setting the burn wrote put back as found. Each burn stands alone on its own scrap of wood in the back-left corner, drawn from the head's home against the gantry stops, and re-runs on the real record, so a completed burn completes again at the same version. +The floor burns at the focus the focus burn measured, so it needs Focus +on the record. The operator's part is the scrap and one press on the machine at the start (the presence check of the ready gate; with the bench actuator up @@ -20,12 +22,15 @@ witnesses, not the operator's eye, and the setting it writes is restored. The page is where the real numbers go in. """ import re +import time +from .. import hw from ..catalog import test -from .setup_dark import run_check, Restore, FALLBACK_WINDOW, LENS_WINDOW_KEYS +from .setup_dark import run_check, Restore, FALLBACK_WINDOW, LENS_WINDOW_KEYS, dark BURN_COVERS = [("forgectrl", "src/wizlive.*"), ("forgectrl", "src/wizrun.h"), ("forgectrl", "src/lens.*"), ("forgectrl", "src/sheet.*"), + ("forgectrl", "src/lenshome.*"), ("forgectrl", "src/font_hershey.*"), ("forgectrl", "src/jobstream.*"), ("forgectrl", "src/jobrun.*"), ("forgectrl", "src/lease.*"), ("forgectrl", "src/curverec.*"), ("forgectrl", "src/wizdark.*"), @@ -78,6 +83,17 @@ def preview_ok(ctx, wid): out = [z for z in zs if z < lo - 0.01 or z > hi + 0.01] ctx.check(not out, "%s commands Z outside the lens window %.2f..%.2f: %s", wid, lo, hi, out) + # The tail the wizard streams after the body: the lens back on the + # reference the program's head declared, the head back to X0 Y0, the + # G92 origin cleared (the controller keeps a G92 offset and restores it + # at its next start, where it would shift a sender's own work offsets), + # and the program end. + lines = [ln.strip() for ln in (body or b"").decode("utf-8", "replace").splitlines()] + tail = [ln for ln in lines if ln and not ln.startswith(";")][-4:] + edge = lens.get("edge_z") + want = ["G0 Z%.2f" % edge, "G0 X0 Y0", "G92.1", "M2"] if isinstance(edge, (int, float)) else None + ctx.evidence.setdefault("program_tail", {})[wid] = tail + ctx.check(tail == want, "the %s program ends %s, not the tail %s", wid, tail, want) def emission_ok(ctx, last): @@ -219,45 +235,97 @@ def focus_window_ok(ctx, r, before): "references the lens, takes the controller in loopback posture, is answered that " "the head is home and that the scrap is in the corner, and takes the thickness " "in the machine's units (3.2 mm or 0.125 in); its program declares X0 Y0 at the " - "head and its tail returns there. With its preview and program served by GET " - "/wiz/sheet.svg and /wiz/sheet.gcode and every Z the program commands inside the " - "reach /status reports, it burns twelve lines after the press, from the top of " + "head, and its tail puts the lens back on the reference, returns the head there, " + "clears the G92 origin (G92.1), and ends (M2). With its preview and program " + "served by GET /wiz/sheet.svg and /wiz/sheet.gcode, every Z the program commands " + "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 " "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 " "reports); every ladder height lies in the reach; the program served afterward " - "agrees with /status; the record carries laser.focus at version 1. The edge is " - "restored at the end.") + "agrees with /status; the record carries laser.focus at version 1. The edge and " + "the window are restored at the end. grblHAL reads them only at its start, and " + "the burn started it before the restore, so the controller is stopped and " + "started again on the settings as found (on every exit path), and it must be " + "running with motion verified under a new process.") def focus(ctx): fc = ctx.forgectrl ev = ctx.evidence + restart = {} + + def restart_on_found(): + """A fresh controller start on the settings as found, once the + burn has ended. Plain waits, not the run's, so an abort gets it + too; a burn still ending starts its own controller after the + restore, which reads them already.""" + try: + t_end = time.time() + 120 + while time.time() < t_end and dark(fc).get("running"): + time.sleep(1) + if dark(fc).get("running"): + restart["skipped"] = "the burn is still ending" + ctx.log("the burn is still ending: its own controller start reads the settings as found") + return + st, m = fc.get("/mode") + m = m if isinstance(m, dict) else {} + restart.update(mode=m.get("mode"), pid=m.get("pid")) + if m.get("mode") != "grbl": + return # the cloud client reads neither the edge nor the window + restart["stop"], body = fc.post("/controller/stop") + ctx.log("restarting the controller on the edge and the window as found: POST " + "/controller/stop -> %s %s", restart["stop"], body) + for _ in range(10): + restart["start"], body = fc.post("/controller/start") + if restart["start"] != 409: + break + time.sleep(1) # the burn's lease goes just after its end + ctx.log("POST /controller/start -> %s %s", restart["start"], body) + except hw.HwError as e: + restart["error"] = str(e) + ctx.log("the controller could not be restarted on the settings as found: %s", e) ctx.ready("The focus burn: twelve lines on the scrap after a press. The bench presses when the " "button lights.") - with Restore(ctx, [EDGE_KEY] + LENS_WINDOW_KEYS): - # A fresh head's focus: the edge unknown, so the burn runs from the - # bench reference's number and writes the head's own. The window - # stays: it is the Lens travel check's. - st, _ = fc.post("/settings", params={EDGE_KEY: ""}) - ctx.check(st == 200, "clearing %s -> %s", EDGE_KEY, st) - before = window_held(ctx) - ev["lens_fresh"] = (fc.status() or {}).get("lens") or {} - r = burn(ctx, "laser.focus", "focus-arm", {"focus-pick": "11"}, FOCUS_KEYS) - ctx.check(r.get("pick") == 11 and r.get("picked_lines") == [11, 11], - "the pick reads %s (lines %s), answered 11", r.get("pick"), r.get("picked_lines")) - ctx.check("stops" not in r, "the focus burn searched for the stops: %s", r.get("stops")) - focus_window_ok(ctx, r, before) - ctx.log("focus: line %s, the edge at %s mm, the window %s/%s", r.get("pick"), r.get("edge_z_mm"), - (r.get("window") or {}).get("below"), (r.get("window") or {}).get("above")) + try: + with Restore(ctx, [EDGE_KEY] + LENS_WINDOW_KEYS): + # A fresh head's focus: the edge unknown, so the burn runs from the + # bench reference's number and writes the head's own. The window + # stays: it is the Lens travel check's. + st, _ = fc.post("/settings", params={EDGE_KEY: ""}) + ctx.check(st == 200, "clearing %s -> %s", EDGE_KEY, st) + before = window_held(ctx) + ev["lens_fresh"] = (fc.status() or {}).get("lens") or {} + r = burn(ctx, "laser.focus", "focus-arm", {"focus-pick": "11"}, FOCUS_KEYS) + ctx.check(r.get("pick") == 11 and r.get("picked_lines") == [11, 11], + "the pick reads %s (lines %s), answered 11", r.get("pick"), r.get("picked_lines")) + ctx.check("stops" not in r, "the focus burn searched for the stops: %s", r.get("stops")) + focus_window_ok(ctx, r, before) + ctx.log("focus: line %s, the edge at %s mm, the window %s/%s", r.get("pick"), r.get("edge_z_mm"), + (r.get("window") or {}).get("below"), (r.get("window") or {}).get("above")) + finally: + restart_on_found() + ev["restart"] = dict(restart) + if restart.get("mode") == "grbl": + ctx.check(restart.get("stop") == 200 and restart.get("start") == 200, + "the controller was not restarted on the settings as found: %s", restart) + + def back(): + st, m = fc.get("/mode") + m = m if isinstance(m, dict) else {} + return (m.get("controller") == "running" and m.get("motion") == "verified" + and (restart.get("start") != 200 or m.get("pid") != restart.get("pid"))) + ok = ctx.wait_for(back, 150) + ctx.check(ok is not None, "the controller did not come back on the settings as found") ctx.log("PASS: the focus burn ran in the window the machine holds and wrote the edge alone; " - "settings restored") + "settings restored and the controller restarted on them") @test("setup.floor", title="The laser floor burn: the rungs and the floor it writes", subsystem="setup", kind="live", hardware="takeover", mode="grbl", est_min=10, covers=BURN_COVERS, - requires=["forgectrl.auth", "setup.check-motion", "laser.emission-witness"], + requires=["forgectrl.auth", "setup.check-motion", "laser.emission-witness", "setup.focus"], actions=["button"], hands=["scrap"], steps=["The gantry stops installed and the head at its home against them. A fresh flat scrap " "of wood about 3 mm (1/8 in) thick and at least 86 x 68 mm (3.4 x 2.7 in), pushed into " @@ -269,20 +337,46 @@ def focus(ctx): "the button again at the arm, as the page asks.", "Nothing else: the test answers every prompt itself and puts back the setting the " "burn writes."], - description="POST /wiz/laser.floor/start on a fresh scrap, set up as the focus burn is: " + description="The floor burns at the focus the focus burn measured, and the wizard refuses " + "to start without Focus on the record, so the record must carry laser.focus " + "(setup.focus, ahead of this test, leaves it there), and the lens reach /status " + "reports must include the scrap's thickness (the wizard refuses a surface the " + "lens cannot focus on); both are read before the scrap is asked for. Then POST " + "/wiz/laser.floor/start on a fresh scrap, set up as the focus burn is: " "answered that the gantry stops are installed, the lens referenced, the " "controller in loopback posture, answered that the head is home and that the " "scrap is in the corner, the thickness in the machine's units (3.2 mm or 0.125 " - "in); its program declares X0 Y0 at the head and its tail returns there. With " - "its preview and program served and every Z the program commands inside the " - "reach /status reports, it burns twelve rungs from 2 to 24 percent density at " + "in); its program declares X0 Y0 at the head, and its tail puts the lens back " + "on the reference, returns the head there, clears the G92 origin (G92.1), and " + "ends (M2). With its preview and program served, every Z the program commands " + "inside the reach /status reports, and the program ending in that tail, it " + "burns twelve rungs from 2 to 24 percent density at " "the focal point on the scrap's surface, the floor and the curve off for the " "job, after the press, seen by the two witnesses. The pick is 8, so the floor " "written is 10 (two above it) and the setting reads it; the result carries its " "keys; the record carries laser.floor at version 1. laser_floor_density is " - "restored at the end.") + "restored at the end (the controller loads it at every arm).") def floor(ctx): fc = ctx.forgectrl + ev = ctx.evidence + # The record and the reach, before the scrap is asked for: the wizard + # refuses the burn without Focus on the record, and it refuses a scrap + # the lens cannot focus on, so the operator's setup would be for + # nothing. + st, w = fc.get("/wiz") + focus_v = ((w if isinstance(w, dict) else {}).get("versions") or {}).get("laser.focus") + ev["focus_on_record"] = focus_v + ctx.check(st == 200 and isinstance(focus_v, int) and focus_v >= 1, + "the record carries no Focus (laser.focus, %s): the floor burns at the focus it measures, " + "and the wizard refuses to start without it", focus_v) + _answer, want_mm = thickness_answer(ctx) + lens = (fc.status() or {}).get("lens") or {} + lo, hi = lens.get("reach_min"), lens.get("reach_max") + ev["floor_reach"] = [lo, hi] + ctx.check(isinstance(lo, (int, float)) and isinstance(hi, (int, float)) + and lo - 0.01 <= want_mm <= hi + 0.01, + "the lens focuses %s to %s mm above the tray, which leaves out the %s mm scrap: the floor " + "refuses a surface the lens cannot focus on", lo, hi, want_mm) ctx.ready("The laser floor burn: twelve rungs on the scrap after a press. The bench presses when " "the button lights.") with Restore(ctx, [FLOOR_KEY]): diff --git a/forgetest/forgetest/suite/setup_dark.py b/forgetest/forgetest/suite/setup_dark.py index 3aea767..c6bb5c2 100644 --- a/forgetest/forgetest/suite/setup_dark.py +++ b/forgetest/forgetest/suite/setup_dark.py @@ -46,8 +46,9 @@ def intakes_counted(): except hw.HwError as e: return "forgectrl unreachable: %s" % e if not s.get(INTAKE_KEY): - return ("%s is unset, so the setup gate holds every controller: count the intake fans " - "first (the Intake fans step of the setup page)" % INTAKE_KEY) + return ("the intake fans are not counted, so the setup gate holds every controller; this " + "test puts the count back as it finds it, and an uncounted machine handed back " + "would hold the controllers of every test after it") return None @@ -211,27 +212,41 @@ def check_sensors(ctx): ("forgectrl", "src/gates.c"), ("forgectrl", "src/super.c")], requires=["forgectrl.auth", "cooling.fan-gate-trips"], precheck=intakes_counted, description="POST /wiz/airflow/start: the controller stops, the fans run at the cut profile " - "for 35 s, the purge is measured on and off, and six settings are written: the " - "four floors at 55 percent of steady, the grace from the slowest spin-up, and the " - "intake fans fitted (2 when both intakes turn; 1 when intake 2 never reads a pulse " - "and intake 1 passes, and then intake 2 is reported not fitted and the intake floor " - "comes from intake 1). Each fan judged must read at least 1000 rpm. The six " - "settings are put back as found at the end, the purge fan is on again and drawing " - "over the floor the check just wrote, and the controller must be running again.") + "for 35 s, the purge is measured on and off, and the five floors are written: the " + "four at 55 percent of steady and the grace from the slowest spin-up. The intake " + "fan count is the machine's: intake 2 turning proves two fans, both intakes are " + "judged, and the count is written as 2; intake 2 silent on a machine counted with " + "one is reported not fitted and not turning, intake 1 alone sets the intake " + "floor, and the count is not written; intake 2 silent on a machine counted with " + "two fails the check. Both intakes turn on the bench reference, so the bench " + "path writes 2. Each fan judged must read at least 1000 rpm. The settings are put " + "back as found at the end, the purge fan is on again and drawing over the floor " + "the check just wrote, and the controller must be running again.") def check_airflow(ctx): keys = ["cool_tach_exhaust_min_rpm", "cool_tach_intake_min_rpm", "cool_tach_air_assist_min_rpm", "cool_purge_min_current", "cool_fan_grace_s"] - with Restore(ctx, keys + ["cool_intake_fans"]): + with Restore(ctx, keys + ["cool_intake_fans"]) as held: + counted = held.before.get("cool_intake_fans") last = run_check(ctx, "airflow", lambda p: None, 240) r = last.get("result") or {} fans = r.get("fans") or {} intakes = r.get("intake_fans") + silent2 = (fans.get("intake 2") or {}).get("turning") is False + ctx.evidence["intakes"] = {"counted": counted, "found": intakes, "intake_2_silent": silent2} ctx.check(intakes in (1, 2), "the result carries no intake fan count: %r", intakes) - ctx.log("intake fans fitted: %s", intakes) + if silent2: + # one intake fan, as the machine is counted (on a machine + # counted with two the check itself fails, and run_check says so) + ctx.check(counted == "1" and intakes == 1, + "intake 2 gave no reading on a machine counted with %r intake fans, and the check " + "found %r", counted, intakes) + else: + ctx.check(intakes == 2, "intake 2 turned, and the check found %r intake fans", intakes) + ctx.log("intake fans: %s (the machine counted %s)", intakes, counted) for name in ("exhaust", "intake 1", "intake 2", "air assist"): f = fans.get(name) or {} - if name == "intake 2" and intakes == 1: - ctx.check(f == {"fitted": False}, "intake 2 with one intake fan: %s", f) + if name == "intake 2" and silent2: + ctx.check(f == {"fitted": False, "turning": False}, "intake 2 with one intake fan: %s", f) continue ctx.check(f.get("ok") is True, "%s: %s", name, f) ctx.log("%s: %s rpm steady, 90 percent at %s s", name, f.get("steady_rpm"), f.get("spinup_s")) @@ -241,8 +256,12 @@ def check_airflow(ctx): s = ctx.forgectrl.settings() or {} for k in keys: ctx.check(s.get(k) == floors.get(k), "%s reads %r, the check wrote %r", k, s.get(k), floors.get(k)) - ctx.check(s.get("cool_intake_fans") == str(intakes), - "cool_intake_fans reads %r, the check found %r", s.get("cool_intake_fans"), intakes) + # The count: 2 written when intake 2 turned; left as counted when it + # was silent. + want = counted if silent2 else "2" + ctx.check(s.get("cool_intake_fans") == want, + "cool_intake_fans reads %r, expected %r (intake 2 %s, the machine counted %r)", + s.get("cool_intake_fans"), want, "silent" if silent2 else "turning", counted) ctx.log("floors: %s", json.dumps(floors)) # The check switches purge air off to read its off current. Leaving # it off costs the next job an airflow hold mid-cut, judged against @@ -281,22 +300,24 @@ def check_airflow(ctx): description="POST /wiz/airflow.intakes/start with the intake fan count set (the precheck " "refuses the start without it: the gate would hold every test after this one). " "The controller stops, the fans run at the run profile for 20 s, and the tachs " - "count the intakes: intake 1 turning and intake 2 never reading a pulse is one, " - "anything else two. No question is asked; the one the step asks when intake 1 " - "does not come up (the operator's count) fails the test. The result carries " - "both intakes' speeds, whether intake 2 was silent, and the count by the tachs, " - "which must be the count the machine held: intake 1 at 1000 rpm or more, intake " - "2 silent exactly when the count is one. The setting reads the count, the record " - "carries airflow.intakes at version 1, the count is put back as found, and the " - "controller must be running again.") + "count the intakes: both turning is two. The step asks the operator to count " + "when the tachs cannot (intake 1 not up to speed, or intake 2 giving no " + "reading, which reads the same as a machine built with one intake fan); both " + "intakes turn on the bench reference, so that question is a fan fault, and it " + "fails the test unanswered. The result carries both intakes' speeds, whether " + "intake 2 was silent, and the count by the tachs, which must be the count the " + "machine held. The setting reads the count, the record carries airflow.intakes " + "at version 1, the count is put back as found, and the controller must be " + "running again.") def check_intakes(ctx): fc = ctx.forgectrl ev = ctx.evidence def on_prompt(p): if p.get("id") == "intake-count": - ctx.fail("intake 1 did not come up to speed, so the tachs could not count the fans: %s", - p.get("text")) + # The tachs could not count, and the step's text says why. + ctx.fail("the intake fans step asked for the count, which the tachs could not make; both " + "intakes turn on the bench reference, so a fan needs attention: %s", p.get("text")) ctx.fail("the intake fans step asked a question: %s", p) with Restore(ctx, [INTAKE_KEY]) as held: found = held.before.get(INTAKE_KEY) @@ -309,13 +330,10 @@ def check_intakes(ctx): "intake_2_silent": r.get("intake_2_silent")} ctx.check(r.get("counted_by") == "tachs", "the fans were not counted by the tachs: %s", r) ctx.check(str(n) == found, "the tachs count %r intake fans, the machine held %s", n, found) - ctx.check((r.get("intake_1_rpm") or 0) >= 1000, "intake 1 turned at %s rpm", r.get("intake_1_rpm")) - ctx.check(r.get("intake_2_silent") is (n == 1), - "intake 2 %s with a count of %s", "silent" if r.get("intake_2_silent") else "turning", n) s = fc.settings() or {} ctx.check(s.get(INTAKE_KEY) == str(n), "%s reads %r, the step counted %r", INTAKE_KEY, s.get(INTAKE_KEY), n) - ctx.log("intake fans: %s by the tachs (intake 1 %s rpm, intake 2 %s rpm%s)", n, r.get("intake_1_rpm"), - r.get("intake_2_rpm"), ", silent" if r.get("intake_2_silent") else "") + ctx.log("intake fans: %s by the tachs (intake 1 %s rpm, intake 2 %s rpm)", n, r.get("intake_1_rpm"), + r.get("intake_2_rpm")) ok = ctx.wait_for(lambda: (fc.get("/mode")[1] or {}).get("controller") == "running", 60) ctx.check(ok is not None, "the controller did not come back after the step") @@ -410,38 +428,88 @@ def check_motion(ctx): "the contact counts, or why not), and the window: the stops found, or the " "fallback (10 half-steps below, 12 above) when they were not. The settings and " "the lens block of /status hold that window, the record carries lens.travel at " - "version 1, the two settings are put back as found, and the controller must be " - "running again.") + "version 1, and the two settings are put back as found. grblHAL reads the " + "window only at its start, and the check started it before the put-back, so in " + "GRBL mode the controller is stopped and started again on the window as found " + "(on every exit path), and it must be running with motion verified under a new " + "process.") def check_lens(ctx): fc = ctx.forgectrl ev = ctx.evidence + restart = {} def on_prompt(p): ctx.fail("the lens travel check asked a question: %s", p) - with Restore(ctx, LENS_WINDOW_KEYS): - last = run_check(ctx, "lens.travel", on_prompt, 300) - r = last.get("result") or {} - ctx.check(len(r.get("z_passes") or []) >= 1, "the lens was not referenced: %s", r.get("z_passes")) - stops, w = r.get("stops") or {}, r.get("window") or {} - ctx.check(isinstance(stops.get("found"), bool), "the result files no stops: %s", r) - window = (w.get("below"), w.get("above")) - if stops.get("found"): - ctx.check(window == (stops.get("below"), stops.get("above")) and w.get("found") is True, - "the window %s is not the stops found %s", w, stops) - else: - ctx.check(window == FALLBACK_WINDOW and w.get("found") is False, - "the stops were not found (%s), but the window is %s", stops.get("why"), w) - s = fc.settings() or {} - held = tuple(str(s.get(k)) for k in LENS_WINDOW_KEYS) - lens = (fc.status() or {}).get("lens") or {} - ev["window"] = {"stops": stops, "window": w, "settings": held, "status": lens} - ctx.check(held == tuple(str(v) for v in window), "the settings hold %s, the check wrote %s", held, window) - ctx.check((lens.get("below"), lens.get("above")) == window, - "/status reads the window %s/%s, the check wrote %s", lens.get("below"), lens.get("above"), window) - ctx.log("lens travel: %s half-steps below the edge, %s above (%s); contact %s/%s", - window[0], window[1], "the stops found" if stops.get("found") else "the fallback: %s" % stops.get("why"), - stops.get("contact_below"), stops.get("contact_above")) - ok = ctx.wait_for(lambda: (fc.get("/mode")[1] or {}).get("controller") == "running", 60) + + def restart_on_found(): + """A fresh controller start on the settings as found, once the + check has ended. Plain waits, not the run's, so an abort gets it + too; a check still ending starts its own controller after the + put-back, which reads them already.""" + try: + t_end = time.time() + 120 + while time.time() < t_end and dark(fc).get("running"): + time.sleep(1) + if dark(fc).get("running"): + restart["skipped"] = "the check is still ending" + ctx.log("the check is still ending: its own controller start reads the settings as found") + return + st, m = fc.get("/mode") + m = m if isinstance(m, dict) else {} + restart.update(mode=m.get("mode"), pid=m.get("pid")) + if m.get("mode") != "grbl": + return # the cloud client reads no lens window + restart["stop"], body = fc.post("/controller/stop") + ctx.log("restarting the controller on the window as found: POST /controller/stop -> %s %s", + restart["stop"], body) + for _ in range(10): + restart["start"], body = fc.post("/controller/start") + if restart["start"] != 409: + break + time.sleep(1) # the check's lease goes just after its end + ctx.log("POST /controller/start -> %s %s", restart["start"], body) + except hw.HwError as e: + restart["error"] = str(e) + ctx.log("the controller could not be restarted on the window as found: %s", e) + try: + with Restore(ctx, LENS_WINDOW_KEYS): + last = run_check(ctx, "lens.travel", on_prompt, 300) + r = last.get("result") or {} + ctx.check(len(r.get("z_passes") or []) >= 1, "the lens was not referenced: %s", r.get("z_passes")) + stops, w = r.get("stops") or {}, r.get("window") or {} + ctx.check(isinstance(stops.get("found"), bool), "the result files no stops: %s", r) + window = (w.get("below"), w.get("above")) + if stops.get("found"): + ctx.check(window == (stops.get("below"), stops.get("above")) and w.get("found") is True, + "the window %s is not the stops found %s", w, stops) + else: + ctx.check(window == FALLBACK_WINDOW and w.get("found") is False, + "the stops were not found (%s), but the window is %s", stops.get("why"), w) + s = fc.settings() or {} + held = tuple(str(s.get(k)) for k in LENS_WINDOW_KEYS) + lens = (fc.status() or {}).get("lens") or {} + ev["window"] = {"stops": stops, "window": w, "settings": held, "status": lens} + ctx.check(held == tuple(str(v) for v in window), "the settings hold %s, the check wrote %s", + held, window) + ctx.check((lens.get("below"), lens.get("above")) == window, + "/status reads the window %s/%s, the check wrote %s", lens.get("below"), lens.get("above"), + window) + ctx.log("lens travel: %s half-steps below the edge, %s above (%s); contact %s/%s", window[0], + window[1], "the stops found" if stops.get("found") else "the fallback: %s" % stops.get("why"), + stops.get("contact_below"), stops.get("contact_above")) + finally: + restart_on_found() + ev["restart"] = dict(restart) + if restart.get("mode") == "grbl": + ctx.check(restart.get("stop") == 200 and restart.get("start") == 200, + "the controller was not restarted on the window as found: %s", restart) + + def back(): + st, m = fc.get("/mode") + m = m if isinstance(m, dict) else {} + return (m.get("controller") == "running" and m.get("motion") == "verified" + and (restart.get("start") != 200 or m.get("pid") != restart.get("pid"))) + ok = ctx.wait_for(back, 150) ctx.check(ok is not None, "the controller did not come back after the check") diff --git a/forgetest/tests/test_baseline.py b/forgetest/tests/test_baseline.py index a6a298a..3a0ef5a 100644 --- a/forgetest/tests/test_baseline.py +++ b/forgetest/tests/test_baseline.py @@ -686,11 +686,65 @@ class BaselineModeTests(BaselineTests): import helpers self.fc = helpers.FakeForgectrl().start() baseline.Baseline._unreachable_until = 0.0 + # the lamp's settle, short: five polls, so a lamp that never holds + # costs the suite no real seconds + self._lamp_timing = (baseline.LID_LAMP_SETTLE_S, baseline.LID_LAMP_POLL_S) + baseline.LID_LAMP_SETTLE_S, baseline.LID_LAMP_POLL_S = 0.05, 0.01 def tearDown(self): + baseline.LID_LAMP_SETTLE_S, baseline.LID_LAMP_POLL_S = self._lamp_timing self.fc.stop() super().tearDown() + def lamp_reads(self, *reads): + """The lid lamp reads `reads` in turn, then what its file holds: + the smooth trigger's fade as the baseline polls it, one read per + poll, so the script does not depend on the clock.""" + seq = list(reads) + real = baseline.hw.sysfs_read + + def read(attr, default=None): + if attr == baseline.LID_LAMP_ATTR and seq: + return seq.pop(0) + return real(attr, default) + baseline.hw.sysfs_read = read + self.addCleanup(setattr, baseline.hw, "sysfs_read", real) + + def lamp_polls(self): + return max(2, int(round(baseline.LID_LAMP_SETTLE_S / baseline.LID_LAMP_POLL_S))) + + def test_a_lamp_still_fading_to_its_idle_level_is_not_a_leftover(self): + # The supervisor re-applied 236 a moment before the check: the lamp + # reads 232 on its way there and then holds 236. Nothing to restore, + # nothing to record. + self._attr("pic/lid_led", "236") + self.lamp_reads("232", "234", "236", "236") + writes = [] + real_write = baseline.hw.sysfs_write + baseline.hw.sysfs_write = lambda attr, value: (writes.append(attr), real_write(attr, value)) + self.addCleanup(setattr, baseline.hw, "sysfs_write", real_write) + left = self.bl().enforce("post", captured=None) + self.assertEqual(left, []) + self.assertNotIn(baseline.LID_LAMP_ATTR, writes) + self.assertTrue(any("the fade toward a level just written" in ln for ln in self.lines), self.lines) + + def test_a_fade_that_only_passes_the_idle_level_is_a_leftover(self): + # A lamp left fading to 77 passes 236 on its way down: one read at + # the level is not the level held. It is restored and recorded at + # the level it settled on. + self.lamp_reads("240", "236", *(["77"] * (self.lamp_polls() - 1))) + left = self.bl().enforce("post", captured=None) + self.assertEqual([(x.item, x.found, x.expected, x.action) for x in left], + [("pic/lid_led", "77", "236", "restored")]) + self.assertEqual(self._read("pic/lid_led"), "236") + + def test_a_lamp_that_never_reaches_the_level_after_the_restore_says_so(self): + # the write does not take: the restore is recorded as failed, with + # the level the lamp holds + self.lamp_reads(*(["77"] * (2 * self.lamp_polls() + 1))) + left = self.bl().enforce("post", captured=None) + self.assertEqual([(x.item, x.action) for x in left], [("pic/lid_led", "failed: reads 77")]) + def run(self, result=None): # only this class's own tests, not the base class's if self._testMethodName not in BaselineModeTests.__dict__: diff --git a/forgetest/tests/test_setup_burns.py b/forgetest/tests/test_setup_burns.py index 34cfc77..33bd63d 100644 --- a/forgetest/tests/test_setup_burns.py +++ b/forgetest/tests/test_setup_burns.py @@ -10,7 +10,9 @@ 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, -and the setting the burn wrote put back as found.""" +the setting the burn wrote put back as found, and for the focus burn the +controller started again on the edge and the window put back. The floor +starts only with Focus on the record and a scrap the lens reaches.""" import os import sys import unittest @@ -18,7 +20,7 @@ import unittest sys.path.insert(0, os.path.dirname(__file__)) import helpers # noqa: E402 from forgetest import catalog # noqa: E402 -from forgetest.runner import Context, Failed, Run # noqa: E402 +from forgetest.runner import Aborted, Context, Failed, Run # noqa: E402 from forgetest.suite import setup_burns, setup_dark # noqa: E402 LIVE = ["laser.focus", "laser.floor"] @@ -78,6 +80,8 @@ WRITES = {"laser.focus": {"lens_hall_edge_z_mm": "6.96"}, "laser.floor": {"laser # travel check wrote, its floor. FOUND = {"lens_hall_edge_z_mm": "4.10", "lens_stop_below_steps": "14", "lens_stop_above_steps": "20", "laser_floor_density": "12", "ui_units": "metric"} +# What grblHAL reads at its start: the edge and the window. +LENS_KEYS = ("lens_hall_edge_z_mm", "lens_stop_below_steps", "lens_stop_above_steps") HOME = "The head is home" @@ -97,8 +101,9 @@ class RegistrationTests(unittest.TestCase): self.assertIsNone(getattr(t, "precheck", None), tid) self.assertEqual(list(self.reg["setup.focus"].requires), ["forgectrl.auth", "setup.check-motion", "laser.emission-witness", "setup.check-lens"]) + # the floor burns at the focus Focus measured, and refuses without it self.assertEqual(list(self.reg["setup.floor"].requires), - ["forgectrl.auth", "setup.check-motion", "laser.emission-witness"]) + ["forgectrl.auth", "setup.check-motion", "laser.emission-witness", "setup.focus"]) def test_the_sheet_is_gone(self): self.assertFalse([i for i in self.reg if i.startswith("setup.sheet")]) @@ -109,6 +114,7 @@ class RegistrationTests(unittest.TestCase): for want in (("forgectrl", "src/wizlive.*"), ("forgectrl", "src/sheet.*"), ("forgectrl", "src/jobstream.*"), ("forgectrl", "src/curverec.*"), ("forgectrl", "src/font_hershey.*"), ("forgectrl", "src/lens.*"), + ("forgectrl", "src/lenshome.*"), # the lens back on its reference at the end ("forgectrl", "src/ui/wizard.*"), ("forgectrl", "src/wizrun.h"), ("grblhal-glowforge", "src/glowforge_laser.c")): self.assertIn(want, covers, "%s lacks %s" % (tid, want)) @@ -124,11 +130,18 @@ class RegistrationTests(unittest.TestCase): class ScriptedDaemon: """The wizard runner as the suite sees it: /wiz//start opens the - first prompt, each answer opens the next (a thickness out of range in - the machine's units opens it again), the press prompt closes on the - poll after the one that served it, the last answer ends the burn with - its result and its writes, /wiz/dark reports it, GET /wiz carries the - versions, and the preview and the program are served per burn.""" + first prompt (the floor ends at once, with the wizard's error, when the + record carries no Focus), each answer opens the next (a thickness that + is not a plain number, out of range in the machine's units, or, for + the floor, outside the lens reach opens it again), the press prompt + closes on the poll after the one that served it, the last answer ends + the burn with its result and its writes, /wiz/dark reports it, GET /wiz + carries the versions (the record carries Focus already, as the bench's + does), the preview and the program are served per burn, the program + ending in the wizard's tail, and the supervisor's stop and start levers + work (a start records the settings the new controller reads).""" + + TAIL = ["G0 X0 Y0", "G92.1", "M2"] # after the lens's return to the edge def __init__(self, fake): self.fake = fake @@ -136,7 +149,9 @@ class ScriptedDaemon: self.dark = {"id": "", "running": False, "phase": "", "progress": 0, "elapsed_s": 0, "log": [], "prompt": None, "result": None, "error": "", "shots": {"lid": False, "head": False}} - self.versions = {} + self.versions = {"laser.focus": 1} + self.starts = [] # the lens settings each controller start read + self.tail = list(self.TAIL) self.answers = {} # wid -> [values] self.order = [] # the burns as started self.aborted = [] @@ -207,7 +222,7 @@ class ScriptedDaemon: lines += ["G0 Z%.2f" % z for z in ladder_z(lens["edge_z"], lens["below"], lens["above"])] if self.stray_z is not None: lines.append("G0 Z%.2f" % self.stray_z) - lines += ["M5", "G0 Z%.2f" % lens["edge_z"], "G0 X0 Y0", "M2"] + lines += ["M5", "G0 Z%.2f" % lens["edge_z"]] + self.tail return 200, ("\n".join(lines) + "\n").encode(), "text/plain" return None @@ -219,6 +234,11 @@ class ScriptedDaemon: return 404, {"error": "no such wizard"} self.order.append(wid) self.settings_at_start[wid] = dict(s) + if wid == "laser.floor" and not self.versions.get("laser.focus"): + # the floor burns at the focus Focus measured: no prompt, no motion + self.dark.update({"id": wid, "running": False, "phase": "", "prompt": None, "result": None, + "error": "run Focus first: the floor burns at the focus it measures"}) + return 200, {"started": True, "id": wid} s.update(self.start_writes.get(wid, {})) self.lens_at_start = lens_block(s) self.queue = list(SCRIPT[wid][0]) @@ -235,8 +255,13 @@ class ScriptedDaemon: self.answers.setdefault(wid, []).append(value) if p["id"] == "thickness": imperial = (self.units or s.get("ui_units")) == "imperial" - mm = float(value) * (25.4 if imperial else 1.0) - if not 0 < mm <= 30: + try: + mm = float(value.strip()) * (25.4 if imperial else 1.0) + except ValueError: + mm = -1 # "1/8" or "3 mm": not a plain number + lens = self.lens_at_start + if not 0 < mm <= 30 or (wid == "laser.floor" + and not lens["reach_min"] <= mm <= lens["reach_max"]): self.open_next() # the same prompt again, a new sequence number return 200, {"ok": True} self.thickness = mm @@ -255,8 +280,19 @@ class ScriptedDaemon: else: s[k] = v return 200, dict(s) + if path == "/controller/stop": + self.fake.state["mode"] = dict(self.fake.state["mode"], controller="standby") + return 200, {"stopped": True} + if path == "/controller/start": + m = self.fake.state["mode"] + self.fake.state["mode"] = dict(m, controller="running", motion="verified", pid=m["pid"] + 1) + self.starts.append({k: s.get(k) for k in LENS_KEYS}) + return 200, {"started": True} return None + def lever_posts(self): + return [p for p, _ in self.fake.posts if p.startswith("/controller/")] + class BurnRunTests(unittest.TestCase): def setUp(self): @@ -300,10 +336,23 @@ class BurnRunTests(unittest.TestCase): self.assertEqual(fw["program_z_before"]["reach"], [-1.44, 10.19]) # the default edge self.assertEqual(fw["program_z_after"]["reach"], [2.17, 13.8]) # the edge the burn wrote self.assertEqual(len(fw["program_z_after"]["z"]), 14) # head, twelve lines, tail + # the program served after the burn ends in the tail at the edge the burn wrote + self.assertEqual(run.evidence["program_tail"]["laser.focus"], ["G0 Z6.96", "G0 X0 Y0", "G92.1", "M2"]) self.assertEqual(run.evidence["settings_before"], {k: FOUND[k] for k in ("lens_hall_edge_z_mm", "lens_stop_below_steps", "lens_stop_above_steps")}) self.assert_found() # restored + self.assert_restarted_on_found() + + def assert_restarted_on_found(self, found=None): + """grblHAL reads the edge and the window only at its start: the + hand-back is one stop and one start after the restore, the start + reading what was found.""" + d = self.daemon + self.assertEqual(d.lever_posts(), ["/controller/stop", "/controller/start"]) + self.assertEqual(d.starts, [found or {k: FOUND[k] for k in LENS_KEYS}]) + self.assertEqual(self.fake.state["mode"]["controller"], "running") + self.assertEqual(self.fake.state["mode"]["pid"], 101) def test_the_focus_burn_in_the_fallback_window(self): # A head the Lens travel check has not measured: the settings hold @@ -318,6 +367,8 @@ class BurnRunTests(unittest.TestCase): self.assertNotIn("lens_stop_below_steps", s) self.assertNotIn("lens_stop_above_steps", s) self.assertEqual(s["lens_hall_edge_z_mm"], "4.10") + self.assert_restarted_on_found({"lens_hall_edge_z_mm": "4.10", "lens_stop_below_steps": None, + "lens_stop_above_steps": None}) def test_the_floor_burn(self): run = self.run_test("setup.floor") @@ -327,6 +378,61 @@ class BurnRunTests(unittest.TestCase): self.assertEqual(d.previews, ["laser.floor"]) self.assertEqual(run.evidence["result"]["floor_density"], 10.0) self.assertEqual(run.evidence["settings_before"], {"laser_floor_density": "12"}) + self.assertEqual(run.evidence["focus_on_record"], 1) + self.assertEqual(run.evidence["program_tail"]["laser.floor"], ["G0 Z4.10", "G0 X0 Y0", "G92.1", "M2"]) + self.assert_found() + # the controller loads the floor at every arm: nothing to restart + self.assertEqual(d.lever_posts(), []) + + def test_the_floor_needs_focus_on_the_record(self): + # The floor burns at the focus Focus measured, and the wizard + # refuses without it: the test says so before the scrap is asked + # for, and nothing starts. + self.daemon.versions = {} + with self.assertRaises(Failed) as cm: + self.run_test("setup.floor") + self.assertIn("the record carries no Focus", str(cm.exception)) + self.assertEqual(self.daemon.order, []) + self.assertEqual(self.daemon.previews, []) + self.assert_found() + + def test_the_floors_own_refusal_without_focus_ends_the_run(self): + # Past the test's own reading (a record that lost Focus in + # between): the wizard's refusal ends the run with its word, before + # any prompt. + real = self.daemon.on_get + + def on_get(path, q): + if path == "/wiz" and not self.daemon.order: + return 200, {"versions": {"laser.focus": 1}, "recommended": [], "completed": True} + return real(path, q) + self.fake.on_get = on_get + self.daemon.versions = {} + with self.assertRaises(Failed) as cm: + self.run_test("setup.floor") + self.assertIn("run Focus first", str(cm.exception)) + self.assertEqual(self.daemon.order, ["laser.floor"]) + self.assertEqual(self.daemon.answers, {}) + self.assert_found() + + def test_a_scrap_the_lens_does_not_reach_fails_before_the_floor(self): + # An edge high enough that the lens cannot focus 3.2 mm above the + # tray: the floor would refuse the thickness, so nothing starts. + self.fake.state["settings"]["lens_hall_edge_z_mm"] = "9.50" + with self.assertRaises(Failed) as cm: + self.run_test("setup.floor") + self.assertIn("leaves out the 3.2 mm scrap", str(cm.exception)) + self.assertEqual(self.daemon.order, []) + + def test_a_program_that_keeps_its_g92_origin_fails_before_the_burn(self): + # The tail clears the G92 origin, which the controller would + # otherwise carry to its next start under a sender's own offsets. + self.daemon.tail = ["G0 X0 Y0", "M2"] + with self.assertRaises(Failed) as cm: + self.run_test("setup.floor") + self.assertIn("not the tail", str(cm.exception)) + self.assertIn("G92.1", str(cm.exception)) + self.assertEqual(self.daemon.order, []) self.assert_found() def test_imperial_units_answer_the_thickness_in_inches(self): @@ -392,6 +498,27 @@ class BurnRunTests(unittest.TestCase): self.run_test("setup.focus") self.assertIn("the focus burn wrote the lens window", str(cm.exception)) self.assert_found() + self.assert_restarted_on_found() # a failed run hands back the controller too + + def test_an_aborted_focus_burn_still_restarts_the_controller_on_the_settings_as_found(self): + # The run aborted at the arm press: the burn is ended, the settings + # go back, and the controller starts again on them. + t = catalog.load_suite()["setup.focus"] + run = Run("test", t.id, t.title) + run.unattended = True + real = self.daemon.on_get + + def on_get(path, q): + p = self.daemon.dark.get("prompt") + if path == "/wiz/dark" and p and p["id"] == "press": + run.aborted.set() + return real(path, q) + self.fake.on_get = on_get + with self.assertRaises(Aborted): + t.fn(Context(run, None, t)) + self.assertEqual(self.daemon.aborted, ["laser.focus"]) + self.assert_found() + self.assert_restarted_on_found() def test_a_focus_ladder_outside_the_window_held_fails(self): self.daemon.patch["laser.focus"] = {"window": {"below": 10, "above": 12, "reach_min": -0.07, diff --git a/forgetest/tests/test_setup_dark.py b/forgetest/tests/test_setup_dark.py index 5e49e76..cee0506 100644 --- a/forgetest/tests/test_setup_dark.py +++ b/forgetest/tests/test_setup_dark.py @@ -8,8 +8,8 @@ takeover ones marked so, every covers map naming the check runner, and the two that put the intake fan count back as found refused unless it is set; and the check driver itself against the fake daemon: a start, the run polled to its end with no question asked, the result taken, the -record read at version 1, and the settings read as before or put back -as found.""" +record read at version 1, the settings read as before or put back as +found, and the controller started again on a lens window put back.""" import os import sys import unittest @@ -48,13 +48,17 @@ class RegistrationTests(unittest.TestCase): self.assertIsNone(getattr(self.reg[tid], "precheck", None), tid) def test_the_intake_count_must_be_set_before_a_check_puts_it_back(self): + # The refusal says why the test cannot start; it never names a + # setting for the operator to change. fake = helpers.FakeForgectrl().start() try: for tid in COUNTED: fake.state["settings"].pop("cool_intake_fans", None) why = self.reg[tid].cannot_start() - self.assertIn("cool_intake_fans is unset", why, tid) - self.assertIn("Intake fans step", why, tid) + self.assertIn("the intake fans are not counted", why, tid) + self.assertIn("the setup gate holds every controller", why, tid) + self.assertNotIn("cool_intake_fans", why, tid) + self.assertNotIn("Intake fans step", why, tid) fake.state["settings"]["cool_intake_fans"] = "2" self.assertIsNone(self.reg[tid].cannot_start(), tid) finally: @@ -168,22 +172,25 @@ class SensorsCheckTests(unittest.TestCase): class AirflowCheckTests(unittest.TestCase): """setup.check-airflow's judge against a scripted daemon: the check - writes its floors and the intake fans fitted, and the test reads the - fans the count says are there. A machine built with one intake fan - reports intake 2 not fitted; the bench reference has two.""" + writes its floors, and the intake fan count is the machine's. Intake + 2 turning proves two fans and the check writes the count as 2; intake + 2 silent on a machine counted with one is reported not fitted and not + turning, and the count is not written; intake 2 silent on a machine + counted with two fails the check. The bench reference has two, both + turning.""" FLOORS = {"cool_tach_exhaust_min_rpm": "6400", "cool_tach_intake_min_rpm": "2290", "cool_tach_air_assist_min_rpm": "6050", "cool_purge_min_current": "345", "cool_fan_grace_s": "12"} FAN = {"steady_rpm": 4150, "spinup_s": 6.8, "ok": True} - def result(self, intakes): + def result(self, silent2): fans = {"exhaust": dict(self.FAN), "intake 1": dict(self.FAN), "intake 2": dict(self.FAN), "air assist": dict(self.FAN)} - if intakes == 1: - fans["intake 2"] = {"fitted": False} - return {"fans": fans, "intake_fans": intakes, "purge_current_on": 628, "purge_current_off": 74, - "floors": dict(self.FLOORS)} + if silent2: + fans["intake 2"] = {"fitted": False, "turning": False} + return {"fans": fans, "intake_fans": 1 if silent2 else 2, "purge_current_on": 628, + "purge_current_off": 74, "floors": dict(self.FLOORS)} def setUp(self): import tempfile @@ -194,25 +201,36 @@ class AirflowCheckTests(unittest.TestCase): self.saved_root = os.environ.get("GF_SYSFS_ROOT") os.environ["GF_SYSFS_ROOT"] = self.sysfs + "/" self.fake = helpers.FakeForgectrl().start() - self.fake.state["settings"].update({"cool_tach_intake_min_rpm": "2300"}) + # the bench reference: two intake fans, counted + self.fake.state["settings"].update({"cool_tach_intake_min_rpm": "2300", "cool_intake_fans": "2"}) self.fake.state["cool"] = {"phase": "idle", "fan_gates": {"purge": {"reading": 628, "floor": 345}}} self.dark = {"id": "", "running": False, "phase": "", "progress": 0, "elapsed_s": 0, "log": [], "prompt": None, "result": None, "error": "", "shots": {"lid": False, "head": False}} self.versions = {} self.polls = 0 - self.wizard_result = self.result(2) - self.wizard_writes = None # what the check writes; None = what its result says + self.silent2 = False # intake 2 gives no tach pulse + self.wizard_result = None # the result the check ends with; None = the daemon's own + self.wizard_writes = None # the count the check writes; None = the daemon's own rule def on_get(path, q): if path == "/wiz/dark": if self.dark["running"]: self.polls += 1 if self.polls >= 2: - r = self.wizard_result s = self.fake.state["settings"] + if self.silent2 and s.get("cool_intake_fans") == "2": + # a silent intake 2 on a machine counted with two + self.dark.update({"running": False, "progress": 100, "result": None, + "error": "a fan did not come up to speed at the run " + "profile: see the log"}) + return 200, dict(self.dark) + r = self.wizard_result or self.result(self.silent2) s.update(r["floors"]) - s["cool_intake_fans"] = self.wizard_writes or str(r["intake_fans"]) + if self.wizard_writes: + s["cool_intake_fans"] = self.wizard_writes + elif not self.silent2: + s["cool_intake_fans"] = "2" # intake 2 turned: two fans self.dark.update({"running": False, "progress": 100, "result": r}) self.versions["airflow"] = 1 return 200, dict(self.dark) @@ -251,34 +269,77 @@ class AirflowCheckTests(unittest.TestCase): t.fn(Context(run, None, t)) return run - def assert_put_back(self): + def count(self, n): + self.fake.state["settings"]["cool_intake_fans"] = n + + def assert_put_back(self, count="2"): s = self.fake.state["settings"] self.assertEqual(s.get("cool_tach_intake_min_rpm"), "2300") - self.assertNotIn("cool_intake_fans", s) + self.assertEqual(s.get("cool_intake_fans"), count) self.assertNotIn("cool_tach_exhaust_min_rpm", s) def test_two_intake_fans_pass_and_are_put_back(self): - self.run_check() + # the bench path: both intakes turn, the count is written as 2 + run = self.run_check() + self.assertEqual(run.evidence["intakes"], {"counted": "2", "found": 2, "intake_2_silent": False}) self.assert_put_back() - def test_one_intake_fan_passes_with_intake_2_not_fitted(self): - self.wizard_result = self.result(1) + def test_a_machine_counted_with_one_whose_intake_2_turns_is_counted_two_and_put_back(self): + # intake 2 turning proves two fans: the check writes 2, and the + # count goes back as found + self.count("1") self.run_check() + self.assert_put_back("1") + self.assertIn(("/settings", {"cool_intake_fans": "1"}), self.fake.posts) + + def test_one_intake_fan_counted_one_passes_and_the_count_is_not_written(self): + self.count("1") + self.silent2 = True + run = self.run_check() + self.assertEqual(run.evidence["intakes"], {"counted": "1", "found": 1, "intake_2_silent": True}) + self.assert_put_back("1") + + def test_a_silent_intake_2_on_a_machine_counted_with_two_fails_the_check(self): + self.silent2 = True + with self.assertRaises(Exception) as cm: + self.run_check() + self.assertIn("a fan did not come up to speed at the run profile", str(cm.exception)) self.assert_put_back() - def test_one_intake_fan_with_the_setting_left_at_two_fails(self): - self.wizard_result = self.result(1) + def test_a_silent_intake_2_whose_count_was_written_fails(self): + # a silent intake 2 never writes the count + self.count("1") + self.silent2 = True self.wizard_writes = "2" with self.assertRaises(Exception) as cm: self.run_check() - self.assertIn("cool_intake_fans reads '2', the check found 1", str(cm.exception)) + self.assertIn("cool_intake_fans reads '2', expected '1' (intake 2 silent", str(cm.exception)) + self.assert_put_back("1") + + def test_a_turning_intake_2_counted_one_fails(self): + r = self.result(False) + r["intake_fans"] = 1 + self.wizard_result = r + with self.assertRaises(Exception) as cm: + self.run_check() + self.assertIn("intake 2 turned, and the check found 1 intake fans", str(cm.exception)) self.assert_put_back() + def test_a_silent_intake_2_reported_fitted_fails(self): + self.count("1") + self.silent2 = True + r = self.result(True) + r["fans"]["intake 2"] = {"fitted": True, "turning": False} + self.wizard_result = r + with self.assertRaises(Exception) as cm: + self.run_check() + self.assertIn("intake 2 with one intake fan", str(cm.exception)) + self.assert_put_back("1") + def test_a_result_without_the_count_fails(self): - r = self.result(2) + r = self.result(False) del r["intake_fans"] self.wizard_result = r - self.wizard_writes = "2" with self.assertRaises(Exception) as cm: self.run_check() self.assertIn("no intake fan count", str(cm.exception)) @@ -289,10 +350,13 @@ class AirflowCheckTests(unittest.TestCase): class ScriptedCheck(unittest.TestCase): """A dark check against the fake daemon: the start, one poll that sees the run, the next that sees its end (`finish` writes the settings and - returns the result), the record at version 1, and /status's lens block - read from the settings as the daemon reads them.""" + returns the result), the record at version 1, /status's lens block + read from the settings as the daemon reads them, and the supervisor's + stop and start levers (a start records the settings the new controller + reads).""" WID = "" + LENS_KEYS = ("lens_stop_below_steps", "lens_stop_above_steps", "lens_hall_edge_z_mm") def setUp(self): self.fake = helpers.FakeForgectrl().start() @@ -303,6 +367,7 @@ class ScriptedCheck(unittest.TestCase): self.polls = 0 self.prompt = None # a prompt the run opens on its first poll, when set self.answers = [] + self.starts = [] # the lens settings each controller start read def on_get(path, q): s = self.fake.state["settings"] @@ -344,6 +409,14 @@ class ScriptedCheck(unittest.TestCase): else: s[k] = v return 200, dict(s) + if path == "/controller/stop": + self.fake.state["mode"] = dict(self.fake.state["mode"], controller="standby") + return 200, {"stopped": True} + if path == "/controller/start": + m = self.fake.state["mode"] + self.fake.state["mode"] = dict(m, controller="running", motion="verified", pid=m["pid"] + 1) + self.starts.append({k: s.get(k) for k in self.LENS_KEYS}) + return 200, {"started": True} return None self.fake.on_get = on_get self.fake.on_post = on_post @@ -351,6 +424,9 @@ class ScriptedCheck(unittest.TestCase): def tearDown(self): self.fake.stop() + def lever_posts(self): + return [p for p, _ in self.fake.posts if p.startswith("/controller/")] + def run_check(self, tid): t = catalog.load_suite()[tid] run = Run("test", t.id, t.title) @@ -381,15 +457,24 @@ class LensCheckTests(ScriptedCheck): "window": {"below": below, "above": above, "found": st["found"]}, "summary": "x"} def assert_put_back(self): + """The window as found, in the settings and in the controller: + grblHAL reads it only at its start, so the hand-back is one stop + and one start after the put-back, the start reading the window + as found.""" s = self.fake.state["settings"] for k, v in self.FOUND.items(): self.assertEqual(s.get(k), v, k) + self.assertEqual(self.lever_posts(), ["/controller/stop", "/controller/start"]) + self.assertEqual([{k: st[k] for k in self.FOUND} for st in self.starts], [self.FOUND]) + self.assertEqual(self.fake.state["mode"]["controller"], "running") + self.assertEqual(self.fake.state["mode"]["pid"], 101) def test_the_stops_found_are_the_window(self): run = self.run_check("setup.check-lens") self.assertEqual(run.evidence["window"]["settings"], ("14", "20")) self.assertEqual(run.evidence["window"]["status"]["below"], 14) self.assertEqual(run.evidence["result"]["stops"]["contact_above"], 22) + self.assertEqual(run.evidence["restart"], {"mode": "grbl", "pid": 100, "stop": 200, "start": 200}) self.assert_put_back() def test_stops_not_found_write_the_fallback(self): @@ -415,54 +500,113 @@ class LensCheckTests(ScriptedCheck): self.assertIn(("/wiz/lens.travel/abort", {}), self.fake.posts) self.assert_put_back() + def test_a_controller_that_does_not_restart_fails_the_hand_back(self): + # The start refused: the controller stays stopped on the stale + # window's heels, and the run says so rather than pass. + real = self.fake.on_post + + def on_post(path, form): + if path == "/controller/start": + return 500, {"error": "no"} + return real(path, form) + self.fake.on_post = on_post + with self.assertRaises(Exception) as cm: + self.run_check("setup.check-lens") + self.assertIn("the controller was not restarted on the window as found", str(cm.exception)) + + def test_cloud_mode_restarts_nothing(self): + # The cloud client reads no lens window: nothing to restart. + self.fake.state["mode"] = dict(self.fake.state["mode"], mode="cloud") + run = self.run_check("setup.check-lens") + self.assertEqual(self.lever_posts(), []) + self.assertEqual(run.evidence["restart"], {"mode": "cloud", "pid": 100}) + class IntakesCheckTests(ScriptedCheck): """setup.check-intakes's judge: the tachs count the intake fans the - machine held, the step writes the count, and it goes back as found.""" + machine held, the step writes the count, and it goes back as found. + The tachs count only two turning fans: intake 2 giving no reading + reads the same as a machine built with one, so the step asks the + operator, and so it does when intake 1 does not come up. Both intakes + turn on the bench reference, so either question fails the test + unanswered.""" WID = "airflow.intakes" + # The step's two questions, as the daemon asks them. + SILENT_2 = ("Intake 1 turns; intake 2 gives no reading. Look at the back of the machine: how many " + "intake fans does it have? If it has two, the second one needs attention before the " + "machine cuts.") + SLOW_1 = ("The intake fans could not be counted: intake 1 did not come up to speed. Look at the back " + "of the machine: how many intake fans does it have? A fan that does not turn needs " + "attention before the machine cuts.") def setUp(self): super().setUp() self.fake.state["settings"]["cool_intake_fans"] = "2" - self.count = 2 def finish(self, s): - s["cool_intake_fans"] = str(self.count) - return {"intake_fans": self.count, "intake_1_rpm": 4150.0, - "intake_2_rpm": 0.0 if self.count == 1 else 4100.0, "intake_2_silent": self.count == 1, + # both intakes turned: two, by the tachs + s["cool_intake_fans"] = "2" + return {"intake_fans": 2, "intake_1_rpm": 4150.0, "intake_2_rpm": 4100.0, "intake_2_silent": False, "counted_by": "tachs", "summary": "x"} + def ask(self, text): + self.prompt = {"id": "intake-count", "kind": "choice", "text": text, "options": ["One", "Two"]} + def test_two_intake_fans_counted_and_put_back(self): run = self.run_check("setup.check-intakes") self.assertEqual(run.evidence["count"]["counted"], 2) + self.assertEqual(run.evidence["count"]["by"], "tachs") self.assertEqual(self.fake.state["settings"]["cool_intake_fans"], "2") self.assertIn(("/settings", {"cool_intake_fans": "2"}), self.fake.posts) # the put-back - def test_one_intake_fan_on_a_machine_built_with_one(self): + def test_a_count_other_than_the_machine_held_fails_and_puts_it_back(self): + # a machine that held one intake fan, whose tachs saw two turn self.fake.state["settings"]["cool_intake_fans"] = "1" - self.count = 1 - run = self.run_check("setup.check-intakes") - self.assertEqual(run.evidence["count"]["counted"], 1) + with self.assertRaises(Exception) as cm: + self.run_check("setup.check-intakes") + self.assertIn("the tachs count 2 intake fans, the machine held 1", str(cm.exception)) self.assertEqual(self.fake.state["settings"]["cool_intake_fans"], "1") - def test_a_count_other_than_the_machine_held_fails_and_puts_it_back(self): - self.count = 1 + def test_a_silent_intake_2_is_a_fault_the_test_does_not_answer(self): + # intake 1 turns, intake 2 gives no reading: the step asks the + # operator, and on the bench reference that is a fan needing attention + self.ask(self.SILENT_2) with self.assertRaises(Exception) as cm: self.run_check("setup.check-intakes") - self.assertIn("the tachs count 1 intake fans, the machine held 2", str(cm.exception)) + self.assertIn("asked for the count, which the tachs could not make", str(cm.exception)) + self.assertIn("intake 2 gives no reading", str(cm.exception)) + self.assertEqual(self.answers, []) + self.assertIn(("/wiz/airflow.intakes/abort", {}), self.fake.posts) self.assertEqual(self.fake.state["settings"]["cool_intake_fans"], "2") - def test_the_operator_question_fails_the_check(self): - # intake 1 never came up: the step asks a person to count, which - # is not a count by the tachs - self.prompt = {"id": "intake-count", "kind": "choice", "text": "how many?", "options": ["One", "Two"]} + def test_a_silent_intake_2_on_a_machine_that_held_one_still_fails(self): + # the test never answers the operator's question, whatever was held + self.fake.state["settings"]["cool_intake_fans"] = "1" + self.ask(self.SILENT_2) with self.assertRaises(Exception) as cm: self.run_check("setup.check-intakes") + self.assertIn("asked for the count", str(cm.exception)) + self.assertEqual(self.answers, []) + self.assertEqual(self.fake.state["settings"]["cool_intake_fans"], "1") + + def test_intake_1_not_up_to_speed_fails_the_check(self): + self.ask(self.SLOW_1) + with self.assertRaises(Exception) as cm: + self.run_check("setup.check-intakes") + self.assertIn("asked for the count", str(cm.exception)) self.assertIn("intake 1 did not come up to speed", str(cm.exception)) self.assertEqual(self.answers, []) self.assertEqual(self.fake.state["settings"]["cool_intake_fans"], "2") + def test_the_tachs_count_is_judged_by_the_count_alone(self): + # counted by the tachs, the count is the whole judgment: the speeds + # are recorded, not judged again + real = self.finish + self.finish = lambda s: dict(real(s), intake_1_rpm=None) + run = self.run_check("setup.check-intakes") + self.assertIsNone(run.evidence["count"]["intake_1_rpm"]) + if __name__ == "__main__": unittest.main()