mirror of
https://github.com/openglow-org/forgefirm.git
synced 2026-09-27 16:51:12 -07:00
ScottW514
9e9d03ba94
Four sessions against the reported messages: arm once per job with M5/M3 persistence and the M2 close (grace pushed beyond the horizon so only the program-end path can disarm); a displaced sender must re-arm; the disarm grace counts down in Hold; a blocking cooling verdict refuses the arm. Test-the-test proven: a build with the job-based window reverted fails at the first discriminating assertion. Wired into the controller repo CI alongside the emission harness.
285 lines
9.8 KiB
Python
285 lines
9.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Host-side verification of the operator-armed window lifecycle.
|
|
|
|
Runs the native grblHAL_glowforge binary in null-sink mode (no
|
|
hardware, no root; arming is automatic without the physical button)
|
|
and drives jobs over TCP, asserting the window's state machine on the
|
|
reported messages:
|
|
|
|
1. the first laser-on of a job arms exactly once, and the window
|
|
persists across M5/M3 toggles within the job (no re-prompt)
|
|
2. program end (M2) closes the window: the disarm lands promptly,
|
|
and the next job arms afresh
|
|
3. a sender change closes the window: a reconnected client must
|
|
re-arm - the displaced session's consent does not carry over
|
|
4. the spindle-off grace counts down in HOLD, not only in Idle - a
|
|
job abandoned in feed hold disarms after laser_disarm_s
|
|
5. arming is refused while the cooling verdict blocks fire, and no
|
|
armed message appears
|
|
|
|
The disarm grace is shortened via a temp config (GFHOME_CONF) and the
|
|
cooling verdict is published hermetically (GF_VERDICT_FILE), the same
|
|
overrides the emission harness uses.
|
|
|
|
Usage: laser_lifecycle_test.py [path-to-binary]
|
|
(default ./build-native/grblHAL_glowforge)
|
|
"""
|
|
import os
|
|
import re
|
|
import shutil
|
|
import signal
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
|
|
BIN = os.path.abspath(sys.argv[1] if len(sys.argv) > 1 else "build-native/grblHAL_glowforge")
|
|
PORT = 2398
|
|
|
|
ARMED = "laser armed"
|
|
DISARMED = "laser disarmed - latch locked"
|
|
BLOCKED = "laser fire blocked"
|
|
|
|
|
|
def fail(msg):
|
|
print("FAIL: %s" % msg)
|
|
sys.exit(1)
|
|
|
|
|
|
def read_avail(sock, log, timeout, until=None):
|
|
end = time.time() + timeout
|
|
buf = b""
|
|
while time.time() < end:
|
|
sock.settimeout(max(0.05, end - time.time()))
|
|
try:
|
|
data = sock.recv(4096)
|
|
except socket.timeout:
|
|
data = b""
|
|
except OSError:
|
|
return None
|
|
if data:
|
|
buf += data
|
|
log.append(data.decode(errors="replace"))
|
|
if until:
|
|
text = buf.decode(errors="replace")
|
|
for token in until:
|
|
if token in text or re.search(r"^%s\b" % re.escape(token), text, re.M):
|
|
return token
|
|
elif until is None:
|
|
return None
|
|
return None
|
|
|
|
|
|
def send_line(sock, line, log, tolerate_error=False):
|
|
sock.sendall((line + "\n").encode())
|
|
r = read_avail(sock, log, 5.0, until=("ok", "error"))
|
|
if r is None:
|
|
fail("no ok/error for %r" % line)
|
|
if r == "error" and not tolerate_error:
|
|
fail("error response to %r" % line)
|
|
return r
|
|
|
|
|
|
def wait_for(log, needle, timeout, sock=None):
|
|
"""Wait until needle appears in the accumulated log (draining the
|
|
socket while waiting when one is given)."""
|
|
end = time.time() + timeout
|
|
while time.time() < end:
|
|
if needle in "".join(log):
|
|
return True
|
|
if sock is not None:
|
|
read_avail(sock, log, 0.2)
|
|
else:
|
|
time.sleep(0.1)
|
|
return False
|
|
|
|
|
|
def wait_idle(sock, log):
|
|
for _ in range(100):
|
|
sock.sendall(b"?")
|
|
read_avail(sock, log, 0.3)
|
|
if re.search(r"<Idle", "".join(log[-3:])):
|
|
return
|
|
time.sleep(0.2)
|
|
fail("controller never returned to Idle")
|
|
|
|
|
|
def publish_verdicts(path, stop, fire_ok):
|
|
while not stop.is_set():
|
|
body = ('{"ts_mono":%.3f,"fire_ok":%s,"hold":false,'
|
|
'"resume_ok":true,"reason":""}'
|
|
% (time.clock_gettime(time.CLOCK_MONOTONIC),
|
|
"true" if fire_ok else "false"))
|
|
tmp = path + ".tmp"
|
|
with open(tmp, "w") as f:
|
|
f.write(body)
|
|
os.replace(tmp, path)
|
|
stop.wait(0.5)
|
|
|
|
|
|
class Session:
|
|
"""One controller process with the lifecycle overrides applied."""
|
|
|
|
def __init__(self, name, fire_ok=True, disarm_s=2):
|
|
self.name = name
|
|
self.workdir = tempfile.mkdtemp(prefix="laser-lifecycle-")
|
|
conf = os.path.join(self.workdir, "forgefirm.conf")
|
|
with open(conf, "w") as f:
|
|
f.write("laser_disarm_s = %d\n" % disarm_s)
|
|
verdict = os.path.join(self.workdir, "cooling.state")
|
|
env = dict(os.environ, GF_VERDICT_FILE=verdict, GFHOME_CONF=conf)
|
|
env.pop("GFSINK", None)
|
|
self.stop = threading.Event()
|
|
self.pub = threading.Thread(target=publish_verdicts,
|
|
args=(verdict, self.stop, fire_ok),
|
|
daemon=True)
|
|
self.pub.start()
|
|
self.proc = subprocess.Popen([BIN, "-p", str(PORT)],
|
|
cwd=self.workdir, env=env,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.PIPE)
|
|
self.log = []
|
|
self.sock = self.connect()
|
|
|
|
def connect(self):
|
|
for _ in range(50):
|
|
try:
|
|
s = socket.create_connection(("127.0.0.1", PORT), timeout=1)
|
|
read_avail(s, self.log, 0.5) # banner
|
|
return s
|
|
except OSError:
|
|
time.sleep(0.1)
|
|
fail("[%s] cannot connect to the controller" % self.name)
|
|
|
|
def armed_count(self):
|
|
return "".join(self.log).count(ARMED)
|
|
|
|
def disarmed_count(self):
|
|
return "".join(self.log).count(DISARMED)
|
|
|
|
def close(self):
|
|
try:
|
|
self.sock.close()
|
|
except OSError:
|
|
pass
|
|
self.proc.send_signal(signal.SIGINT)
|
|
try:
|
|
self.proc.wait(5)
|
|
except subprocess.TimeoutExpired:
|
|
self.proc.kill()
|
|
self.stop.set()
|
|
self.pub.join(2)
|
|
shutil.rmtree(self.workdir, ignore_errors=True)
|
|
|
|
|
|
def test_job_window():
|
|
"""Rules 1 + 2: arm once per job, persist across M5/M3, close at M2,
|
|
fresh consent for the next job. The grace is set far beyond the test
|
|
horizon so only the program-end close can produce the disarm."""
|
|
s = Session("job-window", disarm_s=60)
|
|
try:
|
|
send_line(s.sock, "M4 S100", s.log)
|
|
send_line(s.sock, "G1 X1 F600", s.log)
|
|
if not wait_for(s.log, ARMED, 5, s.sock):
|
|
fail("[job-window] first laser-on did not arm")
|
|
send_line(s.sock, "M5", s.log)
|
|
send_line(s.sock, "M3 S200", s.log)
|
|
send_line(s.sock, "G1 X0 F600", s.log)
|
|
wait_idle(s.sock, s.log)
|
|
if s.armed_count() != 1:
|
|
fail("[job-window] M5/M3 inside one job re-prompted the arm "
|
|
"(armed messages: %d)" % s.armed_count())
|
|
send_line(s.sock, "M2", s.log)
|
|
if not wait_for(s.log, DISARMED, 5, s.sock):
|
|
fail("[job-window] program end (M2) did not close the armed window")
|
|
send_line(s.sock, "M4 S100", s.log)
|
|
send_line(s.sock, "G1 X1 F600", s.log)
|
|
if not wait_for(s.log, ARMED, 5, s.sock) or s.armed_count() != 2:
|
|
fail("[job-window] next job after M2 did not arm afresh")
|
|
wait_idle(s.sock, s.log)
|
|
print("PASS [job-window]: armed once per job, M2 closed the window, "
|
|
"next job re-armed")
|
|
finally:
|
|
s.close()
|
|
|
|
|
|
def test_sender_change():
|
|
"""Rule 3: a reconnected sender must re-arm. The grace is set far
|
|
beyond the test horizon so only the sender change can close the
|
|
window, and the first job ends in M5 so the next M4 is a fresh
|
|
laser-on edge (the arm prompt fires on the off->on edge)."""
|
|
s = Session("sender-change", disarm_s=60)
|
|
try:
|
|
send_line(s.sock, "M4 S100", s.log)
|
|
send_line(s.sock, "G1 X1 F600", s.log)
|
|
if not wait_for(s.log, ARMED, 5, s.sock):
|
|
fail("[sender-change] first laser-on did not arm")
|
|
send_line(s.sock, "M5", s.log)
|
|
wait_idle(s.sock, s.log)
|
|
s.sock.close()
|
|
time.sleep(0.5)
|
|
s.sock = s.connect()
|
|
before = s.armed_count()
|
|
send_line(s.sock, "M4 S100", s.log)
|
|
send_line(s.sock, "G1 X0 F600", s.log)
|
|
end = time.time() + 5
|
|
while time.time() < end and s.armed_count() != before + 1:
|
|
read_avail(s.sock, s.log, 0.2)
|
|
if s.armed_count() != before + 1:
|
|
fail("[sender-change] the reconnected sender inherited the "
|
|
"displaced session's armed window")
|
|
wait_idle(s.sock, s.log)
|
|
print("PASS [sender-change]: displaced consent did not survive; "
|
|
"the new sender re-armed")
|
|
finally:
|
|
s.close()
|
|
|
|
|
|
def test_hold_grace():
|
|
"""Rule 4: the disarm grace counts down in HOLD."""
|
|
s = Session("hold-grace", disarm_s=2)
|
|
try:
|
|
send_line(s.sock, "M4 S100", s.log)
|
|
send_line(s.sock, "G1 X30 F60", s.log) # ~30 s move
|
|
if not wait_for(s.log, ARMED, 5, s.sock):
|
|
fail("[hold-grace] job did not arm")
|
|
time.sleep(1.0)
|
|
s.sock.sendall(b"!") # feed hold
|
|
if not wait_for(s.log, DISARMED, 10, s.sock):
|
|
fail("[hold-grace] a job abandoned in Hold stayed armed past "
|
|
"the disarm grace")
|
|
print("PASS [hold-grace]: the armed window closed while parked in Hold")
|
|
finally:
|
|
s.close()
|
|
|
|
|
|
def test_verdict_blocks_arm():
|
|
"""Rule 5: a blocking cooling verdict refuses the arm."""
|
|
s = Session("verdict-blocked", fire_ok=False)
|
|
try:
|
|
send_line(s.sock, "M4 S100", s.log, tolerate_error=True)
|
|
send_line(s.sock, "G1 X1 F600", s.log, tolerate_error=True)
|
|
if not wait_for(s.log, BLOCKED, 5, s.sock):
|
|
fail("[verdict-blocked] no refusal message for a blocking verdict")
|
|
if ARMED in "".join(s.log):
|
|
fail("[verdict-blocked] armed despite a blocking cooling verdict")
|
|
print("PASS [verdict-blocked]: arming refused while fire_ok=false")
|
|
finally:
|
|
s.close()
|
|
|
|
|
|
def main():
|
|
if not os.path.isfile(BIN):
|
|
fail("controller binary not found at %s" % BIN)
|
|
test_job_window()
|
|
test_sender_change()
|
|
test_hold_grace()
|
|
test_verdict_blocks_arm()
|
|
print("PASS: the armed-window lifecycle holds")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|