vh/forgefirm
1
0
Fork 0
mirror of https://github.com/openglow-org/forgefirm.git synced 2026-09-27 08:41:13 -07:00
Code Issues Packages Projects Releases Wiki Activity

Acceptance catalog: merge the tests that share a setup (39 -> 35, live 10 -> 7)

Browse Source 
A sweep of the whole catalog for the overlap the drill work found. The test
applied was "do these share a SETUP", not "do these share a subsystem":
combining only pays where a human waits - an arm press, scrap, a takeover,
a mode entry, a lid choreography - and it costs failure isolation, because
the campaign inherits per test and a merged test invalidates as a unit. The
17 auto tests were left alone for exactly that reason: they cost no operator
time and separate ids give the coverage map and the domain invalidation
finer teeth.

  kernel.k3-unlock + kernel.fire-abu -> kernel.fire-line
      Four phases behind ONE takeover of the pulse device instead of two:
      A/B/U on the FIRE line, then the mid-ramp unlock. Same HV-not-good
      gate, same zero duty, same safe state on the way out.

  laser.expected-stop + laser.kill-mid-fire -> laser.armed-kill
      Both ways an armed job is killed, on one scrap setup: the supervisor's
      expected stop (with its separate operator-judged restart) and then a
      SIGKILL of the restarted controller. The second phase re-reads the pid
      after the restart, so it kills the process the supervisor just spawned.

  laser.lid-cancel-mid-fire + laser.pause-resume-live -> laser.pause-resume-lid-cancel
      One armed burn in the order the factory uses the machine's controls:
      press (pause - emission stops, latch stays UNLOCKED, armed window
      stays open), press (resume), lid (cancel, reset without alarm, return
      to the job start, button latch SET). One arm press instead of two.

  cloud.lid-abort + cloud.interlock-abort-park -> cloud.lid-interlock-abort
      Two prints in one test: the lid, then the interlock with the lid opened
      during the park. The tail both share - park complete, kernel counters
      back at the job start, ':cancelled', latch locked, armed window closed -
      is one helper now, so both triggers are judged the same way.

Not merged, though they share code: cloud.gfhome-homing and cloud.hunt-lid-open.
Both drive Machine._hunt (gfhome.py and gfcloud.py build the same gfhardware
machine, so the lid ungating is the same lines), but each is about the opposite
value of the same variable - gfhome's homing is camera-corrected and needs the
lid CLOSED in GRBL mode, the hunt test needs it OPEN through a cloud-mode
connect. Merging would put a mid-test mode switch back into the cloud tests.

Shared live-test prologue (the arm cue, the mark job, the wait for the emission
witness) and the post-kill trail/judging are helpers now.

