vh/forgefirm
1
0
Fork 0
mirror of https://github.com/openglow-org/forgefirm.git synced 2026-10-15 17:19:55 -07:00
Code Issues Packages Projects Releases Wiki Activity

forgetest: nineteen tests a host test already proves leave the catalog

Browse Source 
The catalog is for what needs the machine or an operator. Each test
below asserted software rules that a host test in CI already proves,
read assertion by assertion next to that host test before it left; the
record of that reading is kept with the audit.

Safety rules (each approved for deletion by the owner on that reading):
- laser.disarm-in-hold: laser_lifecycle_test test_resume_after_grace.
- laser.arm-wait-lid: laser_arm_test E-G, laser_lifecycle_test
  test_lid_open_in_wait.
- laser.port-dark: ctlport_test test_port_dark (port jogs, a long jog
  and its cancel, four injected words, the panel ops: no FIRE tick
  under an open window with M3 modal).
- motion.respawn-gate: super_test A and D.
- motion.interlock-cancel-home: the lifecycle loop cases, laser_arm_test G.
- motion.lid-policy-hold: laser_lifecycle_test test_lid_policy_hold.
- cloud.lid-during-button-wait, cloud.dark-print, cloud.verdict-refuse,
  cloud.paused-lid-cancel: test_machine_lid_button.py.
- cooling.fan-duty-readback: cool_fanwrite_test.
- cooling.intake-one-fan: cool_intake_test A-E.
- exthost.hold-pause-tier: holdkeep_test, run_test.py, cool_flow_test P,
  holds_test.

Other rules:
- homing.cloud-offsets: manual_home_test [own-offsets],
  test_gfhome_motion; the real session stays in cloud.mode-switch.
- exthost.core-range: version_test, install_test.
- exthost.operator-destinations: run_test.py, netrules_test,
  install_test, extpkg_test.
- exthost.lifecycle: run_test.py's ext.shutdown read before the stop.
- exthost.mcode: the null-sink mcode_test, forgectrl mcode_test,
  sdk_test, install_test.
- motion.soft-limits: z_envelope_test's homed cases, which now assert
  no step before the alarm, and envelope_test.

- cooling_intake.py, extcore.py, extdest.py, extlife.py and extmcode.py
  go whole; homeoff.py keeps only the camera-home helpers that
  cloud.mode-switch, setup.check-envelope and their meta-tests use.
- The helpers only the removed tests used go with them.
- exthost.armed-freeze still runs the cloud suite's dark-print helper,
  whose latch checks still run; its description says so. It now
  requires cooling.floor-and-warm-up, the dark print's own prerequisite,
  in place of the removed cloud.dark-print.
- HOST_TESTED: forgectrl src/holds.* (holds_test) and src/mcode.*
  (mcode_test), whose only catalog home left with these tests.
- The meta-tests that replayed the removed tests go:
  test_cooling_intake_suite.py, the cloud replay cases and their
  fixture. The one that catches a false PASS stays and runs the dark
  print as armed-freeze does: a latch open at the button fails.
