mirror of
https://github.com/openglow-org/forgefirm.git
synced 2026-09-27 08:41:13 -07:00
ScottW514
0870a835c4
Catalog 34 -> 39. Every remaining bench drill of the lid/button parity work
is now a test, with drills that exercise the same path combined:
motion.lid-cancel-home also cancels from a hold - a job paused on the
button is ended by the lid, never resumed - so the
armed window and the hardware button latch stay in
agreement by construction.
motion.cancel-abort also asserts what a sender abort must NOT do: it
stops where it stopped and never returns home. The
return-to-start belongs to the lid policy alone.
motion.interlock-cancel-park (new) the interlock loop cancels like the lid,
and the lid opened during the return home does not
interrupt it.
motion.lid-policy-hold (new) the other policy: Door park, no cancel, no
return, and a cycle start finishes the move. The
setting is restored on the way out.
laser.pause-resume-live (new) the button pause/resume during a live cut:
emission stops, the latch stays UNLOCKED and the
armed window open (a pause is not a cancel), the
next press resumes and the job finishes.
cloud.interlock-abort-park (new) the same interlock/park pair in cloud mode.
cloud.pause-cancel-paths (new) the two non-finishing ends of a print, each
from the state the factory ends it in: paused on the
button then cancelled by the lid, and cancelled from
the app while running.
The shared cancel tail (reason reported, reset without an alarm, position
kept, head back at the job start with the KERNEL counters confirming it) is
now one helper, so every trigger is judged the same way.
scripts/bench/resume_dark_lead.py: samples LASER_ON, FIRE, HV_ENABLE, the
charge-pump watchdog, the button and the doors off the SoC pads at ~2 kHz
through /dev/mem, with motion dated from the kernel step counters, across a
pause and a resume. Levels are taken at idle and everything after is reported
as a change from that baseline, so no polarity assumption is baked in. Dry by
default, with --auto driving the pause and resume through ! / ~ for an
unattended rehearsal; --run live adds the dark lead between FIRE and LASER_ON,
in milliseconds and in millimeters at the job's feed.
Bench registry: argument specs can name a flag (--feed 600) instead of being
positional, and an optional argument with an empty default is left off the
command line entirely.
Host proof: 104 unit tests (20 in the cloud suite - the two new cloud tests
replay the machine's own lid-abort excerpt, with the interlock and the app
cancel substituted for the trigger, and fail for the right reasons), coverage
lint 0 uncovered across 39 tests.
521 lines
26 KiB
Python
521 lines
26 KiB
Python
"""The cloud.* job-behavior tests against a fake forgectrl and a replayed
|
|
gfcloud log: the real test functions run under the real runner Context,
|
|
the operator prompts answered by a script that also drives the replay
|
|
(open the lid -> the machine's lid reads open; close it -> the service's
|
|
re-hunt lines land). The log lines are the machine's own: bench excerpts
|
|
(the lid-abort, button-wait, and lid-open-hunt runs) and, for the pause,
|
|
the run loop's lines as its host test emits them on the print skeleton
|
|
of the lid-abort excerpt.
|
|
|
|
What is proven here: the tests find the right lines in real log noise,
|
|
reuse an existing cloud session and never switch back, restart the client
|
|
for a fresh hunt, judge the print's own finish line (not another action's),
|
|
wait the service's deferred moves out, and fail for the right reasons."""
|
|
import os
|
|
import shutil
|
|
import struct
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import unittest
|
|
|
|
import helpers
|
|
from forgetest.runner import Context, Failed, Run
|
|
from forgetest.suite import cloud
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
FIX = os.path.join(HERE, "fixtures")
|
|
|
|
|
|
def fixture(name):
|
|
with open(os.path.join(FIX, "gfcloud-%s.log" % name), "rb") as f:
|
|
return f.read().decode().splitlines()
|
|
|
|
|
|
def cut(lines, marker, count=1):
|
|
"""(before, after) at the count-th line containing marker (the line
|
|
itself opens `after`)."""
|
|
n = 0
|
|
for i, ln in enumerate(lines):
|
|
if marker in ln:
|
|
n += 1
|
|
if n == count:
|
|
return lines[:i], lines[i:]
|
|
raise KeyError(marker)
|
|
|
|
|
|
class Script:
|
|
"""Answers every prompt with its first option; a hook per prompt
|
|
substring drives the fake machine and the log replay."""
|
|
|
|
def __init__(self, run, hooks=None):
|
|
self.run = run
|
|
self.hooks = hooks or {}
|
|
self.asked = []
|
|
self.th = threading.Thread(target=self._loop, daemon=True)
|
|
self.stop = False
|
|
|
|
def start(self):
|
|
self.th.start()
|
|
return self
|
|
|
|
def _loop(self):
|
|
seen = None
|
|
while not self.stop:
|
|
p = self.run.prompt
|
|
if p and p["id"] != seen:
|
|
seen = p["id"]
|
|
self.asked.append(p["question"])
|
|
for key, fn in self.hooks.items():
|
|
if key in p["question"]:
|
|
fn()
|
|
self.run.answer(p["id"], p["options"][0])
|
|
time.sleep(0.02)
|
|
|
|
|
|
class CloudSuiteTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.mkdtemp(prefix="forgetest-cloud-")
|
|
self.log = os.path.join(self.tmp, "gfcloud.log")
|
|
open(self.log, "wb").close()
|
|
self.sysfs = os.path.join(self.tmp, "sysfs") + os.sep
|
|
os.makedirs(self.sysfs + "cnc")
|
|
self._attr("cnc/interlock_circuit", "45") # latch locked (bit 3)
|
|
self._pos(0, 0, 3)
|
|
os.environ["GF_SYSFS_ROOT"] = self.sysfs
|
|
self.fc = helpers.FakeForgectrl().start()
|
|
self.saved = (cloud.GFCLOUD_LOG, cloud.QUIET_S, cloud.QUIET_TIMEOUT_S, cloud.HUNT_TIMEOUT_S)
|
|
cloud.GFCLOUD_LOG = self.log
|
|
cloud.QUIET_S = 0.4
|
|
cloud.QUIET_TIMEOUT_S = 3
|
|
cloud.HUNT_TIMEOUT_S = 8
|
|
self.script = None
|
|
|
|
def tearDown(self):
|
|
if self.script:
|
|
self.script.stop = True
|
|
self.fc.stop()
|
|
cloud.GFCLOUD_LOG, cloud.QUIET_S, cloud.QUIET_TIMEOUT_S, cloud.HUNT_TIMEOUT_S = self.saved
|
|
os.environ.pop("GF_SYSFS_ROOT", None)
|
|
shutil.rmtree(self.tmp, ignore_errors=True)
|
|
|
|
# -- fakes -----------------------------------------------------------
|
|
def _attr(self, attr, val):
|
|
with open(self.sysfs + attr, "w") as f:
|
|
f.write(val)
|
|
|
|
def _pos(self, x, y, z):
|
|
with open(self.sysfs + "cnc/position", "wb") as f:
|
|
f.write(struct.pack("<3i2I", x, y, z, 0, 0))
|
|
|
|
def append(self, lines, delay=0.0):
|
|
def go():
|
|
if delay:
|
|
time.sleep(delay)
|
|
with open(self.log, "ab") as f:
|
|
f.write(("\n".join(lines) + "\n").encode())
|
|
if delay:
|
|
threading.Thread(target=go, daemon=True).start()
|
|
else:
|
|
go()
|
|
|
|
def in_cloud(self, pid=2278):
|
|
self.fc.state["mode"] = {"mode": "cloud", "controller": "running", "pid": pid, "motion": "verified"}
|
|
self.fc.state["settings"]["controller_mode"] = "cloud"
|
|
|
|
def lid(self, closed):
|
|
self.fc.state["status"]["switches"]["lid"] = bool(closed)
|
|
|
|
def run_test(self, fn, hooks=None, test_id="cloud.x"):
|
|
run = Run("test", test_id, test_id)
|
|
run.baseline_captured = {"mode": self.fc.state["mode"]["mode"], "position": [0, 0, 0]}
|
|
ctx = Context(run, None, helpers.make_test(test_id, []))
|
|
self.script = Script(run, hooks).start()
|
|
try:
|
|
fn(ctx)
|
|
finally:
|
|
self.script.stop = True
|
|
return run
|
|
|
|
def assertFails(self, fn, needle, hooks=None):
|
|
with self.assertRaises(Failed) as cm:
|
|
self.run_test(fn, hooks)
|
|
self.assertIn(needle, str(cm.exception))
|
|
return cm.exception
|
|
|
|
# -- session detection -------------------------------------------------
|
|
def test_session_live_is_per_client(self):
|
|
lines = fixture("huntlid")
|
|
# the old client (1927) closed and left; the new one (2278) is ready
|
|
self.append(lines)
|
|
self.assertTrue(cloud.session_live(2278)[0])
|
|
self.assertFalse(cloud.session_live(1927)[0])
|
|
# an unknown pid falls back to the newest lines (ready)
|
|
self.assertTrue(cloud.session_live(9999)[0])
|
|
# a drop after the ready line: not live until it is ready again
|
|
self.append(["2026-08-17T10:00:00.000000+00:00 gfcloud[2278] INFO websocket:_on_close RX-EVENT: closed (1006, )",
|
|
"2026-08-17T10:00:00.100000+00:00 gfcloud[2278] INFO websocket:run RECONNECTING"])
|
|
live, detail = cloud.session_live(2278)
|
|
self.assertFalse(live)
|
|
self.assertIn("RECONNECTING", detail)
|
|
self.append(["2026-08-17T10:00:06.000000+00:00 gfcloud[2278] INFO websocket:_on_open RX-EVENT: ready"])
|
|
self.assertTrue(cloud.session_live(2278)[0])
|
|
|
|
def test_enter_cloud_reuses_a_live_session_and_never_switches(self):
|
|
self.in_cloud()
|
|
self.append(fixture("huntlid"))
|
|
run = self.run_test(lambda ctx: cloud.enter_cloud(ctx))
|
|
self.assertTrue(any("reusing it" in l for l in run.lines), run.lines)
|
|
self.assertEqual(self.fc.posts, [])
|
|
self.assertEqual(run.baseline_captured["mode"], "cloud")
|
|
|
|
def test_enter_cloud_refuses_a_dead_session(self):
|
|
self.in_cloud(pid=1927) # the client that shut down in the excerpt
|
|
self.append(fixture("huntlid"))
|
|
self.assertFails(cloud.enter_cloud, "no live service session")
|
|
|
|
def test_enter_cloud_switches_once_from_grbl_and_declares_it(self):
|
|
lines = fixture("huntlid")
|
|
pre, post = cut(lines, "gfuiservice:__init__ INITIALIZED")
|
|
self.append(pre)
|
|
|
|
def on_post(path, form):
|
|
if path == "/mode":
|
|
self.append(post, delay=0.2) # the new client's session + hunt + moves
|
|
return None
|
|
self.fc.on_post = on_post
|
|
run = self.run_test(lambda ctx: cloud.enter_cloud(ctx))
|
|
self.assertEqual([p for p, _ in self.fc.posts], ["/mode"])
|
|
self.assertEqual(self.fc.state["mode"]["mode"], "cloud")
|
|
self.assertEqual(run.baseline_captured["mode"], "cloud") # declared: the baseline keeps it
|
|
self.assertTrue(any("connect-time hunt" in l and ":completed" in l for l in run.lines), run.lines)
|
|
|
|
def test_enter_cloud_waits_for_the_service_to_stop_moving(self):
|
|
self.in_cloud()
|
|
self.append(fixture("huntlid"))
|
|
# a motion in flight (never idle within the timeout) fails, and says so
|
|
self.fc.state["status"]["state"] = "running"
|
|
self.assertFails(cloud.enter_cloud, "still running service moves")
|
|
|
|
# -- the hunt with the lid open ------------------------------------------
|
|
def test_hunt_lid_open_restarts_the_client_in_cloud_mode(self):
|
|
self.in_cloud(pid=1927)
|
|
lines = fixture("huntlid")
|
|
pre, post = cut(lines, "gfuiservice:__init__ INITIALIZED")
|
|
hunt_part, close_part = cut(post, "_switch_event lid closed")
|
|
self.append(pre)
|
|
|
|
def on_post(path, form):
|
|
if path == "/controller/stop":
|
|
self.fc.state["mode"] = dict(self.fc.state["mode"], controller="standby", pid=0)
|
|
elif path == "/controller/start":
|
|
self.fc.state["mode"] = dict(self.fc.state["mode"], controller="running", pid=2278)
|
|
self.append(hunt_part, delay=0.2)
|
|
return None
|
|
self.fc.on_post = on_post
|
|
|
|
def lid_closed():
|
|
self.lid(True)
|
|
self.append(close_part, delay=0.2)
|
|
run = self.run_test(cloud.hunt_lid_open,
|
|
hooks={"Open the lid": lambda: self.lid(False), "Close the lid": lid_closed},
|
|
test_id="cloud.hunt-lid-open")
|
|
self.assertEqual([p for p, _ in self.fc.posts], ["/controller/stop", "/controller/start"])
|
|
ev = run.evidence
|
|
self.assertIn(":completed", ev["hunt_line"])
|
|
self.assertEqual(ev["refusals_before_hunt_end"], 0)
|
|
self.assertEqual(ev["motions_after_lid_close"], 3)
|
|
self.assertEqual(self.fc.state["mode"]["mode"], "cloud") # still cloud
|
|
self.assertTrue(any("PASS:" in l for l in run.lines))
|
|
# the prompts, in order: open, confirm lens, close
|
|
self.assertEqual(len(self.script.asked), 3)
|
|
|
|
def test_hunt_lid_open_from_grbl_switches_and_stays(self):
|
|
lines = fixture("huntlid")
|
|
pre, post = cut(lines, "gfuiservice:__init__ INITIALIZED")
|
|
hunt_part, close_part = cut(post, "_switch_event lid closed")
|
|
self.append(pre)
|
|
|
|
def on_post(path, form):
|
|
if path == "/mode":
|
|
self.append(hunt_part, delay=0.2)
|
|
return None
|
|
self.fc.on_post = on_post
|
|
run = self.run_test(cloud.hunt_lid_open,
|
|
hooks={"Open the lid": lambda: self.lid(False),
|
|
"Close the lid": lambda: (self.lid(True), self.append(close_part, delay=0.2))})
|
|
self.assertEqual([p for p, _ in self.fc.posts], ["/mode"])
|
|
self.assertEqual(self.fc.state["mode"]["mode"], "cloud")
|
|
self.assertEqual(run.baseline_captured["mode"], "cloud")
|
|
self.assertIn(":completed", run.evidence["hunt_line"])
|
|
|
|
def test_hunt_lid_open_needs_the_lid_open(self):
|
|
self.in_cloud()
|
|
self.append(fixture("huntlid"))
|
|
self.assertFails(cloud.hunt_lid_open, "lid reads closed")
|
|
|
|
def test_hunt_refused_for_the_lid_fails(self):
|
|
self.in_cloud(pid=1927)
|
|
lines = fixture("huntlid")
|
|
pre, post = cut(lines, "gfuiservice:__init__ INITIALIZED")
|
|
# a refusal ahead of the hunt's end (the lid gating the hunt would look like this)
|
|
i = next(i for i, l in enumerate(post) if "z_axis:home starting z homing cycle" in l)
|
|
post = post[:i] + ["2026-08-17T09:45:10.595000+00:00 gfcloud[2278] INFO machine:_safe_to_move lid opened, unsafe to move"] + post[i:]
|
|
self.append(pre)
|
|
|
|
def on_post(path, form):
|
|
if path == "/controller/stop":
|
|
self.fc.state["mode"] = dict(self.fc.state["mode"], controller="standby", pid=0)
|
|
elif path == "/controller/start":
|
|
self.fc.state["mode"] = dict(self.fc.state["mode"], controller="running", pid=2278)
|
|
self.append(post, delay=0.1)
|
|
return None
|
|
self.fc.on_post = on_post
|
|
self.assertFails(cloud.hunt_lid_open, "refused for the lid", hooks={"Open the lid": lambda: self.lid(False)})
|
|
|
|
# -- pause / resume ---------------------------------------------------------
|
|
def replay_print(self, name, at_run, at_end, tail_delay=0.3):
|
|
"""Split a print excerpt into: up to the run's RUNNING line
|
|
(lands when the operator says Print is done), the block landing at
|
|
the mid-run prompt, and the rest."""
|
|
lines = fixture(name)
|
|
pre, rest = cut(lines, "waiting for button")
|
|
run_pre, rest = cut(rest, "current state: MachineState.RUNNING") # the PRINT's run
|
|
pre, rest = pre + run_pre + [rest[0]], rest[1:]
|
|
mid, tail = cut(rest, at_end)
|
|
return {"Click Done here": lambda: self.append(pre, delay=0.1),
|
|
at_run: lambda: (self.append(mid, delay=0.05), self.append(tail, delay=tail_delay))}
|
|
|
|
def test_pause_resume_passes_on_the_machines_lines(self):
|
|
self.in_cloud(pid=1522)
|
|
self.append(["2026-08-17T09:41:00.000000+00:00 gfcloud[1522] INFO authentication:authenticate_machine SUCCESS",
|
|
"2026-08-17T09:41:00.500000+00:00 gfcloud[1522] INFO websocket:_on_open RX-EVENT: ready",
|
|
"2026-08-17T09:41:00.900000+00:00 gfcloud[1522] INFO websocket:ws_connect ESTABLISHED"])
|
|
hooks = self.replay_print("pause", "Press the button once NOW", "current state: MachineState.IDLE")
|
|
self.fc.state["cool"]["armed"] = True
|
|
run = self.run_test(cloud.pause_resume, hooks=hooks, test_id="cloud.pause-resume")
|
|
ev = run.evidence
|
|
self.assertEqual(ev["log"], {"button pressed mid-run; pausing": True, "paused at": True,
|
|
"button pressed while paused; resuming": True})
|
|
self.assertTrue(ev["armed_after_resume"])
|
|
self.assertIn(":completed", ev["log_end"]["print finished"])
|
|
self.assertTrue(ev["log_end"]["return home complete"])
|
|
self.assertEqual(ev["relock_or_cancel_lines"], 0)
|
|
self.assertEqual(self.fc.posts, [])
|
|
self.assertTrue(any("PASS: button pause/resume" in l for l in run.lines))
|
|
# the post-print hunt was waited out
|
|
self.assertTrue(any("machine is quiet" in l for l in run.lines))
|
|
|
|
def test_pause_resume_fails_when_the_print_is_cancelled_instead(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"])
|
|
lines = fixture("pause")
|
|
lines = [l.replace('print [1576550507]: finished with event ":completed"',
|
|
'print [1576550507]: finished with event ":cancelled"') for l in lines]
|
|
pre, rest = cut(lines, "current state: MachineState.RUNNING")
|
|
pre, rest = pre + [rest[0]], rest[1:]
|
|
hooks = {"Click Done here": lambda: self.append(pre, delay=0.1),
|
|
"Press the button once NOW": lambda: self.append(rest, delay=0.05)}
|
|
self.assertFails(cloud.pause_resume, "did not complete after the resume", hooks=hooks)
|
|
|
|
def test_pause_resume_fails_without_the_pause_line(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"])
|
|
lines = [l for l in fixture("pause") if "button pressed mid-run" not in l]
|
|
pre, rest = cut(lines, "current state: MachineState.RUNNING")
|
|
pre, rest = pre + [rest[0]], rest[1:]
|
|
# the wait for the pause lines is 90 s: shorten it through the module's poll by
|
|
# ending the log early - the finish line arrives, but the pause never does
|
|
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:
|
|
hooks = {"Click Done here": lambda: self.append(pre, delay=0.1),
|
|
"Press the button once NOW": lambda: self.append(rest, delay=0.05)}
|
|
self.assertFails(cloud.pause_resume, "did not pause the run", hooks=hooks)
|
|
finally:
|
|
cloud.wait_log = saved
|
|
|
|
# -- the two lid tests, on their bench excerpts --------------------------------
|
|
def test_lid_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")
|
|
ev = run.evidence
|
|
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.assertEqual(self.fc.posts, [])
|
|
self.assertTrue(any("PASS: lid open" in l for l in run.lines))
|
|
|
|
def test_lid_during_button_wait_on_the_bench_excerpt(self):
|
|
self.in_cloud(pid=1927)
|
|
self.append(["2026-08-17T09:41:00.500000+00:00 gfcloud[1927] INFO websocket:_on_open RX-EVENT: ready"])
|
|
lines = fixture("buttonwait")
|
|
pre, rest = cut(lines, "waiting for button")
|
|
pre, rest = pre + [rest[0]], rest[1:]
|
|
hooks = {"Click Done here": lambda: self.append(pre, delay=0.1),
|
|
"Open the lid now": lambda: self.append(rest, delay=0.05)}
|
|
run = self.run_test(cloud.lid_during_button_wait, hooks=hooks, test_id="cloud.lid-during-button-wait")
|
|
ev = run.evidence
|
|
self.assertEqual(ev["runs_started_after_wait"], 0)
|
|
self.assertIn(":cancelled", ev["log"]["print finished"])
|
|
self.assertTrue(ev["latch_locked"])
|
|
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,
|
|
the same tail with the app's cancel as the trigger)."""
|
|
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:]
|
|
paused = ["2026-08-17T09:42:25.500000+00:00 gfcloud[1522] INFO machine:_run_loop "
|
|
"button pressed mid-run; pausing",
|
|
"2026-08-17T09:42:26.100000+00:00 gfcloud[1522] INFO machine:_run_loop "
|
|
"paused at Position(x=41.2, y=17.0, z=0.0)"]
|
|
app_cancel = [l.replace("lid opened mid-run; stopping motion",
|
|
"action cancelled mid-run; stopping motion") for l in rest]
|
|
return pre, paused, rest, app_cancel
|
|
|
|
def test_pause_cancel_paths_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, paused, lid_tail, app_tail = self.cancel_parts()
|
|
prints = []
|
|
|
|
def next_print():
|
|
prints.append(1)
|
|
self.append(pre, delay=0.1)
|
|
run = self.run_test(cloud.pause_cancel_paths,
|
|
hooks={"Click Done here": next_print,
|
|
"Press the button once NOW": lambda: self.append(paused, delay=0.05),
|
|
"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),
|
|
"Cancel the print from the app now": lambda: self.append(app_tail,
|
|
delay=0.05)},
|
|
test_id="cloud.pause-cancel-paths")
|
|
ev = run.evidence
|
|
self.assertEqual(len(prints), 2) # two prints, one cue each
|
|
self.assertEqual(ev["paused"], {"button pressed mid-run; pausing": True, "paused at": True})
|
|
self.assertIn(":cancelled", ev["lid_from_pause"]["print finished"])
|
|
self.assertIn(":cancelled", ev["service_cancel"]["print finished"])
|
|
self.assertTrue(ev["lid_from_pause"]["return home complete"])
|
|
self.assertTrue(ev["service_cancel"]["return home complete"])
|
|
self.assertEqual(ev["counters_after_print1"], [0, 0, 3])
|
|
self.assertEqual(ev["counters_after_print2"], [0, 0, 3])
|
|
self.assertTrue(ev["latch_locked_after_print1"] and ev["latch_locked_after_print2"])
|
|
self.assertFalse(ev["armed_after_print1"] or ev["armed_after_print2"])
|
|
self.assertEqual(self.fc.posts, [])
|
|
self.assertTrue(any("PASS: a paused print cancelled by the lid" in l for l in run.lines), run.lines)
|
|
|
|
def test_pause_cancel_paths_fails_when_the_paused_print_resumes_instead(self):
|
|
# a lid that resumed (or was ignored) leaves the print ':completed'
|
|
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, paused, lid_tail, _app = self.cancel_parts()
|
|
lid_tail = [l.replace(':cancelled"', ':completed"') for l in lid_tail]
|
|
self.assertFails(
|
|
cloud.pause_cancel_paths, "print 1 did not end ':cancelled'",
|
|
hooks={"Click Done here": lambda: self.append(pre, delay=0.1),
|
|
"Press the button once NOW": lambda: self.append(paused, delay=0.05),
|
|
"Open the lid NOW": lambda: (self.lid(False), self.append(lid_tail, delay=0.05))})
|
|
|
|
def test_print_finish_is_the_prints_not_another_actions(self):
|
|
# a motion that completes before the print must not satisfy the print's finish
|
|
lines = fixture("lidabort")
|
|
i = cloud.action_finish_index(lines, "print")
|
|
j = cloud.action_finish_index(lines, "motion")
|
|
self.assertIsNotNone(i)
|
|
self.assertIsNotNone(j)
|
|
self.assertLess(j, i)
|
|
self.assertIn(":cancelled", lines[i])
|
|
self.assertIn(":completed", lines[j])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|