Host proof: 104 unit tests - the merged cloud test replays the machine's own
lid-abort excerpt twice, once with the interlock substituted for the trigger,
and three negative cases hold it honest (loop already open, a park the lid can
interrupt, a stop that is not edge-driven) - and the coverage lint at 0
uncovered across 35 tests. Catalog time 137 -> 125 minutes, 96 of it attended.
This commit is contained in:
ScottW514
2026-08-17 08:39:35 -04:00
parent e48b2c8e62
commit cd5098b6b3
4 changed files with 406 additions and 486 deletions
Download Patch File Download Diff File Expand all files Collapse all files
forgetest/forgetest/suite/cloud.py
+91 -103
View File
@@ -500,22 +500,55 @@ CLOUD_STEP = ("Cloud credentials configured; the machine in cloud mode (the test
"GRBL mode and stays in cloud mode; switch back on the panel when done).")
@test("cloud.lid-abort", title="Lid open during a cloud print: stop, park with the lid open, cancelled",
subsystem="cloud", kind="live", est_min=8,
covers=_CLOUD_COVERS, requires=["laser.emission-witness"],
steps=[CLOUD_STEP, "The app open in a browser; scrap on the bed and a small engrave/score job ready.",
"Print from the app and press the button when it lights; open the lid a few seconds "
"into the run."],
description="A cloud print aborted by the lid behaves as the factory's does: the edge "
"reaches the controlled stop within milliseconds, the head returns home at "
"once with the lid still open, the laser latch relocks and the armed window "
"closes, and the job ends ':cancelled'.")
def lid_abort(ctx):
def judge_abort_tail(ctx, ev, offset, tag, fin):
"""The tail every cloud abort shares: the park ran to completion, the head
is back at the job start by the KERNEL counters (cloud clears them at every
job start, so a completed park reads back at zero - stale ring bytes played
ahead of it would not), the print ended ':cancelled', the latch is locked
and the armed window closed."""
ctx.check(ctx.forgectrl.wait_idle(15, abort=ctx.aborted), "[%s] machine not idle after the park", tag)
kpos = read_position()
ev[tag + "_counters_after_park"] = kpos
ctx.log("[%s] kernel counters after the park: %s", tag, kpos)
ctx.check(kpos is not None and abs(kpos[0]) <= 3 and abs(kpos[1]) <= 3,
"[%s] the head did not come back to the job start (kernel counters %s)", tag, kpos)
ctx.check(fin and CANCELLED in fin, "[%s] the print did not end ':cancelled': %s", tag,
message(fin) or "no finish line")
st, cs = ctx.forgectrl.get("/cool/status")
ev[tag + "_armed_after"] = cs.get("armed") if isinstance(cs, dict) else None
ev[tag + "_latch_locked"] = latch_locked()
ctx.check(not ev[tag + "_armed_after"], "[%s] armed window still open after the abort", tag)
ctx.check(ev[tag + "_latch_locked"], "[%s] kernel latch not locked after the abort", tag)
@test("cloud.lid-interlock-abort", title="Lid and interlock each abort a cloud print; the park ignores "
"an open lid",
subsystem="cloud", kind="live", est_min=11,
covers=_CLOUD_COVERS + [("forgectrl", "src/super.c")],
requires=["laser.emission-witness"],
steps=[CLOUD_STEP, "The app open in a browser; scrap on the bed and a small engrave/score job "
"ready - the test runs TWO prints.",
"Be able to open the remote-interlock loop for the second print: unplug the Pro's "
"interlock plug, or pull the jumper at J8 on a Basic/Plus. Restore it at the end.",
"Print 1: press the button to start, open the lid a few seconds in. Print 2: press the "
"button to start, open the interlock a few seconds in, then open the lid as well while "
"the head is parking."],
description="Both enclosure triggers, on one setup, behaving as the factory's do. The lid: the "
"edge reaches the controlled stop within milliseconds, the head returns home at "
"once with the lid still open, the latch relocks and the armed window closes, and "
"the job ends ':cancelled'. The remote-interlock loop: the same tail, and opening "
"the lid while THAT park is running does not interrupt it - the park is deliberately "
"immune, so it always reports complete.")
def lid_interlock_abort(ctx):
ev = ctx.evidence
sw = (ctx.forgectrl.status().get("switches") or {})
ctx.check(sw.get("interlock_ok"), "the interlock loop already reads open - close it before this test")
offset = enter_cloud(ctx)
# -- print 1: the lid ----------------------------------------------------
ctx.instruct(APP_PRINT_CUE)
got = wait_print_running(ctx, offset, 300)
ctx.check(got, "the print never reached its run within 300 s (not started, or the button not pressed)")
ctx.check(got, "print 1 never reached its run within 300 s (not started, or the button not pressed)")
ctx.instruct("The head is moving. Open the lid NOW, then click Done. Leave it open until the head "
"has returned to the corner.")
needles = ["lid opened", "lid opened mid-run; stopping motion", "start return home",
@@ -523,9 +556,9 @@ def lid_abort(ctx):
got = wait_log(ctx, offset, needles, 90)
fin = wait_action_finished(ctx, offset, "print", 60)
got["print finished"] = fin
ev["log"] = {k: message(v) for k, v in got.items()}
ev["lid_log"] = {k: message(v) for k, v in got.items()}
for k, v in got.items():
ctx.log(" %s: %s", k, "seen" if v else "MISSING")
ctx.log(" [lid] %s: %s", k, "seen" if v else "MISSING")
ctx.check(got["lid opened mid-run; stopping motion"], "the lid open did not stop the run")
# The lid edge that stopped the run is the LAST "lid opened" edge line
# before the stop line (an earlier open, e.g. to place the scrap, is
@@ -545,27 +578,53 @@ def lid_abort(ctx):
ev["edge_to_stop_ms"])
ctx.check(got["start return home"] and got["return home complete"],
"the park did not run to completion with the lid open")
# the machine, not the client: the job started at counters (0,0,0)
# (cloud clears them at every job start), so a completed park reads
# back there - stale ring bytes replayed ahead of the park would not
ctx.check(ctx.forgectrl.wait_idle(15, abort=ctx.aborted), "machine not idle after the park")
kpos = read_position()
ev["kernel_counters_after_park"] = kpos
ctx.log("kernel counters after the park: %s", kpos)
ctx.check(kpos is not None and abs(kpos[0]) <= 3 and abs(kpos[1]) <= 3,
"the head did not come back to the job start (kernel counters %s)", kpos)
ctx.check(fin and CANCELLED in fin, "the print did not end ':cancelled': %s",
message(fin) or "no finish line")
st, cs = ctx.forgectrl.get("/cool/status")
ev["armed_after"] = cs.get("armed") if isinstance(cs, dict) else None
ev["latch_locked"] = latch_locked()
ctx.check(not ev["armed_after"], "armed window still open after the abort")
ctx.check(ev["latch_locked"], "kernel latch not locked after the abort")
judge_abort_tail(ctx, ev, offset, "lid", fin)
ctx.confirm("Did the head stop as soon as the lid opened and go straight home with the lid "
"still open, and does the app show the print as cancelled?")
ctx.instruct("Close the lid, then click Done.")
settle_cloud(ctx, offset)
ctx.log("PASS: lid open -> stop in %s ms, park completed with the lid open, ':cancelled'",
# -- print 2: the interlock, with the lid opened during the park ---------
offset = log_size(GFCLOUD_LOG)
ctx.instruct(APP_PRINT_CUE)
got = wait_print_running(ctx, offset, 300)
ctx.check(got, "print 2 never reached its run within 300 s (not started, or the button not pressed)")
ctx.instruct("The head is moving. Open the INTERLOCK loop now (unplug it / pull the jumper), then "
"click Done.")
sw = (ctx.forgectrl.status().get("switches") or {})
ev["interlock_ok_after_pull"] = sw.get("interlock_ok")
ctx.check(sw.get("interlock_ok") is False,
"the interlock still reads closed - the loop was not opened (switches: %s)", sw)
stop_line = "interlock opened mid-run; stopping motion"
got = wait_log(ctx, offset, [stop_line, "start return home"], 90)
ctx.check(got[stop_line], "the interlock open did not stop the run")
ctx.check(got["start return home"], "the abort did not start the return home")
ctx.instruct("The head is on its way back. Open the LID now as well, then click Done - leave both "
"open until the head has stopped.")
done = wait_log(ctx, offset, ["return home complete"], 90)
fin = wait_action_finished(ctx, offset, "print", 60)
ev["interlock_log"] = {stop_line: message(got[stop_line]),
"start return home": message(got["start return home"]),
"return home complete": message(done["return home complete"]),
"print finished": message(fin)}
for k, v in ev["interlock_log"].items():
ctx.log(" [interlock] %s: %s", k, "seen" if v else "MISSING")
ctx.check(done["return home complete"],
"the park did not run to completion with the lid opened during it")
sw = (ctx.forgectrl.status().get("switches") or {})
ev["switches_at_return"] = {"lid": sw.get("lid"), "interlock_ok": sw.get("interlock_ok")}
ctx.check(sw.get("lid") is False,
"the lid was not open at the end of the park - the park's immunity was not exercised")
judge_abort_tail(ctx, ev, offset, "interlock", fin)
ctx.confirm("Did the head stop when the interlock opened and go back to the corner without the "
"open lid interrupting it, and does the app show the print as cancelled?")
ctx.instruct("Close the lid and restore the interlock loop (plug/jumper back in), then click Done.")
sw = (ctx.forgectrl.status().get("switches") or {})
ev["restored"] = {"lid": sw.get("lid"), "interlock_ok": sw.get("interlock_ok")}
ctx.check(sw.get("interlock_ok"), "the interlock loop is still open - restore it before continuing")
settle_cloud(ctx, offset)
ctx.log("PASS: lid open -> stop in %s ms and park with the lid open; interlock open -> the same "
"tail with the park running through a lid edge; both prints ':cancelled'",
ev.get("edge_to_stop_ms"))
@@ -704,81 +763,10 @@ def pause_resume(ctx):
ctx.log("PASS: button pause/resume mid-print, job completed and parked")
@test("cloud.interlock-abort-park", title="The interlock loop aborts a cloud print, and the park ignores "
"the lid",
subsystem="cloud", kind="live", est_min=8,
covers=_CLOUD_COVERS, requires=["cloud.lid-abort"],
steps=[CLOUD_STEP, "The app open in a browser; scrap on the bed and a small engrave/score job ready.",
"Be able to open the remote-interlock loop: unplug the Pro's interlock plug, or pull the "
"jumper at J8 on a Basic/Plus. Restore it at the end.",
"Print from the app and press the button when it lights; open the interlock a few seconds "
"into the run, then open the lid while the head is parking."],
description="In cloud mode the remote-interlock loop is the lid's equal, and the park is immune to "
"both: opening the loop mid-print stops the run and cancels the job, the head returns "
"to the job start, and opening the lid while that park is running does not interrupt "
"it - the print still ends ':cancelled' with the park reported complete.")
def interlock_abort_park(ctx):
ev = ctx.evidence
sw = (ctx.forgectrl.status().get("switches") or {})
ctx.check(sw.get("interlock_ok"), "the interlock loop already reads open - close it before this test")
offset = enter_cloud(ctx)
ctx.instruct(APP_PRINT_CUE)
got = wait_print_running(ctx, offset, 300)
ctx.check(got, "the print never reached its run within 300 s (not started, or the button not pressed)")
ctx.instruct("The head is moving. Open the INTERLOCK loop now (unplug it / pull the jumper), then "
"click Done.")
sw = (ctx.forgectrl.status().get("switches") or {})
ev["interlock_ok_after_pull"] = sw.get("interlock_ok")
ctx.check(sw.get("interlock_ok") is False,
"the interlock still reads closed - the loop was not opened (switches: %s)", sw)
stop_line = "interlock opened mid-run; stopping motion"
got = wait_log(ctx, offset, [stop_line, "start return home"], 90)
ctx.check(got[stop_line], "the interlock open did not stop the run")
ctx.check(got["start return home"], "the abort did not start the return home")
# The lid goes up while the park is running: it must change nothing.
ctx.instruct("The head is on its way back. Open the LID now as well, then click Done - leave both "
"open until the head has stopped.")
done = wait_log(ctx, offset, ["return home complete"], 90)
fin = wait_action_finished(ctx, offset, "print", 60)
ev["log"] = {stop_line: message(got[stop_line]),
"start return home": message(got["start return home"]),
"return home complete": message(done["return home complete"]),
"print finished": message(fin)}
for k, v in ev["log"].items():
ctx.log(" %s: %s", k, "seen" if v else "MISSING")
ctx.check(done["return home complete"],
"the park did not run to completion with the lid opened during it")
sw = (ctx.forgectrl.status().get("switches") or {})
ev["switches_at_return"] = {"lid": sw.get("lid"), "interlock_ok": sw.get("interlock_ok")}
ctx.check(sw.get("lid") is False,
"the lid was not open at the end of the park - the park's immunity was not exercised")
ctx.check(ctx.forgectrl.wait_idle(15, abort=ctx.aborted), "machine not idle after the park")
kpos = read_position()
ev["kernel_counters_after_park"] = kpos
ctx.log("kernel counters after the park: %s", kpos)
ctx.check(kpos is not None and abs(kpos[0]) <= 3 and abs(kpos[1]) <= 3,
"the head did not come back to the job start (kernel counters %s)", kpos)
ctx.check(fin and CANCELLED in fin, "the print did not end ':cancelled': %s",
message(fin) or "no finish line")
st, cs = ctx.forgectrl.get("/cool/status")
ev["armed_after"] = cs.get("armed") if isinstance(cs, dict) else None
ev["latch_locked"] = latch_locked()
ctx.check(not ev["armed_after"], "armed window still open after the abort")
ctx.check(ev["latch_locked"], "kernel latch not locked after the abort")
ctx.confirm("Did the head stop when the interlock opened and go back to the corner without the open "
"lid interrupting it, and does the app show the print as cancelled?")
ctx.instruct("Close the lid and restore the interlock loop (plug/jumper back in), then click Done.")
sw = (ctx.forgectrl.status().get("switches") or {})
ev["restored"] = {"lid": sw.get("lid"), "interlock_ok": sw.get("interlock_ok")}
ctx.check(sw.get("interlock_ok"), "the interlock loop is still open - restore it before continuing")
settle_cloud(ctx, offset)
ctx.log("PASS: interlock open -> stop, cancelled, park completed through an open lid")
@test("cloud.pause-cancel-paths", title="A paused cloud print is cancelled by the lid, and a running one "
"by the app",
subsystem="cloud", kind="live", est_min=12,
covers=_CLOUD_COVERS, requires=["cloud.pause-resume", "cloud.lid-abort"],
covers=_CLOUD_COVERS, requires=["cloud.pause-resume", "cloud.lid-interlock-abort"],
steps=[CLOUD_STEP, "The app open in a browser; scrap on the bed and a small engrave/score job "
"ready - the test runs TWO prints.",
"Print 1: press the button to start, press it again a few seconds in (pause), then open "
forgetest/forgetest/suite/kernel.py
+52 -55
View File
@@ -294,54 +294,44 @@ def k1_k2(ctx):
# ---------------------------------------------------------------- K3
@test("kernel.k3-unlock", title="Mid-run latch unlock never re-arms FIRE", subsystem="kernel",
kind="operator", hardware="takeover", always=True, est_min=2,
covers=_KERNEL_COVERS,
requires=["kernel.k1-k2"],
steps=["If the HV supply reports good the drill asks you to open the lid first (the "
"safety chain holds HV off); zero duty throughout."],
description="Stream of FIRE bits run with the latch locked (laser-less by the run-start "
"guard); the latch is unlocked during the accel ramp. The unlock drives the "
"latch pin (interlock bit 3 clears) but must not restore the FIRE drive while "
"the run is in flight: laser_enable stays 0 for the entire run.")
def k3_unlock(ctx):
def _k3_phase(ctx):
"""K3: the latch unlocked during the accel ramp must not restore the FIRE
drive while the run is in flight. Runs inside the caller's takeover."""
ev = ctx.evidence
require_hv_not_good(ctx)
with ctx.takeover():
check_hv_not_good(ctx)
stream = POWER0 + FIRE * (3 * TICK_HZ) + PAD * (TICK_HZ // 2)
ctx.log("K3: %d bytes = %.1f s of FIRE bits; ramp_rate 10000 Hz/s (~0.9 s accel "
"window); unlock at t=+0.15 s", len(stream), len(stream) / TICK_HZ)
snap(ctx, "K3 pre")
wr("cnc/motor_lock", 15)
check_hv_not_good(ctx)
stream = POWER0 + FIRE * (3 * TICK_HZ) + PAD * (TICK_HZ // 2)
ctx.log("K3: %d bytes = %.1f s of FIRE bits; ramp_rate 10000 Hz/s (~0.9 s accel "
"window); unlock at t=+0.15 s", len(stream), len(stream) / TICK_HZ)
snap(ctx, "K3 pre")
wr("cnc/motor_lock", 15)
wr("cnc/laser_latch", 1)
wr("cnc/step_freq", TICK_HZ)
wr("cnc/ramp_rate", 10000)
try:
with PulseDevice(ctx) as dev:
dev.rewind()
wr("cnc/enable", 1)
ctx.sleep(0.5)
dev.write(stream)
wr("cnc/run", 1)
time.sleep(0.15) # inside the accel ramp
wr("cnc/laser_latch", 0)
ilk = rd("cnc/interlock_circuit")
ctx.log("K3: latch UNLOCKED mid-ramp; interlock=%s (bit 3 should read 0)", ilk)
hits, state = watch_laser_until_idle(ctx, 20)
snap(ctx, "K3 post")
ev["k3"] = {"interlock_after_unlock": ilk, "hits": hits[:10], "end_state": state,
"laser_on_sampled": rd("cnc/laser_on_sampled"),
"underruns": rd("cnc/underruns"), "faults": rd("cnc/faults")}
finally:
wr("cnc/laser_latch", 1)
wr("cnc/step_freq", TICK_HZ)
wr("cnc/ramp_rate", 10000)
try:
with PulseDevice(ctx) as dev:
dev.rewind()
wr("cnc/enable", 1)
ctx.sleep(0.5)
dev.write(stream)
wr("cnc/run", 1)
time.sleep(0.15) # inside the accel ramp
wr("cnc/laser_latch", 0)
ilk = rd("cnc/interlock_circuit")
ctx.log("K3: latch UNLOCKED mid-ramp; interlock=%s (bit 3 should read 0)", ilk)
hits, state = watch_laser_until_idle(ctx, 20)
snap(ctx, "K3 post")
ev["k3"] = {"interlock_after_unlock": ilk, "hits": hits[:10], "end_state": state,
"laser_on_sampled": rd("cnc/laser_on_sampled"),
"underruns": rd("cnc/underruns"), "faults": rd("cnc/faults")}
finally:
wr("cnc/laser_latch", 1)
try:
wr("cnc/ramp_rate", 125000)
except OSError:
ctx.log("WARNING: could not restore ramp_rate=125000")
ctx.check((int(ilk) & LATCH_BIT) == 0, "K3: the unlock did not drive the latch pin (interlock=%s)", ilk)
ctx.check(not hits, "K3: FIRE drive re-armed by a mid-run unlock: %s", hits[:10])
ctx.log("K3 PASS: laser_enable stayed 0 for the entire run after the mid-ramp unlock")
wr("cnc/ramp_rate", 125000)
except OSError:
ctx.log("WARNING: could not restore ramp_rate=125000")
ctx.check((int(ilk) & LATCH_BIT) == 0, "K3: the unlock did not drive the latch pin (interlock=%s)", ilk)
ctx.check(not hits, "K3: FIRE drive re-armed by a mid-run unlock: %s", hits[:10])
ctx.log("K3 PASS: laser_enable stayed 0 for the entire run after the mid-ramp unlock")
# ---------------------------------------------------------------- FIRE A/B/U
@@ -433,18 +423,23 @@ def _fire_phase(ctx, mode):
return ev
@test("kernel.fire-abu", title="FIRE line: latch locked, unlocked-unarmed, true underrun",
subsystem="kernel", kind="operator", hardware="takeover", always=True, est_min=3,
@test("kernel.fire-line", title="FIRE line: latch locked, unlocked-unarmed, true underrun, and a "
"mid-run unlock",
subsystem="kernel", kind="operator", hardware="takeover", always=True, est_min=4,
covers=_KERNEL_COVERS,
requires=["kernel.k1-k2"],
steps=["Phases B and U unlock the latch with a zero-duty stream: if the HV supply reports "
"good the drill asks you to open the lid first (the safety chain holds HV off)."],
description="A: latch locked, 40 000 streamed FIRE bits, nothing on the FIRE/LASER_ON "
"nets. B: latch unlocked with the chain unarmed, the FIRE line is driven "
"mid-window and LASER_ON stays off (the safety AND-gate holds), FIRE clear at "
"end-of-data. U: streaming declared, the terminal end-of-data is a true "
"underrun, the backstop drops FIRE and stop acks it.")
def fire_abu(ctx):
steps=["Phases B, U and K3 unlock the latch with a zero-duty stream: if the HV supply "
"reports good the drill asks you to open the lid first (the safety chain holds HV "
"off)."],
description="Four phases behind one takeover of the pulse device, all zero duty. A: latch "
"locked, 40 000 streamed FIRE bits, nothing on the FIRE/LASER_ON nets. B: latch "
"unlocked with the chain unarmed, the FIRE line is driven mid-window and "
"LASER_ON stays off (the safety AND-gate holds), FIRE clear at end-of-data. U: "
"streaming declared, the terminal end-of-data is a true underrun, the backstop "
"drops FIRE and stop acks it. K3: the latch unlocked during the accel ramp "
"drives the latch pin but never restores the FIRE drive to a run already in "
"flight - laser_enable stays 0 for the whole run.")
def fire_line(ctx):
ev = ctx.evidence
require_hv_not_good(ctx)
with ctx.takeover():
@@ -479,6 +474,8 @@ def fire_abu(ctx):
ctx.check(u["underruns_after"] == u["underruns_before"] + 1,
"U: underrun counter %d -> %d", u["underruns_before"], u["underruns_after"])
ctx.log("fire U PASS: true underrun, backstop dropped FIRE, stop acked")
_k3_phase(ctx)
finally:
wr("cnc/laser_latch", 1)
try:
forgetest/forgetest/suite/laser.py
+175 -244
View File
@@ -75,6 +75,52 @@ def stream(g, lines):
g.send_raw((ln + "\n").encode())
MARK_JOB = ["G91", "G21", "M4", "S400",
"G1 X40 F200", "G1 Y40 F200", "G1 X-40 F200", "G1 Y-40 F200",
"M5", "G90", "M2"]
def arm_and_fire(ctx, g, room="40 mm +X and +Y", job=None, timeout=240):
"""The arm cue, the job, and the wait for the emission witness - the
prologue every live test shares. Returns the first sample with emission,
or soft-resets and fails: no emission means the arm was refused or the
button was never pressed."""
ctx.instruct(ARM_CUE % room)
stream(g, job or MARK_JOB)
t0 = time.time()
while time.time() - t0 < timeout:
ctx.checkpoint()
smp = sample(ctx)
if smp and smp["emission"] and smp["emission"] > 0:
return smp
time.sleep(0.15)
g.realtime(0x18)
raise Failed("no emission seen within %d s (arm refused, or no button press)" % timeout)
def kill_trail(ctx, t0, seconds=5.0):
"""Sample emission / kernel state / armed for `seconds` after a kill."""
trail = []
for _ in range(int(seconds / 0.12)):
s = sample(ctx)
if s:
trail.append((round(time.time() - t0, 2), s["emission"], s["kstate"], s["armed"]))
time.sleep(0.12)
return trail
def judge_kill(ctx, trail, what):
"""(first zero, tail stayed zero, kernel stopped running) from a trail."""
for t in trail:
ctx.log(" post-%s %s", what, t)
zero_at = next((t for t, e, _, _ in trail if e == 0), None)
tail_zero = all(e == 0 for _, e, _, _ in trail[-16:])
not_running = all(k != "running" for _, _, k, _ in trail[-16:])
ctx.log("emission first 0 at +%s s; last 2 s all zero: %s; kernel not running: %s",
zero_at, tail_zero, not_running)
return zero_at, tail_zero, not_running
def run_and_sample(ctx, g, job, sample_hz=8, overall_timeout=200):
"""Stream the job; sample forgectrl through arm -> fire -> disarm.
Completes on: emission seen then Idle > 3 s; or armed then disarmed
@@ -276,63 +322,42 @@ def disarm_in_hold(ctx):
"going dark on its own about a minute later?")
@test("laser.expected-stop", title="Armed kill on the expected-stop path (POST /controller/stop)",
subsystem="laser", kind="live", est_min=4,
@test("laser.armed-kill", title="Armed kill mid-fire: the expected stop, then a SIGKILL",
subsystem="laser", kind="live", est_min=6,
covers=_LASER_COVERS + [("forgectrl", "src/main.c")],
requires=["laser.emission-witness"],
requires=["laser.emission-witness", "motion.deadman"],
steps=["Scrap under the head with 40 mm of free +X and +Y travel; lid closed; exhaust on.",
"Press the physical button when it lights white.",
"The controller is left stopped until you judge the stop; the test then restarts it."],
description="Start a mark job at S400/F200; once emission is live, POST /controller/stop. "
"Emission must drop to 0 within 2.5 s and stay 0, the kernel must not be "
"running, and the supervisor's restart is a separate, operator-judged step "
"(POST /controller/start, no motion, no laser).")
def expected_stop(ctx):
"Press the physical button when it lights white - twice over the test, once per burn.",
"After the first burn the controller is left stopped until you judge the stop; the "
"test then restarts it and runs the second burn."],
description="Both ways an armed job is killed, on one setup. Expected: mid-burn "
"POST /controller/stop - the supervisor writes cnc/stop and relocks before the "
"SIGTERM, so emission drops within 2.5 s and stays 0, the kernel is not running, "
"and the restart is a separate operator-judged step. Unexpected: mid-burn SIGKILL "
"of the controller - the supervisor's exit safing must end the fire tail inside "
"the ring's in-flight window, leave the latch locked, and respawn the controller.")
def armed_kill(ctx):
ev = ctx.evidence
fc = ctx.forgectrl
# -- 1. the expected stop -------------------------------------------------
with ctx.grbl() as g, LiveJob(ctx, g):
prepare(ctx, g)
ctx.instruct(ARM_CUE % "40 mm +X and +Y")
stream(g, ["G91", "G21", "M4", "S400",
"G1 X40 F200", "G1 Y40 F200", "G1 X-40 F200", "G1 Y-40 F200",
"M5", "G90", "M2"])
t0 = time.time()
smp = None
seen = False
while time.time() - t0 < 240:
ctx.checkpoint()
smp = sample(ctx)
if smp and smp["emission"] and smp["emission"] > 0:
seen = True
break
time.sleep(0.15)
if not seen:
g.realtime(0x18)
raise Failed("no emission seen within 240 s (arm refused, or no button press)")
smp = arm_and_fire(ctx, g)
ctx.log("emission live (%s) - stopping the controller NOW", smp["emission"])
t_stop = time.time()
code, body = fc.post("/controller/stop")
post_dt = time.time() - t_stop
ctx.log("POST /controller/stop -> %s %s (%.2f s)", code, body, post_dt)
trail = []
for _ in range(40): # ~5 s at 8 Hz
s = sample(ctx)
if s:
trail.append((round(time.time() - t_stop, 2), s["emission"], s["kstate"], s["armed"]))
time.sleep(0.12)
for t in trail:
ctx.log(" post-stop %s", t)
zero_at = next((t for t, e, _, _ in trail if e == 0), None)
tail_zero = all(e == 0 for _, e, _, _ in trail[-16:])
not_running = all(k != "running" for _, _, k, _ in trail[-16:])
trail = kill_trail(ctx, t_stop)
zero_at, tail_zero, not_running = judge_kill(ctx, trail, "stop")
st_mode, mode = fc.get("/mode")
ev.update({"post_status": code, "post_s": round(post_dt, 2), "zero_at_s": zero_at,
"tail_zero": tail_zero, "kernel_not_running": not_running, "mode_after_stop": mode,
"trail": trail})
ctx.log("emission first 0 at +%s s; last 2 s all zero: %s; kernel not running: %s; /mode %s",
zero_at, tail_zero, not_running, mode)
ev["expected"] = {"post_status": code, "post_s": round(post_dt, 2), "zero_at_s": zero_at,
"tail_zero": tail_zero, "kernel_not_running": not_running,
"mode_after_stop": mode, "trail": trail}
ctx.check(code == 200, "POST /controller/stop -> %s", code)
ctx.check(zero_at is not None and zero_at < 2.5, "emission did not drop within 2.5 s (first 0 at %s)", zero_at)
ctx.check(zero_at is not None and zero_at < 2.5,
"emission did not drop within 2.5 s of the stop (first 0 at %s)", zero_at)
ctx.check(tail_zero, "emission returned after the stop")
ctx.check(not_running, "the kernel was still running after the stop")
ctx.instruct("The controller is STOPPED (supervision held). Judge the stop on the scrap - a "
@@ -347,67 +372,32 @@ def expected_stop(ctx):
ctx.log("/mode after start: %s", mode)
ctx.check(isinstance(mode, dict) and mode.get("controller") == "running",
"controller not running after the restart: %s", mode)
ctx.log("PASS: stop in %.2f s, emission 0 at +%s s, controller restarted", post_dt, zero_at)
ctx.log("expected stop PASS: returned in %.2f s, emission 0 at +%s s, controller restarted",
post_dt, zero_at)
@test("laser.kill-mid-fire", title="Armed kill: SIGKILL of the controller while emitting",
subsystem="laser", kind="live", est_min=4,
covers=_LASER_COVERS,
requires=["laser.expected-stop", "motion.deadman"],
steps=["Scrap under the head with 40 mm of free +X and +Y travel; lid closed; exhaust on.",
"Press the physical button when it lights white."],
description="Start a mark job at S400/F200; once emission is live, SIGKILL the controller. "
"The supervisor's exit safing must end the fire tail within the ring's in-flight "
"window: emission drops to 0 within 2.5 s and stays 0, the kernel is not "
"running, the latch reads locked, and the controller is respawned.")
def kill_mid_fire(ctx):
# -- 2. the SIGKILL -------------------------------------------------------
# The pid is the RESTARTED controller's, not the one phase 1 stopped.
import os as _os
import signal as _signal
ev = ctx.evidence
fc = ctx.forgectrl
st, m0 = fc.get("/mode")
ctx.check(st == 200 and isinstance(m0, dict) and m0.get("controller") == "running", "controller not running: %s", m0)
ctx.check(st == 200 and isinstance(m0, dict) and m0.get("controller") == "running",
"controller not running before the kill: %s", m0)
pid = m0.get("pid")
with ctx.grbl() as g, LiveJob(ctx, g):
prepare(ctx, g)
ctx.instruct(ARM_CUE % "40 mm +X and +Y")
stream(g, ["G91", "G21", "M4", "S400",
"G1 X40 F200", "G1 Y40 F200", "G1 X-40 F200", "G1 Y-40 F200",
"M5", "G90", "M2"])
t0 = time.time()
smp = None
seen = False
while time.time() - t0 < 240:
ctx.checkpoint()
smp = sample(ctx)
if smp and smp["emission"] and smp["emission"] > 0:
seen = True
break
time.sleep(0.15)
if not seen:
g.realtime(0x18)
raise Failed("no emission seen within 240 s (arm refused, or no button press)")
smp = arm_and_fire(ctx, g)
ctx.log("emission live (%s) - SIGKILL controller pid %s NOW", smp["emission"], pid)
t_kill = time.time()
_os.kill(pid, _signal.SIGKILL)
trail = []
for _ in range(40): # ~5 s at 8 Hz
s = sample(ctx)
if s:
trail.append((round(time.time() - t_kill, 2), s["emission"], s["kstate"], s["armed"]))
time.sleep(0.12)
for t in trail:
ctx.log(" post-kill %s", t)
zero_at = next((t for t, e, _, _ in trail if e == 0), None)
tail_zero = all(e == 0 for _, e, _, _ in trail[-16:])
not_running = all(k != "running" for _, _, k, _ in trail[-16:])
trail = kill_trail(ctx, t_kill)
zero_at, tail_zero, not_running = judge_kill(ctx, trail, "kill")
ilk = hw.sysfs_int("cnc/interlock_circuit")
locked = ilk is not None and bool(ilk & (1 << 3))
ev.update({"pid": pid, "zero_at_s": zero_at, "tail_zero": tail_zero, "kernel_not_running": not_running,
"latch_locked": locked, "trail": trail})
ctx.log("emission first 0 at +%s s; last 2 s all zero: %s; kernel not running: %s; latch locked: %s",
zero_at, tail_zero, not_running, locked)
ctx.check(zero_at is not None and zero_at < 2.5, "emission did not drop within 2.5 s (first 0 at %s)", zero_at)
ev["sigkill"] = {"pid": pid, "zero_at_s": zero_at, "tail_zero": tail_zero,
"kernel_not_running": not_running, "latch_locked": locked, "trail": trail}
ctx.log("latch locked after the kill: %s", locked)
ctx.check(zero_at is not None and zero_at < 2.5,
"emission did not drop within 2.5 s of the kill (first 0 at %s)", zero_at)
ctx.check(tail_zero, "emission returned after the kill")
ctx.check(not_running, "the kernel was still running after the kill")
ctx.check(locked, "latch not locked after the kill")
@@ -418,13 +408,14 @@ def kill_mid_fire(ctx):
if isinstance(m1, dict) and m1.get("controller") == "running" and m1.get("pid") != pid:
break
ctx.sleep(1)
ev["mode_after"] = m1
ev["mode_after_kill"] = m1
ctx.log("/mode after the kill: %s", m1)
ctx.check(m1 and m1.get("controller") == "running" and m1.get("pid") != pid,
"supervisor did not respawn the controller: %s", m1)
ctx.confirm("Did the cut end abruptly at the kill (a short line, no run-on), with the machine "
"quiet and the button dark now?")
ctx.log("PASS: emission 0 at +%s s after SIGKILL, latch locked, controller respawned", zero_at)
ctx.confirm("Did both burns end abruptly where they were killed (a short line, no run-on), "
"with the machine quiet and the button dark now?")
ctx.log("PASS: expected stop 0 at +%s s and SIGKILL 0 at +%s s, latch locked, controller "
"respawned", ev["expected"]["zero_at_s"], zero_at)
@test("laser.arm-wait-lid", title="Lid open during the arm wait cancels the job",
@@ -497,142 +488,39 @@ def arm_wait_lid(ctx):
ctx.log("PASS: lid open during the arm wait cancelled the job (clean reset, no alarm), armed=false, latch locked")
@test("laser.lid-cancel-mid-fire", title="Lid open mid-burn: beam off in hardware, job cancelled, head home",
subsystem="laser", kind="live", est_min=5,
@test("laser.pause-resume-lid-cancel", title="One live cut: the button pauses and resumes it, the lid "
"cancels it and sends the head home",
subsystem="laser", kind="live", est_min=7,
covers=_LASER_COVERS + [("grblhal-glowforge", "src/glowforge_switches.c"),
("grblhal-glowforge", "src/glowforge_switch_map.h")],
requires=["laser.emission-witness", "motion.lid-cancel-home"],
requires=["laser.emission-witness", "motion.lid-cancel-home", "motion.button-hold-resume"],
steps=["Scrap under the head with 40 mm of free +X and +Y travel; lid closed; exhaust on.",
"Press the physical button when it lights white; open the lid once the burn is under way "
"and leave it open until the head has come back."],
description="Start the S400 square; once emission is live, open the lid. Emission stops in "
"hardware, the job parks and is cancelled (reason reported, controller reset "
"with the position kept), the armed window closes and the kernel latch relocks, "
"the hardware button latch reads SET, and the head returns to the job start "
"with the lid still open.")
def lid_cancel_mid_fire(ctx):
"Press the physical button when it lights white (arm). Once the cut is under way press "
"it again (pause), wait about 3 seconds, press it once more (resume), and then open the "
"lid and leave it open until the head has come back.",
"Keep the pause short: the armed window's idle grace closes it after about a minute in "
"a hold, and a job that disarms cannot resume its emission."],
description="The machine's own controls during one armed burn, in the order the factory uses "
"them. Press: the job feed-holds, emission stops, and the latch stays UNLOCKED "
"with the armed window open - a pause is not a cancel. Press again: the cut "
"resumes from where it stopped (GRBL has no backtrack; the kernel refuses one on "
"a live-streamed ring). Lid: emission stops in hardware, the job is cancelled "
"with the reason reported, the controller resets with the position kept and no "
"alarm, the armed window closes and the kernel latch relocks, the hardware button "
"latch reads SET, and the head returns to the job start with the lid still open.")
def pause_resume_lid_cancel(ctx):
ev = ctx.evidence
with ctx.grbl() as g, LiveJob(ctx, g):
prepare(ctx, g)
start = g.status_report()["MPos"]
k0 = kernel_xy_mm(ctx)
ev["kernel_start"] = k0
ctx.instruct(ARM_CUE % "40 mm +X and +Y")
stream(g, ["G91", "G21", "M4", "S400",
"G1 X40 F200", "G1 Y40 F200", "G1 X-40 F200", "G1 Y-40 F200",
"M5", "G90", "M2"])
t0 = time.time()
smp = None
seen = False
while time.time() - t0 < 240:
ctx.checkpoint()
smp = sample(ctx)
if smp and smp["emission"] and smp["emission"] > 0:
seen = True
break
time.sleep(0.15)
if not seen:
g.realtime(0x18)
raise Failed("no emission seen within 240 s (arm refused, or no button press)")
ctx.log("emission live (%s) - asking the operator to open the lid", smp["emission"])
ctx.instruct("The laser is cutting. Open the lid NOW and leave it open, then click Done.")
t_lid = time.time()
trail = []
text = ""
while time.time() - t_lid < 8:
s = sample(ctx)
if s:
trail.append((round(time.time() - t_lid, 2), s["emission"], s["kstate"], s["armed"]))
text += g.drain()
time.sleep(0.12)
for t in trail:
ctx.log(" post-lid %s", t)
ev["messages"] = [ln for ln in text.splitlines() if ln.startswith("[MSG:") or "help]" in ln
or ln.startswith("ALARM")]
ctx.log("controller: %s", ev["messages"])
zero_at = next((t for t, e, _, _ in trail if e == 0), None)
tail_zero = all(e == 0 for _, e, _, _ in trail[-16:])
ev.update({"zero_at_s": zero_at, "tail_zero": tail_zero})
ctx.check(zero_at is not None and zero_at < 3.0, "emission did not stop after the lid opened (first 0 at %s)", zero_at)
ctx.check(tail_zero, "emission returned after the lid opened")
ctx.check("lid opened - job cancelled" in text, "the lid open was not reported as cancelling the job")
ctx.check("help]" in text, "no reset banner after the cancel")
ctx.check("ALARM" not in text, "an alarm was raised on the cancel (position should be kept)")
# the head returns to the job start on its own
t1 = time.time()
returned = "returned to the job start" in text
while not returned and time.time() - t1 < 30:
ctx.checkpoint()
text += g.drain()
returned = "returned to the job start" in text
time.sleep(0.2)
ev["returned_message"] = returned
ctx.check(returned, "the head did not report returning to the job start")
st = g.status_report()
drift = max(abs(st["MPos"][i] - start[i]) for i in range(2))
ev["drift_mm"] = round(drift, 3)
s = sample(ctx)
ev["armed_after"] = s["armed"] if s else None
ilk = hw.sysfs_int("cnc/interlock_circuit")
ev["latch_locked"] = ilk is not None and bool(ilk & (1 << 3))
ev["button_latch"] = hw.sysfs_int("cnc/button_latch")
ctx.log("returned: drift %.3f mm; armed=%s latch_locked=%s button_latch=%s", drift,
ev["armed_after"], ev["latch_locked"], ev["button_latch"])
ctx.check(drift <= 0.05, "head not back at the job start (drift %.3f mm)", drift)
# what the MACHINE did: the kernel counters must agree
ctx.check(ctx.forgectrl.wait_idle(10, abort=ctx.aborted), "machine not idle after the return")
check_kernel_returned(ctx, ev, k0)
ctx.check(not ev["armed_after"], "armed window still open after the cancel")
ctx.check(ev["latch_locked"], "kernel latch not locked after the cancel")
ctx.check(ev["button_latch"] == 1, "hardware button latch not SET after the lid open (%s)", ev["button_latch"])
ctx.confirm("Did the burn stop the instant the lid opened, and did the head then go straight back "
"to where the job started with the lid still open, dark?")
ctx.instruct("Close the lid, then click Done.")
ctx.sleep(1)
ctx.log("PASS: lid open mid-burn -> emission 0 at +%s s, cancelled, reset without alarm, returned (drift %.3f mm), "
"button latch SET", zero_at, drift)
@test("laser.pause-resume-live", title="Button pause and resume mid-burn: emission stops, the window stays "
"open, the cut continues",
subsystem="laser", kind="live", est_min=6,
covers=_LASER_COVERS + [("grblhal-glowforge", "src/glowforge_switches.c"),
("grblhal-glowforge", "src/glowforge_switch_map.h")],
requires=["laser.emission-witness", "motion.button-hold-resume"],
steps=["Scrap under the head with 60 mm of free +X travel; lid closed; exhaust on.",
"Press the physical button when it lights white (arm). A few seconds into the cut press "
"it once (pause), wait about 3 seconds, and press it once more (resume). Let the job finish.",
"Keep the pause short: the armed window's idle grace closes it after about a minute in a "
"hold, and a job that disarms cannot resume its emission."],
description="The button's pause/resume during a live cut, the factory's behavior on the machine: "
"the first press feed-holds the job and emission stops, the laser latch stays "
"UNLOCKED and the armed window stays open (a pause is not a cancel), and the second "
"press resumes the cut from where it stopped and the job runs to its end. GRBL mode "
"has no backtrack - the kernel refuses one on a live-streamed ring - so the restart "
"picks up where the deceleration ended.")
def pause_resume_live(ctx):
ev = ctx.evidence
with ctx.grbl() as g, LiveJob(ctx, g):
prepare(ctx, g)
start = g.status_report()["MPos"]
ev["start"] = start
ctx.instruct(ARM_CUE % "60 mm +X")
stream(g, ["G91", "G21", "M3", "S400", "G1 X60 F300", "M5", "G90", "M2"])
t0 = time.time()
smp = None
seen = False
while time.time() - t0 < 240:
ctx.checkpoint()
smp = sample(ctx)
if smp and smp["emission"] and smp["emission"] > 0:
seen = True
break
time.sleep(0.15)
if not seen:
g.realtime(0x18)
raise Failed("no emission seen within 240 s (arm refused, or no button press)")
ev["kernel_start"] = k0
smp = arm_and_fire(ctx, g)
ev["emission_running"] = smp["emission"]
ctx.log("emission live (%s) - asking the operator to pause", smp["emission"])
# -- the button pauses ------------------------------------------------
ctx.instruct("The laser is cutting. Press the button ONCE now (pause), then click Done - "
"do not wait long before the next step.")
st = wait_state(ctx, g, "Hold", 8)
@@ -640,8 +528,6 @@ def pause_resume_live(ctx):
text = drain_text(g, 0.5)
ev["hold_state"] = st["state"]
ev["pause_message"] = "job paused" in text
# emission has to be gone, and the window has to still be open: a
# pause that relocked would need a fresh arm press to go on.
paused = []
t1 = time.time()
while time.time() - t1 < 4:
@@ -657,12 +543,15 @@ def pause_resume_live(ctx):
ilk = hw.sysfs_int("cnc/interlock_circuit")
ev["latch_locked_while_paused"] = ilk is not None and bool(ilk & (1 << 3))
ctx.log("paused: %s, emission 0 = %s, armed = %s, latch locked = %s", ev["hold_state"],
ev["emission_zero_when_paused"], ev["armed_while_paused"], ev["latch_locked_while_paused"])
ev["emission_zero_when_paused"], ev["armed_while_paused"],
ev["latch_locked_while_paused"])
ctx.check(ev["emission_zero_when_paused"], "emission did not stop while the job was paused: %s", paused)
ctx.check(ev["armed_while_paused"], "the armed window closed on the pause (a pause is not a cancel)")
ctx.check(not ev["latch_locked_while_paused"],
"the kernel latch relocked on the pause - the resume could not fire without a new arm press")
ctx.instruct("Press the button once more now (resume), then click Done and let the job finish.")
# -- the button resumes -----------------------------------------------
ctx.instruct("Press the button once more now (resume), then click Done.")
st = wait_state(ctx, g, "Run", 10)
ev["resumed_state"] = st["state"] if st else g.status_report()["state"]
ctx.check(st is not None, "the second press did not resume the job (state %s)", ev["resumed_state"])
@@ -682,22 +571,64 @@ def pause_resume_live(ctx):
ev["resume_trail"] = trail[-6:]
ctx.log("emission after the resume: %s", trail[-6:])
ctx.check(back, "emission did not return after the resume: %s", trail[-6:])
# the job runs to its end on its own
peak, states, st = wait_idle(ctx, g, 60)
ev["states_after_resume"] = states
ctx.check("TIMEOUT" not in states, "the resumed job did not finish: %s", states)
moved = st["MPos"][0] - start[0]
ev["moved_mm"] = round(moved, 3)
ctx.log("finished: moved %.3f mm of 60 (states %s)", moved, states)
ctx.check(abs(moved - 60.0) <= 0.1, "the resumed job did not finish its move (%.3f mm of 60)", moved)
ctx.check(ctx.forgectrl.wait_idle(15, abort=ctx.aborted), "machine not idle after the job")
# -- the lid cancels --------------------------------------------------
ctx.instruct("The cut is running again. Open the lid NOW and leave it open, then click Done.")
t_lid = time.time()
lid_trail = []
text = ""
while time.time() - t_lid < 8:
s = sample(ctx)
if s:
lid_trail.append((round(time.time() - t_lid, 2), s["emission"], s["kstate"], s["armed"]))
text += g.drain()
time.sleep(0.12)
for t in lid_trail:
ctx.log(" post-lid %s", t)
ev["messages"] = [ln for ln in text.splitlines() if ln.startswith("[MSG:") or "help]" in ln
or ln.startswith("ALARM")]
ctx.log("controller: %s", ev["messages"])
zero_at = next((t for t, e, _, _ in lid_trail if e == 0), None)
tail_zero = all(e == 0 for _, e, _, _ in lid_trail[-16:])
ev.update({"lid_zero_at_s": zero_at, "lid_tail_zero": tail_zero})
ctx.check(zero_at is not None and zero_at < 3.0,
"emission did not stop after the lid opened (first 0 at %s)", zero_at)
ctx.check(tail_zero, "emission returned after the lid opened")
ctx.check("lid opened - job cancelled" in text, "the lid open was not reported as cancelling the job")
ctx.check("help]" in text, "no reset banner after the cancel")
ctx.check("ALARM" not in text, "an alarm was raised on the cancel (position should be kept)")
t3 = time.time()
returned = "returned to the job start" in text
while not returned and time.time() - t3 < 30:
ctx.checkpoint()
text += g.drain()
returned = "returned to the job start" in text
time.sleep(0.2)
ev["returned_message"] = returned
ctx.check(returned, "the head did not report returning to the job start")
st = g.status_report()
drift = max(abs(st["MPos"][i] - start[i]) for i in range(2))
ev["drift_mm"] = round(drift, 3)
s = sample(ctx)
ev["armed_after_job"] = s["armed"] if s else None
ctx.confirm("Did the burn stop on the first press and start again on the second, continuing the "
"same line to its end? (A small mark where it restarted is expected - GRBL mode does "
"not backtrack.)")
g.command("$J=G91X-60F1200")
wait_idle(ctx, g, 30)
g.command("G90")
ctx.log("PASS: button paused the burn (emission 0, armed kept, latch unlocked), the next press resumed "
"it and the job finished (%.3f mm)", ev["moved_mm"])
ev["armed_after"] = s["armed"] if s else None
ilk = hw.sysfs_int("cnc/interlock_circuit")
ev["latch_locked"] = ilk is not None and bool(ilk & (1 << 3))
ev["button_latch"] = hw.sysfs_int("cnc/button_latch")
ctx.log("returned: drift %.3f mm; armed=%s latch_locked=%s button_latch=%s", drift,
ev["armed_after"], ev["latch_locked"], ev["button_latch"])
ctx.check(drift <= 0.05, "head not back at the job start (drift %.3f mm)", drift)
ctx.check(ctx.forgectrl.wait_idle(10, abort=ctx.aborted), "machine not idle after the return")
check_kernel_returned(ctx, ev, k0)
ctx.check(not ev["armed_after"], "armed window still open after the cancel")
ctx.check(ev["latch_locked"], "kernel latch not locked after the cancel")
ctx.check(ev["button_latch"] == 1, "hardware button latch not SET after the lid open (%s)",
ev["button_latch"])
ctx.confirm("Did the burn stop on the first press, start again on the second, then stop the "
"instant the lid opened - and did the head go straight back to where the job "
"started, with the lid still open and dark? (A small mark where the cut "
"restarted is expected - GRBL mode does not backtrack.)")
ctx.instruct("Close the lid, then click Done.")
ctx.sleep(1)
ctx.log("PASS: button paused the burn (emission 0, armed kept, latch unlocked) and resumed it; "
"the lid then cancelled it - emission 0 at +%s s, reset without alarm, returned (drift "
"%.3f mm), button latch SET", zero_at, drift)
forgetest/tests/test_cloud_suite.py
+88 -84
View File
@@ -338,20 +338,99 @@ class CloudSuiteTests(unittest.TestCase):
finally:
cloud.wait_log = saved
# -- the two lid tests, on their bench excerpts --------------------------------
def test_lid_abort_on_the_bench_excerpt(self):
# -- the lid/interlock abort and the button-wait tests, on their excerpts ------
# -- the merged lid + interlock abort test -------------------------------
def abort_parts(self):
"""The lid-abort excerpt split for the merged test: the print
prologue (replayed for both prints), the lid stop + park + cancel,
and the same tail with the interlock as the trigger, cut where the
test stops to have the lid opened during the park."""
lines = fixture("lidabort")
pre, rest = cut(lines, "waiting for button")
run_pre, rest = cut(rest, "machine:_run_loop starting run")
pre, rest = pre + run_pre + [rest[0]], rest[1:]
ilk = [l.replace("lid opened mid-run; stopping motion",
"interlock opened mid-run; stopping motion") for l in rest]
stop, tail = cut(ilk, "start return home")
return pre, rest, stop + [tail[0]], tail[1:]
def abort_hooks(self, pre, lid_tail, ilk_stop, ilk_park, prints):
def next_print():
prints.append(1)
self.append(pre, delay=0.1)
def pull_interlock():
self.fc.state["status"]["switches"]["interlock_ok"] = False
self.append(ilk_stop, delay=0.05)
def restore():
self.lid(True)
self.fc.state["status"]["switches"]["interlock_ok"] = True
return {"Click Done here": next_print,
"Open the lid NOW": lambda: (self.lid(False), self.append(lid_tail, delay=0.05)),
"Close the lid, then click Done": lambda: self.lid(True),
"Open the INTERLOCK loop now": pull_interlock,
"Open the LID now as well": lambda: (self.lid(False),
self.append(ilk_park, delay=0.05)),
"Close the lid and restore the interlock": restore}
def test_lid_interlock_abort_on_the_bench_excerpt(self):
self.in_cloud(pid=1522)
self.append(["2026-08-17T09:41:00.500000+00:00 gfcloud[1522] INFO websocket:_on_open RX-EVENT: ready"])
hooks = self.replay_print("lidabort", "Open the lid NOW", "start cool down")
run = self.run_test(cloud.lid_abort, hooks=hooks, test_id="cloud.lid-abort")
prints = []
run = self.run_test(cloud.lid_interlock_abort,
hooks=self.abort_hooks(*self.abort_parts(), prints),
test_id="cloud.lid-interlock-abort")
ev = run.evidence
self.assertEqual(len(prints), 2) # two prints, one cue each
# print 1: the lid
self.assertLess(ev["edge_to_stop_ms"], 60)
self.assertIn(":cancelled", ev["log"]["print finished"])
self.assertEqual(ev["kernel_counters_after_park"], [0, 0, 3])
self.assertTrue(ev["latch_locked"])
self.assertFalse(ev["armed_after"])
self.assertIn(":cancelled", ev["lid_log"]["print finished"])
self.assertEqual(ev["lid_counters_after_park"], [0, 0, 3])
self.assertTrue(ev["lid_latch_locked"])
self.assertFalse(ev["lid_armed_after"])
# print 2: the interlock, with the lid opened during the park
self.assertFalse(ev["interlock_ok_after_pull"])
self.assertIn(":cancelled", ev["interlock_log"]["print finished"])
self.assertTrue(ev["interlock_log"]["return home complete"])
self.assertEqual(ev["switches_at_return"], {"lid": False, "interlock_ok": False})
self.assertEqual(ev["interlock_counters_after_park"], [0, 0, 3])
self.assertTrue(ev["interlock_latch_locked"])
self.assertEqual(ev["restored"], {"lid": True, "interlock_ok": True})
self.assertEqual(self.fc.posts, [])
self.assertTrue(any("PASS: lid open" in l for l in run.lines))
self.assertTrue(any("PASS: lid open" in l for l in run.lines), run.lines)
def test_lid_interlock_abort_refuses_when_the_loop_is_already_open(self):
self.in_cloud(pid=1522)
self.fc.state["status"]["switches"]["interlock_ok"] = False
self.assertFails(cloud.lid_interlock_abort, "already reads open")
def test_lid_interlock_abort_fails_when_the_park_stops_at_the_lid(self):
# the regression this guards: a park an open lid can interrupt
self.in_cloud(pid=1522)
self.append(["2026-08-17T09:41:00.500000+00:00 gfcloud[1522] INFO websocket:_on_open RX-EVENT: ready"])
pre, lid_tail, ilk_stop, ilk_park = self.abort_parts()
ilk_park = [l for l in ilk_park if "return home complete" not in l]
saved = cloud.wait_log
def fast_wait_log(ctx, offset, needles, timeout, poll=0.5):
return saved(ctx, offset, needles, min(timeout, 1.5), poll=0.1)
cloud.wait_log = fast_wait_log
try:
self.assertFails(cloud.lid_interlock_abort, "did not run to completion",
hooks=self.abort_hooks(pre, lid_tail, ilk_stop, ilk_park, []))
finally:
cloud.wait_log = saved
def test_lid_interlock_abort_fails_when_the_lid_stop_is_not_edge_driven(self):
# a polled stop (the pre-parity behavior) shows up as a long edge->stop gap
self.in_cloud(pid=1522)
self.append(["2026-08-17T09:41:00.500000+00:00 gfcloud[1522] INFO websocket:_on_open RX-EVENT: ready"])
pre, lid_tail, ilk_stop, ilk_park = self.abort_parts()
lid_tail = [l.replace("2026-08-17T09:42:30.838627", "2026-08-17T09:42:31.838627")
if "lid opened mid-run; stopping motion" in l else l for l in lid_tail]
self.assertFails(cloud.lid_interlock_abort, "not edge-driven",
hooks=self.abort_hooks(pre, lid_tail, ilk_stop, ilk_park, []))
def test_lid_during_button_wait_on_the_bench_excerpt(self):
self.in_cloud(pid=1927)
@@ -369,81 +448,6 @@ class CloudSuiteTests(unittest.TestCase):
self.assertEqual(self.fc.posts, [])
self.assertTrue(any("PASS: lid open at the button prompt" in l for l in run.lines))
# -- interlock abort, and the park through an open lid --------------------
def interlock_parts(self):
"""The lid-abort excerpt with the interlock as the trigger, split at
the three operator steps: up to the print's run, the stop through
'start return home', and the park itself."""
lines = [l.replace("lid opened mid-run; stopping motion",
"interlock opened mid-run; stopping motion")
for l in fixture("lidabort")]
pre, rest = cut(lines, "waiting for button")
run_pre, rest = cut(rest, "machine:_run_loop starting run")
pre, rest = pre + run_pre + [rest[0]], rest[1:]
stop, tail = cut(rest, "start return home")
return pre, stop + [tail[0]], tail[1:]
def test_interlock_abort_park_on_the_bench_excerpt(self):
self.in_cloud(pid=1522)
self.append(["2026-08-17T09:41:00.500000+00:00 gfcloud[1522] INFO websocket:_on_open RX-EVENT: ready"])
pre, stop, park = self.interlock_parts()
def pull_interlock():
self.fc.state["status"]["switches"]["interlock_ok"] = False
self.append(stop, delay=0.05)
def open_lid():
self.lid(False)
self.append(park, delay=0.05)
def restore():
self.lid(True)
self.fc.state["status"]["switches"]["interlock_ok"] = True
run = self.run_test(cloud.interlock_abort_park,
hooks={"Click Done here": lambda: self.append(pre, delay=0.1),
"Open the INTERLOCK loop now": pull_interlock,
"Open the LID now as well": open_lid,
"Close the lid and restore the interlock": restore},
test_id="cloud.interlock-abort-park")
ev = run.evidence
self.assertFalse(ev["interlock_ok_after_pull"])
self.assertIn(":cancelled", ev["log"]["print finished"])
self.assertEqual(ev["switches_at_return"], {"lid": False, "interlock_ok": False})
self.assertEqual(ev["kernel_counters_after_park"], [0, 0, 3])
self.assertTrue(ev["latch_locked"])
self.assertFalse(ev["armed_after"])
self.assertEqual(ev["restored"], {"lid": True, "interlock_ok": True})
self.assertEqual(self.fc.posts, [])
self.assertTrue(any("PASS: interlock open" in l for l in run.lines), run.lines)
def test_interlock_abort_refuses_when_the_loop_is_already_open(self):
self.in_cloud(pid=1522)
self.fc.state["status"]["switches"]["interlock_ok"] = False
self.assertFails(cloud.interlock_abort_park, "already reads open")
def test_interlock_abort_fails_when_the_park_stops_at_the_lid(self):
# the regression this guards: a park that the lid can interrupt
self.in_cloud(pid=1522)
self.append(["2026-08-17T09:41:00.500000+00:00 gfcloud[1522] INFO websocket:_on_open RX-EVENT: ready"])
pre, stop, park = self.interlock_parts()
park = [l for l in park if "return home complete" not in l]
saved = cloud.wait_log
def fast_wait_log(ctx, offset, needles, timeout, poll=0.5):
return saved(ctx, offset, needles, min(timeout, 1.5), poll=0.1)
cloud.wait_log = fast_wait_log
try:
self.assertFails(
cloud.interlock_abort_park, "did not run to completion",
hooks={"Click Done here": lambda: self.append(pre, delay=0.1),
"Open the INTERLOCK loop now": lambda: (
self.fc.state["status"]["switches"].__setitem__("interlock_ok", False),
self.append(stop, delay=0.05)),
"Open the LID now as well": lambda: (self.lid(False),
self.append(park, delay=0.05))})
finally:
cloud.wait_log = saved
# -- a paused print cancelled by the lid, a running one by the app --------
def cancel_parts(self):
"""(print prologue, the pause lines, the lid stop + park + cancel,