diff --git a/forgetest/forgetest/bench.py b/forgetest/forgetest/bench.py index e4a113d..11ba279 100644 --- a/forgetest/forgetest/bench.py +++ b/forgetest/forgetest/bench.py @@ -273,6 +273,16 @@ TOOLS = [ "(a CR LF with a header behind it included) never reaches the wire, and the homing runner the " "controller starts does not inherit it. A CI harness (the grblHAL repo): needs the host-built " "null-sink controller, not the machine, so it is not a bench-page tool."}, + {"id": "mcode-test", "title": "Package M-code barrier harness", "script": "mcode_test.py", + "safety": "dry", "where": "host", "ported": False, "args": [], + "desc": "The M-codes extension packages answer (M160 to M179), on the null-sink controller with a " + "scripted sender and a stand-in for the machine daemon on the port: the table and its refusals, " + "a number nothing answers is error:20 where it is parsed, an answered one waits with the head " + "still and a port jog refused while the port's state names it, the answer's words, an answer " + "that the work was not done and no answer in time each hold the job, a soft reset ends the " + "wait, and a wait under an open armed window with M3 S500 ships no FIRE tick. Imports its " + "sender and port client from ctlport_test.py. A CI harness (the grblHAL repo): needs the " + "host-built null-sink controller, not the machine, so it is not a bench-page tool."}, {"id": "manual-home-test", "title": "Manual home and motor release harness", "script": "manual_home_test.py", "safety": "dry", "where": "host", "ported": False, "args": [], "desc": "The manual homing provider and the motor release on the null-sink controller: $H under manual " diff --git a/forgetest/forgetest/suite/__init__.py b/forgetest/forgetest/suite/__init__.py index 759b0e2..f08cfa9 100644 --- a/forgetest/forgetest/suite/__init__.py +++ b/forgetest/forgetest/suite/__init__.py @@ -27,3 +27,4 @@ from . import extdest # noqa: F401,E402 from . import extlife # noqa: F401,E402 from . import evmore # noqa: F401,E402 from . import extcat # noqa: F401,E402 +from . import extmcode # noqa: F401,E402 diff --git a/forgetest/forgetest/suite/extmcode.py b/forgetest/forgetest/suite/extmcode.py new file mode 100644 index 0000000..4423319 --- /dev/null +++ b/forgetest/forgetest/suite/extmcode.py @@ -0,0 +1,238 @@ +# Copyright 2026 514 LLC d/b/a OpenGlow +# Written by Scott Wiederhold +# https://community.openglow.org +# SPDX-License-Identifier: MIT + +"""An M-code a package answers, on the machine. + +Its own module, for the reason extcore.py gives. The package is the +reference package's id and key with a service of its own that answers +M160 on its call socket, so exthost's put-back takes it away like the +others. The jobs are dark: they move and never command the laser. +""" + +import json +import os +import subprocess + +from ..catalog import test +from .exthost import (EXT_ROOT, FWUP, REF_ID, REF_KEY, _as_found, _forgeext, _put_back, _svc, _tree, _until, _write) +from .motion import _job, _job_post, _job_wait, _words, kernel_start, kernel_xy_mm, machine_idle +from .setup import SAFETY_PHRASE, read_file, record_path, request + +# The service: it answers M160 on the call socket the host hands it (fd 4), and keeps every call it was +# asked, with the monotonic time it was asked, in its data directory. P1 takes 2 s and is done; any +# other P is refused. +MC_SERVICE = r''' +import json, os, socket, time +data = os.environ["FFX_DATA"] +lst = socket.socket(fileno=int(os.environ["FFX_CALL_FD"])) +calls = [] +while True: + c, _ = lst.accept() + try: + buf = b"" + while b"\r\n\r\n" not in buf: + k = c.recv(4096) + if not k: + break + buf += k + head, _, body = buf.partition(b"\r\n\r\n") + n = 0 + for line in head.split(b"\r\n")[1:]: + k, _, v = line.partition(b":") + if k.strip().lower() == b"content-length": + n = int(v) + while len(body) < n: + k = c.recv(4096) + if not k: + break + body += k + req = json.loads(body[:n] or b"{}") + path = head.split(b" ")[1].decode() + t0 = time.monotonic() + words = req.get("words", {}) + if path == "/mcode" and words.get("P") == 1: + time.sleep(2.0) + status, ans = 200, {"message": "forgetest P1"} + elif path == "/mcode": + status, ans = 409, {"error": "the stand-in refused"} + else: + status, ans = 404, {"error": "no such call"} + calls.append({"path": path, "req": req, "t": t0, "status": status}) + with open(os.path.join(data, "calls.json.new"), "w") as f: + json.dump(calls, f) + os.rename(os.path.join(data, "calls.json.new"), os.path.join(data, "calls.json")) + out = json.dumps(ans).encode() + c.sendall(b"HTTP/1.1 %d X\r\nContent-Type: application/json\r\nContent-Length: %d\r\nConnection: close\r\n\r\n" + % (status, len(out)) + out) + except Exception: + pass + finally: + c.close() +''' + + +def _pack_mcode(work): + import io + import tarfile + manifest = {"manifest": 1, "id": REF_ID, "name": "forgetest M-code", "version": "1.0.0", + "author": "forgetest", "license": "MIT", "api": "0.1", "runtime": "python", + "service": {"exec": "bin/mcode.py"}, "capabilities": ["mcode:160", "job_time.run"]} + payload = os.path.join(work, "payload.tar.gz") + with tarfile.open(payload, "w:gz") as t: + for name, text, mode in (("manifest.json", json.dumps(manifest), 0o644), ("bin/mcode.py", MC_SERVICE, 0o755)): + info = tarfile.TarInfo(name) + data = text.encode() + info.size, info.mode = len(data), mode + t.addfile(info, io.BytesIO(data)) + conf = os.path.join(work, "fwup.conf") + _write(conf, 'meta-product = "ForgeFIRM extension"\nmeta-description = "%s"\nmeta-version = "1.0.0"\n' + 'meta-platform = "forgefirm-ext"\nfile-resource payload.tar.gz {\n host-path = "%s"\n}\n' + % (REF_ID, payload)) + key = os.path.join(work, REF_KEY) + raw, signed = os.path.join(work, "raw.ffx"), os.path.join(work, "mcode.ffx") + for cmd in ([FWUP, "-g", "-o", key], [FWUP, "-c", "-f", conf, "-o", raw], + [FWUP, "-S", "-s", key + ".priv", "-i", raw, "-o", signed]): + subprocess.run(cmd, check=True, capture_output=True, timeout=60, cwd=work) + return signed, key + ".pub" + + +def _port_state(fc): + st, body = fc.get("/motion/state") + return body if st == 200 and isinstance(body, dict) else {} + + +@test("exthost.mcode", title="A job waits at an M-code a package answers, dark and still", + subsystem="exthost", kind="auto", mode="grbl", est_min=4, + covers=[("grblhal-glowforge", "src/glowforge_mcode.*"), ("grblhal-glowforge", "src/ctlport.*"), + ("forgectrl", "src/mcode.*"), ("forgectrl", "src/extpkg.*"), ("forgectrl", "src/grblport.*"), + ("forgeext", "src/main.c"), ("forgeext", "src/run.*"), ("forgeext", "src/caps.*"), + ("forgeext", "src/manifest.*"), ("forgeext", "src/call.*")], + requires=["exthost.service", "motion.job"], + steps=["Bed clear; the head needs 10 mm of free travel toward +X. Nobody touches the gantry."], + description="A package that asks for mcode:160 and job_time.run, the one granted, is installed and " + "turned on; its service answers POST /mcode on its call socket. The host's status names it " + "as the one that answers M160. A dark job (out 5 mm, M160 P1, back) plays through POST " + "/job: at the M-code the controller's port state names M160 with P 1, the head stands " + "still across the wait (the kernel's counters and the port's position unchanged over " + "1 s), and the service answers after 2 s; the job then goes on and ends done, every line " + "acknowledged, the head back where it began, the service asked once with the code and " + "its words, no discharge. A job that names M161, which nothing answers, fails at that " + "line with nothing moved. A job at M160 P2, which the service refuses, is held: the " + "port's state goes to Hold with the head still, and the job is aborted from there. " + "Everything is put back as exthost.service puts it back; the refusals the controller " + "makes of a table or an answer out of form, the timeout, the reset, and the barrier's " + "darkness under an open armed window are the driver's mcode_test harness's.") +def mcode(ctx): + import tempfile + fc = ctx.forgectrl + ev = ctx.evidence + machine_idle(ctx) + prior = fc.settings().get("ext_enabled") or "" + raw = read_file(record_path()) + found_tree = _tree(EXT_ROOT) + dir_mode = os.stat(os.path.dirname(EXT_ROOT)).st_mode & 0o7777 + st, body, hdrs = request(fc.base, "GET", "/advisories/extensions", headers={"Host": fc.host_header()}) + etag = hdrs.get("etag") + work = tempfile.mkdtemp(prefix="forgetest-ffx.") + owner_key = os.path.join(EXT_ROOT, "keys", REF_KEY + ".pub") + calls_file = os.path.join(EXT_ROOT, "data", REF_ID, "calls.json") + + def calls(): + try: + return json.loads(read_file(calls_file) or "[]") + except ValueError: + return [] + + try: + archive, pub = _pack_mcode(work) + import shutil + shutil.copy(pub, owner_key) + os.chmod(owner_key, 0o644) + r = _forgeext("install", archive, "--consent-community", "--grant", "job_time.run") + ctx.check(r.get("ok") is True, "the install with job_time.run granted -> %s", r.get("error")) + st, reply = fc.post("/settings", data={"ext_enabled": "1", "advisory": etag, "phrase": SAFETY_PHRASE}) + ctx.check(st == 200, "ext_enabled=1 -> %s %r", st, reply) + x = _until(ctx, lambda: _svc(REF_ID) if _svc(REF_ID).get("state") == "running" else None, 90, poll=0.5) + ctx.check(x, "the service is not running: %s", _svc(REF_ID)) + + def listed(): + try: + doc = json.loads(read_file("/run/forgefirm/ext/status.json") or "{}") + except ValueError: + return None + return doc.get("mcodes") if doc.get("mcodes") == [{"code": 160, "id": REF_ID}] else None + ev["status_mcodes"] = _until(ctx, listed, 30, poll=0.5) + ctx.check(ev["status_mcodes"], "the host's status does not name the package as the one that answers M160") + ctx.sleep(3.0) # the relay tells the controller within its 2 s + + # A dark job that waits at M160 P1 for 2 s. + x0, _y0 = kernel_start(ctx) + st, body = _job_post(fc, "(forgetest exthost.mcode)\nG21\nG91\nG1 X5 F1200\nM160 P1\nG1 X-5 F1200\nG90\n") + ctx.check(st == 200 and isinstance(body, dict) and body.get("state") == "running", + "POST /job -> %s %s", st, _words(body)[:200]) + waiting = _until(ctx, lambda: _port_state(fc) if isinstance(_port_state(fc).get("mcode"), dict) else None, + 20, poll=0.1) + ctx.check(waiting and waiting["mcode"].get("code") == 160 and waiting["mcode"].get("words") == {"P": 1}, + "the port's state does not name M160 P1 while the job waits: %s", waiting) + k1 = kernel_xy_mm(ctx) + ctx.sleep(1.0) + k2 = kernel_xy_mm(ctx) + still = _port_state(fc) + ev["wait"] = {"port": waiting, "kernel": [k1, k2], "port_after_1s": still} + ctx.log("the job waits at M160: port %s, kernel %s -> %s", waiting, k1, k2) + ctx.check(abs(k2[0] - k1[0]) < 0.001 and abs(k2[1] - k1[1]) < 0.001, + "the head moved while the job waited: %s -> %s", k1, k2) + ctx.check(still.get("mpos") == waiting.get("mpos") and still.get("state") == "Idle", + "the port's position or state moved while the job waited: %s -> %s", waiting, still) + ctx.check(abs(k1[0] - x0 - 5.0) < 0.05, "the job did not wait at the M-code 5 mm out: %.3f", k1[0] - x0) + rec, far = _job_wait(ctx, fc, 40, x0) + x1 = kernel_xy_mm(ctx)[0] + got = calls() + ev["job"] = {"record": rec, "farthest_mm": round(far, 3), "end_mm": round(x1 - x0, 3), "calls": got} + ctx.log("the job at M160 P1: %s", ev["job"]) + ctx.check(rec["state"] == "done" and rec["reason"] == "", "the job did not end well: %s", rec) + ctx.check(rec["sent"] == rec["acked"] == rec["lines"] + 1, "lines %s, sent %s, acked %s", + rec["lines"], rec["sent"], rec["acked"]) + ctx.check(rec["lit"] is False and rec["emission"]["laser_on_samples"] == 0, "a dark job's witnesses: %s", + rec["emission"]) + ctx.check(abs(x1 - x0) < 0.1, "the job did not end where it began: %.3f mm off", x1 - x0) + ctx.check(len(got) == 1 and got[0]["path"] == "/mcode" and got[0]["req"] == {"code": 160, "words": {"P": 1}} + and got[0]["status"] == 200, "the service was not asked M160 P1 once: %s", got) + + # M161: nothing answers it, and the job stops where it is parsed. + x0 = kernel_xy_mm(ctx)[0] + st, body = _job_post(fc, "G21\nG91\nM161\nG1 X5 F1200\nG1 X-5 F1200\nG90\n") + rec, far = _job_wait(ctx, fc, 20, x0) + ev["unanswered"] = {"record": rec, "farthest_mm": round(far, 3)} + ctx.log("the job at M161: %s", ev["unanswered"]) + ctx.check(rec["state"] == "failed" and "answered error:20" in (rec.get("reason") or ""), + "a job at M161 did not fail with error:20: %s", rec) + ctx.check(far < 0.01, "a job at an M-code nothing answers moved the head %.3f mm", far) + + # M160 P2: the service refuses it, and the job is held there. + x0 = kernel_xy_mm(ctx)[0] + st, body = _job_post(fc, "G21\nG91\nM160 P2\nG1 X5 F1200\nG1 X-5 F1200\nG90\n") + held = _until(ctx, lambda: _port_state(fc) if _port_state(fc).get("state") == "Hold" else None, 30, poll=0.2) + ev["refused"] = {"port": held, "calls": calls()} + ctx.log("the job at M160 P2: %s", ev["refused"]) + ctx.check(held and held.get("mcode") is None, "a job at a refused M160 is not held: %s", _port_state(fc)) + ctx.sleep(1.0) + ctx.check(abs(kernel_xy_mm(ctx)[0] - x0) < 0.01, "the held job moved the head") + st, body = fc.post("/job/abort") + ctx.check(st == 200, "POST /job/abort -> %s %s", st, _words(body)[:160]) + rec, _far = _job_wait(ctx, fc, 20, x0) + ctx.check(rec["state"] == "failed", "the held job did not end aborted: %s", rec) + if _port_state(fc).get("state") == "Alarm": + st, body = _job_post(fc, "G21\n", unlock="1") + _job_wait(ctx, fc, 20, x0) + ctx.check(_port_state(fc).get("state") == "Idle", "the controller is not idle after the abort: %s", + _port_state(fc)) + finally: + # A check that failed mid-job leaves the job running; it holds the settings until it ends. + if _job(fc).get("state") == "running": + fc.post("/job/abort") + _job_wait(ctx, fc, 20) + _put_back(ctx, fc, work, prior, etag, raw, dir_mode) + _as_found(ctx, fc, prior, raw, dir_mode, found_tree) diff --git a/scripts/bench/mcode_test.py b/scripts/bench/mcode_test.py new file mode 100644 index 0000000..4348971 --- /dev/null +++ b/scripts/bench/mcode_test.py @@ -0,0 +1,265 @@ +#!/usr/bin/env python3 +# Copyright 2026 514 LLC d/b/a OpenGlow +# Written by Scott Wiederhold +# https://community.openglow.org +# SPDX-License-Identifier: MIT +"""Host-side verification of the M-codes an extension package answers. + +Runs the native grblHAL_glowforge binary in null-sink mode, with the +scripted sender and the port client of ctlport_test.py standing in for a +sender and for the machine daemon: + + 1. the table: the port's mcodes op takes "-" or numbers from M160 to + M179, each once, the whole range included, and refuses every other + form, leaving the table as it was + 2. a number nothing answers is an unsupported command where the line is + parsed (error:20), and so is every number outside the table + 3. an answered M-code is a barrier: the sender's line waits for its ok, + the port's state names the M-code with its P, Q and R words and its + seq, the head does not move, a port jog is refused (busy:mcode), and + the answer lets the line finish with a [MSG:] of its words; a second + answer, or one under another seq, is stale, and one out of form is + refused + 4. an answer that says the work was not done holds the job with a + [MSG:] of its words, and a cycle start resumes it + 5. a soft reset ends the wait at once, and the controller answers the + next line + 6. no answer in GFMCODE_WAIT_S holds the job + 7. the barrier is dark: under an open armed window with M3 S500, a job + waiting at an M-code ships no FIRE tick, the window stays open across + the wait, and the next laser move fires again + +Usage: mcode_test.py [path/to/grblHAL_glowforge] +""" +import os +import sys +import time + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import ctlport_test as ct # noqa: E402 + +fail = ct.fail +WAIT_S = 30.0 + + +def send_nowait(s, line): + """A sender line whose response is not waited for; the count before it.""" + n = s.sender.count() + s.sender.sock.sendall((line + "\n").encode()) + return n + + +def wait_pending(s, timeout=5.0): + end = time.time() + timeout + while time.time() < end: + st = s.port.state() + if st.get("mcode"): + return st + time.sleep(0.05) + fail("no M-code waits in the port's state: %r" % s.port.state()) + + +def test_table(s): + if s.port.state().get("mcode", "missing") is not None: + fail("[table] the state carries no mcode null when nothing waits: %r" % s.port.state()) + for good in ("-", ",".join(str(n) for n in range(160, 180)), "160", "160,179,165"): + r = s.port.request("mcodes " + good) + if r != "ok": + fail("[table] mcodes %s -> %r" % (good, r)) + for bad in ("159", "180", "160,160", "160,", ",160", "16a", "0160", "160 161", "1600", "160,,161", ""): + r = s.port.request("mcodes " + bad) + if r != "error:invalid": + fail("[table] mcodes %r -> %r, not error:invalid" % (bad, r)) + # the refusals left the last good table: 160, 179 and 165 + n = send_nowait(s, "M165") + m = wait_pending(s)["mcode"] + if m.get("code") != 165: + fail("[table] a refused list changed the table: %r" % m) + s.port.request("mcode_result %d ok" % m["seq"]) + end = time.time() + 3 + while s.sender.count() == n and time.time() < end: + time.sleep(0.02) + if s.sender.responses[n:n + 1] != ["ok"]: + fail("[table] M165 got %r" % s.sender.responses[n:]) + print("PASS [table]: the port takes '-' and numbers from M160 to M179, the whole range; 11 other forms " + "refused, the table left as it was") + + +def test_unanswered(s): + s.port.request("mcodes 160") + # The core holds a parser error against every later block until an empty + # line (the sync) or a reset: after an unsupported command a job halts. + for line in ("M161", "M179", "M159", "M180 P1"): + r = s.sender.send(line) + if r != "error:20": + fail("[unanswered] %s with only M160 answered got %r, not error:20" % (line, r)) + s.sender.send("") + s.port.request("mcodes -") + r = s.sender.send("M160") + if r != "error:20": + fail("[unanswered] M160 with nothing answered got %r" % r) + s.sender.send("") + print("PASS [unanswered]: a number nothing answers is error:20 where it is parsed") + + +def test_answered(s): + s.port.request("mcodes 160,161") + s.sender.send("G91") + s.sender.send("G0 X3") + s.sender.wait_state("Idle") + n = send_nowait(s, "M160 P2 Q3.5 R-1") + st = wait_pending(s) + m = st["mcode"] + if m.get("code") != 160 or m.get("words") != {"P": 2, "Q": 3.5, "R": -1} or not m.get("seq"): + fail("[answered] the state names %r" % m) + x0 = st["mpos"] + time.sleep(1.0) + st2 = s.port.state() + if st2["mpos"] != x0 or st2["state"] != "Idle": + fail("[answered] the head moved or the state is %s while the job waits: %r -> %r" + % (st2["state"], x0, st2["mpos"])) + if s.sender.count() != n: + fail("[answered] the M-code line got its response before the answer: %r" % s.sender.responses[n:]) + r = s.port.request("jog G91 X1 F600") + if r != "busy:mcode": + fail("[answered] a port jog while the job waits -> %r, not busy:mcode" % r) + for bad in ("mcode_result %d maybe" % m["seq"], "mcode_result x ok", "mcode_result 0 ok", + "mcode_result %d ok [bracket]" % m["seq"], "mcode_result %d ok %s" % (m["seq"], "w" * 97)): + r = s.port.request(bad) + if r not in ("error:invalid", "error:stale"): + fail("[answered] %r -> %r" % (bad, r)) + if s.port.request("mcode_result %d ok" % (m["seq"] + 7)) != "error:stale": + fail("[answered] an answer under another seq was taken") + if s.port.state().get("mcode") is None: + fail("[answered] a refused answer ended the wait") + r = s.port.request("mcode_result %d ok all set" % m["seq"]) + if r != "ok": + fail("[answered] the answer -> %r" % r) + end = time.time() + 3 + while s.sender.count() == n and time.time() < end: + time.sleep(0.02) + if s.sender.responses[n:n + 1] != ["ok"]: + fail("[answered] the M-code line got %r after the answer" % s.sender.responses[n:]) + if not s.sender.saw("M160: done: all set"): + fail("[answered] the sender got no [MSG:] of the answer's words") + if s.port.request("mcode_result %d ok" % m["seq"]) != "error:stale": + fail("[answered] a second answer was taken") + if s.port.state().get("mcode") is not None: + fail("[answered] the state still names an M-code") + if s.sender.send("G0 X-3") != "ok": + fail("[answered] the job did not go on") + print("PASS [answered]: the line waited for the answer with the head still and a port jog " + "refused; the state named M160 with P, Q and R; the answer let it go on with its words") + + +def test_fail_holds(s): + s.sender.wait_state("Idle") + n = send_nowait(s, "M161") + m = wait_pending(s)["mcode"] + s.port.request("mcode_result %d fail the exhaust did not start" % m["seq"]) + end = time.time() + 3 + while s.sender.count() == n and time.time() < end: + time.sleep(0.02) + s.sender.wait_state("Hold", timeout=5) + if not s.sender.saw("M161: the exhaust did not start: the job is held"): + fail("[fail] the sender got no [MSG:] of why the job is held") + s.sender.realtime(b"~") + s.sender.wait_state("Idle", timeout=5) + if s.sender.send("G0 X1") != "ok": + fail("[fail] the job did not go on after the resume") + print("PASS [fail]: an answer that the work was not done held the job with its words; a cycle " + "start resumed it") + + +def test_reset_ends_wait(s): + s.sender.wait_state("Idle") + send_nowait(s, "M160") + wait_pending(s) + t0 = time.time() + s.sender.realtime(b"\x18") + end = time.time() + 3 + while time.time() < end and s.port.state().get("mcode") is not None: + time.sleep(0.02) + dt = time.time() - t0 + if s.port.state().get("mcode") is not None: + fail("[reset] the wait outlived a soft reset") + time.sleep(0.5) + s.sender.send("$X") + if s.sender.send("G0 X1") != "ok": + fail("[reset] the controller did not answer after the reset") + print("PASS [reset]: a soft reset ended the wait in %.2f s" % dt) + + +def test_timeout_holds(s): + s.port.request("mcodes 160") + s.sender.wait_state("Idle") + t0 = time.time() + send_nowait(s, "M160") + wait_pending(s) + s.sender.wait_state("Hold", timeout=WAIT_S + 5) + dt = time.time() - t0 + if dt < WAIT_S - 1: + fail("[timeout] the job was held after %.1f s, not after the wait" % dt) + if not s.sender.saw("M160 had no answer from its extension in 30 s: the job is held"): + fail("[timeout] the sender got no [MSG:] of the timeout") + s.sender.realtime(b"~") + s.sender.wait_state("Idle", timeout=5) + print("PASS [timeout]: with no answer the job was held after %.1f s, and resumed" % dt) + + +def test_dark(): + s = ct.Session(laser=True) + try: + s.port.request("mcodes 160") + s.sender.send("G91") + s.sender.send("M3 S500") + s.sender.send("G1 X10 F1500") + s.sender.wait_state("Idle") + time.sleep(0.5) + if not s.sender.saw("laser armed"): + fail("[dark] the laser line did not arm: the case would prove nothing") + steps0, fire0 = s.ticks() + if fire0 == 0: + fail("[dark] the armed G1 shipped no FIRE tick: the dump cannot see emission") + send_nowait(s, "M160") + m = wait_pending(s)["mcode"] + time.sleep(3.0) + steps1, fire1 = s.ticks() + s.port.request("mcode_result %d ok" % m["seq"]) + time.sleep(0.3) + s.sender.send("G1 X5 F1500") + s.sender.wait_state("Idle") + time.sleep(0.5) + steps2, fire2 = s.ticks() + if fire1 != fire0 or steps1 != steps0: + fail("[dark] the job waiting at M160 under M3 S500 shipped %d FIRE ticks and %d steps" + % (fire1 - fire0, steps1 - steps0)) + if s.sender.saw("laser disarmed"): + fail("[dark] the armed window closed across the wait") + if fire2 <= fire1: + fail("[dark] the laser move after the answer shipped no FIRE: the case would prove nothing") + print("PASS [dark]: 3 s waiting at M160 under an open window with M3 S500 shipped no FIRE tick " + "and no step; the window stayed open and the next move fired (%d ticks)" % (fire2 - fire1)) + finally: + s.close() + + +def main(): + if not os.path.exists(ct.BIN): + fail("no controller binary at %s" % ct.BIN) + s = ct.Session() + try: + test_table(s) + test_unanswered(s) + test_answered(s) + test_fail_holds(s) + test_reset_ends_wait(s) + test_timeout_holds(s) + finally: + s.close() + test_dark() + print("ALL PASS") + + +if __name__ == "__main__": + main()