diff --git a/forgetest/forgetest/suite/cloud.py b/forgetest/forgetest/suite/cloud.py index 0bcd7fc..0671d93 100644 --- a/forgetest/forgetest/suite/cloud.py +++ b/forgetest/forgetest/suite/cloud.py @@ -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 " diff --git a/forgetest/forgetest/suite/kernel.py b/forgetest/forgetest/suite/kernel.py index cd9c60d..1c48628 100644 --- a/forgetest/forgetest/suite/kernel.py +++ b/forgetest/forgetest/suite/kernel.py @@ -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: diff --git a/forgetest/forgetest/suite/laser.py b/forgetest/forgetest/suite/laser.py index 2096dff..726b027 100644 --- a/forgetest/forgetest/suite/laser.py +++ b/forgetest/forgetest/suite/laser.py @@ -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) diff --git a/forgetest/tests/test_cloud_suite.py b/forgetest/tests/test_cloud_suite.py index 5c2e1a4..28db395 100644 --- a/forgetest/tests/test_cloud_suite.py +++ b/forgetest/tests/test_cloud_suite.py @@ -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,