- Text that named a removed test (exthost.motion-job's description and
  comment, test_baseline's docstring) no longer does.

Proof: the coverage lint on a manifest from the recipe pins reports 0
uncovered paths and 0 host-tested row problems over 111 tests (its one
failure is setup.check-lens's src/lensfind.* entry, which the forgectrl
pin bump clears); pyflakes adds no finding; the gate self-check refuses
an empty artifact cleanly; the forgetest host suite passes under Linux
(547 tests); the catalog loads with 111 tests after armed-freeze's
description edit.

Acceptance: the catalog hash moves, so the next image opens a new
campaign. Every remaining test in laser.py, motion.py, cloud.py,
cooling.py, exthost.py, homeoff.py and bedsize.py moves its
fingerprint (their modules' shared text changed); exthost.armed-freeze
moves by its requires.
This commit is contained in:
ScottW514
2026-10-13 16:05:14 -04:00
parent bef1bf9902
commit 6a434bbf77
18 changed files with 41 additions and 2446 deletions
Download Patch File Download Diff File Expand all files Collapse all files
forgetest/forgetest/manifest.py
+6
View File
@@ -125,6 +125,12 @@ BEHAVIORAL = [
# catalog test exercises on the hardware belongs in that test's covers, not
# here.
HOST_TESTED = [
# An extension package's holds as the engine reads them (forgeext's
# holdkeep_test and run_test prove the writer's side).
("forgectrl", "src/holds.*", "forgectrl:tests/holds_test.c", "forgectrl:.github/workflows/build.yml"),
# The M-code relay to the package that answers (the controller's side is
# the null-sink mcode_test harness, under grblhal-glowforge's src/**).
("forgectrl", "src/mcode.*", "forgectrl:tests/mcode_test.c", "forgectrl:.github/workflows/build.yml"),
]
# This repository's root, when the package runs from a checkout of it; a
forgetest/forgetest/suite/__init__.py
-5
View File
@@ -15,20 +15,15 @@ from . import setup_burns # noqa: F401,E402
from . import logs # noqa: F401,E402
from . import motion # noqa: F401,E402
from . import cooling # noqa: F401,E402
from . import cooling_intake # noqa: F401,E402
from . import laser # noqa: F401,E402
from . import camera # noqa: F401,E402
from . import update # noqa: F401,E402
from . import cloud # noqa: F401,E402
# Last: it stands on the setup record's consent and on the cloud suite's dark print.
from . import exthost # noqa: F401,E402
from . import extcore # noqa: F401,E402
from . import extcall # noqa: F401,E402
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
from . import extwizard # noqa: F401,E402
from . import updlock # noqa: F401,E402
from . import homeoff # noqa: F401,E402
forgetest/forgetest/suite/bedsize.py
+1 -1
View File
@@ -9,7 +9,7 @@ Its own module, so that no other test's fingerprint moves. The check is
driven from here inside the travel the machine already knows: a few jogs
out from the camera home, never toward an end, so it needs no operator.
Measuring the real ends is the operator's own run of the check. The
homing is the web-service session homing.cloud-offsets runs.
homing is the web-service session cloud.mode-switch runs.
"""
import json
forgetest/forgetest/suite/cloud.py
+1 -267
View File
@@ -1262,123 +1262,7 @@ def lid_interlock_abort_body(ctx, ev, off, job, offset):
ev.get("edge_to_stop_ms"))
@test("cloud.lid-during-button-wait", title="Lid open at the cloud button prompt cancels the print",
subsystem="cloud", kind="operator", est_min=8,
covers=_MACHINE_RUN, requires=[], actions=["lid"],
steps=[OFFLINE_STEP, LID_STEP,
"The job is longer than the ring holds, so the ring is filled before the button "
"lights; that takes a minute. When the button lights white, do NOT press it - open "
"the lid, and close it when told. Nothing moves and nothing fires."],
description="A cloud print waiting for the button is canceled by the lid: the wait ends "
"with the lid named as the reason, the laser latch relocks, the armed window "
"closes, no run starts, and the job ends ':cancelled'. The job is longer than "
"the ring, so its feeder is alive through the wait with the rest of the print "
"in hand: the cancel has to stop that feeder before the park clears the ring, "
"and the ring has to stay empty afterward. A feeder left alive would refill "
"it and the park, which ignores the lid, would play the print's path.")
def lid_during_button_wait(ctx):
ev = ctx.evidence
offset = enter_offline(ctx)
job = offline_job(ctx, "wait.puls", seconds=3500)
off = Offline().__enter__()
try:
lid_during_button_wait_body(ctx, ev, off, job, offset)
finally:
off.__exit__(None, None, None)
offline_cleanup(ctx)
def lid_during_button_wait_body(ctx, ev, off, job, offset):
off.print_ready(9003, job)
got = wait_button_wait(ctx, offset, 180, also=["job is longer than the ring"])
ctx.check(got["job is longer than the ring"],
"the job fit the ring: this test needs a job the ring cannot hold")
ctx.act("lid", "open", text="The button is lit: do NOT press it.", timeout=120)
relock = "button wait lid opened - relocking the laser"
got = wait_log(ctx, offset, [relock], 60)
fin = wait_action_finished(ctx, offset, "print", 60)
ev["log"] = {relock: bool(got[relock]), "print finished": message(fin)}
ctx.check(got[relock], "the lid did not end the button wait")
ctx.check(fin and CANCELLED in fin, "the print did not end ':cancelled': %s",
message(fin) or "no finish line")
# No run may start between the button wait and the cancel: the print
# itself, or a park (the head never moved, there is nothing to park).
# The service's moves BEFORE the print are legitimate runs and are
# outside this window.
lines = log_lines_since(GFCLOUD_LOG, offset)
wait_i = next((i for i, ln in enumerate(lines) if "waiting for button" in ln), None)
end_i = action_finish_index(lines, "print")
if end_i is None or wait_i is None or end_i < wait_i:
end_i = len(lines)
ran = ([ln for ln in lines[wait_i:end_i] if "starting run" in ln]
if wait_i is not None else [])
ev["runs_started_after_wait"] = len(ran)
ctx.check(not ran, "a run started after the lid-open cancel (%d)", len(ran))
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 cancel")
ctx.check(ev["latch_locked"], "kernel latch not locked after the cancel")
ev["button_dark"] = hw.button_lit()
ctx.check(ev["button_dark"] is False, "the button is still lit after the cancel (%s)", ev["button_dark"])
# The park cleared the ring and nothing refilled it: the feeder is gone,
# the device is out of live-feed mode, and the program total stands at
# zero and stays there. A feeder left alive shows here as a total that
# comes back within its retry period.
total = read_program_total()
ctx.sleep(2)
later = read_program_total()
streaming = hw.sysfs_int("cnc/streaming", 0)
ev["program_total_after"] = {"first": total, "later": later}
ev["streaming_after"] = streaming
ctx.check(total == 0 and later == 0,
"the ring is not empty after the cancel (program total %s, then %s): the feeder "
"outlived the job", total, later)
ctx.check(streaming == 0, "cnc/streaming is %s after the cancel", streaming)
ctx.act("lid", "close")
settle_cloud(ctx, offset)
ev["events"] = offline_events(off)
ctx.log("PASS: lid open at the button prompt canceled the print; latch locked, armed=false, "
"button dark, ring empty with the feed stopped")
# ---------------------------------------------------------------- the cooling verdict
HOLD_KEYS = ("cool_temp_min", "cool_temp_start")
REFUSE_KEYS = HOLD_KEYS + ("cloud_hold_max_s",)
HOLD_REFUSE_S = 60 # the shortest bound the client takes
HOLD_REFUSE_LINE = "cooling hold held 60 s - relocking the laser"
WARMUP_WAIT_LINE = "waiting on the cooling engine: WARMUP"
@test("cloud.dark-print", title="A dark cloud print: the latch locked until the run, unlocked "
"for it, locked after it",
subsystem="cloud", kind="operator", est_min=3,
covers=_MACHINE_RUN + [("python3-gfhardware", "gfhardware/coolsvc.py"),
("forgectrl", "src/cool.*"), ("forgectrl", "src/main.c")],
requires=["cloud.lid-during-button-wait", "cooling.floor-and-warm-up"], actions=["button"],
steps=[OFFLINE_STEP,
"Bed clear (the job is dark: a 30 s square at S0, nothing fires). Press the button "
"when it lights: the print waits for the engine, runs, and finishes."],
description="The cooling-engine contract in cloud mode at its ordinary end, on the engine "
"itself: the print arms on the press, waits for the engine's acknowledgment (the "
"fans at their floors), and runs. The laser latch is locked at the button and "
"through the wait, unlocks only for the run (immediately before it starts), and "
"is locked again when the job ends; the print completes. A hold the engine never "
"clears is cloud.verdict-refuse; the engine's own warm-up release is "
"cooling.floor-and-warm-up.")
def dark_print(ctx):
ev = ctx.evidence
fc = ctx.forgectrl
offset = enter_offline(ctx)
job = offline_job(ctx, "dark.puls", seconds=30)
off = Offline().__enter__()
try:
dark_print_body(ctx, ev, off, job, offset, fc)
finally:
off.__exit__(None, None, None)
offline_cleanup(ctx)
# ---------------------------------------------------------------- the dark print (exthost.armed-freeze)
def dark_print_body(ctx, ev, off, job, offset, fc):
off.print_ready(9005, job)
@@ -1404,91 +1288,6 @@ def dark_print_body(ctx, ev, off, job, offset, fc):
ctx.log("PASS: locked at the button, unlocked for the run, locked after it; the print completed")
@test("cloud.verdict-refuse", title="A cloud print held by the engine past the bound is canceled with "
"the latch never unlocked",
subsystem="cloud", kind="operator", est_min=4,
covers=_MACHINE_RUN + [("python3-gfhardware", "gfhardware/coolsvc.py"),
("forgectrl", "src/cool.*")],
requires=["cloud.dark-print"], actions=["button"],
steps=[OFFLINE_STEP,
"Coolant at room temperature (10 to 30 C). The test sets the warm-up gate well above "
"the coolant and the hold bound to its minimum, and restores both. Press the button "
"when it lights: nothing runs; a minute later the print cancels by itself."],
description="The other end of the cooling-engine contract before a run: a verdict that "
"refuses fire and never clears. The start gate far above the coolant keeps the "
"armed session under the warm-up for longer than the hold bound (cloud_hold_max_s "
"at its minimum, 60 s). The client waits with the laser latch locked the whole "
"time - it is never unlocked before the verdict passes, so there is nothing to "
"relock - then cancels the print at the bound: no run, the armed window closed, "
"the print ':cancelled'. Dark by construction.")
def verdict_refuse(ctx):
ev = ctx.evidence
fc = ctx.forgectrl
before = fc.settings()
orig = {k: before.get(k, "") for k in REFUSE_KEYS}
ev["orig"] = orig
up = (fc.status().get("coolant") or {}).get("up_c")
ctx.check(up is not None and 10.0 <= up <= 30.0, "coolant outside the test's range (up_c %s)", up)
offset = enter_offline(ctx)
job = offline_job(ctx, "refuse.puls", seconds=30)
off = Offline().__enter__()
try:
verdict_refuse_body(ctx, ev, off, job, offset, fc, up, orig)
finally:
st, body = fc.post("/settings", params=orig)
ctx.log("restore: POST /settings %s -> %s", orig, st)
off.__exit__(None, None, None)
offline_cleanup(ctx)
def verdict_refuse_body(ctx, ev, off, job, offset, fc, up, orig):
gate = round(up + 8.0, 1) # eight degrees: the heater cannot get there in a minute
st, body = fc.post("/settings", params={"cool_temp_start": str(gate),
"cool_temp_min": orig["cool_temp_min"],
"cloud_hold_max_s": str(HOLD_REFUSE_S)})
ctx.check(st == 200, "POST /settings cool_temp_start=%s cloud_hold_max_s=%s -> %s %s",
gate, HOLD_REFUSE_S, st, body)
ev["gate"] = gate
off.print_ready(9006, job)
wait_button_wait(ctx, offset, 120)
ev["latch_locked_at_button"] = latch_locked()
ctx.check(ev["latch_locked_at_button"], "the latch is unlocked at the button wait")
ctx.act("button", "press", text="The button is lit: press it. Nothing runs; the print cancels "
"itself a minute later.",
until=lambda: log_has(offset, WARMUP_WAIT_LINE), timeout=PRESS_TIMEOUT_S)
got = wait_log(ctx, offset, [WARMUP_WAIT_LINE], 30)
ctx.check(got[WARMUP_WAIT_LINE], "the armed print did not wait on the warm-up")
t0 = time.time()
samples = []
ctx.notice("The engine holds the print; the client gives up at the bound. Do nothing.")
try:
while time.time() - t0 < HOLD_REFUSE_S + 30:
ctx.checkpoint()
samples.append(latch_locked())
if log_has(offset, HOLD_REFUSE_LINE):
break
ctx.sleep(2)
finally:
ctx.clear_notice()
ev["hold_s"] = round(time.time() - t0)
ev["latch_samples"] = {"n": len(samples), "unlocked": samples.count(False)}
ctx.check(log_has(offset, HOLD_REFUSE_LINE), "the client did not cancel at the hold bound (%d s)",
ev["hold_s"])
ctx.check(all(samples), "the latch unlocked during the hold (%d of %d samples)",
samples.count(False), len(samples))
ctx.check(wait_print_running(ctx, offset, 1) is None, "the run started under the refused verdict")
fin = wait_action_finished(ctx, offset, "print", 120)
ev["print finished"] = message(fin)
ctx.check(fin and CANCELLED in fin, "the held print did not end ':cancelled': %s",
message(fin) or "no finish line")
st, cs = fc.get("/cool/status")
ev["armed_after"] = cs.get("armed") if isinstance(cs, dict) else None
ctx.check(not ev["armed_after"], "armed window still open after the cancel")
ctx.check(latch_locked(), "the latch is unlocked after the cancel")
ev["events"] = offline_events(off)
ctx.log("PASS: held %s s with the latch locked throughout, then canceled at the bound", ev["hold_s"])
@test("cloud.pause-resume", title="Button pauses and resumes a cloud print (factory backtrack + lead)",
subsystem="cloud", kind="live", est_min=8,
covers=_CLOUD_ALL,
@@ -1722,68 +1521,3 @@ def oversize_stream_body(ctx, ev, off, job, offset, before):
"stopped it, parked, relocked and reported ':cancelled'")
@test("cloud.paused-lid-cancel", title="A paused cloud print is canceled by the lid",
subsystem="cloud", kind="live", est_min=6,
covers=_MACHINE_RUN, requires=["cloud.pause-resume", "cloud.lid-interlock-abort"],
actions=["button", "lid"],
steps=[OFFLINE_STEP, ARM_STEP, LID_STEP,
"The head needs 40 mm of free +X and +Y travel.",
"Press the button to start; when asked, press it again (pause), then open the lid and "
"leave it open until the head is back; close it when told."],
description="A job paused on the button is canceled by the lid, from the state the factory "
"ends it in - there is no resume past a lid open. The same tail as every abort: "
"the motion stops, the head parks back at the job start, the latch relocks and "
"the armed window closes, the button goes dark, and the print ends ':cancelled'. "
"(The app's own cancel of a running print is exercised where a print has to be "
"ended that way: cloud.oversize-stream.)")
def paused_lid_cancel(ctx):
ev = ctx.evidence
offset = enter_offline(ctx)
job = offline_job(ctx, "paused.puls", seconds=60)
off = Offline().__enter__()
try:
paused_lid_cancel_body(ctx, ev, off, job, offset)
finally:
off.__exit__(None, None, None)
offline_cleanup(ctx)
def paused_lid_cancel_body(ctx, ev, off, job, offset):
# -- paused on the button, then the lid ---------------------------------
offline_start_print(ctx, off, 9005, job, offset)
ctx.act("button", "press", text="The head is moving: the press pauses the print.",
until=lambda: log_has(offset, PAUSE_LINES[0]), timeout=PRESS_TIMEOUT_S, fail=False)
got = wait_log(ctx, offset, list(PAUSE_LINES), 10)
ev["paused"] = {k: bool(v) for k, v in got.items()}
ctx.check(got["button pressed mid-run; pausing"], "the press did not pause the print")
ctx.check(got["paused at"], "the pause did not settle (no 'paused at')")
ctx.act("lid", "open", text="The print is paused: leave the lid open until the head has returned "
"to the corner.", timeout=60)
lid_stop = "lid opened mid-run; stopping motion"
got = wait_log(ctx, offset, [lid_stop, "start return home", "return home complete"], 120)
fin1 = wait_action_finished(ctx, offset, "print", 60)
ev["lid_from_pause"] = {k: message(v) for k, v in got.items()}
ev["lid_from_pause"]["print finished"] = message(fin1)
for k, v in ev["lid_from_pause"].items():
ctx.log(" [print] %s: %s", k, "seen" if v else "MISSING")
ctx.check(got[lid_stop], "the lid did not end the paused print")
ctx.check(got["return home complete"], "the paused print did not park to completion")
ctx.check(fin1 and CANCELLED in fin1, "the print did not end ':cancelled': %s",
message(fin1) or "no finish line")
ctx.check(ctx.forgectrl.wait_idle(15, abort=ctx.aborted), "machine not idle after the park")
kpos = read_position()
ev["counters_after"] = 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)
st, cs = ctx.forgectrl.get("/cool/status")
ev["armed_after"] = cs.get("armed") if isinstance(cs, dict) else None
ev["latch_locked_after"] = latch_locked()
ctx.check(not ev["armed_after"], "armed window still open after the paused print was canceled")
ctx.check(ev["latch_locked_after"],
"kernel latch not locked after the paused print was canceled")
ev["button_dark"] = hw.button_lit()
ctx.check(ev["button_dark"] is False, "the button is still lit after the cancel (%s)", ev["button_dark"])
ctx.act("lid", "close")
settle_cloud(ctx, offset)
ev["events"] = offline_events(off)
ctx.log("PASS: a paused print canceled by the lid stopped, parked, relocked and reported ':cancelled'")
forgetest/forgetest/suite/cooling.py
-67
View File
@@ -1404,73 +1404,6 @@ def flow_under_load(ctx):
"(dT %.1f), window closed %.1f s after M2", raw, share, rise, float(limit), float(dtc), dt)
@test("cooling.fan-duty-readback", title="A fan duty a device lost is read back and put back inside a tick",
subsystem="cooling", kind="auto", mode="grbl", est_min=2,
covers=_COOL_COVERS + [("kernel-module-glowforge", "src/head*")],
requires=["kernel.latch-locked-idle"],
description="M8 opens a run session and every fan reads its run duty back from its device. "
"The head's air-assist register is then written back to the idle duty behind the "
"engine's back (what a head reset or a dropped transaction leaves) and the exhaust "
"PWM to zero: within a few ticks each reads its run duty again, the log names each "
"loss and its put-back, and the session stays OK with no airflow fault (a fan put "
"back gets its spin-up grace). Then M9 and the idle duties.")
def fan_duty_readback(ctx):
from .motion import FORGECTRL_LOG, _log_lines, _log_offset
fc = ctx.forgectrl
ev = ctx.evidence
run = {"head/air_assist_pwm": 1023, "thermal/exhaust_pwm": 65535, "thermal/intake_pwm": 43278}
with ctx.grbl() as g:
c = _hold_session(ctx, g, fc, lambda c: c.get("phase") == "run", "readback", 20)
try:
ctx.check(c.get("phase") == "run", "no run session opened: %s", c)
ctx.sleep(2)
before = {k: hw.sysfs_int(k) for k in run}
ctx.log("duties at run: %s", before)
ev["run_duties"] = before
ctx.check(all(before.get(k) == v for k, v in run.items()),
"not every fan reads its run duty: %s", before)
at_loss = _log_offset(FORGECTRL_LOG) # the put-back lines are ones written after this
hw.sysfs_write("head/air_assist_pwm", 204)
hw.sysfs_write("thermal/exhaust_pwm", 0)
lost = {k: hw.sysfs_int(k) for k in ("head/air_assist_pwm", "thermal/exhaust_pwm")}
ctx.log("written behind the engine's back: %s", lost)
back = {}
t0 = time.time()
while time.time() - t0 < 6:
ctx.sleep(0.5)
back = {k: hw.sysfs_int(k) for k in ("head/air_assist_pwm", "thermal/exhaust_pwm")}
if back.get("head/air_assist_pwm") == 1023 and back.get("thermal/exhaust_pwm") == 65535:
break
took = round(time.time() - t0, 1)
ev["put_back"] = {"duties": back, "after_s": took}
ctx.log("put back after %s s: %s", took, back)
ctx.check(back.get("head/air_assist_pwm") == 1023, "the air-assist duty was not put back: %s", back)
ctx.check(back.get("thermal/exhaust_pwm") == 65535, "the exhaust duty was not put back: %s", back)
ctx.sleep(2)
line = "head/air_assist_pwm read 204 with 1023 commanded: put back"
ctx.wait_for(lambda: bool(_log_lines(FORGECTRL_LOG, at_loss, line)), 5, poll=0.25)
ctx.check(_log_lines(FORGECTRL_LOG, at_loss, line),
"the air-assist loss is not named in the log")
line = "thermal/exhaust_pwm read 0 with 65535 commanded: put back"
ctx.wait_for(lambda: bool(_log_lines(FORGECTRL_LOG, at_loss, line)), 5, poll=0.25)
ctx.check(_log_lines(FORGECTRL_LOG, at_loss, line),
"the exhaust loss is not named in the log")
for _ in range(8):
ctx.sleep(1)
c = _cool(fc)
if c.get("verdict") != "OK":
break
ev["verdict_after"] = {"verdict": c.get("verdict"), "hold": c.get("hold"), "reason": c.get("reason")}
ctx.check(c.get("verdict") == "OK" and not c.get("hold"),
"the session did not stay OK after the put-back: %s %s", c.get("verdict"), c.get("reason"))
finally:
g.command("M9")
_session_ended(ctx, fc, "readback")
idle = {k: hw.sysfs_int(k) for k in run}
ev["idle_duties"] = idle
ctx.log("duties at idle: %s", idle)
@test("cooling.fail-tier-stop", title="A fail-tier verdict ends the controller: CRASH stops motion, "
"locks the laser, and the supervisor restarts the controller",
subsystem="cooling", kind="operator", mode="grbl", est_min=4,
forgetest/forgetest/suite/cooling_intake.py
-114
View File
@@ -1,114 +0,0 @@
# Copyright 2026 514 LLC d/b/a OpenGlow
# Written by Scott Wiederhold
# https://community.openglow.org
# SPDX-License-Identifier: MIT
"""cooling.intake-one-fan - a machine built with one intake fan
(cool_intake_fans 1, which the setup's Intake fans step writes when the
operator counts one): the second
intake tach is published as not fitted and never judged, and intake 1
still holds the intake floor. On a machine with two intake fans the
setting alone proves both halves: intake 2 turns and is not judged, and
an intake floor no fan can meet trips with intake 2 still out of it."""
from ..catalog import test
from .cooling import _COOL_COVERS, _after_session, _cool, _fan_gate, _run_session, _set_gates
INTAKE_KEYS = ("cool_intake_fans", "cool_tach_intake_min_rpm", "cool_fan_grace_s")
INTAKE_GRACE_S = "8" # past the intakes' 7 s to 90 percent, so only the leg's floor trips
INTAKE_TRIP_WAIT_S = 20 # grace + three ticks, with slack for the 1 Hz pipeline
@test("cooling.intake-one-fan", title="With one intake fan set, intake 2 is not judged and intake 1 still is",
subsystem="cooling", kind="auto", mode="grbl", est_min=3,
covers=_COOL_COVERS + [("forgectrl", "src/main.c")], requires=["cooling.fan-gate-trips"],
steps=["Machine idle, fans quiet. The test sets one intake fan, the intake floor and the spin-up "
"grace and restores all three; three short M8/M9 cycles spin the fans."],
description="cool_intake_fans 1 must take intake 2 out of the airflow gates and leave intake 1 in "
"them. With it set, a session at the cut fan profile must publish intake_2 as not fitted "
"with no floor while it reads its real speed, run OK, and judge intake_1 ok at the intake "
"floor; an intake floor at its legal maximum must then trip AIRFLOW (hold, fire blocked) "
"with intake_1 TRIPPED and intake_2 still not fitted, and the fault must end with the "
"session; restored, the next session judges intake_2 again and every fan reads at or "
"above its floor.")
def intake_one_fan(ctx):
fc = ctx.forgectrl
ev = ctx.evidence
before = fc.settings()
orig = {k: before.get(k, "") for k in INTAKE_KEYS}
ev["orig"] = orig
ctx.log("original: %s", orig)
inl = (before.get("gates") or {}).get("cool_tach_intake_min_rpm") or {}
ctx.check(inl.get("gate") == "intake" and inl.get("off") == "low",
"settings reply does not describe the intake floor as the intake gate, off at zero: %s", inl)
ctx.check(orig["cool_intake_fans"] != "1",
"cool_intake_fans is already 1: the test starts from a machine with both intakes judged")
c0 = _cool(fc)
ctx.check(c0.get("verdict") == "OK" and c0.get("gates_off") == [],
"engine is not at OK with every gate on before the test: %s", c0)
restored = False
with ctx.grbl() as grbl:
st = grbl.status_report()
ctx.check(st["state"].startswith("Idle"), "controller is %s", st["state"])
try:
# Leg 1: one intake fan. Intake 2 still turns on this machine; it
# is measured and published not fitted, and nothing judges it.
_set_gates(ctx, fc, {"cool_intake_fans": "1", "cool_fan_grace_s": INTAKE_GRACE_S})
c = _run_session(ctx, grbl, fc,
lambda c: c.get("verdict") == "OK" and _fan_gate(c, "intake_1").get("state") == "ok",
"one intake fan leg", wait=INTAKE_TRIP_WAIT_S + 10)
g1, g2 = _fan_gate(c, "intake_1"), _fan_gate(c, "intake_2")
ev["one_fan"] = {"cool": c, "intake_1": g1, "intake_2": g2}
ctx.check(g2.get("state") == "not fitted" and (g2.get("floor") or 0) == 0,
"intake_2 is not published not fitted with no floor: %s", g2)
ctx.check((g2.get("reading") or 0) > 0,
"intake_2 reads %s: the test needs a machine whose second intake turns", g2.get("reading"))
ctx.check(c.get("verdict") == "OK", "the one intake fan session is not OK: %s", c)
ctx.check(g1.get("state") == "ok" and (g1.get("floor") or 0) > 0,
"intake_1 is not judged ok at the intake floor: %s", g1)
# Leg 2: the intake floor at its legal maximum. Intake 1 is the
# one intake left in the gates; intake 2 stays out of them.
_set_gates(ctx, fc, {"cool_tach_intake_min_rpm": str(inl.get("hi"))})
c = _run_session(ctx, grbl, fc, lambda c: c.get("verdict") == "AIRFLOW", "intake 1 trip leg",
wait=INTAKE_TRIP_WAIT_S)
g1, g2 = _fan_gate(c, "intake_1"), _fan_gate(c, "intake_2")
ev["intake_1_trip"] = {"cool": c, "intake_1": g1, "intake_2": g2}
ctx.check(c.get("verdict") == "AIRFLOW", "an intake floor of %s rpm did not trip: %s", inl.get("hi"), c)
ctx.check(c.get("fire_ok") is False and c.get("hold") is True,
"AIRFLOW without fire blocked and a hold: %s", c)
ctx.check(g1.get("state") == "TRIPPED", "intake_1 does not read TRIPPED: %s", g1)
ctx.check(g2.get("state") == "not fitted", "intake_2 was judged with one intake fan set: %s", g2)
ctx.check("intake_1" in (c.get("reason") or ""), "the reason does not name intake_1: %r", c.get("reason"))
c = _after_session(ctx, fc)
ev["intake_1_trip_after"] = c
ctx.check(c.get("verdict") == "OK" and c.get("hold") is False,
"the fan fault outlived its run session: %s", c)
# Restore, and prove it: intake 2 judged again, every fan at or
# above its floor. The restored grace is the shipped one.
_set_gates(ctx, fc, orig)
restored = True
c = _run_session(ctx, grbl, fc,
lambda c: c.get("verdict") == "OK" and not c.get("gates_off")
and all(g.get("state") == "ok" for g in (c.get("fan_gates") or {}).values()),
"restored", wait=INTAKE_TRIP_WAIT_S + 20)
ev["restored"] = c.get("fan_gates")
ctx.check(c.get("verdict") == "OK" and c.get("gates_off") == [],
"engine did not return to OK with every gate on after the restore: %s", c)
bad = {k: g for k, g in (c.get("fan_gates") or {}).items() if g.get("state") != "ok"}
ctx.check(not bad, "fans not at or above their floors after the restore: %s", bad)
finally:
if not restored:
st, body = fc.post("/settings", params=orig)
ctx.log("restore on failure: POST /settings %s -> %s", orig, st)
try:
c = _run_session(ctx, grbl, fc, lambda c: c.get("verdict") == "OK" and not c.get("gates_off"),
"restore on failure")
ctx.log("restore on failure: engine %s gates_off %s", c.get("verdict"), c.get("gates_off"))
except Exception as e: # the original failure is the one to report
ctx.log("restore on failure: run session did not complete (%s)", e)
after = fc.settings()
ctx.check(all(after.get(k, "") == orig[k] for k in INTAKE_KEYS),
"settings not restored: %s", {k: after.get(k) for k in INTAKE_KEYS})
ctx.check(fc.wait_idle(60, abort=ctx.aborted), "machine did not return to idle")
forgetest/forgetest/suite/extcore.py
-117
View File
@@ -1,117 +0,0 @@
# Copyright 2026 514 LLC d/b/a OpenGlow
# Written by Scott Wiederhold
# https://community.openglow.org
# SPDX-License-Identifier: MIT
"""What the extension host makes of the firmware's own version file.
A campaign only ever runs on a dev image - the release image does not
ship this suite - and a dev image stamps ${DATETIME} (dev), which is no
version at all. So the branch where a core range is actually judged
cannot be reached from here by any machine this suite runs on, and it is
not this module's to prove: that is forgeext's version_test, over the
file as each image build writes it. What is left here is the half a
machine can show, and it is the half that ties the daemon to the image:
that the host reads the image's own file, finds a build stamp, and
judges nothing rather than guessing.
Its own module rather than a test in exthost.py: a test's fingerprint
holds the shared text of the module it lives in, and the blank lines
between tests are shared text, so a test added to exthost.py moves the
fingerprint of every other test there and costs a campaign a re-run of
each - exthost.panel-install among them, which is an operator drill with
the button held. The helpers come from exthost, and importing them puts
that module's shared text in this test's fingerprint and not the other
way round.
"""
import json
import os
import subprocess
from ..catalog import test
from .exthost import FWUP, _forgeext, _read, _write
@test("exthost.core-range", title="The host reads the image's own version file and judges by it",
subsystem="exthost", kind="auto", est_min=2,
covers=[("forgeext", "src/install.*"), ("forgeext", "src/manifest.*"), ("forgeext", "src/main.c")],
description="The firmware's version reaches the host from /etc/forgefirm-version, which the image "
"build writes. This test asks the host what it made of that file rather than reading the "
"file and believing it. On the dev image a campaign runs on, the first word is a build "
"stamp and no version: inspect answers core_checked false, and a package asking for a "
"minimum far above or a maximum far below is accepted all the same, because a range "
"cannot be judged against a stamp and the host does not guess one. A dev image that "
"began stamping something version-shaped would fail here, which is the regression this "
"test is for; the judging branch belongs to forgeext's version_test, since no image "
"carrying this suite can reach it. Nothing is installed, nothing runs, and nothing "
"moves: inspect changes nothing and the work directory goes at the end.")
def core_range(ctx):
import io
import re
import shutil
import tarfile
import tempfile
ev = ctx.evidence
stamp = (_read("/etc/forgefirm-version", "").split() or [""])[0]
ctx.check(stamp, "/etc/forgefirm-version says nothing: the image did not stamp itself")
version = stamp[1:] if stamp[:1] == "v" and stamp[1:2].isdigit() else stamp
looks_like_a_version = bool(re.match(r"^\d+\.\d+\.\d+(-[0-9A-Za-z.-]+)?$", version))
ctx.log("/etc/forgefirm-version: %r -> %r", stamp, version)
ev["firmware_version"] = {"file": stamp, "version": version, "looks_like_a_version": looks_like_a_version}
# The suite ships on the dev image alone, so this is the only shape a
# campaign can meet. If it ever is not, the image changed and the
# judging path needs a proof that is not here.
ctx.check(not looks_like_a_version,
"this image stamped %r, which is a version: a campaign cannot run on a release image, so "
"either the dev image's stamp changed or this is not the image this suite belongs on", stamp)
work = tempfile.mkdtemp(prefix="forgetest-core-")
try:
def packed(core):
"""A data package that asks for nothing, with the core range given."""
m = {"manifest": 1, "id": "org.forgetest.corerange", "name": "forgetest core range",
"version": "1.0.0", "author": "forgetest", "license": "MIT", "api": "0.1",
"runtime": "data", "capabilities": []}
if core:
m["core"] = core
payload = os.path.join(work, "payload.tar.gz")
with tarfile.open(payload, "w:gz") as t:
data = json.dumps(m).encode()
info = tarfile.TarInfo("manifest.json")
info.size, info.mode = len(data), 0o644
t.addfile(info, io.BytesIO(data))
conf = os.path.join(work, "fwup.conf")
_write(conf, 'meta-product = "ForgeFIRM extension"\nmeta-description = "core range"\n'
'meta-version = "1.0.0"\nmeta-platform = "forgefirm-ext"\n'
'file-resource payload.tar.gz {\n host-path = "%s"\n}\n' % payload)
out = os.path.join(work, "corerange.ffx")
if os.path.exists(out):
os.remove(out)
subprocess.run([FWUP, "-c", "-f", conf, "-o", out], check=True, capture_output=True,
timeout=60, cwd=work)
return out
r = _forgeext("inspect", packed(None))
ctx.check(r.get("ok") is True, "a package with no core range was refused: %s", r.get("error"))
ctx.check(r.get("core_checked") is False,
"the host judged a range against the stamp %r: core_checked %s", stamp, r.get("core_checked"))
ev["no_range"] = r
# A stamp is no version, so neither end of a range is judged and
# neither refuses the package. The host says so rather than
# refusing on a comparison it cannot make.
ev["ranges"] = []
for core in ({"min": "99.0.0"}, {"max": "0.0.1"}, {"min": "0.0.1", "max": "0.0.2"}):
r = _forgeext("inspect", packed(core))
ctx.check(r.get("ok") is True and r.get("core_checked") is False,
"the host judged %s against the stamp anyway: %s", core, r)
ev["ranges"].append({"core": core, "inspect": r})
ctx.check("org.forgetest.corerange" not in
[x.get("id") for x in (_forgeext("list").get("packages") or [])],
"inspect installed the package: it must change nothing")
finally:
shutil.rmtree(work, ignore_errors=True)
forgetest/forgetest/suite/extdest.py
-237
View File
@@ -1,237 +0,0 @@
# Copyright 2026 514 LLC d/b/a OpenGlow
# Written by Scott Wiederhold
# https://community.openglow.org
# SPDX-License-Identifier: MIT
"""The destinations the operator names for a package that asks for them.
Its own module rather than a test in exthost.py, for the reason extcore.py
gives. The package here is the reference package's id and key with a
service of its own, so exthost's put-back takes it away like the others.
What it dials is the machine's own DNS resolver on TCP port 53: an address
every machine on a network has, that is not the machine, and that answers.
Named by address, it gives the service no resolver of its own, so the
only way through to it is the one the operator named.
"""
import json
import os
import socket
import subprocess
from ..catalog import test
from .exthost import (EXT_ROOT, FWUP, NFT, REF_ID, REF_KEY, _as_found, _forgeext, _put_back, _read, _svc, _tree,
_until, _write)
from .forgectrl import lan_ip
from .setup import SAFETY_PHRASE, read_file, record_path, request
# The service: at every start, where it may connect and what one dial does.
DEST_SERVICE = r'''
import errno, json, os, socket, sys, time
data, target, port = os.environ["FFX_DATA"], sys.argv[1], int(sys.argv[2])
def api(method, path):
s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
s.settimeout(5)
try:
s.connect(os.environ["FFX_API"])
s.sendall(("%s %s HTTP/1.1\r\nHost: forgeext\r\n\r\n" % (method, path)).encode())
buf = b""
while True:
c = s.recv(65536)
if not c:
break
buf += c
return json.loads(buf.partition(b"\r\n\r\n")[2] or b"null")
finally:
s.close()
def dial(p):
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.settimeout(4)
try:
s.connect((target, p))
return "connected"
except socket.timeout:
return "timeout"
except OSError as e:
return errno.errorcode.get(e.errno, str(e.errno))
finally:
s.close()
report = {"pid": os.getpid(), "destinations": (api("GET", "/v0/self") or {}).get("destinations"),
"dial": dial(port), "dial_other_port": dial(port + 1)}
with open(os.path.join(data, "report.json.new"), "w") as f:
json.dump(report, f)
os.rename(os.path.join(data, "report.json.new"), os.path.join(data, "report.json"))
print("destinations service up: %s" % report, flush=True)
while True:
time.sleep(60)
'''
def _resolver():
"""The first IPv4 nameserver of /etc/resolv.conf, or None."""
for line in _read("/etc/resolv.conf").splitlines():
parts = line.split()
if len(parts) >= 2 and parts[0] == "nameserver" and parts[1].count(".") == 3:
return parts[1]
return None
def _pack_dest(work, target, port):
"""The reference package's id and key, asking for the operator's destinations: (archive, public key)."""
import io
import tarfile
manifest = {"manifest": 1, "id": REF_ID, "name": "forgetest destinations", "version": "1.0.0",
"author": "forgetest", "license": "MIT", "api": "0.1", "runtime": "python",
"service": {"exec": "bin/dest.py", "args": [target, str(port)]},
"capabilities": ["net.outbound.operator"]}
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/dest.py", DEST_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, "dest.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"
@test("exthost.operator-destinations", title="The operator names where a package may connect, and takes it away",
subsystem="exthost", kind="auto", hardware="api", est_min=4,
covers=[("forgeext", "src/install.*"), ("forgeext", "src/state.*"), ("forgeext", "src/run.*"),
("forgeext", "src/super.*"), ("forgeext", "src/api.*"), ("forgeext", "src/caps.*"),
("forgeext", "src/manifest.*"), ("forgeext", "src/netrules.*"), ("forgeext", "src/main.c"),
("forgectrl", "src/extpkg.*"), ("forgectrl", "src/main.c")],
requires=["exthost.service"],
description="A package that asks for net.outbound.operator may reach the places the operator names for it "
"and nothing else. Installed, it has none: its own GET /v0/self names no destination and its "
"dial to the machine's DNS resolver on TCP port 53 is refused inside the sandbox (EACCES). "
"The operator names that resolver through POST /ext/dest; the host starts the service again "
"(said in its log, and no crash), its answer names the destination, its dial connects, the "
"next port up is still refused, and the account's chain in the rule table carries the "
"address. The machine's own LAN address and loopback are refused in words, a destination "
"out of form is 400, and taking away one that was never named is refused. Taken away, the "
"service starts again with no destination, the dial is refused again, and the chain no longer "
"carries the address. The package, the key, the setting, and the setup record are put back as "
"found.")
def operator_destinations(ctx):
import shutil
import tempfile
fc = ctx.forgectrl
ev = ctx.evidence
target = _resolver()
ctx.check(target, "/etc/resolv.conf names no IPv4 resolver: there is nothing to dial")
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.settimeout(4)
try:
s.connect((target, 53))
reachable = True
except OSError as e:
reachable = str(e)
finally:
s.close()
ev["resolver"] = {"address": target, "tcp_53_from_root": reachable}
ctx.check(reachable is True, "the resolver %s does not take a TCP connection on port 53 even from outside the "
"sandbox (%s): this network cannot show a way out opening", target, reachable)
ctx.check(fc.wait_idle(timeout=30, abort=ctx.aborted), "machine not idle: settings are locked")
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-dest.")
owner_key = os.path.join(EXT_ROOT, "keys", REF_KEY + ".pub")
report = os.path.join(EXT_ROOT, "data", REF_ID, "report.json")
where = "%s:53" % target
def seen():
try:
with open(report) as f:
return json.load(f)
except (OSError, ValueError):
return None
def fresh(than):
r = seen()
return r if r and r.get("pid") != than and _svc(REF_ID).get("state") == "running" else None
def chain_has(addr):
"""Does the package's own chain let it through to addr on TCP 53? No chain is no."""
acct = _svc(REF_ID).get("account") or ""
if not acct.startswith("ffx"):
return False
out = subprocess.run([NFT, "list", "chain", "inet", "ffx", "u%d" % (800 + int(acct[3:]))], capture_output=True,
text=True, timeout=30).stdout
return ("daddr %s tcp dport 53 accept" % addr) in out
try:
archive, pub = _pack_dest(work, target, 53)
shutil.copy(pub, owner_key)
os.chmod(owner_key, 0o644)
r = _forgeext("install", archive, "--consent-community")
ctx.check(r.get("ok") is True, "the install of a package asking for the operator's destinations -> %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)
first = _until(ctx, lambda: fresh(None), 60)
ctx.check(first, "the service did not come up: %s", _svc(REF_ID))
ev["none_named"] = first
ctx.log("none named: %s", first)
ctx.check(first.get("destinations") == [] and first.get("dial") == "EACCES",
"with nothing named it may reach nothing, and the sandbox refuses the dial: %s", first)
ctx.check(not chain_has(target), "the rule table carries %s before the operator named it", target)
# Refusals, in the host's words or the relay's.
refusals = []
for dest, want, words in (("%s:80" % lan_ip(), 409, "is this machine"), ("127.0.0.1:80", 409, "is this machine"),
("--root", 400, "host:port"), ("%s" % target, 409, "host:port")):
st_, why = fc.post("/ext/dest", data={"id": REF_ID, "action": "add", "dest": dest})
refusals.append([dest, st_, why])
ctx.check(st_ == want and words in json.dumps(why), "add %s -> %s %s, wanted %s %s", dest, st_, why, want, words)
st_, why = fc.post("/ext/dest", data={"id": REF_ID, "action": "remove", "dest": where})
refusals.append(["remove " + where, st_, why])
ctx.check(st_ == 409 and "is not one of" in json.dumps(why), "taking away one never named -> %s %s", st_, why)
ev["refusals"] = refusals
# Named: started again with its way out.
st_, doc = fc.post("/ext/dest", data={"id": REF_ID, "action": "add", "dest": where})
mine = [p for p in (doc or {}).get("packages", []) if p.get("id") == REF_ID] if isinstance(doc, dict) else []
ctx.check(st_ == 200 and mine and mine[0].get("destinations") == [where], "POST /ext/dest add %s -> %s %s",
where, st_, doc if st_ != 200 else mine)
named = _until(ctx, lambda: fresh(first.get("pid")), 60)
ev["named"] = named
ctx.log("named: %s", named)
ctx.check(named and named.get("destinations") == [where] and named.get("dial") == "connected"
and named.get("dial_other_port") == "EACCES",
"named, it is started again, names it, reaches it, and still not the next port: %s", named)
ctx.check(chain_has(target), "the rule table does not carry %s once named", target)
svc = _svc(REF_ID)
ctx.check(svc.get("state") == "running" and "quarantined" not in (svc.get("reason") or ""),
"the restart was no crash: %s", svc)
# Taken away: started again with none.
st_, doc = fc.post("/ext/dest", data={"id": REF_ID, "action": "remove", "dest": where})
ctx.check(st_ == 200, "POST /ext/dest remove %s -> %s %s", where, st_, doc)
gone = _until(ctx, lambda: fresh(named.get("pid") if named else None), 60)
ev["taken_away"] = gone
ctx.check(gone and gone.get("destinations") == [] and gone.get("dial") == "EACCES",
"taken away, it starts again and reaches nothing: %s", gone)
ctx.check(not chain_has(target), "the rule table still carries %s after it was taken away", target)
finally:
_put_back(ctx, fc, work, prior, etag, raw, dir_mode)
_as_found(ctx, fc, prior, raw, dir_mode, found_tree)
forgetest/forgetest/suite/exthost.py
+7 -192
View File
@@ -1214,13 +1214,14 @@ def _frozen(id_):
covers=[("forgeext", "src/super.*"), ("forgeext", "src/run.*"), ("forgeext", "src/machine.*"),
("forgeext", "src/cgroup.*"), ("forgectrl", "src/cool.*"), ("forgectrl", "src/cam.*"),
("forgectrl", "src/main.c")],
requires=["exthost.service", "cloud.dark-print"], actions=["button"],
requires=["exthost.service", "cooling.floor-and-warm-up"], actions=["button"],
steps=[OFFLINE_STEP,
"Bed clear (the job is dark: a 30 s square at S0, nothing fires). Press the button "
"when it lights."],
description="The reference package runs, its heartbeat advancing twice a second, and a dark "
"cloud print (cloud.dark-print's own job and checks) opens a real armed window over "
"it. Sampled five times a second from before the button to after the end: the "
"cloud print opens a real armed window over it: the cloud suite's dark-print "
"helper, whose latch checks still run (locked at the button, unlocked for the run, "
"locked after it), and the print completes. Sampled five times a second from before the button to after the end: the "
"engine's armed flag, the group's frozen state as the kernel reports it, and the "
"heartbeat. The window is open for at least 10 s. From 2 s after it opens to its "
"close the group reads frozen in every sample and the heartbeat does not move, the "
@@ -1350,195 +1351,9 @@ def armed_freeze(ctx):
# ------------------------------------------------------------- the hold
HOLDS_DIR = "/run/forgefirm/holds"
REQUIRED_HOLDS = EXT_ROOT + "/required-holds"
@test("exthost.hold-pause-tier", title="A package's hold withholds fire, and the operator's exits end it",
subsystem="exthost", kind="auto", hardware="takeover", est_min=7,
covers=[("forgeext", "src/holdkeep.*"), ("forgeext", "src/run.*"), ("forgeext", "src/install.*"),
("forgeext", "src/state.*"), ("forgeext", "src/main.c"), ("forgectrl", "src/holds.*"),
("forgectrl", "src/cool.*")],
requires=["exthost.service"],
description="The reference package is installed with the hold grant and its hold is marked required "
"(the package is then named under required-holds for forgectrl). Until its service has "
"run healthy (60 s) the required hold stands in the host's words; from then on the host "
"keeps its hold file fresh and clear and the engine's verdict is OK. The package then raises "
"its hold itself over its API socket (POST /v0/hold): EXT with the package's own words, "
"words the form does not take refused with the hold unchanged, and cleared again. The test then "
"makes the service end at every start: the host raises the hold in its own words and GET "
"/cool/status reads verdict EXT, fire_ok false, hold true, with the reason naming the "
"package. Safe mode ends the hold within two ticks and leaving it brings the hold back; "
"marked advisory the hold is dropped and the verdict is OK; marked required again, with "
"the host suspended the file goes stale and the reason becomes that the host is not "
"answering, and resumed it is the package's again; ext_enabled=0 ends it. Nothing moves "
"and nothing fires: the verdict is read at idle. Everything is put back as "
"exthost.service puts it back, and the required holds are empty at the end.")
def hold_pause_tier(ctx):
import tempfile
fc = ctx.forgectrl
ev = ctx.evidence
ctx.check(len(_host_pids()) == 1, "the extension host is not one running process: %s", _host_pids())
ctx.check(fc.wait_idle(timeout=30, abort=ctx.aborted), "machine not idle: settings are locked")
prior = fc.settings().get("ext_enabled") or ""
raw = read_file(record_path())
ctx.check(raw, "no setup record at %s", record_path())
ctx.check(not os.path.exists(SAFE_FILE), "%s exists: the machine is in safe mode", SAFE_FILE)
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")
ctx.check(st == 200 and etag, "GET /advisories/extensions -> %s", st)
def cool():
st_, c = fc.get("/cool/status")
return c if st_ == 200 and isinstance(c, dict) else {}
def verdict_is(name, seconds, reason=None):
def ok():
c = cool()
return c if c.get("verdict") == name and (reason is None or reason in (c.get("reason") or "")) else None
return _until(ctx, ok, seconds, poll=0.25)
def hold_file():
try:
h = json.loads(_read("%s/%s.json" % (HOLDS_DIR, REF_ID)))
except ValueError:
return None
h["age"] = time.monotonic() - h["ts_mono"]
return h
start = cool()
ev["verdict_at_start"] = {k: start.get(k) for k in ("verdict", "fire_ok", "hold", "reason")}
ctx.check(start.get("verdict") == "OK", "the engine's own verdict %r stands above a hold: the test needs OK",
start.get("verdict"))
work = tempfile.mkdtemp(prefix="forgetest-ffx.")
stop_path = os.path.join(EXT_ROOT, "data", REF_ID, "stop")
why_not_running = REF_ID + ": the extension is not running"
try:
archive, pub = _pack_reference(work, lan_ip(), more_caps=["hold"])
with open(pub, "rb") as f:
write_file(os.path.join(EXT_ROOT, "keys", REF_KEY + ".pub"), f.read())
os.chmod(os.path.join(EXT_ROOT, "keys", REF_KEY + ".pub"), 0o644)
r = _forgeext("install", archive, "--consent-community")
ev["install_without_the_grant"] = r.get("error")
ctx.check(r.get("ok") is False and "hold" in (r.get("error") or ""), "a hold was installed without the operator's grant: %s", r)
r = _forgeext("install", archive, "--consent-community", "--grant", "hold")
ctx.check(r.get("ok") is True, "the install -> %s", r.get("error"))
ctx.check(not os.listdir(REQUIRED_HOLDS), "a granted hold is required before the operator marks it: %s",
os.listdir(REQUIRED_HOLDS))
r = _forgeext("hold", REF_ID, "required")
ctx.check(r.get("ok") is True and os.listdir(REQUIRED_HOLDS) == [REF_ID], "marked required -> %s, named %s",
r.get("error"), os.listdir(REQUIRED_HOLDS))
st, reply = fc.post("/settings", data={"ext_enabled": "1", "advisory": etag, "phrase": SAFETY_PHRASE})
ctx.check(st == 200, "ext_enabled=1 over the advisory -> %s %r", st, reply)
beat_path = os.path.join(EXT_ROOT, "data", REF_ID, "beat")
seen = set()
ctx.check(_until(ctx, lambda: seen.add(_read(beat_path)) or len(seen) >= 3, 120, poll=0.2),
"the reference service's heartbeat does not advance: %s", _svc(REF_ID) or _host_status())
h = hold_file()
ev["running"] = {"hold_file": h, "verdict": cool().get("verdict"), "status_hold": _svc(REF_ID).get("hold")}
ctx.check(h and h["required"] is True and h["age"] < 1.5, "a running package's required hold is not fresh: %s", h)
# a required hold stands until its package has run healthy (60 s): one that ends at every
# start reads as running for a moment each time, and must not flicker clear
ctx.check(cool().get("verdict") == "EXT" and "only just started" in (cool().get("reason") or ""),
"a required hold is clear before its package has run healthy: %s", cool())
ctx.check(verdict_is("OK", 100), "with the package running healthy the verdict is %s (%s)", cool().get("verdict"),
cool().get("reason"))
h = hold_file()
ctx.check(h and h["raised"] is False and h["age"] < 1.5, "healthy: the hold file is not fresh and clear: %s", h)
ctx.check(_svc(REF_ID).get("hold") == "required", "the status file does not say the hold is required: %s", _svc(REF_ID))
# the package's own word, over its API socket: raised in its words, and cleared
def tell(words):
said = os.path.join(EXT_ROOT, "data", REF_ID, "said")
if os.path.exists(said):
os.remove(said)
_write(os.path.join(EXT_ROOT, "data", REF_ID, "say"), json.dumps(words))
ctx.check(_until(ctx, lambda: os.path.exists(said), 10), "the service did not pass %s on", words)
return json.loads(_read(said))
r = tell({"raised": True, "reason": "no badge presented"})
ctx.check(r[0] == 200 and r[1] == {"raised": True, "reason": "no badge presented"}, "POST /v0/hold raise -> %s", r)
t0 = time.time()
c = verdict_is("EXT", 6, REF_ID + ": no badge presented")
ev["raised_by_the_package"] = {"after_s": round(time.time() - t0, 1), "cool": {k: (c or cool()).get(k) for k in ("verdict", "fire_ok", "hold", "reason")}}
ctx.log("the package raised its hold: %s", ev["raised_by_the_package"])
ctx.check(c and c.get("fire_ok") is False and c.get("hold") is True, "the package's own hold does not withhold fire: %s", c or cool())
r = tell({"raised": True, "reason": "a \"quote\""})
ctx.check(r[0] == 400 and verdict_is("EXT", 3, REF_ID + ": no badge presented"), "words the form does not take -> %s, and the "
"hold now reads %s", r, cool().get("reason"))
r = tell({"raised": False})
ctx.check(r[0] == 200 and r[1].get("raised") is False and verdict_is("OK", 6), "POST /v0/hold clear -> %s, verdict %s", r,
cool().get("verdict"))
# the package can no longer speak for itself
_write(stop_path, "")
os.kill(_svc(REF_ID)["pid"], signal.SIGKILL)
t0 = time.time()
c = verdict_is("EXT", 40, why_not_running)
ev["held"] = {"after_s": round(time.time() - t0, 1), "cool": {k: (c or {}).get(k) for k in ("verdict", "fire_ok", "hold", "reason")},
"hold_file": hold_file()}
ctx.log("the service ends at every start: %s", ev["held"])
ctx.check(c and c.get("fire_ok") is False and c.get("hold") is True,
"the hold does not withhold fire: %s", c or cool())
# it reads as running for a moment at every start on its way to quarantine: never clear
looks = []
for _ in range(24):
ctx.sleep(0.5)
c = cool()
looks.append((c.get("verdict"), c.get("fire_ok")))
ev["looks_over_the_crash_loop"] = sorted(set(looks), key=str)
ctx.check(all(v == "EXT" and ok is False for v, ok in looks),
"the required hold flickered clear while its package kept ending: %s", ev["looks_over_the_crash_loop"])
# the operator's exits
_write(SAFE_FILE, "")
t0 = time.time()
ctx.check(verdict_is("OK", 4), "safe mode did not end the hold within four seconds: %s", cool())
ev["safe_mode_released_s"] = round(time.time() - t0, 1)
os.remove(SAFE_FILE)
ctx.check(verdict_is("EXT", 30, why_not_running), "out of safe mode the hold did not come back: %s", cool())
r = _forgeext("hold", REF_ID, "advisory")
ctx.check(r.get("ok") is True and not os.listdir(REQUIRED_HOLDS), "marked advisory -> %s, still named %s",
r.get("error"), os.listdir(REQUIRED_HOLDS))
ctx.check(verdict_is("OK", 6), "an advisory hold of a package that is not running was not dropped: %s", cool())
r = _forgeext("hold", REF_ID, "required")
ctx.check(r.get("ok") is True and verdict_is("EXT", 6, why_not_running), "marked required again: %s", cool())
# the host itself gone quiet: the file goes stale, and that is what stands
host = _host_pids()[0]
os.kill(host, signal.SIGSTOP)
try:
c = verdict_is("EXT", 8, REF_ID + ": the extension host is not answering")
ev["host_suspended"] = {k: (c or cool()).get(k) for k in ("verdict", "fire_ok", "reason")}
ctx.check(c and c.get("fire_ok") is False, "with the host suspended the required hold does not stand on its "
"stale file: %s", c or cool())
finally:
os.kill(host, signal.SIGCONT)
ctx.check(verdict_is("EXT", 8, why_not_running), "the host resumed: the hold is not the package's again: %s", cool())
st, reply = fc.post("/settings", data={"ext_enabled": "0"})
ctx.check(st == 200, "ext_enabled=0 -> %s %r", st, reply)
t0 = time.time()
ctx.check(verdict_is("OK", 4), "ext_enabled=0 did not end the hold within four seconds: %s", cool())
ev["ext_off_released_s"] = round(time.time() - t0, 1)
ctx.log("PASS: held in %s s; safe mode released it in %s s, extensions off in %s s", ev["held"]["after_s"],
ev["safe_mode_released_s"], ev["ext_off_released_s"])
finally:
for pid in _host_pids():
try:
os.kill(pid, signal.SIGCONT)
except OSError:
pass
_put_back(ctx, fc, work, prior, etag, raw, dir_mode)
_as_found(ctx, fc, prior, raw, dir_mode, found_tree)
ctx.check(not os.listdir(REQUIRED_HOLDS), "a required hold is still named at the end: %s", os.listdir(REQUIRED_HOLDS))
ctx.check(_until(ctx, lambda: cool().get("verdict") == "OK", 6), "the verdict at the end is %s", cool().get("verdict"))
# ------------------------------------------------- the operator's door
@test("exthost.events", title="The machine's events reach a package, over one subscription",
@@ -1932,8 +1747,8 @@ def ui_delivery(ctx):
"package, under the machine lease, and the machine's own job record names it as the "
"owner and counts the program's lines. A dark program commands no laser, so no armed "
"window opens and there is nothing for a press to arm: the arm gates are in force and "
"are never reached, and a job that fires is cloud.dark-print's and "
"laser.emission-witness's. Every line is sent and acknowledged, and the runner ends the "
"are never reached, and a job that fires is laser.emission-witness's. "
"Every line is sent and acknowledged, and the runner ends the "
"job as one that ended without a discharge - which is what a program that commands no "
"laser is, and is the outcome this test wants. The job's own emission witnesses are all "
"zero. The head accelerometer sees the motion and the "
@@ -2003,7 +1818,7 @@ def motion_job(ctx):
# it as this package. A dark program commands no laser, so no
# armed window opens and there is nothing for a press to arm -
# the arm gates are in force and simply never reached. A job
# that fires is cloud.dark-print's and laser.emission-witness's.
# that fires is laser.emission-witness's.
def job_rec():
st_, j_ = fc.get("/job")
return j_ if isinstance(j_, dict) else {}
forgetest/forgetest/suite/extlife.py
-154
View File
@@ -1,154 +0,0 @@
# Copyright 2026 514 LLC d/b/a OpenGlow
# Written by Scott Wiederhold
# https://community.openglow.org
# SPDX-License-Identifier: MIT
"""The host's own events: a service is told before it is stopped.
Its own module, for the reason extcore.py gives. The package here is the
reference package's id and key with a service of its own, so exthost's
put-back takes it away like the others.
"""
import json
import os
import subprocess
import time
from ..catalog import test
from .exthost import (EXT_ROOT, FWUP, REF_ID, REF_KEY, _as_found, _forgeext, _put_back, _svc, _tree, _until, _write)
from .setup import SAFETY_PHRASE, read_file, record_path, request
# The service: it follows the host's event feed and keeps every host event it reads, with the
# monotonic time it read it, in its data directory.
LIFE_SERVICE = r'''
import json, os, socket, time
data = os.environ["FFX_DATA"]
def api(body):
s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
s.settimeout(40)
try:
s.connect(os.environ["FFX_API"])
payload = json.dumps(body).encode()
s.sendall(b"POST /v0/events HTTP/1.1\r\nHost: forgeext\r\nContent-Type: application/json\r\n"
b"Content-Length: %d\r\n\r\n" % len(payload) + payload)
buf = b""
while True:
c = s.recv(65536)
if not c:
break
buf += c
return json.loads(buf.partition(b"\r\n\r\n")[2] or b"null")
finally:
s.close()
def keep(name, doc):
with open(os.path.join(data, name + ".new"), "w") as f:
json.dump(doc, f)
os.rename(os.path.join(data, name + ".new"), os.path.join(data, name))
place = api({})["next"]
keep("ready.json", {"place": place, "t": time.monotonic()})
got = []
while True:
ans = api({"since": place, "wait": 20})
for e in ans.get("events") or []:
if e.get("event", "").startswith("ext."):
got.append({"event": e["event"], "data": e.get("data"), "t": time.monotonic()})
keep("host-events.json", got)
place = ans.get("next", place)
'''
def _pack_life(work):
import io
import tarfile
manifest = {"manifest": 1, "id": REF_ID, "name": "forgetest lifecycle", "version": "1.0.0",
"author": "forgetest", "license": "MIT", "api": "0.1", "runtime": "python",
"service": {"exec": "bin/life.py"}, "capabilities": ["events"]}
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/life.py", LIFE_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, "life.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"
@test("exthost.lifecycle", title="A package is told, in the feed it reads, before the host stops it",
subsystem="exthost", kind="auto", hardware="api", est_min=3,
covers=[("forgeext", "src/run.*"), ("forgeext", "src/machine.*"), ("forgeext", "src/evfeed.*"),
("forgeext", "src/api.*")],
requires=["exthost.service"],
description="A package that holds events reads the host's own events beside the machine's. With the "
"package running and following its feed, extensions are turned off: the host puts "
"ext.shutdown into the feed with the reason (\"extensions are off (ext_enabled)\"), and "
"stops the service a second later, so that the package has read it - the event is in its "
"data directory, read before its process ended. Holds do not wait for that second: they "
"follow extensions off at once, as exthost.hold-pause-tier holds. ext.will_freeze and "
"ext.thawed, which an armed window brings, are the automation package's host test's "
"(automation_test.py, where a service with job_time.run reads both). The package, the "
"key, the setting, and the setup record are put back as found.")
def lifecycle(ctx):
import shutil
import tempfile
fc = ctx.forgectrl
ev = ctx.evidence
ctx.check(fc.wait_idle(timeout=30, abort=ctx.aborted), "machine not idle: settings are locked")
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-life.")
owner_key = os.path.join(EXT_ROOT, "keys", REF_KEY + ".pub")
data = os.path.join(EXT_ROOT, "data", REF_ID)
def doc(name):
try:
with open(os.path.join(data, name)) as f:
return json.load(f)
except (OSError, ValueError):
return None
try:
archive, pub = _pack_life(work)
shutil.copy(pub, owner_key)
os.chmod(owner_key, 0o644)
r = _forgeext("install", archive, "--consent-community")
ctx.check(r.get("ok") is True, "the install -> %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)
ready = _until(ctx, lambda: _svc(REF_ID).get("state") == "running" and doc("ready.json"), 60)
ctx.check(ready, "the service did not come up and follow its feed: %s", _svc(REF_ID))
time.sleep(1.0)
pid = _svc(REF_ID).get("pid")
st, reply = fc.post("/settings", data={"ext_enabled": "0"})
ctx.check(st == 200, "ext_enabled=0 -> %s %r", st, reply)
stopped = _until(ctx, lambda: _svc(REF_ID).get("state") != "running" and time.monotonic(), 20)
ctx.check(stopped, "the service was not stopped: %s", _svc(REF_ID))
ctx.check(not os.path.exists("/proc/%s" % pid), "its process %s outlived the stop", pid)
got = doc("host-events.json") or []
ev["host_events"] = got
ctx.log("what it read: %s", got)
ctx.check(len(got) >= 1 and got[0].get("event") == "ext.shutdown"
and (got[0].get("data") or {}).get("reason") == "extensions are off (ext_enabled)",
"it read ext.shutdown, with the reason, before it was stopped: %s", got)
finally:
_put_back(ctx, fc, work, prior, etag, raw, dir_mode)
_as_found(ctx, fc, prior, raw, dir_mode, found_tree)
forgetest/forgetest/suite/extmcode.py
-238
View File
@@ -1,238 +0,0 @@
# 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)
forgetest/forgetest/suite/homeoff.py
+7 -80
View File
@@ -3,13 +3,12 @@
# https://community.openglow.org
# SPDX-License-Identifier: MIT
"""A camera home's offset pair, on the machine.
"""The helpers for a test that lets the web service move the head.
Its own module, so that no other test's fingerprint moves. The homing is
the web-service session cloud.mode-switch runs. The helpers for a test
that lets the service move the head live here too: the motions ran whole,
and the head goes back to where the test found it, by the travel the
service's motions logged. No counter reading can say where that was: every
The homing is the web-service session cloud.mode-switch runs. These judge
that its motions ran whole, and take the head back to where the test found
it by the travel the service's motions logged. No counter reading can say
where that was: every
service motion zeroes the step counters at its start, and a home and every
controller start zero them again.
"""
@@ -18,13 +17,11 @@ import re
import sys
from ..baseline import XY_STEPS_PER_MM, counter_frame, counter_steps_per_mm, read_position
from ..catalog import test
from ..runner import Failed
from . import cloud as logs # the log paths, read at each call
from .cloud import _HOMING_PATH, gfhome_homing, log_lines_since, log_size
from .motion import _drop_reference, clean_slate, machine_idle, wait_idle, wait_state
from .cloud import log_lines_since, log_size
from .motion import _drop_reference, clean_slate, machine_idle, wait_idle
HOME_X, HOME_Y = 4.5, -3.25
RETURN_MAX_MM = 600.0 # no hand-back jogs farther on an axis: the bed is smaller
# The client's own record of a motion's end: the kernel counters, which
# the motion zeroed at its start (the "(actual/expected)" line of the same
@@ -169,73 +166,3 @@ def _jog_back(ctx, rec, travel, since):
_drop_reference(ctx, ctx.forgectrl)
ctx.counters_rezeroed()
ctx.log("head returned to where the test found it (%.3f/%.3f mm)", -dx, -dy)
@test("homing.cloud-offsets", title="A camera home declares its offset pair, and the envelope reaches back to a "
"negative one",
subsystem="homing", kind="auto", mode="grbl", est_min=7,
covers=_HOMING_PATH + [("grblhal-glowforge", "src/glowforge_homing.*"),
("grblhal-glowforge", "src/glowforge_cooling.*")],
requires=["forgectrl.auth", "motion.pacing"],
steps=["Bed clear, lid closed; cloud credentials configured; the machine on the network. The head "
"travels to the camera home and back to where it was."],
description="With cloud mode on, homing_mode = gfcloud and the camera-home offsets set to (4.5, -3.25), "
"$H runs the web-service homing session. Every service motion of the session runs whole: "
"the runner reports to the cooling engine in the controller's place, and one stopped short "
"(the engine's stop, a fault) leaves the head short of the home and the position declared "
"there false. When the session ends, the controller declares the position (4.5, -3.25): a "
"camera-home offset may be negative, and the declared position is snapped to the step "
"grid. The work envelope reaches back to the home on the negative axis, since the head "
"stands there: a jog 2 mm out in +Y and 2 mm back ends at the home, accepted, and the same "
"in X. The homing mode and the offsets are put back as found, the camera home is dropped, "
"and the head is jogged back to where the test found it by the travel the session's "
"motions logged; a travel that cannot be known fails the test and moves nothing. That an "
"offset past the axis travel is refused, and the envelope's edge on the grid, are the "
"driver's unit and harness cases.")
def cloud_offsets(ctx):
fc = ctx.forgectrl
ev = ctx.evidence
s0 = fc.settings() or {}
ctx.check(s0.get("cloud_enabled") == "1", "cloud mode is off: a camera home needs it on")
found = {k: s0.get(k) or "" for k in ("homing_mode", "gfcloud_home_x", "gfcloud_home_y")}
ev["found"] = found
session_at = None
try:
for k, v in (("homing_mode", "gfcloud"), ("gfcloud_home_x", str(HOME_X)), ("gfcloud_home_y", str(HOME_Y))):
st, body = fc.post("/settings", data={k: v})
ctx.check(st == 200, "%s=%s -> %s %s", k, v, st, body)
with ctx.grbl() as g:
session_at = session_mark()
gfhome_homing(ctx, ev, g)
judge_whole_motions(ctx, ev, session_at)
mpos = g.status_report().get("MPos") or (None, None, None)
ev["declared"] = mpos
ctx.log("the camera home declared %s", mpos)
ctx.check(mpos[0] is not None and abs(mpos[0] - HOME_X) < 0.02 and abs(mpos[1] - HOME_Y) < 0.02,
"the camera home did not declare (%.2f, %.2f): %s", HOME_X, HOME_Y, mpos)
jogs = {}
for axis, home in (("Y", HOME_Y), ("X", HOME_X)):
out = g.command("$J=G91 G21 %s2 F1000" % axis, timeout=10)
wait_state(ctx, g, "Idle", 20)
back = g.command("$J=G91 G21 %s-2 F1000" % axis, timeout=10)
wait_state(ctx, g, "Idle", 20)
at = g.status_report().get("MPos")
jogs[axis] = {"out": out[-1:], "back": back[-1:], "at": at}
ctx.check(out[-1:] == ["ok"] and back[-1:] == ["ok"],
"a jog out and back to the home in %s was refused: %s", axis, jogs[axis])
i = "XY".index(axis)
ctx.check(at and abs(at[i] - home) < 0.02, "the jog back in %s did not end at the home: %s", axis, at)
ev["jogs"] = jogs
ctx.log("out and back to the home: %s", jogs)
finally:
# The settings are refused for a moment after the suite's Grbl client closes; each is put back when taken.
for k, v in found.items():
def put():
st_, body_ = (fc.post("/settings", params={k: ""}) if v == "" else fc.post("/settings", data={k: v}))
if st_ != 200:
ctx.log("restore %s=%r -> %s %s", k, v, st_, body_)
return st_ == 200 or None
took = ctx.wait_for(put, 20, poll=0.5)
ctx.log("restore %s=%r -> %s", k, v, "taken" if took is not None else "not taken")
if session_at is not None:
camera_home_return(ctx, ev, session_at)
forgetest/forgetest/suite/laser.py
+1 -248
View File
@@ -30,7 +30,7 @@ from ..catalog import test
from .. import hw
from ..runner import Failed
from .motion import (kernel_xy_mm, kernel_start, check_kernel_returned, wait_state,
wait_state_text, wait_left_state, wait_idle, drain_text, Watch, machine_idle,
wait_state_text, wait_left_state, drain_text, Watch, machine_idle,
_job, _job_post, _job_wait, _lease_holder, _port_state, _words)
_LASER_COVERS = [("grblhal-glowforge", "src/**"), ("kernel-module-glowforge", "**"),
@@ -652,86 +652,6 @@ def m5_rapid_dark(ctx):
peak, hv_tail)
@test("laser.disarm-in-hold", title="Disarm grace counts down in Hold", subsystem="laser",
kind="live", mode="grbl", est_min=4,
covers=_LASER_COVERS,
requires=["laser.emission-witness"], actions=["button"],
steps=["Scrap under the head with 40 mm of free +X travel; lid closed; exhaust on.",
"Press the physical button when it lights white. Nothing else: the test holds the "
"job itself and waits about a minute for the disarm."],
description="Arm and start a +X move at S400/F300, feed-hold it after ~2 s of motion, and "
"hold: the disarm grace must count down while held and close the armed window "
"(armed -> false) without the job resuming, and the button goes dark with it.")
def disarm_in_hold(ctx):
ev = ctx.evidence
with ctx.grbl() as g, LiveJob(ctx, g):
prepare(ctx, g)
k0 = kernel_start(ctx)
ctx.ready(ARM_CUE % "40 mm +X")
stream(g, ["G91", "G21", "M4", "S400", "G1 X40 F300"])
ctx.log("armed; waiting for motion to start (arm + your button press)...")
w = Watch(g)
# The press waits for the button to light: pressed before the job
# reaches its arm wait, it is lost, and the move never starts.
ctx.arm_press()
ctx.wait_for(w.in_state("Run"), 180)
ctx.clear_notice()
st = w.last["state"] if w.last else None
ctx.check(st and st.startswith("Run"), "motion never started (state=%s) - arm refused or no press", st)
ctx.log("moving under laser: %s; feed-hold in 2 s", st)
ctx.sleep(2)
g.realtime(ord("!"))
t1 = time.time()
while time.time() - t1 < 5:
st = g.status_report()["state"]
if st.startswith("Hold"):
break
time.sleep(0.1)
ev["held_state"] = st
ctx.log("feed-held mid-move: %s; watching the disarm grace count down IN HOLD", st)
ctx.check(st.startswith("Hold"), "feed hold did not park (state %s)", st)
t0 = time.time()
disarmed_at = None
left_hold = None
while time.time() - t0 < 120:
ctx.checkpoint()
s = sample(ctx)
held = g.status_report()["state"].startswith("Hold")
if s and not s["armed"]:
disarmed_at = time.time() - t0
break
if not held and left_hold is None:
left_hold = g.status_report()["state"]
ctx.log("note: left Hold (state=%s) before disarm", left_hold)
time.sleep(1)
ev["disarmed_after_s"] = round(disarmed_at, 1) if disarmed_at is not None else None
ev["left_hold"] = left_hold
# recover: laser off, abort out of hold
g.command("M5", timeout=1)
g.realtime(0x18)
ctx.sleep(1)
if "Alarm" in g.status_report()["state"]:
g.command("$X")
# The hold parks the head part-way through the move and the reset
# abandons the rest of it, so the job ends wherever the feed hold
# caught it - about 11 mm in, on the bench reference. That is the
# test's own doing and it goes back: the kernel counters say how
# far the head actually went, which is what a jog has to undo.
wait_idle(ctx, g, 15)
dx, dy = kernel_xy_mm(ctx)
back_x, back_y = k0[0] - dx, k0[1] - dy
ev["left_at_mm"] = [round(dx - k0[0], 3), round(dy - k0[1], 3)]
if abs(back_x) >= 0.02 or abs(back_y) >= 0.02:
ctx.log("returning the head: %+.3f/%+.3f mm", back_x, back_y)
g.command("$J=G91X%.3fY%.3fF1200" % (back_x, back_y))
wait_idle(ctx, g, 30)
ev["returned_mm"] = list(kernel_xy_mm(ctx))
ctx.check(disarmed_at is not None, "still armed after 120 s in Hold")
ctx.check(left_hold is None, "the job left Hold (%s) before the disarm", left_hold)
check_button_dark(ctx, ev)
ctx.log("PASS: disarmed in Hold after %.1f s, button dark", disarmed_at)
@test("laser.armed-kill", title="Armed kill mid-fire: the expected stop, then a SIGKILL",
subsystem="laser", kind="live", mode="grbl", est_min=6,
covers=_LASER_COVERS + [("forgectrl", "src/main.c")],
@@ -1090,78 +1010,6 @@ def verdict_cut(ctx):
DISARMED_MSG = "laser disarmed - latch locked"
@test("laser.arm-wait-lid", title="Lid open during the arm wait cancels the job",
subsystem="laser", kind="operator", mode="grbl", est_min=3,
covers=_LASER_COVERS + [("grblhal-glowforge", "src/glowforge_laser.c"),
("grblhal-glowforge", "src/glowforge_switches.c"),
("grblhal-glowforge", "src/glowforge_switch_map.h")],
requires=["kernel.latch-locked-idle", "motion.jog-roundtrip"], actions=["lid"],
steps=["Lid closed; nothing under the head needs to be in place - the machine will not fire.",
"When the white button lights, do NOT press it: open the lid instead, then close it "
"when told."],
description="Start a laser job so the controller unlocks the latch and lights the button, "
"then open the lid while it waits. The wait must abort with the lid named as the "
"reason, canceled with a soft reset (no alarm - nothing to unlock), the armed "
"window closed (armed -> false), the kernel latch back to locked, and no emission; "
"the controller ends Idle. No press is given, so nothing can fire.")
def arm_wait_lid(ctx):
ev = ctx.evidence
with ctx.grbl() as g, LiveJob(ctx, g):
prepare(ctx, g)
base = sample(ctx)
ctx.check(base, "forgectrl /status or /cool/status unavailable")
ctx.check(not base["armed"], "armed window already open before the job")
stream(g, ["G91", "G21", "M4", "S400", "G1 X5 F600"])
# The prompt: the latch is unlocked and the button lit from here on.
t0 = time.time()
text = ""
while time.time() - t0 < 30 and "press the button" not in text:
ctx.checkpoint()
text += g.drain()
time.sleep(0.1)
ctx.check("press the button" in text, "no arm prompt within 30 s (job did not reach the arm)")
s = sample(ctx)
ev["armed_during_wait"] = s["armed"] if s else None
ctx.log("arm prompt seen; armed=%s; asking the operator to open the lid", ev["armed_during_wait"])
ctx.act("lid", "open", text="The button is lit white: do NOT press it.", timeout=60)
t1 = time.time()
while time.time() - t1 < 15:
ctx.checkpoint()
text += g.drain()
if "job canceled" in text and "help]" in text:
break
time.sleep(0.1)
ev["messages"] = [ln for ln in text.splitlines() if ln.startswith("[MSG:") or ln.startswith("ALARM")]
ctx.log("controller: %s", ev["messages"])
ctx.check("lid opened during arm - job canceled" in text,
"the lid open was not reported as canceling the arm")
ctx.check("help]" in text, "no reset banner after the lid-open cancel (it must be a clean cancel)")
ctx.check("ALARM" not in text, "an alarm was raised on the lid-open cancel")
# Armed window closed and the kernel latch locked.
t2 = time.time()
s = None
while time.time() - t2 < 10:
s = sample(ctx)
if s and not s["armed"]:
break
time.sleep(0.25)
ev["armed_after"] = s["armed"] if s else None
ilk = hw.sysfs_int("cnc/interlock_circuit")
locked = ilk is not None and bool(ilk & (1 << 3))
ev["latch_locked"] = locked
ev["emission"] = s["emission"] if s else None
ctx.check(s and not s["armed"], "the armed window stayed open after the lid-open cancel")
ctx.check(locked, "kernel latch not locked after the lid-open cancel (interlock_circuit=%s)", ilk)
ctx.check(not ev["emission"], "emission_samples nonzero (%s) - nothing may have fired", ev["emission"])
check_button_dark(ctx, ev)
ctx.act("lid", "close")
ctx.sleep(1)
st = g.status_report()["state"]
ev["state_after"] = st
ctx.check(st.startswith("Idle"), "controller is %s after the cancel, expected Idle (no unlock needed)", st)
ctx.log("PASS: lid open during the arm wait canceled the job (clean reset, no alarm), armed=false, latch locked")
@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", mode="grbl", est_min=7,
@@ -1327,101 +1175,6 @@ def pause_resume_lid_cancel(ctx):
"%.3f mm), button latch SET", zero_at, drift)
def _port_jog_sampled(ctx, fc, params, samples, timeout=30.0):
"""One jog through the controller port, sampled at 8 Hz until the port
says it is over. Returns the HTTP status and body of the jog request."""
st, body = fc.post("/motion/jog", params=params)
t0 = time.time()
while time.time() - t0 < timeout:
ctx.checkpoint()
smp = sample(ctx)
if smp:
samples.append(smp)
try:
ps = fc.get("/motion/state")[1]
except hw.HwError:
ps = None
if st != 200 or (isinstance(ps, dict) and ps.get("state") == "Idle" and not ps.get("port_jog")):
break
time.sleep(0.125)
return st, body
@test("laser.port-dark", title="A panel jog under an open armed window ships dark",
subsystem="laser", kind="live", mode="grbl", est_min=3,
covers=_LASER_COVERS + [("forgectrl", "src/grblport.*")],
requires=["laser.emission-witness", "motion.port-jog"], actions=["button"],
steps=["Scrap under the head with 20 mm of free +X travel; lid closed; exhaust on.",
"Press the physical button when it lights white (the arm)."],
description="A 20 mm line at M3 S400/F600 with no M5 and no program end after it, so the "
"armed window stays open with M3 modal and S above zero: the state in which an "
"injected G1 would fire. The line itself must be witnessed lit, or the case "
"proves nothing. Then POST /motion/jog takes the head back over the line and out "
"again, through the controller port. The window must read armed before and "
"after the jogs, and through both the kernel's LASER_ON sample count stays 0, "
"the HV current stays at its idle reading, and the head's beam detector does "
"not rise over its level before the jogs. The client then ends the job (M5, "
"M2) and the window closes.")
def port_dark(ctx):
ev = ctx.evidence
fc = ctx.forgectrl
with ctx.grbl() as g, LiveJob(ctx, g):
prepare(ctx, g)
base = sample(ctx)
ctx.check(base, "forgectrl /status or /cool/status unavailable")
ctx.check(not base["emission"], "emission_samples nonzero before the job (%s)", base["emission"])
ctx.ready(ARM_CUE % "20 mm +X")
ctx.arm_press()
jog_samples = []
try:
# No M5 and no M2: the window stays open, M3 S400 stays modal.
lit = run_and_sample(ctx, g, ["G91", "G21", "M3", "S400", "G1 X20 F600"])
peak = max((s["emission"] or 0 for s in lit), default=0)
ctx.check(peak > 0, "no emission witnessed on the G1: the dark jogs would prove nothing")
# The cut is over and dark again; the window is still open.
ok = ctx.wait_for(lambda: (sample(ctx) or {}).get("emission") == 0, 10, poll=0.125)
ctx.check(ok is not None, "emission_samples did not return to 0 after the G1")
before = sample(ctx)
ctx.check(before and before["armed"], "the armed window closed before the jogs: %s",
before and before["phase"])
ctx.sleep(0.6) # past the port's sender-quiet interval
for params in ({"x": "-20", "feed": "600"}, {"x": "20", "feed": "2400"},
{"x": "-20", "feed": "2400"}):
st, body = _port_jog_sampled(ctx, fc, params, jog_samples)
ctx.check(st == 200, "POST /motion/jog %s under the open window -> %s %s", params, st, body)
ctx.sleep(0.6)
after = sample(ctx)
ctx.check(after and after["armed"],
"the armed window closed during the jogs: the dark samples were not under it")
finally:
ctx.clear_notice()
r = g.command("M5")
r += g.command("M2")
ctx.check(not any(x.startswith("error") for x in r), "the client could not end the job: %s", r)
ev["disarm_s"] = wait_disarm(ctx, 30)
ctx.check(ev["disarm_s"] is not None, "the armed window did not close after the program end")
lit_in_jog = [s for s in jog_samples if s["emission"]]
hv_max = max((s["hv"] for s in jog_samples if s["hv"] is not None), default=0)
beams = [s["beam"] for s in jog_samples if s.get("beam") is not None]
beam_rise = (max(beams) - before["beam"]) if beams and before.get("beam") is not None else None
ev.update({"emission_peak_g1": peak, "jog_samples": len(jog_samples),
"jog_emission_samples": len(lit_in_jog), "jog_hv_max": hv_max,
"jog_beam_rise": beam_rise, "trail": trail(jog_samples)})
ctx.log("G1 emission peak %s; %d samples across three port jogs: %d with emission, HV max %s, "
"beam detector rise %s", peak, len(jog_samples), len(lit_in_jog), hv_max, beam_rise)
ctx.check(len(jog_samples) >= 16, "only %d samples across the jogs", len(jog_samples))
ctx.check(not lit_in_jog, "a port jog fired under the open window: %d samples with emission, "
"first emission_samples %s", len(lit_in_jog), lit_in_jog[0]["emission"] if lit_in_jog else None)
ctx.check(hv_max <= HV_DARK_MAX, "HV current %s during a port jog (idle reads ~0)", hv_max)
ctx.check(beam_rise is not None and beam_rise < BEAM_DELTA_MIN,
"the head's beam detector rose %s during a port jog", beam_rise)
ctx.log("PASS: lit on the G1 (peak %s), and three port jogs under the same open window with M3 S400 "
"modal shipped dark by all three witnesses", peak)
# ---------------------------------------------------------------- the recorder, as a job of the runner's
def _laser_keys(fc):
forgetest/forgetest/suite/motion.py
-324
View File
@@ -1146,192 +1146,6 @@ def homing_kill(ctx, wait_running):
ctx.log("restore homing_mode=%r -> %s", hm, st)
@test("motion.respawn-gate", title="A respawn waits for the lid the way a first spawn does",
subsystem="motion", kind="operator", mode="grbl", est_min=2,
covers=_MOTION_COVERS, requires=["motion.deadman", "motion.gate-waits-for-lid"], actions=["lid"],
steps=["Bed clear. Open the lid when told and close it when told; nothing moves."],
description="The enclosure check runs before every controller spawn, respawns included. "
"With the lid open, the controller is killed: the supervisor safes the machine "
"(latch locked), reports waiting with why naming the lid, and starts no controller "
"while the lid stays open. When the lid closes the controller comes back verified, "
"without a second motion probe (the probe is once per broker hold).")
def respawn_gate(ctx):
import os as _os
import signal as _signal
fc = ctx.forgectrl
ev = ctx.evidence
ctx.check(fc.wait_idle(15, abort=ctx.aborted), "machine not idle at the start")
st, m0 = fc.get("/mode")
ctx.check(st == 200 and isinstance(m0, dict) and m0.get("controller") == "running", "controller not running: %s", m0)
pid0 = m0.get("pid")
off = _log_offset(FORGECTRL_LOG)
ctx.act("lid", "open")
try:
ctx.sleep(1.0)
_os.kill(pid0, _signal.SIGKILL)
t_kill = time.time()
ctx.log("SIGKILL sent to controller pid %d with the lid open", pid0)
waited = ctx.wait_for(lambda: (fc.get("/mode")[1] or {}).get("controller") == "waiting", 15)
st, m1 = fc.get("/mode")
ev["after_kill"] = {"waiting_s": waited, "mode": m1}
ctx.log("mode %.1f s after the kill: %s", time.time() - t_kill, m1)
ctx.check(waited is not None, "the supervisor did not report waiting for the lid: %s", m1)
ctx.check("lid" in (m1.get("why") or ""), "why does not name the lid: %r", m1.get("why"))
ilk = hw.sysfs_int("cnc/interlock_circuit")
ctx.check(ilk is not None and ilk & (1 << 3), "latch not locked after the kill")
ctx.sleep(4.0)
st, m2 = fc.get("/mode")
ctx.check(isinstance(m2, dict) and m2.get("controller") == "waiting" and not m2.get("pid"),
"the wait did not hold with the lid open: %s", m2)
finally:
ctx.act("lid", "close")
t_close = time.time()
came = ctx.wait_for(lambda: ((fc.get("/mode")[1] or {}).get("controller") == "running"
and (fc.get("/mode")[1] or {}).get("pid") != pid0), 60)
st, m3 = fc.get("/mode")
ev["after_close"] = {"running_s": came, "mode": m3}
ctx.log("mode %.1f s after the lid closed: %s", time.time() - t_close, m3)
ctx.check(came is not None and m3.get("motion") == "verified",
"the controller did not come back verified after the lid closed: %s", m3)
lines = _probe_lines(FORGECTRL_LOG, off)
ev["probe_lines"] = lines[-2:]
ctx.check(not lines, "the motion probe ran again for the respawn: %s", lines[:1])
ctx.check(fc.wait_idle(15, abort=ctx.aborted), "machine not idle at the end")
ctx.log("PASS: the respawn waited %.1f s for the lid, then came back verified in %.1f s with no probe",
t_close - t_kill, came)
@test("motion.soft-limits", title="After a home the bed is the X/Y envelope",
subsystem="motion", kind="auto", mode="grbl", est_min=5,
covers=_MOTION_COVERS, requires=["motion.jog-roundtrip", "cloud.mode-switch"],
steps=["Bed clear, lid closed; cloud credentials configured; the machine on the network. The test "
"homes the machine through the web service (about a minute), sends moves the controller "
"must refuse, and hands the head back where it found it."],
description="The machine has no limit switches, so the core's $20 cannot be turned on and "
"the X/Y soft limits are the driver's: off while the position is not trusted, on "
"after a home, when the envelope is the bed from the home corner: the far edges the "
"bed check measured (envelope_x_mm, envelope_y_mm, held to 50 mm up to the travel "
"plus 30 mm), or $130 by $131 on a machine that was never measured. "
"With homing_mode = gfcloud and the camera-home offsets unset for the run, $H runs "
"the web-service homing session, every motion of it run whole. Homed, a program "
"move 5 mm past the far edge in X or in Y, or past the near edge, raises ALARM:2 "
"before any motion (the kernel counters do not move), a jog past the far edge is "
"refused with error 15, and a move inside the bed runs. A move or jog that is not "
"refused is stopped before the test fails. The Z envelope is the lens window, as "
"before. The settings are put back as found, the camera home is dropped, and the "
"head is jogged back to where the test found it by the travel the session's "
"motions logged.")
def soft_limits(ctx):
# The camera-home helpers, imported here: homeoff imports this module.
from .cloud import gfhome_homing
from .homeoff import camera_home_return, judge_whole_motions, session_mark
fc = ctx.forgectrl
ev = ctx.evidence
s0 = fc.settings() or {}
ctx.check(s0.get("cloud_enabled") == "1", "cloud mode is off: a camera home needs it on")
found = {k: s0.get(k) or "" for k in ("homing_mode", "gfcloud_home_x", "gfcloud_home_y")}
ev["found"] = found
session_at = None
def put(k, v):
# The controller refuses a setting for a moment after a Grbl client
# closes: each is written when it is taken. An empty value unsets it.
def write():
st_, body_ = fc.post("/settings", params={k: ""}) if v == "" else fc.post("/settings", data={k: v})
if st_ != 200:
ctx.log("%s=%r -> %s %s", k, v, st_, body_)
return st_ == 200 or None
return ctx.wait_for(write, 20, poll=0.5) is not None
try:
# A camera home at the origin: the envelope is the bed from the home corner.
for k, v in (("homing_mode", "gfcloud"), ("gfcloud_home_x", ""), ("gfcloud_home_y", "")):
ctx.check(put(k, v), "%s=%r was not taken", k, v)
with ctx.grbl() as g:
clean_slate(ctx, g)
session_at = session_mark()
gfhome_homing(ctx, ev, g)
judge_whole_motions(ctx, ev, session_at)
ctx.check(fc.status().get("homed"), "the machine is not homed")
x_travel = float(grbl_setting(g, "$130"))
y_travel = float(grbl_setting(g, "$131"))
ev["travel"] = {"x": x_travel, "y": y_travel}
# The far edges the home armed, as the driver takes them
# (gfhome_clamp_envelope_mm): the measured key held to 50 mm up
# to the travel plus 30 mm, or the travel when it is unset.
def far_edge(key, travel):
try:
mm = float(s0.get(key) or 0)
except ValueError:
mm = 0.0
if not mm > 0:
return travel
return min(max(mm, 50.0), travel + 30.0)
x_edge = far_edge("envelope_x_mm", x_travel)
y_edge = far_edge("envelope_y_mm", y_travel)
ev["envelope"] = {"x": x_edge, "y": y_edge}
k0 = kernel_xy_mm(ctx)
def refused(cmd):
lines = g.command(cmd, timeout=2)
ctx.sleep(0.5)
text = "\n".join(lines) + g.drain()
k1 = kernel_xy_mm(ctx)
moved = max(abs(k1[0] - k0[0]), abs(k1[1] - k0[1]))
st = g.status_report()["state"]
rec = {"cmd": cmd, "reply": lines, "alarm": "ALARM:2" in text, "moved_mm": round(moved, 3),
"state": st}
ctx.log("%s", rec)
# Stopped first: a move that was not refused must not run on
# to its target while the test fails.
g.realtime(0x18)
ctx.sleep(1.5)
g.drain()
ctx.check(rec["alarm"], "%s was not refused with ALARM:2: %s", cmd, lines)
ctx.check(moved < 0.05, "%s moved the kernel %.3f mm before the alarm", cmd, moved)
unlock = g.command("$X")
ctx.check(unlock and unlock[-1] == "ok", "$X after the soft-limit alarm: %s", unlock)
st = g.status_report()["state"]
ctx.check(st.startswith("Idle"), "controller is %s after the recovery", st)
return rec
ev["refused"] = [refused("G90 G1 X%.1f F600" % (x_edge + 5)),
refused("G90 G1 Y%.1f F600" % (y_edge + 5)),
refused("G90 G1 X-1 F600")]
# Homed still: the soft-limit alarm and its reset keep the reference.
ctx.check(fc.status().get("homed"), "the soft-limit alarm and the reset un-homed the machine")
jog = g.command("$J=G91X%.1fF1200" % (x_edge + 5))
ev["jog"] = jog
if not any(l.startswith("error:15") for l in jog):
g.realtime(0x85) # the jog cancel, before the test fails
ctx.sleep(1.5)
g.drain()
ctx.check(any(l.startswith("error:15") for l in jog), "a jog past the bed was not refused with error 15: %s",
jog)
# The core answers the line after an error with that error again
# until an empty line clears it (the sender's acknowledgment).
g.command("")
g.command("G90 G1 X10 Y10 F600", timeout=0.5)
peak, states, st = wait_idle(ctx, g, 20)
ev["inside"] = states
ctx.check("TIMEOUT" not in states, "a move inside the bed did not run")
g.command("G90 G1 X0 Y0 F600", timeout=0.5)
wait_idle(ctx, g, 20)
k1 = kernel_start(ctx) # at rest: grbl's Idle leads the kernel's tail
ev["kernel_back_mm"] = [round(k1[0] - k0[0], 3), round(k1[1] - k0[1], 3)]
ctx.check(max(abs(k1[0] - k0[0]), abs(k1[1] - k0[1])) < 0.1, "the head did not come back to the corner: %s",
ev["kernel_back_mm"])
finally:
for k, v in found.items():
ctx.log("restore %s=%r -> %s", k, v, "taken" if put(k, v) else "not taken")
if session_at is not None:
camera_home_return(ctx, ev, session_at)
ctx.log("PASS: X max, Y max and X min refused with ALARM:2 and no motion, the jog refused with "
"error 15, a move inside the bed ran")
# ------------------------------------------------- lid / button (the factory's)
_LID_COVERS = _MOTION_COVERS + [("grblhal-glowforge", "src/glowforge_switches.c"),
@@ -1597,144 +1411,6 @@ def lid_cancel_home(ctx):
"reset without alarm, head returned to the start both times", drift, hold_drift)
@test("motion.interlock-cancel-home", title="The interlock loop cancels a job like the lid and returns to "
"the job start",
subsystem="motion", kind="operator", mode="grbl", est_min=4,
covers=_LID_COVERS, requires=["motion.lid-cancel-home"], actions=["interlock"],
steps=["Bed clear; the head needs 60 mm of free +X travel. No laser is involved.",
"Be able to open the remote-interlock loop: unplug the Pro's interlock plug, or pull the "
"jumper at J8 on a Basic/Plus.",
"On Ready the head starts a 12 s move: open the interlock while it moves and leave it "
"open until the head has come back, then restore it."],
description="The remote-interlock loop is the lid's equal in the cancel policy: opening it mid-job "
"cancels the job with 'interlock open' named as the reason - the lid's own message would "
"be wrong here - and sends the head back to the job start with the loop still open. "
"(An open lid not stopping the park is the lid test's park, which runs with the lid open "
"throughout; in GRBL mode the park is a rapid too short to open a lid during, so the "
"mid-park lid edge is the cloud test's.)")
def interlock_cancel_home(ctx):
ev = ctx.evidence
policy = (ctx.forgectrl.settings() or {}).get("lid_policy") or "cancel"
ev["lid_policy"] = policy
ctx.check(policy == "cancel", "lid_policy is %r; this test needs cancel", policy)
sw = (ctx.forgectrl.status().get("switches") or {})
ctx.check(sw.get("interlock_ok"), "the interlock loop already reads open - close it before this test")
with ctx.grbl() as g:
clean_slate(ctx, g)
k0 = kernel_start(ctx)
start = g.status_report()["MPos"]
ev["start"] = start
ev["kernel_start"] = k0
ctx.ready("On Ready the head starts a 12 s move along +X. Open the interlock loop while it "
"moves and leave it open until the head has come back on its own.")
g.command("M5")
g.command("G91")
start_move(ctx, g, "G1X60F300") # a 12 s move
g.drain()
ctx.act("interlock", "open", text="Leave it open until the head has come back.", timeout=16)
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)
drift = expect_cancel_and_return(ctx, g, ev, start, k0, "interlock open", "interlock")
sw = (ctx.forgectrl.status().get("switches") or {})
ev["switches_at_return"] = {"lid": sw.get("lid"), "interlock_ok": sw.get("interlock_ok")}
ctx.log("at the end of the park: %s", ev["switches_at_return"])
ctx.check(sw.get("interlock_ok") is False,
"the interlock was closed again before the park finished - the park ran with the loop "
"restored, not open")
ctx.act("interlock", "close")
ctx.sleep(1)
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")
r = g.command("$J=G91X5F1200")
ctx.check(not any(x.startswith("error") for x in r), "jog refused after the cancel: %s", r)
wait_idle(ctx, g, 15)
g.command("$J=G91X-5F1200")
wait_idle(ctx, g, 15)
g.command("G90")
machine_idle(ctx)
ctx.log("PASS: interlock open canceled the job with its own reason and the head returned to the "
"start (drift %.3f mm) with the loop still open", drift)
@test("motion.lid-policy-hold", title="lid_policy=hold parks the job in Door and a cycle start resumes it",
subsystem="motion", kind="operator", mode="grbl", est_min=4,
covers=_LID_COVERS + [("forgectrl", "src/settings.*")], requires=["motion.lid-cancel-home"],
actions=["lid"],
steps=["Bed clear; the head needs 40 mm of free +X travel. No laser is involved.",
"On Ready the head starts an 8 s move: open the lid while it moves (the job parks in "
"Door), then close it when told; the job finishes after that."],
description="The other lid policy, kept for senders that expect stock grblHAL: with "
"lid_policy=hold a lid open parks the job in the door state and holds it there - "
"no cancel, no return home - and once the lid is closed a cycle start finishes the "
"move with its position intact. The setting is restored to cancel at the end.")
def lid_policy_hold(ctx):
ev = ctx.evidence
fc = ctx.forgectrl
# What the machine had, exactly: an unset lid_policy reads as the empty
# string and behaves as cancel, and writing the word back where the
# machine had nothing is a leftover the hand-back reports. The default
# is for reading the value, never for restoring it.
was = (fc.settings() or {}).get("lid_policy", "")
ev["lid_policy_before"] = was
ev["lid_policy_in_force"] = was or "cancel"
st, _b = fc.post("/settings", data={"lid_policy": "hold"})
ctx.check(st == 200, "could not set lid_policy=hold (%s)", st)
ctx.check(((fc.settings() or {}).get("lid_policy")) == "hold", "lid_policy did not take")
try:
with ctx.grbl() as g:
clean_slate(ctx, g)
start = g.status_report()["MPos"]
ctx.ready("On Ready the head starts an 8 s move along +X. Open the lid while it moves; "
"the job parks in the door state and waits.")
g.command("M5")
g.command("G91")
start_move(ctx, g, "G1X40F300") # an 8 s move
g.drain()
ctx.act("lid", "open", timeout=12)
st, text = wait_state_text(ctx, g, "Door", 8)
ev["door_state"] = st["state"] if st else g.status_report()["state"]
ctx.check(st is not None, "the lid did not park the job in Door (state %s)", ev["door_state"])
ev["messages"] = [ln for ln in text.splitlines() if ln.startswith("[MSG:")]
ctx.check("job canceled" not in text, "the job was canceled under lid_policy=hold: %s", ev["messages"])
ctx.check("returned to the job start" not in text,
"the head returned to the job start under lid_policy=hold")
held = g.status_report()["MPos"]
ev["parked_at"] = held
ctx.log("parked in %s at %s", ev["door_state"], held)
ctx.act("lid", "close")
ctx.sleep(1)
ev["state_after_close"] = g.status_report()["state"]
ctx.log("after the lid closed: %s (a cycle start is needed)", ev["state_after_close"])
g.realtime(0x7E) # ~ cycle start
peak, states, st = wait_idle(ctx, g, 30)
ev["states_after_resume"] = states
ctx.check("TIMEOUT" not in states, "the job did not finish after the resume: %s", states)
final = st["MPos"]
moved = final[0] - start[0]
ev["moved_mm"] = round(moved, 3)
ctx.log("finished: moved %.3f mm of 40 (states %s)", moved, states)
ctx.check(abs(moved - 40.0) <= 0.05,
"the resumed job did not finish its move (%.3f mm of 40)", moved)
g.command("$J=G91X-40F1200")
wait_idle(ctx, g, 30)
g.command("G90")
machine_idle(ctx)
finally:
# An empty value clears the key, and only the query-string form
# carries one: an empty form field never reaches the request, and
# the write is refused with "no known setting in request".
st, _b = (fc.post("/settings", params={"lid_policy": ""}) if not was
else fc.post("/settings", data={"lid_policy": was}))
ev["lid_policy_restored"] = (fc.settings() or {}).get("lid_policy", "")
ctx.log("lid_policy restored to %s", ev["lid_policy_restored"])
ctx.check(ev["lid_policy_restored"] == was, "lid_policy was not restored to %r", was)
ctx.log("PASS: lid_policy=hold parked the job in Door and the cycle start finished it (%.3f mm)",
ev["moved_mm"])
GRBLHAL_LOG = "/data/log/forgefirm/grblhal/grblhal.log"
SCHED_FIFO = 1
forgetest/tests/fixtures/gfcloud-buttonwait.log
Vendored
-41
View File
@@ -1,41 +0,0 @@
2026-08-17T09:44:00.417337+00:00 gfcloud[1927] INFO cnc:set_step_freq 28160
2026-08-17T09:44:00.418627+00:00 gfcloud[1927] INFO cnc:set_x_decay 1
2026-08-17T09:44:00.419858+00:00 gfcloud[1927] INFO cnc:set_x_mode 8
2026-08-17T09:44:00.421083+00:00 gfcloud[1927] INFO cnc:set_x_current 135
2026-08-17T09:44:00.422535+00:00 gfcloud[1927] INFO cnc:set_y_decay 1
2026-08-17T09:44:00.423794+00:00 gfcloud[1927] INFO cnc:set_y_mode 8
2026-08-17T09:44:00.425014+00:00 gfcloud[1927] INFO cnc:set_y_current 22
2026-08-17T09:44:00.432591+00:00 gfcloud[1927] INFO machine:_run_loop starting run
2026-08-17T09:44:00.434005+00:00 gfcloud[1927] INFO machine:_run_loop current state: MachineState.IDLE
2026-08-17T09:44:00.438277+00:00 gfcloud[1927] INFO machine:_run_loop current state: MachineState.RUNNING
2026-08-17T09:44:01.785230+00:00 gfcloud[1927] INFO machine:_run_loop current state: MachineState.IDLE
2026-08-17T09:44:01.797728+00:00 gfcloud[1927] INFO machine:_run_loop finished run
2026-08-17T09:44:01.800661+00:00 gfcloud[1927] INFO cnc:laser_latch 1
2026-08-17T09:44:01.808504+00:00 gfcloud[1927] INFO machine:_motion_locked end positions (actual/expected): X (-10313/-10313), Y (-649/-649), Z (0/0)
2026-08-17T09:44:01.809326+00:00 gfcloud[1927] INFO machine:_motion_locked motion bytes actual:34952, expected: 34952
2026-08-17T09:44:01.810066+00:00 gfcloud[1927] INFO machine:_motion_locked start idle
2026-08-17T09:44:01.811352+00:00 gfcloud[1927] INFO cnc:set_x_current 33
2026-08-17T09:44:01.812874+00:00 gfcloud[1927] INFO cnc:set_y_current 5
2026-08-17T09:44:01.820631+00:00 gfcloud[1927] INFO machine:_motion_locked end positions (-10313, -649, 0)
2026-08-17T09:44:01.822062+00:00 gfcloud[1927] INFO machine:_motion end motion
2026-08-17T09:44:01.822829+00:00 gfcloud[1927] INFO basemachine:_finish_action motion [1576550614]: finished with event ":completed"
2026-08-17T09:44:01.824200+00:00 gfcloud[1927] INFO cnc:laser_latch 1
2026-08-17T09:44:07.993647+00:00 gfcloud[1927] INFO gfuiservice:run service action request: print (ready)
2026-08-17T09:44:07.998655+00:00 gfcloud[1927] INFO machine:_motion start motion
2026-08-17T09:44:07.999643+00:00 gfcloud[1927] INFO gfuiservice:run print (ready) service action print
2026-08-17T09:44:14.199666+00:00 gfcloud[1927] INFO cnc:laser_latch 0
2026-08-17T09:44:14.208291+00:00 gfcloud[1927] INFO machine:_button_wait waiting for button
2026-08-17T09:44:21.270846+00:00 gfcloud[1927] INFO machine:_switch_event lid opened
2026-08-17T09:44:21.275703+00:00 gfcloud[1927] WARNING machine:_button_wait button wait lid opened - relocking the laser
2026-08-17T09:44:21.278615+00:00 gfcloud[1927] INFO cnc:laser_latch 1
2026-08-17T09:44:21.288305+00:00 gfcloud[1927] INFO machine:_return_home start return home
2026-08-17T09:44:21.291373+00:00 gfcloud[1927] INFO machine:_return_home return home: already at the job start
2026-08-17T09:44:21.292501+00:00 gfcloud[1927] INFO machine:_motion_locked start cool down
2026-08-17T09:44:21.305578+00:00 gfcloud[1927] INFO machine:_motion_locked start idle
2026-08-17T09:44:21.307322+00:00 gfcloud[1927] INFO cnc:set_x_current 33
2026-08-17T09:44:21.309185+00:00 gfcloud[1927] INFO cnc:set_y_current 5
2026-08-17T09:44:21.316340+00:00 gfcloud[1927] INFO machine:_motion_locked end positions (0, 0, 0)
2026-08-17T09:44:21.318139+00:00 gfcloud[1927] INFO machine:_motion end motion
2026-08-17T09:44:21.318860+00:00 gfcloud[1927] INFO basemachine:_finish_action print [1576550621]: finished with event ":cancelled"
2026-08-17T09:44:21.320214+00:00 gfcloud[1927] INFO cnc:laser_latch 1
2026-08-17T09:44:34.058320+00:00 gfcloud[1927] INFO machine:_switch_event lid closed
forgetest/tests/test_baseline.py
+3 -3
View File
@@ -531,9 +531,9 @@ class ExtensionFreeTests(BaselineTests):
class TransientNotLeftoverTests(BaselineTests):
"""A leftover is what a run left behind, not the machine part-way
through its own work. Found on the bench reference, where
laser.arm-wait-lid passed every check it makes and failed its
hand-back on the post-job smoke clear."""
through its own work. Found on the bench reference, where a laser
test passed every check it makes and failed its hand-back on the
post-job smoke clear."""
def test_a_fading_button_led_is_not_a_leftover(self):
# the trigger fades brightness toward target: target 0 with
forgetest/tests/test_cloud_suite.py
+15 -182
View File
@@ -69,6 +69,19 @@ def cut(lines, marker, count=1):
raise KeyError(marker)
def dark_print(ctx):
"""The dark print as exthost.armed-freeze runs it: the cloud suite's
dark_print_body between the offline service's setup and cleanup."""
offset = cloud.enter_offline(ctx)
job = cloud.offline_job(ctx, "dark.puls", seconds=30)
off = cloud.Offline().__enter__()
try:
cloud.dark_print_body(ctx, ctx.evidence, off, job, offset, ctx.forgectrl)
finally:
off.__exit__(None, None, None)
cloud.offline_cleanup(ctx)
class Script:
"""Answers every prompt with its first option; a hook per prompt or
notice substring drives the fake machine and the log replay (a
@@ -1040,30 +1053,7 @@ class CloudSuiteTests(unittest.TestCase):
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_offline()
lines = fixture("buttonwait")
pre, rest = cut(lines, "waiting for button")
# The bench excerpt is a print that fit the ring; the test now wants
# one that did not, so the feeder is alive through the wait.
long_job = ("2026-08-17T09:44:14.100000+00:00 gfcloud[1927] INFO machine:_feed_and_run "
"job is longer than the ring: 33554432 bytes enqueued, feeding the rest as it plays")
pre, rest = pre + [long_job, rest[0]], rest[1:]
self.offline_print_hooks(pre)
hooks = {"do NOT press it": lambda: (self.lid(False), self.append(rest, delay=0.05)),
"Close the lid.": lambda: self.lid(True)}
run = self.run_test(cloud.lid_during_button_wait, hooks=hooks, test_id="cloud.lid-during-button-wait")
ev = run.evidence
self.assertEqual([m["status"] for m in FakeOffline.sent], ["ready"])
self.assertEqual(ev["runs_started_after_wait"], 0)
self.assertIn(":cancelled", ev["log"]["print finished"])
self.assertTrue(ev["latch_locked"])
self.assertEqual(ev["program_total_after"], {"first": 0, "later": 0})
self.assertEqual(ev["streaming_after"], 0)
self.assertEqual(self.fc.posts, [])
self.assertTrue(any("PASS: lid open at the button prompt" in l for l in run.lines))
# -- the cooling verdict: locked until the run, a hold that never clears --
# -- the dark print exthost.armed-freeze runs: locked at the button --------
def dark_parts(self):
"""(the print prologue through the button prompt, the run and the
print's end with no pause in it)."""
@@ -1074,170 +1064,13 @@ class CloudSuiteTests(unittest.TestCase):
run = [l for l in run if "paus" not in l and "resum" not in l]
return pre, run
def test_dark_print_on_the_bench_excerpt(self):
self.in_offline()
self.fc.state["cool"].update({"verdict": "OK", "hold": False, "fire_ok": True, "armed": False})
pre, run = self.dark_parts()
self.offline_print_hooks(pre)
def press():
# the run: the client unlocks the latch right before it, and
# relocks when the job ends
self._attr("cnc/interlock_circuit", "37")
self.append(run[:1], delay=0.05)
def ended():
self._attr("cnc/interlock_circuit", "45")
self.append(run[1:], delay=0.0)
threading.Timer(2.0, ended).start()
hooks = {"press it. The print runs dark": press}
run_ = self.run_test(cloud.dark_print, hooks=hooks, test_id="cloud.dark-print")
ev = run_.evidence
self.assertTrue(ev["latch_locked_at_button"])
self.assertFalse(ev["latch_locked_in_run"])
self.assertTrue(ev["latch_locked_after"])
self.assertIn(":completed", ev["print finished"])
self.assertTrue(any("PASS: locked at the button" in l for l in run_.lines))
def test_dark_print_fails_when_the_latch_is_open_at_the_button(self):
self.in_offline()
pre, run = self.dark_parts()
self._attr("cnc/interlock_circuit", "37") # unlocked before the press
self.offline_print_hooks(pre)
hooks = {"press it. The print runs dark": lambda: self.append(run, delay=0.05)}
self.assertFails(cloud.dark_print, "the latch is unlocked at the button wait", hooks=hooks)
def test_a_print_turned_away_before_the_button_fails_at_once_with_the_reason(self):
"""The bench reference's own lines, 2026-09-20: the coolant read 27.4
against the client's start ceiling of 27, and the test then sat out
its two minutes and said only that the button wait never came."""
self.in_offline()
stamp = "2026-09-20T21:28:16.83+00:00 gfcloud[782] INFO "
refused = [stamp + "gfuiservice:run service action request: print (ready)",
stamp + "machine:_motion start motion",
stamp + "machine:_safe_to_move machine temp is too high, temp: 27.4",
stamp + "machine:_motion end motion",
stamp + 'basemachine:_finish_action print [9005]: finished with event ":cancelled"']
self.offline_print_hooks(refused)
t0 = time.time()
self.assertFails(cloud.dark_print, "turned the print away before the button wait: "
"INFO machine:_safe_to_move machine temp is too high, temp: 27.4")
self.assertLess(time.time() - t0, 30, "the refusal was not seen until the wait timed out")
def test_verdict_refuse_on_the_bench_excerpt(self):
self.in_offline()
self.fc.state["status"]["coolant"] = {"up_c": 21.5, "down_c": 21.4}
self.fc.state["settings"].update({"cool_temp_min": "5", "cool_temp_start": "16",
"cloud_hold_max_s": ""})
self.fc.state["cool"].update({"verdict": "OK", "hold": False, "fire_ok": True, "armed": False})
pre, run = self.dark_parts()
stamp = "2026-08-17T09:44:14.%06d+00:00 gfcloud[1927] %s machine:_verdict_wait "
wait = [stamp % (100000, "INFO") + "waiting on the cooling engine: WARMUP (WARM-UP: coolant "
"21.5 C under the 29.5 C start gate - heater on, hold)"]
refuse = [stamp % (200000, "WARNING") + "cooling hold held 60 s - relocking the laser",
"2026-08-17T09:44:15.000000+00:00 gfcloud[1927] INFO basemachine:_finish_action print "
"[9006]: finished with event \":cancelled\""]
def on_post(path, form):
if path == "/settings" and form.get("cool_temp_start") not in (None, "0", "16"):
self.fc.state["cool"].update({"verdict": "WARMUP", "hold": True, "fire_ok": False,
"armed": True, "phase": "warm-up"})
return None
self.fc.on_post = on_post
self.offline_print_hooks(pre)
def press():
self.append(wait, delay=0.05)
def refused():
self.fc.state["cool"].update({"verdict": "OK", "hold": False, "fire_ok": True,
"armed": False, "phase": "idle"})
self.append(refuse, delay=0.0)
threading.Timer(2.5, refused).start()
hooks = {"press it. Nothing runs": press}
run_ = self.run_test(cloud.verdict_refuse, hooks=hooks, test_id="cloud.verdict-refuse")
ev = run_.evidence
self.assertEqual(ev["gate"], 29.5)
self.assertTrue(ev["latch_locked_at_button"])
self.assertEqual(ev["latch_samples"]["unlocked"], 0)
self.assertGreaterEqual(ev["latch_samples"]["n"], 1)
self.assertIn(":cancelled", ev["print finished"])
posted = [f for p_, f in self.fc.posts if p_ == "/settings"]
self.assertEqual(posted[0]["cool_temp_start"], "29.5")
self.assertEqual(posted[0]["cloud_hold_max_s"], "60")
self.assertEqual(posted[-1], {"cool_temp_min": "5", "cool_temp_start": "16", "cloud_hold_max_s": ""})
self.assertTrue(any("PASS: held" in l for l in run_.lines))
def test_verdict_refuse_fails_when_the_latch_unlocks_during_the_hold(self):
self.in_offline()
self.fc.state["status"]["coolant"] = {"up_c": 21.5, "down_c": 21.4}
self.fc.state["settings"].update({"cool_temp_min": "5", "cool_temp_start": "16"})
self.fc.state["cool"].update({"verdict": "WARMUP", "hold": True, "fire_ok": False, "armed": True})
pre, run = self.dark_parts()
stamp = "2026-08-17T09:44:14.%06d+00:00 gfcloud[1927] %s machine:_verdict_wait "
wait = [stamp % (100000, "INFO") + "waiting on the cooling engine: WARMUP (no reason given)"]
refuse = [stamp % (200000, "WARNING") + "cooling hold held 60 s - relocking the laser",
"2026-08-17T09:44:15.000000+00:00 gfcloud[1927] INFO basemachine:_finish_action print "
"[9006]: finished with event \":cancelled\""]
self.offline_print_hooks(pre)
def press():
self.append(wait, delay=0.05)
self._attr("cnc/interlock_circuit", "37") # unlocked under the hold
threading.Timer(2.5, lambda: self.append(refuse, delay=0.0)).start()
hooks = {"press it. Nothing runs": press}
self.assertFails(cloud.verdict_refuse, "the latch unlocked during the hold", hooks=hooks)
# -- a paused print canceled 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 - what
cloud.oversize-stream's ending looks like)."""
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 canceled mid-run; stopping motion") for l in rest]
return pre, paused, rest, app_cancel
def test_paused_lid_cancel_on_the_bench_excerpt(self):
self.in_offline()
pre, paused, lid_tail, _app = self.cancel_parts()
self.offline_print_hooks(pre)
run = self.run_test(cloud.paused_lid_cancel,
hooks={"the press pauses the print": lambda: self.append(paused, delay=0.05),
"The print is paused: leave the lid open": lambda: (
self.lid(False), self.append(lid_tail, delay=0.05)),
"Close the lid.": lambda: self.lid(True)},
test_id="cloud.paused-lid-cancel")
ev = run.evidence
self.assertEqual(len(FakeOffline.sent), 1) # one print, handed over the socket
self.assertEqual(ev["paused"], {"button pressed mid-run; pausing": True, "paused at": True})
self.assertIn(":cancelled", ev["lid_from_pause"]["print finished"])
self.assertTrue(ev["lid_from_pause"]["return home complete"])
self.assertEqual(ev["counters_after"], [0, 0, 3])
self.assertTrue(ev["latch_locked_after"])
self.assertFalse(ev["armed_after"])
self.assertFalse(ev["button_dark"])
self.assertEqual(self.fc.posts, [])
self.assertTrue(any("PASS: a paused print canceled by the lid" in l for l in run.lines), run.lines)
def test_paused_lid_cancel_fails_when_the_paused_print_resumes_instead(self):
# a lid that resumed (or was ignored) leaves the print ':completed'
self.in_offline()
pre, paused, lid_tail, _app = self.cancel_parts()
lid_tail = [l.replace(':cancelled"', ':completed"') for l in lid_tail]
self.offline_print_hooks(pre)
self.assertFails(
cloud.paused_lid_cancel, "the print did not end ':cancelled'",
hooks={"the press pauses the print": lambda: self.append(paused, delay=0.05),
"The print is paused: leave the lid open": lambda: (
self.lid(False), self.append(lid_tail, delay=0.05))})
self.assertFails(dark_print, "the latch is unlocked at the button wait", hooks=hooks)
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
forgetest/tests/test_cooling_intake_suite.py
-176
View File
@@ -1,176 +0,0 @@
# Copyright 2026 514 LLC d/b/a OpenGlow
# Written by Scott Wiederhold
# https://community.openglow.org
# SPDX-License-Identifier: MIT
"""cooling.intake-one-fan replayed host-side under the real runner
Context against a scripted engine: the fake reads cool_intake_fans and
the floors at every M8 (the engine reloads at run start), publishes
intake_2 not fitted with no floor when one intake fan is set, holds every
fan in grace for the configured seconds, then judges the bench readings
against the floors. The negative engines are the two ways the change
could go wrong: the setting ignored (intake 2 still judged) and the
setting taking intake 1 out with it."""
import threading
import time
import unittest
import helpers
from forgetest.runner import Context, Failed, Run
from forgetest.suite import cooling, cooling_intake
from test_cooling_suite import FakeGrbl
class IntakeOneFanTests(unittest.TestCase):
READINGS = {"exhaust": 11640, "intake_1": 4157, "intake_2": 4158, "air_assist": 11050, "purge": 625}
FLOORS = {"exhaust": ("cool_tach_exhaust_min_rpm", 6400.0, 20000.0),
"intake_1": ("cool_tach_intake_min_rpm", 2290.0, 20000.0),
"intake_2": ("cool_tach_intake_min_rpm", 2290.0, 20000.0),
"air_assist": ("cool_tach_air_assist_min_rpm", 6000.0, 30000.0),
"purge": ("cool_purge_min_current", 300.0, 1023.0)}
GATE_OF = {"exhaust": "exhaust", "intake_1": "intake", "intake_2": "intake",
"air_assist": "air_assist", "purge": "purge"}
def setUp(self):
self.fc = helpers.FakeForgectrl().start()
self.grbl = FakeGrbl()
self.saved = (cooling.VERDICT_WAIT_S, cooling.SESSION_END_WAIT_S, cooling_intake.INTAKE_TRIP_WAIT_S)
cooling.VERDICT_WAIT_S = 4
cooling.SESSION_END_WAIT_S = 3
cooling_intake.INTAKE_TRIP_WAIT_S = 4
self.fc.state["status"] = dict(self.fc.state["status"], gates_off=[])
self.fc.state["cool"] = {"phase": "idle", "verdict": "OK", "fire_ok": False, "hold": False,
"gates_off": [], "fan_gates": {}}
for key, _d, _h in self.FLOORS.values():
self.fc.state["settings"].setdefault(key, "")
self.fc.state["settings"].setdefault("cool_fan_grace_s", "")
self.fc.state["settings"].setdefault("cool_intake_fans", "")
self.honors_setting = True # the engine takes intake 2 out with one intake fan
self.drops_both = False # the engine takes intake 1 out with it
self.grace_scale = 0.1 # seconds of fake grace per configured second
self._describe()
self.grbl.on_command = self._engine
self.fc.on_post = self._on_post
def tearDown(self):
cooling.VERDICT_WAIT_S, cooling.SESSION_END_WAIT_S, cooling_intake.INTAKE_TRIP_WAIT_S = self.saved
self.grbl.close()
self.fc.stop()
def setting(self, key, default):
v = self.fc.state["settings"].get(key) or ""
return float(v) if v else default
def _describe(self):
gates = {}
for fan, (key, default, hi) in self.FLOORS.items():
v = self.setting(key, default)
gates[key] = {"gate": self.GATE_OF[fan], "def": default, "lo": 0.0, "hi": hi,
"band": [default * 0.9, default * 1.1], "off": "low", "value": v,
"state": "off" if v <= 0 else "ok"}
self.fc.state["settings"]["gates"] = gates
def _on_post(self, path, form):
if path != "/settings":
return None
self.fc.state["settings"].update(form)
self._describe()
return (200, self.fc.state["settings"])
def _engine(self, line):
self._describe()
cool = self.fc.state["cool"]
if line == "M9":
def end():
time.sleep(0.3)
cool["phase"] = "idle"
cool.update(verdict="OK", fire_ok=True, hold=False, reason="")
for g in cool.get("fan_gates", {}).values():
if g["state"] != "not fitted":
g["state"] = "idle"
threading.Thread(target=end, daemon=True).start()
return
if line != "M8":
return
cool["phase"] = "run"
one = self.fc.state["settings"].get("cool_intake_fans") == "1" and self.honors_setting
out = {"intake_2"} | ({"intake_1"} if self.drops_both else set()) if one else set()
floors = {fan: 0.0 if fan in out else self.setting(key, default)
for fan, (key, default, _h) in self.FLOORS.items()}
cool.update(verdict="OK", fire_ok=True, hold=False, resume_ok=True, reason="", gates_off=[],
fan_gates={f: {"reading": self.READINGS[f], "floor": floors[f],
"state": "not fitted" if f in out else "grace"} for f in floors})
grace = self.setting("cool_fan_grace_s", 15.0) * self.grace_scale
def judge():
time.sleep(grace)
if cool["phase"] != "run":
return
for f in floors:
if f not in out:
cool["fan_gates"][f]["state"] = "ok" if self.READINGS[f] >= floors[f] else "under"
time.sleep(0.3)
if cool["phase"] != "run":
return
for f in floors:
if f not in out and floors[f] > self.READINGS[f]:
cool["fan_gates"][f]["state"] = "TRIPPED"
if cool["verdict"] != "AIRFLOW":
cool.update(verdict="AIRFLOW", fire_ok=False, hold=True, resume_ok=False,
reason="AIRFLOW: %s %d under the %d floor for 3 s - hold, no resume this job"
% (f, self.READINGS[f], floors[f]))
threading.Thread(target=judge, daemon=True).start()
def run_test(self):
run = Run("test", "cooling.intake-one-fan", "t")
ctx = Context(run, None, helpers.make_test("cooling.intake-one-fan", []))
cooling_intake.intake_one_fan(ctx)
return run
def settings_posts(self):
return [f for p, f in self.fc.posts if p == "/settings"]
def restored(self):
return {"cool_intake_fans": "", "cool_tach_intake_min_rpm": "", "cool_fan_grace_s": ""}
def test_one_fan_then_intake_1_trip_then_restore_pass(self):
run = self.run_test()
ev = run.evidence
self.assertEqual(ev["one_fan"]["intake_2"]["state"], "not fitted")
self.assertEqual(ev["one_fan"]["intake_1"]["state"], "ok")
self.assertEqual(ev["intake_1_trip"]["intake_1"]["state"], "TRIPPED")
self.assertEqual(ev["intake_1_trip"]["intake_2"]["state"], "not fitted")
self.assertTrue(all(g["state"] == "ok" for g in ev["restored"].values()))
posts = self.settings_posts()
self.assertEqual(posts[0], {"cool_intake_fans": "1", "cool_fan_grace_s": "8"})
self.assertEqual(posts[1], {"cool_tach_intake_min_rpm": "20000.0"})
self.assertEqual(posts[-1], self.restored())
self.assertEqual(self.grbl.commands.count("M8"), 3)
self.assertEqual(self.grbl.commands.count("M9"), 3)
def test_an_engine_that_ignores_the_setting_fails_and_restores(self):
self.honors_setting = False
with self.assertRaises(Failed) as cm:
self.run_test()
self.assertIn("intake_2 is not published not fitted", str(cm.exception))
self.assertEqual(self.settings_posts()[-1], self.restored())
self.assertEqual(self.fc.state["cool"]["verdict"], "OK")
def test_an_engine_that_drops_intake_1_too_fails_and_restores(self):
self.drops_both = True
with self.assertRaises(Failed) as cm:
self.run_test()
self.assertIn("intake_1 is not judged ok at the intake floor", str(cm.exception))
self.assertEqual(self.settings_posts()[-1], self.restored())
def test_a_machine_already_at_one_intake_fan_is_refused_untouched(self):
self.fc.state["settings"]["cool_intake_fans"] = "1"
with self.assertRaises(Failed) as cm:
self.run_test()
self.assertIn("already 1", str(cm.exception))
self.assertEqual(self.settings_posts(), [])
self.assertEqual(self.grbl.commands.count("M8"), 0)
if __name__ == "__main__":
unittest.main()