vh/forgefirm
1
0
Fork 0
mirror of https://github.com/openglow-org/forgefirm.git synced 2026-09-27 16:51:12 -07:00
Code Issues Packages Projects Releases Wiki Activity

commissioning: the layer, the acceptance tests, the harness rule, the docs, and the bench drills

Browse Source 
meta-forgefirm: the forgefirm-users init replays the account at boot;
sshd refuses root and empty passwords and runs only while the panel
turns it on; the release image keeps an empty root password for the
console; the console banner; avahi announces forgefirm.local; https in
libmicrohttpd and ulfius; the panel on 80 and 443; the license bundle on
the rootfs; release.sh checks the root policy on the built rootfs.

forgetest: the commission suites (commission, commission_dark,
commission_sheet: 23 cases); the runner turns cloud mode on with the
typed phrase for a test that declares it; the baseline's motor_lock is
0; the log-export test checks the bundle for the camera key; the record
helpers write bytes as given and join the daemon's paths as POSIX. The
stream harness gains rule 24: a hold verdict is held again after a
resume. Bench drills: lens_travel.py and lens_stop_accel.py.

Docs: BRINGUP carries the present state; CAMPAIGN-LOG carries the dated
record.
This commit is contained in:
ScottW514
2026-09-06 19:56:05 -04:00
parent ff9796cde6
commit 97287aa6a9
59 changed files with 15113 additions and 9139 deletions
Download Patch File Download Diff File Expand all files Collapse all files
forgetest/README.md
+3 -2
View File
@@ -16,7 +16,7 @@ inheritance, the gate, the coverage rule - is
## Run the daemon on a workstation (against a mock or a manifest file)
FORGETEST_DATA=/tmp/ft FORGETEST_MANIFEST=../tree-manifest.json \
FORGECTRL_URL=http://<machine>:8080 python3 -m forgetest --port 8090
FORGECTRL_URL=http://<machine> python3 -m forgetest --port 8090
`scripts/manifest-from-tree.py` produces `tree-manifest.json` from the recipe
pins; the coverage lint is `python3 -m forgetest.coverage --manifest ...`.
@@ -31,7 +31,8 @@ pins; the coverage lint is `python3 -m forgetest.coverage --manifest ...`.
| `FORGETEST_BENCH_DIR` | `/usr/share/forgetest/bench` | the installed bench scripts |
| `FORGETEST_BENCH_DATA` | `<FORGETEST_DATA>/bench` | passed to bench tools: where they keep their data files (with `GF_HOST=127.0.0.1` and the panel token in `GF_TOKEN`) |
| `FORGETEST_MARKER` | `/run/forgetest.active` | takeover marker |
| `FORGECTRL_URL`, `FORGECTRL_TOKEN_FILE` | `http://127.0.0.1:8080`, `/data/forgefirm/panel.token` | forgectrl client |
| `FORGECTRL_URL`, `FORGECTRL_TOKEN_FILE` | `http://127.0.0.1`, `/data/forgefirm/panel.token` | forgectrl client (HTTP; the token authorizes writes from the board) |
| `FORGECTRL_TLS_URL` | `https://127.0.0.1` | forgectrl over HTTPS (self-signed, unverified), for the login test |
| `GF_SYSFS_ROOT` | `/sys/glowforge/` | kernel module sysfs |
| `GRBL_HOST`, `GRBL_PORT` | 127.0.0.1, 23 | Grbl TCP |
forgetest/forgetest.init
+7
View File
@@ -22,6 +22,13 @@ case "$1" in
start)
echo "Starting forgetest"
mkdir -p "$DATA"
# The bench seed: the commissioning override lifts the wizard part of
# the controller gate until the next reboot (/run is tmpfs), so a
# campaign runs on a machine whose required wizards are not complete.
# Nothing lifts the agreements or the account: the bench accepts them
# once, and that stays in /data.
mkdir -p /run/forgefirm
: > /run/forgefirm/commissioning-override
start-stop-daemon -S -q -p $PIDFILE -m -b -x /bin/sh -- -c \
"exec /usr/bin/python3 -m forgetest >> $LOG 2>&1"
;;
forgetest/forgetest/baseline.py
+34 -15
View File
@@ -44,12 +44,14 @@ from .log import now_ts
# GRBL-mode resting values (kernel attribute -> value as read back), as a
# fresh boot of the dev image leaves them (2026-08-16 bench dump).
# motor_lock/x_mode/y_mode/x_decay/y_decay and the hold currents are the
# GRBL controller's init writes (glowforge_io.c), step_freq its default
# machine tick, ramp_rate the module default; streaming is only ever 1
# inside a live job; the head white LED is a camera lamp, off at idle; the
# loop heater and TEC are the diagnostics' tools, off at idle.
# GRBL controller's init writes (glowforge_io.c): motor_lock 0, every axis
# in the pulse path (a job's Z moves the lens; the driver's Z soft limit
# guards it), step_freq its default machine tick, ramp_rate the module
# default; streaming is only ever 1 inside a live job; the head white LED
# is a camera lamp, off at idle; the loop heater and TEC are the
# diagnostics' tools, off at idle.
FIXED_SYSFS = [
("cnc/motor_lock", "8"),
("cnc/motor_lock", "0"),
("cnc/x_mode", "8"),
("cnc/y_mode", "8"),
("cnc/x_decay", "1"),
@@ -229,8 +231,10 @@ class Baseline:
def wait_settled(self, timeout=SETTLE_S, unreachable_s=10):
"""Block until forgectrl reports a settled supervisor: motion
verified (the probe passed), motion-fault (the ladder exhausted),
or standby (the manual stop lever). Gives up after unreachable_s
without an answer. Returns the last /mode body (None if unreachable)."""
standby (the manual stop lever), or gated (the commissioning gate
is closed: no controller spawns until it opens). Gives up after
unreachable_s without an answer. Returns the last /mode body (None
if unreachable)."""
t0 = time.time()
deadline = t0 + timeout
last = seen = heard = None
@@ -252,7 +256,7 @@ class Baseline:
seen = key
self.log("/mode controller=%s motion=%s" % key)
ctl = body.get("controller")
if ctl in ("motion-fault", "standby") or (ctl == "running" and body.get("motion") == "verified"):
if ctl in ("motion-fault", "standby", "gated") or (ctl == "running" and body.get("motion") == "verified"):
if ctl == "motion-fault":
self.log("WARNING - motion liveness ladder failed, controllers are "
"down (motion-fault); retry via POST /mode")
@@ -304,6 +308,20 @@ class Baseline:
return True, "already in %s mode" % want
self.log("switching to %s mode (found %s, controller %s)"
% (want, mode.get("mode"), mode.get("controller")))
if want == "cloud":
# Cloud mode exists only while cloud_enabled is 1. A test that
# needs it gets it turned on here, with a line in the log; it
# stays on afterward, as a cloud job the owner ran would leave it.
st, settings = self.fc_get("/settings")
if st == 200 and isinstance(settings, dict) and settings.get("cloud_enabled") != "1":
# the typed phrase the cloud step asks for: the runner
# gives it under the operator's rule for cloud tests
st, body = self.fc_post("/settings", data={"cloud_enabled": "1",
"phrase": "I UNDERSTAND"})
if st != 200:
return False, "cloud_enabled=1 for the test -> %s %s" % (st, body)
self.log("cloud mode turned on for the test (cloud_enabled was %r)"
% (settings.get("cloud_enabled") or ""))
self.nohunt_on(want)
st, body = self.fc_post("/mode", data={"controller": want})
if st != 200:
@@ -697,19 +715,20 @@ def check_fixed_against(ref, log):
# The attributes the GRBL controller writes at its own start (its analog
# config + machine tick): once they read the fixed values the controller
# has configured the machine. Before that the kernel shows the supervisor's
# motion-probe leftovers (motor_lock 0, step_freq 10000, y_mode at the
# module default) - the state /mode already calls "running", because
# "running" is the spawn, not the config.
CONFIGURED_MARKERS = [(a, dict(FIXED_SYSFS)[a]) for a in ("cnc/step_freq", "cnc/motor_lock", "cnc/y_mode")]
# motion-probe leftovers (step_freq 10000, y_mode at the module default) -
# the state /mode already calls "running", because "running" is the spawn,
# not the config. motor_lock is no marker: the probe and the controller
# both leave it 0.
CONFIGURED_MARKERS = [(a, dict(FIXED_SYSFS)[a]) for a in ("cnc/step_freq", "cnc/y_mode")]
CONFIGURED_TIMEOUT_S = 20
CONFIGURED_SETTLE_S = 1.0
def reference_preconfig(ref):
"""True when a saved reference shows the pre-controller state: every
marker present differs from its fixed value (the probe's step_freq /
motor_lock and the module's y_mode together), i.e. it was dumped
before the controller's init writes landed."""
marker present differs from its fixed value (the probe's step_freq and
the module's y_mode together), i.e. it was dumped before the
controller's init writes landed."""
sysfs = (ref or {}).get("sysfs") or {}
seen = [(sysfs.get(a), want) for a, want in CONFIGURED_MARKERS if sysfs.get(a) is not None]
return bool(seen) and all(got != want for got, want in seen)
forgetest/forgetest/hw.py
+2 -2
View File
@@ -3,7 +3,7 @@ kernel module's sysfs, the init scripts, and the Grbl TCP port.
Everything is reachable through environment overrides so the suite can be
exercised against a mock on a host:
FORGECTRL_URL default http://127.0.0.1:8080
FORGECTRL_URL default http://127.0.0.1
FORGECTRL_TOKEN_FILE default /data/forgefirm/panel.token
GF_SYSFS_ROOT default /sys/glowforge/ (must end with '/')
GRBL_HOST / GRBL_PORT default 127.0.0.1 / 23
@@ -41,7 +41,7 @@ class Forgectrl:
"""Thin client for the machine-services daemon."""
def __init__(self, base=None, token=None, timeout=10.0, slow_timeout=SLOW_TIMEOUT_S):
self.base = (base or os.environ.get("FORGECTRL_URL") or "http://127.0.0.1:8080").rstrip("/")
self.base = (base or os.environ.get("FORGECTRL_URL") or "http://127.0.0.1").rstrip("/")
self.timeout = timeout
self.slow_timeout = slow_timeout
self._token = token
forgetest/forgetest/runner.py
+10 -8
View File
@@ -424,15 +424,17 @@ class Context:
self.log("ACT %s %s: done after %.1f s", channel, state, dt)
return dt
def arm_press(self, text="The button lights white: press it to arm. The machine fires after your press."):
def arm_press(self, text="The button lights white: press it to arm. The machine fires after your press.",
lit_timeout=60):
"""The arm cue of a live test. A person's press by default: a
standing notice until the caller clears it. The fixture presses
only where the bench opted in (arm_press in its config) and its
button channel is enabled: a thread waits for the button to light
(the job may still be on its way to the arm wait) and presses
once, recorded as the fixture's; if the button never lights or
the press fails, the notice goes up for a person. Returns True
when the fixture has been asked."""
button channel is enabled: a thread waits up to `lit_timeout`
seconds for the button to light (the job may still be on its way
to the arm wait; a card that settles the coolant first takes
minutes) and presses once, recorded as the fixture's; if the
button never lights or the press fails, the notice goes up for a
person. Returns True when the fixture has been asked."""
fixture = getattr(self.runner, "fixture", None) if self.runner is not None else None
rec = {"channel": "button", "state": "arm", "by": "operator", "ts": now_ts()}
self.evidence.setdefault("actions", []).append(rec)
@@ -452,9 +454,9 @@ class Context:
self.log("ARM: the fixture presses when the button lights (the bench's arm_press opt-in)")
def press():
lit = self.wait_for(hw.button_lit, 60)
lit = self.wait_for(hw.button_lit, lit_timeout)
if lit is None:
self.log("ARM: the button never lit within 60 s - asking the operator")
self.log("ARM: the button never lit within %d s - asking the operator", lit_timeout)
self.notice(text)
return
try:
forgetest/forgetest/suite/__init__.py
+3
View File
@@ -4,6 +4,9 @@ order. Each module registers its tests with @catalog.test."""
from . import image # noqa: F401,E402
from . import kernel # noqa: F401,E402
from . import forgectrl # noqa: F401,E402
from . import commission # noqa: F401,E402
from . import commission_dark # noqa: F401,E402
from . import commission_sheet # noqa: F401,E402
from . import logs # noqa: F401,E402
from . import motion # noqa: F401,E402
from . import cooling # noqa: F401,E402
forgetest/forgetest/suite/camera.py
+65
View File
@@ -1,4 +1,6 @@
"""camera.* - the lid camera pipeline through forgectrl."""
import json
from ..catalog import test
from .. import hw
from ..baseline import LID_LAMP_ATTR
@@ -380,3 +382,66 @@ def lid_privacy(ctx):
ctx.check(st == 200 and data[:2] == b"\xff\xd8",
"snapshot after closing the lid -> %s", st)
ctx.log("lid closed again: snapshot %d bytes", len(data))
@test("camera.key-read", title="The camera key reads without a login", subsystem="camera",
kind="auto", est_min=1,
covers=[("forgectrl", "src/camkey.*"), ("forgectrl", "src/auth.c"),
("forgectrl", "src/main.c"), ("forgectrl", "src/ui/panel.js"),
("forgectrl", "src/ui/index.html"), ("forgectrl", "src/ui/help.js")],
requires=["forgectrl.auth"],
description="With the read routes closed to the network (panel_open_reads=0), a camera "
"status read from the board's LAN address is refused without the key and "
"served with it, on HTTP and on HTTPS; a wrong key is refused; the key never "
"authorizes a write; a rotation from the panel route stops the old key and "
"the new one reads. The setting is put back after.")
def key_read(ctx):
from .commission import request
from .forgectrl import lan_ip
fc = ctx.forgectrl
ev = ctx.evidence
ip = lan_ip()
ctx.check(ip, "cannot determine the board's LAN address")
http_base = "http://%s" % ip
tls_base = "https://%s" % ip
token = {"X-ForgeFIRM-Token": fc.token}
st, body, _ = request(fc.base, "GET", "/system/camera-key", headers=token)
ctx.check(st == 200, "GET /system/camera-key from loopback -> %s", st)
key = json.loads(body)["key"]
ctx.check(len(key) == 32, "the key is not 32 characters: %r", key)
ev["key_len"] = len(key)
before = fc.settings().get("panel_open_reads", "")
st, _ = fc.post("/settings", params={"panel_open_reads": "0"})
ctx.check(st == 200, "closing the reads -> %s", st)
try:
for base in (http_base, tls_base):
st, _, _ = request(base, "GET", "/cam/status")
ctx.log("GET %s/cam/status (no key) -> %s", base, st)
ctx.check(st == 403, "a closed read without the key -> %s, expected 403", st)
st, _, _ = request(base, "GET", "/cam/status?key=" + key)
ctx.log("GET %s/cam/status?key=... -> %s", base, st)
ctx.check(st == 200, "a read with the key -> %s, expected 200", st)
st, _, _ = request(base, "GET", "/cam/status", headers={"X-ForgeFIRM-Camera-Key": key})
ctx.check(st == 200, "a read with the key header -> %s, expected 200", st)
wrong = ("0" if key[0] != "0" else "1") + key[1:]
st, _, _ = request(base, "GET", "/cam/status?key=" + wrong)
ctx.check(st == 403, "a wrong key -> %s, expected 403", st)
# the key is a read credential only
st, _, _ = request(tls_base, "POST", "/settings", data={"panel_open_reads": "0"},
headers={"X-ForgeFIRM-Camera-Key": key})
ctx.log("POST /settings with the camera key -> %s", st)
ctx.check(st == 403, "the camera key authorized a write (%s)", st)
# rotation
st, body, _ = request(fc.base, "POST", "/system/camera-key?rotate=1", headers=token)
ctx.check(st == 200, "rotate -> %s", st)
new = json.loads(body)["key"]
ctx.check(new != key, "rotation kept the same key")
st, _, _ = request(http_base, "GET", "/cam/status?key=" + key)
ctx.check(st == 403, "the old key still reads after rotation (%s)", st)
st, _, _ = request(http_base, "GET", "/cam/status?key=" + new)
ctx.check(st == 200, "the new key does not read (%s)", st)
ev["rotated"] = True
finally:
fc.post("/settings", params={"panel_open_reads": before})
forgetest/forgetest/suite/cloud.py
+14 -17
View File
@@ -165,17 +165,6 @@ GFHOME_LOG = "/data/log/forgefirm/gfhome/gfhome.log"
HOMING_TIMEOUT_S = 600
def homing_mode_is_gfcloud():
"""Precheck: the web-service homing needs homing_mode = gfcloud."""
try:
hm = (hw.Forgectrl().settings() or {}).get("homing_mode")
except hw.HwError as e:
return "forgectrl unreachable: %s" % e
if hm != "gfcloud":
return "homing_mode is %r; the web-service homing needs gfcloud" % (hm,)
return None
def judge_hunt_with_lid_open(ctx, ev, offset):
"""The connect-time hunt from `offset` on: its terminal line is
:completed, nothing before it was refused for the lid, and the lens
@@ -246,9 +235,9 @@ def gfhome_homing(ctx, ev, g):
covers=_HOMING_PATH + [("forgectrl", "src/cool.*"), ("forgectrl", "src/airflow.*"),
("grblhal-glowforge", "src/**")],
requires=["forgectrl.auth", "motion.pacing"], actions=["lid"],
precheck=homing_mode_is_gfcloud,
steps=["Bed clear; cloud credentials configured and homing_mode = gfcloud; the machine on "
"the network.",
steps=["Bed clear; cloud credentials configured; the machine on the network. The test "
"turns cloud mode and the gfcloud homing on itself when they are off, and puts "
"the settings back at the end.",
"Open the lid when told and leave it open through the cloud client's connect and its "
"hunt; close it when told. Nothing else: the switch back and the $H homing run on "
"their own, and the head ends parked at the home corner."],
@@ -374,9 +363,17 @@ def mode_switch(ctx):
# -- the web-service homing from grbl mode --------------------------------
ev["homing_mode"] = (fc.settings() or {}).get("homing_mode")
ctx.check(ev["homing_mode"] == "gfcloud", "homing_mode is %r; $H needs gfcloud", ev["homing_mode"])
with ctx.grbl() as g:
gfhome_homing(ctx, ev, g)
if ev["homing_mode"] != "gfcloud":
st, body = fc.post("/settings", data={"homing_mode": "gfcloud"})
ctx.log("homing_mode=gfcloud for the homing -> %s %s", st, body if isinstance(body, str) else "")
ctx.check(st == 200, "homing_mode=gfcloud -> %s %s", st, body)
try:
with ctx.grbl() as g:
gfhome_homing(ctx, ev, g)
finally:
if ev["homing_mode"] != "gfcloud":
st, body = fc.post("/settings", data={"homing_mode": ev["homing_mode"] or "none"})
ctx.log("restore homing_mode=%r -> %s", ev["homing_mode"], st)
ctx.log("PASS: grbl -> cloud (session, hunt with the lid open, lens homed, airflow unjudged) -> "
"grbl (port open, %s), then $H homed in %.1f s", ev["grbl_state"], ev["homing_s"])
forgetest/forgetest/suite/commission.py
+1752
View File
File diff suppressed because it is too large Load Diff
forgetest/forgetest/suite/commission_dark.py
+327
View File
@@ -0,0 +1,327 @@
"""commission.check-* - the setup's checks (the dark wizards), driven the
way the page drives them: POST /wiz/<id>/start, GET /wiz/dark polled,
the prompts answered from here (the bench fixture works the lid and the
button; a confirmation is answered yes once the snapshot exists), the
result judged, and every setting a check wrote put back as found. Each
check re-runs on the real record: a completed check completes again at
the same version, so the record reads as before.
"""
import json
import os
import time
from ..catalog import test
from .. import hw
from ..baseline import read_position
from .commission import wiz, Restore # noqa: F401 - Restore is re-exported for the sheet
POLL_S = 1.0
DARK_COVERS = [("forgectrl", "src/wizdark.*"), ("forgectrl", "src/wizcalc.*"),
("forgectrl", "src/wiz.*"), ("forgectrl", "src/commission.*"),
("forgectrl", "src/main.c"), ("forgectrl", "src/ui/wizard.*")]
def dark(fc):
st, body = fc.get("/wiz/dark")
if st != 200 or not isinstance(body, dict):
raise hw.HwError("GET /wiz/dark -> %s" % st)
return body
def start(ctx, fc, wid):
ev = ctx.evidence
st, body = fc.post("/wiz/%s/start" % wid)
if st == 409 and dark(fc).get("id") == wid and dark(fc).get("running"):
# The same check left running by an earlier run of this test (an
# abort mid-wait): this run owns the machine, so it ends that one
# and starts its own.
ctx.log("the %s check is still running from an earlier run: aborting it", wid)
fc.post("/wiz/%s/abort" % wid)
ctx.wait_for(lambda: not dark(fc).get("running"), 30)
st, body = fc.post("/wiz/%s/start" % wid)
ctx.log("POST /wiz/%s/start -> %s %s", wid, st, body if isinstance(body, dict) else "")
ctx.check(st == 200, "the %s check did not start (%s %s)", wid, st, body)
ev["started"] = st == 200
def answer(fc, wid, prompt, value):
return fc.post("/wiz/%s/answer" % wid, data={"seq": str(prompt["seq"]), "value": value})
def run_check(ctx, wid, on_prompt, timeout_s):
"""Start `wid` and follow it to its end. on_prompt(prompt) is called
once per prompt (by sequence number) and returns the value to answer
with, or None when it handled the prompt another way (a fixture
action). Returns the final status document."""
fc = ctx.forgectrl
ev = ctx.evidence
start(ctx, fc, wid)
t0 = time.time()
seen = set()
phases = []
last = None
try:
while time.time() - t0 < timeout_s:
ctx.checkpoint()
d = dark(fc)
if d.get("id") != wid:
raise hw.HwError("another check took the slot: %s" % d.get("id"))
if d.get("phase") and (not phases or phases[-1] != d["phase"]):
phases.append(d["phase"])
ctx.log("%s: %s", wid, d["phase"])
p = d.get("prompt")
if p and p.get("seq") not in seen:
seen.add(p["seq"])
ctx.log("prompt %s (%s): %s", p.get("id"), p.get("kind"), p.get("text"))
value = on_prompt(p)
if value is not None:
st, body = answer(fc, wid, p, value)
ctx.check(st == 200, "the answer to %s -> %s %s", p.get("id"), st, body)
if not d.get("running"):
last = d
break
time.sleep(POLL_S)
except BaseException:
# An abort or a failure on the way out leaves no check running
# behind this test: the next run must be able to start its own.
fc.post("/wiz/%s/abort" % wid)
raise
ev["elapsed_s"] = round(time.time() - t0, 1)
ev["phases"] = phases[-12:]
ctx.check(last is not None, "the %s check did not end within %d s", wid, timeout_s)
if last is None:
fc.post("/wiz/%s/abort" % wid)
return {}
ev["log"] = last.get("log", [])[-12:]
ev["error"] = last.get("error")
ev["result"] = last.get("result")
ctx.check(not last.get("error"), "the %s check failed: %s", wid, last.get("error"))
ctx.check(isinstance(last.get("result"), dict), "the %s check ended without a result", wid)
w = wiz(fc)
ctx.check((w.get("versions") or {}).get(wid) == 1, "the record does not carry %s at version 1", wid)
return last
@test("commission.check-switches", title="The switches check follows the lid and the button",
subsystem="commission", kind="operator", est_min=3,
covers=DARK_COVERS + [("forgectrl", "src/status.c")],
requires=["forgectrl.auth"], actions=["lid", "button"],
description="POST /wiz/switches/start; the check asks for the lid to open and close and for "
"a press, each as a 'wait' prompt that the switch edge itself answers; the bench "
"fixture works the lid and the button. On a Basic or Plus the interlock loop reads "
"satisfied without a prompt. The result says lid, button, head present, and "
"whether the HV enable readback followed the lid; the record carries switches at "
"version 1.")
def check_switches(ctx):
def on_prompt(p):
pid = p.get("id", "")
if pid in ("lid-open", "lid-close-first", "lid-close"):
ctx.act("lid", "open" if pid == "lid-open" else "close",
until=lambda: (dark(ctx.forgectrl).get("prompt") or {}).get("seq") != p["seq"],
text=p.get("text", ""))
elif pid == "button-press":
ctx.act("button", "press",
until=lambda: (dark(ctx.forgectrl).get("prompt") or {}).get("id") != "button-press",
text=p.get("text", ""))
elif pid == "button-release":
pass # the fixture's press releases on its own
elif pid.startswith("interlock-"):
ctx.act("interlock", "open" if pid == "interlock-open" else "close",
until=lambda: (dark(ctx.forgectrl).get("prompt") or {}).get("seq") != p["seq"],
text=p.get("text", ""))
return None
last = run_check(ctx, "switches", on_prompt, 600)
r = last.get("result") or {}
ctx.check(r.get("lid") is True and r.get("button") is True, "lid or button not proven: %s", r)
ctx.check(r.get("head_present") is True, "the head did not read present: %s", r)
ctx.log("switches: %s", json.dumps(r))
@test("commission.check-sensors", title="The sensors check reads a plausible machine at rest",
subsystem="commission", kind="auto", est_min=2,
covers=DARK_COVERS + [("forgectrl", "src/status.c"), ("forgectrl", "src/accel.c"),
("forgectrl", "src/cool.c")],
requires=["forgectrl.auth"],
description="POST /wiz/sensors/start; ten seconds of readings, then the room-temperature "
"prompt, answered Skip so no offset is written. The result carries both coolant "
"temperatures, the chassis and SoC, the lid IR maxima, the accelerometer event "
"count, the supply power-good, the HV current, and the idle fan speeds; the "
"record carries sensors at version 1 and cool_temp_offset_c reads as before.")
def check_sensors(ctx):
with Restore(ctx, ["cool_temp_offset_c"]):
last = run_check(ctx, "sensors", lambda p: "Skip" if p.get("id") == "room-temp" else None, 120)
r = last.get("result") or {}
for k in ("coolant_down_c", "coolant_up_c", "lid_ir_max", "laser_pgood", "hv_current_max",
"exhaust_rpm_idle", "intake_rpm_idle"):
ctx.check(k in r, "the result lacks %s: %s", k, r)
ctx.check(r.get("laser_pgood") == 1, "power-good read %s", r.get("laser_pgood"))
ctx.log("sensors: coolant %s/%s C, IR %s, HV %s", r.get("coolant_down_c"), r.get("coolant_up_c"),
r.get("lid_ir_max"), r.get("hv_current_max"))
@test("commission.check-airflow", title="The airflow check measures the fans and sets the floors",
subsystem="commission", kind="auto", hardware="takeover", est_min=3,
covers=DARK_COVERS + [("forgectrl", "src/cool.c"), ("forgectrl", "src/airflow.*"),
("forgectrl", "src/gates.c"), ("forgectrl", "src/super.c")],
requires=["forgectrl.auth", "cooling.fan-gate-trips"],
description="POST /wiz/airflow/start: the controller stops, the fans run at the cut profile "
"for 35 s, the purge is measured on and off, and five settings are written: the "
"four floors at 55 percent of steady and the grace from the slowest spin-up. Each "
"fan must read at least 1000 rpm. The five settings are put back as found at the "
"end and the controller must be running again.")
def check_airflow(ctx):
keys = ["cool_tach_exhaust_min_rpm", "cool_tach_intake_min_rpm", "cool_tach_air_assist_min_rpm",
"cool_purge_min_current", "cool_fan_grace_s"]
with Restore(ctx, keys):
last = run_check(ctx, "airflow", lambda p: None, 240)
r = last.get("result") or {}
fans = r.get("fans") or {}
for name in ("exhaust", "intake 1", "intake 2", "air assist"):
f = fans.get(name) or {}
ctx.check(f.get("ok") is True, "%s: %s", name, f)
ctx.log("%s: %s rpm steady, 90 percent at %s s", name, f.get("steady_rpm"), f.get("spinup_s"))
floors = r.get("floors") or {}
for k in keys:
ctx.check(k in floors, "no floor for %s: %s", k, floors)
s = ctx.forgectrl.settings() or {}
for k in keys:
ctx.check(s.get(k) == floors.get(k), "%s reads %r, the check wrote %r", k, s.get(k), floors.get(k))
ctx.log("floors: %s", json.dumps(floors))
ok = ctx.wait_for(lambda: (ctx.forgectrl.get("/mode")[1] or {}).get("controller") == "running", 60)
ctx.check(ok is not None, "the controller did not come back after the check")
@test("commission.check-cameras", title="The cameras check captures both cameras",
subsystem="commission", kind="operator", est_min=2,
covers=DARK_COVERS + [("forgectrl", "src/cam.c")],
requires=["forgectrl.auth", "camera.snapshot"], actions=["lid"],
description="POST /wiz/cameras/start with the lid closed: a lid snapshot, the question, a "
"head snapshot, the question. Each question is answered yes once GET /wiz/shot "
"serves the snapshot as image/jpeg. The result names the sensor and says both "
"views were accepted.")
def check_cameras(ctx):
fc = ctx.forgectrl
ev = ctx.evidence
ctx.act("lid", "close")
shots = {}
def on_prompt(p):
pid = p.get("id", "")
if pid in ("lid-view", "head-view"):
cam = "lid" if pid == "lid-view" else "head"
st, body = fc.get("/wiz/shot", params={"cam": cam}, raw=True)
shots[cam] = {"status": st, "bytes": len(body) if body else 0,
"jpeg": bool(body) and body[:2] == b"\xff\xd8"}
ctx.check(st == 200 and shots[cam]["jpeg"], "no %s snapshot to judge: %s", cam, shots[cam])
return "Yes"
if pid == "lid-close":
ctx.act("lid", "close", until=lambda: (dark(fc).get("prompt") or {}).get("seq") != p["seq"])
return None
last = run_check(ctx, "cameras", on_prompt, 180)
ev["shots"] = shots
r = last.get("result") or {}
ctx.check(r.get("lid_ok") is True and r.get("head_ok") is True, "a view was not accepted: %s", r)
ctx.check(bool(r.get("sensor")) and r.get("sensor") != "unknown", "no sensor named: %s", r)
ctx.log("cameras: %s, lid %d bytes, head %d bytes", r.get("sensor"),
shots.get("lid", {}).get("bytes", 0), shots.get("head", {}).get("bytes", 0))
@test("commission.check-motion", title="The motion check proves the rail, the lens reference, and the jogs",
subsystem="commission", kind="auto", hardware="takeover", est_min=5,
covers=DARK_COVERS + [("forgectrl", "src/super.c"), ("forgectrl", "src/liveness.c"),
("forgectrl", "src/accel.c"), ("forgectrl", "src/cool.c")],
requires=["forgectrl.auth", "motion.pacing"],
description="POST /wiz/motion/start: the controller stops, the liveness probe runs, the lens "
"finds the hall reference in five agreeing passes, the controller comes back in "
"loopback posture, and the head jogs 50 mm each way on X and Y with the "
"accelerometer as the witness. The one prompt (the jogs are about to move) is "
"answered Continue. The result carries the probe line, the passes, and a "
"witnessed reading for each jog; the controller must be back in its normal "
"posture, and the kernel position counters must read what they read before: "
"every move played to its end (the probe's return leg under the cooling "
"engine's dead-man, the last jog under the controller stop).")
def check_motion(ctx):
fc = ctx.forgectrl
before = read_position()
ctx.notice("The head moves 50 mm each way on X and on Y. Keep the bed clear.")
last = run_check(ctx, "motion", lambda p: "Continue" if p.get("id") == "jogs" else None, 420)
ctx.clear_notice()
r = last.get("result") or {}
ctx.check(r.get("z_referenced") is True, "the lens was not referenced: %s", r)
ctx.check(len(r.get("z_passes") or []) == 5, "not five reference passes: %s", r.get("z_passes"))
moves = r.get("moves") or {}
for name in ("+X", "-X", "+Y", "-Y"):
m = moves.get(name) or {}
ctx.check(m.get("witnessed") is True, "the %s jog was not witnessed: %s", name, m)
ok = ctx.wait_for(lambda: (fc.get("/mode")[1] or {}).get("local") is False
and (fc.get("/mode")[1] or {}).get("controller") == "running", 90)
ctx.check(ok is not None, "the controller is not back in its normal posture")
after = read_position()
if before is not None and after is not None:
ctx.check(all(abs(a - b) <= 2 for a, b in zip(after[:2], before[:2])),
"the check lost motion: position %s before, %s after (steps)", before, after)
ctx.log("motion: probe '%s', passes %s, rest %s, moves %s", r.get("probe"), r.get("z_passes"),
r.get("rest"), {k: (v.get("p2p_lp_x"), v.get("p2p_lp_y")) for k, v in moves.items()})
@test("commission.check-flow-verify", title="The flow check runs as a setup check",
subsystem="commission", kind="auto", hardware="takeover", est_min=5,
covers=DARK_COVERS + [("forgectrl", "src/diag.c")],
requires=["forgectrl.auth", "cooling.aa-offset-calibrate", "cooling.flow-verify"],
description="POST /wiz/cooling.flow-verify/start drives the flow-verify diagnostic through the "
"check runner: the diagnostic's phases show as the check's, its result becomes the "
"check's result with pass true, and the record carries cooling.flow-verify at "
"version 1. A thin margin recommends the calibration in the record's flags. The "
"air-assist offset calibration is named first among the prerequisites so a queue "
"keeps it ahead of every heater tool this check pulls forward.")
def check_flow_verify(ctx):
last = run_check(ctx, "cooling.flow-verify", lambda p: None, 600)
r = last.get("result") or {}
ctx.check(r.get("pass") is True, "the flow check did not pass: %s", r)
ctx.log("flow check: threshold %s, flow %s, no-flow %s, thin %s", r.get("threshold"),
r.get("flow_rise"), r.get("noflow_rise"), r.get("thin_margin"))
@test("commission.cloud-header-capture", title="The cloud header check takes one print's envelope",
subsystem="commission", kind="operator", hardware="takeover", mode="grbl", est_min=8,
covers=DARK_COVERS + [("forgectrl", "src/super.c"),
("python3-gfhardware", "gfhardware/machine.py"),
("python3-gfhardware", "forgefirm-app/gfcloud.py"),
("python3-gfhardware", "forgefirm-app/ffmachine.py")],
requires=["forgectrl.auth", "cloud.mode-switch"], hands=["workstation"],
steps=["The Glowforge app open on the workstation, signed in to this machine. The test "
"turns cloud mode on itself and puts it back.",
"Once the app shows the machine online, place any small design on the bed image "
"and press Print. That is the only action: the check takes the print's header and "
"cancels the print before it arms. Do not press the machine's button.",
"The machine returns to GRBL mode on its own."],
description="POST /wiz/cloud.header/start with cloud mode on: the wizard writes the "
"client's one-start capture marker, starts cloud mode in the wizard's posture, "
"asks for a print, and takes the header the client writes to the run directory "
"before it cancels the print. The result carries the tag count, the "
"calibration-bearing tags, the limits, and the machine's own numbers; the log "
"has the tagged block; the machine is back in GRBL mode with the marker gone.")
def cloud_header_capture(ctx):
fc = ctx.forgectrl
ev = ctx.evidence
with Restore(ctx, ["cloud_enabled", "homing_mode", "controller_mode"]):
s = fc.settings() or {}
if s.get("cloud_enabled") != "1":
st, body = fc.post("/settings", data={"cloud_enabled": "1"})
ctx.check(st == 200, "cloud_enabled=1 -> %s %s", st, body)
ctx.log("cloud mode turned on for the test")
ctx.notice("Print any small design from the Glowforge app once it shows the machine "
"online. Do not press the machine's button.")
last = run_check(ctx, "cloud.header",
lambda p: "Continue" if p.get("id") == "print" else None, 900)
ctx.clear_notice()
r = last.get("result") or {}
ctx.check(isinstance(r.get("tag_count"), int) and r["tag_count"] > 100,
"the header carried %s tags", r.get("tag_count"))
kept = r.get("calibration_tags") or {}
ctx.check(len(kept) > 10, "only %d calibration-bearing tags kept", len(kept))
ev["kept"] = sorted(kept)[:40]
ctx.check(not os.path.exists("/run/gfcloud-capture"), "the capture marker is still down")
ok = ctx.wait_for(lambda: (fc.get("/mode")[1] or {}).get("mode") == "grbl"
and (fc.get("/mode")[1] or {}).get("controller") == "running", 120)
ctx.check(ok is not None, "the machine did not return to GRBL mode")
ctx.log("header: %d tags, %d kept; job %s", r.get("tag_count", 0), len(kept), r.get("job_id"))
forgetest/forgetest/suite/commission_sheet.py
+195
View File
@@ -0,0 +1,195 @@
"""commission.sheet - the setup's sheet (the live wizards) in one run,
driven the way the page drives them: POST /wiz/<id>/start, GET /wiz/dark
polled, the prompts answered from here, each burn armed by a press, the
result judged, and every setting a wizard wrote put back as found. One
piece of wood at least 200 x 150 mm (8 x 6 in) carries the placement,
the frame, and the five cards. Each wizard re-runs on the real record,
so a completed one completes again at the same version.
The operator's part is one press on the machine at the start (the
presence check of the ready gate; with the bench actuator up it then
makes every arm press) and nothing else. The judgments a person makes
by eye on the sheet (the thinnest line, the faintest rung, the best
corner, the frame's fit) are answered here with a middle option or the
witnesses: the test proves each wizard's mechanics and the emission
witnesses, not the operator's eye, and the settings it writes are
restored. The page is where the real numbers go in.
"""
import re
from ..catalog import test
from .commission_dark import run_check, Restore
SHEET_COVERS = [("forgectrl", "src/wizlive.*"), ("forgectrl", "src/sheet.*"),
("forgectrl", "src/font_hershey.*"), ("forgectrl", "src/jobstream.*"),
("forgectrl", "src/curverec.*"), ("forgectrl", "src/wizdark.*"),
("forgectrl", "src/wiz.*"), ("forgectrl", "src/commission.*"),
("forgectrl", "src/main.c"), ("forgectrl", "src/ui/wizard.*"),
("forgectrl", "tools/hershey_gen.py"), ("forgectrl", "src/super.c"),
("forgectrl", "src/status.c"), ("forgectrl", "src/cool.c"),
("forgectrl", "src/accel.*"), ("forgectrl", "src/wizcalc.*"),
("grblhal-glowforge", "src/glowforge_laser.c"),
("grblhal-glowforge", "src/glowforge_status.c"),
("grblhal-glowforge", "src/glowforge_homing.*"),
("grblhal-glowforge", "src/glowforge_io.*")]
# The live wizards in the order the sheet burns them: the wizard id, the
# answers to its prompts (a middle pick where a person would look at the
# sheet), the keys its result must carry, and how long the bench actuator
# waits for the button to light before the arm press goes to a person.
# The flow-load card settles the coolant loop before it lights the button
# (60 to 240 s); the others light within seconds of the lens reference.
CARDS = [
("sheet.frame", {"frame-arm": "Continue", "frame-ok": "Yes"}, ["mark_s", "mark_feed"], 120),
("laser.focus", {"focus-arm": "Continue", "focus-pick": "11", "thickness": "Keep"},
["pick", "thickness_mm", "pick_half_steps", "edge_z_mm", "steps_per_mm", "max_height_mm",
"focus_range_mm", "stops"], 120),
("laser.floor", {"floor-arm": "Continue", "floor-pick": "8"}, ["faintest_density", "floor_density"], 120),
("laser.dose-curve", {"dose-arm": "Continue"}, ["points", "curve"], 120),
("laser.corner", {"corner-arm": "Continue", "corner-pick": "1.50"}, ["gamma"], 120),
("cooling.flow-load", {"load-arm": "Continue"}, ["lit_s", "dose_raw_s", "peak_c", "k_density", "k_cw"], 420),
]
# Every setting a card writes, restored as found when the run ends.
CARD_SETTINGS = ["lens_hall_edge_z_mm", "lens_stop_below_steps", "lens_stop_above_steps",
"laser_floor_density", "laser_dose_curve", "laser_corner_gamma",
"cool_laser_heat_density", "cool_laser_heat_cw"]
# The evidence run_check and the witnesses write at the top level, moved
# under the card's own key once it is done.
CARD_EVIDENCE = ("started", "elapsed_s", "phases", "log", "error", "result",
"preview_bytes", "program_lines", "emission")
def preview_ok(ctx, wid):
"""The page's preview and the program for the wizard are served."""
fc = ctx.forgectrl
st, body = fc.get("/wiz/sheet.svg?card=%s" % wid, raw=True)
ctx.check(st == 200 and body.lstrip().startswith(b"<svg"), "no preview for %s (%s)", wid, st)
ctx.evidence["preview_bytes"] = len(body)
st, body = fc.get("/wiz/sheet.gcode?card=%s" % wid, raw=True)
# The text and the frame are M4 at the mark dose; the patterns bring
# their own M3 or M4 lines. A program is one with a laser-on command.
ctx.check(st == 200 and re.search(rb"^M[34] S\d", body or b"", re.M) is not None,
"no program for %s (%s)", wid, st)
ctx.evidence["program_lines"] = body.count(b"\n")
def emission_ok(ctx, last):
"""The three witnesses of the burn, from the wizard's result."""
e = (last.get("result") or {}).get("emission") or {}
ctx.evidence["emission"] = e
ctx.check(e.get("hv_max", 0) > 30, "the tube current never rose (%s)", e.get("hv_max"))
ctx.check(e.get("laser_on_samples", 0) > 0, "the kernel never sampled LASER_ON")
ctx.check(e.get("thermopile_delta", 0) >= 100, "the thermopile did not see the beam (%s)",
e.get("thermopile_delta"))
def file_card(ctx, wid):
"""Move the card's evidence under its own key."""
ev = ctx.evidence
ev.setdefault("cards", {})[wid] = {k: ev.pop(k) for k in CARD_EVIDENCE if k in ev}
def thickness_answer(ctx):
"""The thickness the placement is answered with, in the machine's
units (the number prompts read ui_units), and the millimeters the
record must carry for it."""
imperial = (ctx.forgectrl.settings() or {}).get("ui_units") == "imperial"
return ("0.125", 3.175) if imperial else ("3.2", 3.2)
def place(ctx):
"""The placement: the lens reference, two jogs, the origin at the
head's position, the full sheet, a thickness. Nothing fires."""
jogs = ["X+10", "X-10"]
answer, want_mm = thickness_answer(ctx)
ctx.evidence["thickness_answer"] = answer
def on_prompt(p):
# Each jog answer closes the prompt and the wizard asks again with
# the next sequence number, so the pad is answered one move at a
# time until the origin is set.
if p.get("id") == "place":
return jogs.pop(0) if jogs else "Set origin"
if p.get("id") == "sheet-kind":
return "Full sheet"
if p.get("id") == "thickness":
return answer
return None
last = run_check(ctx, "sheet.place", on_prompt, 300)
r = last.get("result") or {}
ctx.check("origin_x" in r and "origin_y" in r, "no origin in the result: %s", r)
ctx.check((r.get("steps_per_mm") or {}).get("x", 0) > 0, "no steps per mm: %s", r)
ctx.check(r.get("z_referenced") is True, "the lens was not referenced")
ctx.check(abs((r.get("thickness_mm") or 0) - want_mm) < 0.01,
"thickness %s mm, expected %s (answered %s)", r.get("thickness_mm"), want_mm, answer)
ctx.check(r.get("alone") is False, "alone %s", r.get("alone"))
ctx.log("placed: datum at %s, %s; steps per mm %s", r.get("origin_x"), r.get("origin_y"),
r.get("steps_per_mm"))
file_card(ctx, "sheet.place")
def burn(ctx, wid, answers, want, lit_s):
"""One live card: the preview, the burn after the press with the
prompts answered from `answers`, the witnesses, the keys of the
result in `want`. The actuator waits `lit_s` for the button to light."""
preview_ok(ctx, wid)
ctx.arm_press(lit_timeout=lit_s)
def on_prompt(p):
return answers.get(p.get("id"))
try:
last = run_check(ctx, wid, on_prompt, 900)
finally:
ctx.clear_notice()
emission_ok(ctx, last)
r = last.get("result") or {}
missing = [k for k in want if k not in r]
ctx.check(not missing, "the result lacks %s: %s", missing, sorted(r))
ctx.log("%s: %s, emission %s", wid, {k: r.get(k) for k in want}, ctx.evidence.get("emission"))
file_card(ctx, wid)
@test("commission.sheet", title="The commissioning sheet: the placement, the frame, and the five cards",
subsystem="commission", kind="live", hardware="takeover", mode="grbl", est_min=35,
covers=SHEET_COVERS,
requires=["forgectrl.auth", "commission.check-motion", "laser.emission-witness", "cooling.flow-verify"],
actions=["button"], hands=["scrap"],
steps=["A piece of wood at least 200 x 150 mm (8 x 6 in) on the bed, pushed as far left as "
"it goes with its top edge at the top of the cut area (the head's home corner); "
"lid closed. Eye protection on, exhaust on, extinguisher in reach.",
"Start the test and press the machine's button once. That press says you are at the "
"machine, and the bench actuator makes every arm press after it. With no actuator "
"the start is the Ready answer instead, and you press the button each time it lights "
"white: six times, the frame and the five cards.",
"Nothing else: the test answers every prompt itself and puts back every setting the "
"cards write. It runs about half an hour, mostly the coolant settle and the dark "
"tails."],
description="One run over the sheet's seven wizards on one piece of wood. "
"POST /wiz/sheet.place/start references the lens, takes the controller in "
"loopback posture, jogs +X 10 and -X 10, and sets the origin at the head "
"(Full sheet; the thickness answered in the machine's units, 3.2 mm or "
"0.125 in). Then, each after the press and each with its preview "
"and program served by GET /wiz/sheet.svg and /wiz/sheet.gcode: the frame "
"(180 x 130 mm) and the header band at the mark dose; the focus card (the "
"lens homed on its bottom stop, the hall edge counted, twelve lines down the "
"travel; the pick is line 11, the thickness kept); the floor card (twelve "
"rungs from 2 to 24 percent with the floor and the curve off for the job; the "
"pick is 8, the floor written is 10); the dose card (seven rungs sampled at "
"25 Hz, the curve fit on 20 s of dark); the corner card (five gammas reloaded "
"through M102 inside one armed job; the pick is 1.50); and the flow-load card "
"(the coolant loop settled, the box and the patch at 60 percent, the "
"downstream peak over 100 s of dark as the coefficient). Every burn must be "
"seen by the three witnesses (the tube current, the kernel's LASER_ON samples, "
"the thermopile rise); every result must carry its keys; the record must "
"carry each wizard at version 1; the seven settings the cards write are "
"restored at the end.")
def sheet(ctx):
ctx.ready("The sheet: the placement, then the frame and the five cards burn after a press "
"each. The bench presses when the button lights.")
place(ctx)
with Restore(ctx, CARD_SETTINGS):
for wid, answers, want, lit_s in CARDS:
burn(ctx, wid, answers, want, lit_s)
ctx.log("PASS: the sheet's seven wizards ran on one piece; settings restored")
forgetest/forgetest/suite/forgectrl.py
+49 -13
View File
@@ -6,7 +6,12 @@ import time
from ..catalog import test
from .. import hw
_COVERS_AUTH = [("forgectrl", "src/auth.*"), ("forgectrl", "src/peer.*"), ("forgectrl", "src/main.c")]
# The write guard reads the account and the session store (a machine with
# an account needs a session from the LAN; this host and the dev image
# write with the token), so both are the guard's domain.
_COVERS_AUTH = [("forgectrl", "src/auth.*"), ("forgectrl", "src/peer.*"), ("forgectrl", "src/main.c"),
("forgectrl", "src/session.*"), ("forgectrl", "src/users.*"),
("forgectrl", "src/ui/login.js"), ("forgectrl", "src/ui/wizard.js")]
def lan_ip():
@@ -27,12 +32,15 @@ def lan_ip():
@test("forgectrl.auth", title="API access control", subsystem="forgectrl", kind="auto", est_min=1,
covers=_COVERS_AUTH,
description="Every state-changing endpoint refuses an unauthenticated write; a non-literal "
description="Every state-changing endpoint refuses an unauthenticated write (the factory "
"return, the SSH switch and the wizard's own routes included); a non-literal "
"Host, a non-literal Origin and a cross-site Sec-Fetch-Site are refused; the "
"cooling report channel accepts the loopback peer and refuses a non-loopback "
"one; the fuse view is two-factor "
"one (over HTTP the write is sent to HTTPS first, 302; over HTTPS the route "
"answers 403 loopback only); the fuse view is two-factor "
"(token and the physical button) and refused without either; "
"the flash and factory-restore chain is refused unauthenticated.")
"the flash and factory-restore chain is refused unauthenticated; a page "
"asked for without a session is sent to the login carrying its path.")
def auth(ctx):
fc = ctx.forgectrl
ev = ctx.evidence
@@ -46,7 +54,10 @@ def auth(ctx):
("/mode", {"controller": "grbl"}), ("/settings", {"ui_units": "mm"}),
("/diag/flow-verify", None), ("/diag/abort", None),
("/update/apply", None), ("/boot", {"target": "a"}),
("/system/reboot", None), ("/restore/factory", None)):
("/system/reboot", None), ("/restore/factory", None),
("/restore/factory-return", {"confirm": "1"}), ("/system/ssh", {"enable": "1"}),
("/wiz/advisories/accept", None), ("/wiz/account", None),
("/wiz/complete", None)):
st, body = fc.post(path, params=params, auth=False)
ctx.log("POST %s (no token) -> %s %s", path, st, body if isinstance(body, dict) else "")
ev["noauth " + path] = st
@@ -92,7 +103,7 @@ def auth(ctx):
ev["sfs_cross"] = st
ctx.log("GET /status Sec-Fetch-Site=cross-site -> %s", st)
ctx.check(st == 403, "a cross-site fetch was accepted (%s)", st)
st, body = fc.get("/status", headers={"Sec-Fetch-Site": "same-origin", "Origin": "http://127.0.0.1:8080"})
st, body = fc.get("/status", headers={"Sec-Fetch-Site": "same-origin", "Origin": "http://127.0.0.1"})
ctx.check(st == 200, "same-origin literal Origin refused (%s)", st)
# the fuse view is two-factor: the token AND the physical button held.
@@ -122,17 +133,42 @@ def auth(ctx):
ctx.check(st == 200, "/cool/state refused the loopback peer (%s %r): the controller's "
"reports never reach the engine", st, body)
# ...and a non-loopback peer is refused, even with a token
# ...and a non-loopback peer is refused, even with a token. Over HTTP
# the write is sent to HTTPS first (302, the listener's rule); over
# HTTPS the route itself refuses the peer (403 loopback only). Neither
# request follows the redirect, and the self-signed certificate is
# not verified.
from .commission import request, decode
ip = lan_ip()
ev["lan_ip"] = ip
ctx.check(ip, "cannot determine the board's LAN address")
port = fc.base.rsplit(":", 1)[-1]
lan = hw.Forgectrl("http://%s:%s" % (ip, port), token=fc.token)
st, body = lan.post("/cool/state", params={"mode": "idle", "armed": "0"})
token = {"X-ForgeFIRM-Token": fc.token}
report = {"mode": "idle", "armed": "0"}
st, body, hdrs = request("http://%s" % ip, "POST", "/cool/state", data=report, headers=token)
loc = hdrs.get("location", "")
ev["cool_state_from_lan_http"] = {"status": st, "location": loc}
ctx.log("POST http://%s/cool/state -> %s %s", ip, st, loc)
ctx.check(st == 302 and loc.startswith("https://"),
"a LAN write over HTTP -> %s %r, expected a 302 to HTTPS", st, loc)
st, body, hdrs = request("https://%s" % ip, "POST", "/cool/state", data=report, headers=token)
b = decode(body)
ev["cool_state_from_lan"] = st
ctx.log("POST /cool/state from %s -> %s %s", ip, st, body if isinstance(body, dict) else "")
ctx.check(st == 403 and isinstance(body, dict) and body.get("error") == "loopback only",
"/cool/state accepted a non-loopback peer (%s %r)", st, body)
ctx.log("POST https://%s/cool/state -> %s %s", ip, st, b if isinstance(b, dict) else "")
ctx.check(st == 403 and isinstance(b, dict) and b.get("error") == "loopback only",
"/cool/state accepted a non-loopback peer (%s %r)", st, b)
# the login return path: a page asked for from the LAN without a
# session is sent to the login carrying the path it asked for (with
# its query), so the login can come back to it. Before the setup has
# made the account the page is served instead (200).
st, body, hdrs = request("https://%s" % ip, "GET", "/setup?step=laser.focus")
loc = hdrs.get("location", "")
ev["setup_from_lan"] = {"status": st, "location": loc}
ctx.log("GET https://%s/setup?step=laser.focus (no session) -> %s %s", ip, st, loc)
ctx.check(st in (200, 302), "the setup page from the LAN -> %s", st)
if st == 302:
ctx.check(loc == "/login?next=/setup%3Fstep%3Dlaser.focus",
"the login redirect does not carry the path: %r", loc)
@test("forgectrl.settings-bounds", title="Settings validation and restore", subsystem="forgectrl",
forgetest/forgetest/suite/image.py
+54
View File
@@ -252,3 +252,57 @@ def image_health(ctx):
ctx.check("kernel-module-glowforge" in manifest.components, "manifest lacks kernel-module-glowforge")
ctx.check("linux-fslc" in manifest.components, "manifest lacks the kernel entry")
ctx.log("manifest %s: %d components", manifest.content_sha[:12], len(manifest.components))
@test("image.license-bundle", title="The license texts travel with the image",
subsystem="image", kind="auto", est_min=1,
covers=[("forgectrl", "src/main.c"), ("forgectrl", "src/ui/index.html"),
("forgectrl", "src/ui/wizard.html"), ("forgectrl", "src/ui/login.html")],
requires=["forgectrl.panel-serves"],
description="The image carries /usr/share/forgefirm/licenses.tar.gz, a gzip tar with "
"the license manifest (every installed package with its license) and the "
"license texts, and no loose common-licenses tree; the panel serves the "
"bundle, the manifest as text, and the Licenses page every footer links, "
"each without a login.")
def license_bundle(ctx):
import io
import tarfile
fc = ctx.forgectrl
ev = ctx.evidence
path = "/usr/share/forgefirm/licenses.tar.gz"
ctx.check(os.path.isfile(path), "%s is missing", path)
ctx.check(not os.path.isdir("/usr/share/common-licenses"),
"the loose common-licenses tree is still on the rootfs")
size = os.path.getsize(path)
ev["bundle_bytes"] = size
ctx.log("%s: %d bytes", path, size)
with tarfile.open(path, "r:gz") as t:
names = t.getnames()
ctx.check("common-licenses/license.manifest" in names, "no license.manifest in the bundle")
manifest = t.extractfile("common-licenses/license.manifest").read().decode("utf-8", "replace")
packages = manifest.count("PACKAGE NAME: ")
texts = sum(1 for n in names if "/generic_" in n and not n.endswith("/"))
ev["packages"] = packages
ev["generic_texts"] = texts
ctx.log("manifest lists %d packages; %d generic license entries", packages, texts)
ctx.check(packages >= 40, "only %d packages in the manifest", packages)
ctx.check(texts >= 5, "only %d generic license entries", texts)
for name in ("forgectrl", "grblhal-glowforge", "kernel-module-glowforge"):
ctx.check("RECIPE NAME: %s\n" % name in manifest, "%s is not in the manifest", name)
st, body = fc.get("/system/licenses/manifest", raw=True)
ctx.check(st == 200, "GET /system/licenses/manifest -> %s", st)
ctx.check(isinstance(body, (bytes, bytearray)) and b"PACKAGE NAME: " in body,
"the served manifest is not the manifest")
st, body = fc.get("/system/licenses", raw=True)
ctx.check(st == 200, "GET /system/licenses -> %s", st)
ctx.check(isinstance(body, (bytes, bytearray)) and len(body) == size,
"the served bundle is %s bytes, the file %d", len(body) if body else None, size)
with tarfile.open(fileobj=io.BytesIO(bytes(body)), mode="r:gz") as t:
ctx.check("common-licenses/license.manifest" in t.getnames(),
"the served bundle holds no manifest")
st, body = fc.get("/licenses", raw=True)
ctx.check(st == 200, "GET /licenses -> %s", st)
text = bytes(body).decode("utf-8", "replace") if body else ""
ctx.check("/system/licenses" in text and "PACKAGE NAME: " in text,
"the Licenses page lacks the download link or the manifest")
forgetest/forgetest/suite/laser.py
+16
View File
@@ -93,6 +93,7 @@ def sample(ctx):
"verdict": cs.get("verdict"),
"beam": hw.sysfs_int("head/beam_detect_analog"),
"beam_d": hw.sysfs_int("head/beam_detect_digital"),
"exhaust": (st.get("fans") or {}).get("exhaust"),
"button_lit": hw.button_lit(),
"hv_enable": (st.get("switches") or {}).get("hv_enable"),
"button_latch": hw.sysfs_int("cnc/button_latch"),
@@ -494,6 +495,18 @@ def emission_witness(ctx):
"HV_ENABLE dipped=%s, back with emission after it=%s", gap["button_latch_unlocked_max"],
gap["button_latch_unlocked_samples"], gap["button_latch_set_at"], gap["hv_enable_dipped"],
gap["hv_enable_back_lit"])
# The first fire waits for the airflow: the exhaust reads at or
# above its floor on the first sample that shows emission.
floor = (ctx.forgectrl.settings() or {}).get("cool_tach_exhaust_min_rpm") or ""
try:
floor = float(floor) if floor else 6400.0
except ValueError:
floor = 6400.0
first_lit = next((s for s in samples if s["emission"]), None)
ev["exhaust_at_first_fire"] = first_lit["exhaust"] if first_lit else None
ev["exhaust_floor"] = floor
ctx.log("exhaust at the first emission sample: %s rpm (floor %.0f)",
ev["exhaust_at_first_fire"], floor)
# X-3: job-based disarm at Idle after M2
dt = wait_disarm(ctx, 75)
ev["disarm_after_idle_s"] = round(dt, 1) if dt is not None else None
@@ -506,6 +519,9 @@ def emission_witness(ctx):
"their idle duty (phase %s at t=%s s, emission %s)",
first_idle_fire[1], first_idle_fire[0], first_idle_fire[2])
ctx.check(end == 0, "emission_samples did not return to 0 at Idle (%s)", end)
ctx.check(ev["exhaust_at_first_fire"] is not None and ev["exhaust_at_first_fire"] >= floor,
"the beam started before the exhaust reached its floor (%s rpm at the first "
"emission sample, floor %.0f)", ev["exhaust_at_first_fire"], floor)
ctx.check(hv and max(hv) > min(hv), "HV current did not rise during the burn (%s..%s)",
ev["hv_min"], ev["hv_max"])
ctx.check(dt is not None and dt < 10.0,
forgetest/forgetest/suite/logs.py
+16 -11
View File
@@ -28,7 +28,7 @@ _LOG_COVERS = [("forgectrl", "src/logs.*"), ("forgectrl", "src/fflog.*"), ("forg
covers=_LOG_COVERS, requires=["forgectrl.auth"],
description="/logs lists the loggers with their levels and files, /logs/tail returns the "
"forgectrl logger's tail, and POST /logs/export streams a sanitized tar.gz "
"bundle that contains no panel token.")
"bundle that contains neither the panel token nor the camera key.")
def tree_tail_export(ctx):
fc = ctx.forgectrl
ev = ctx.evidence
@@ -66,16 +66,21 @@ def tree_tail_export(ctx):
ctx.log("bundle: %d members, e.g. %s", len(members), members[:5])
ctx.check(members, "empty bundle")
ctx.check(any(m.endswith("README.txt") for m in members), "sanitized bundle lacks README.txt")
token = fc.token
if token:
leaked = []
for m in tf.getmembers():
if m.isfile():
content = tf.extractfile(m).read()
if token.encode() in content:
leaked.append(m.name)
ev["token_leaks"] = leaked
ctx.check(not leaked, "the sanitized bundle contains the panel token: %s", leaked)
# the two secrets the sanitizer knows by value and a bundle could
# carry: the panel token and the camera key (a URL parameter that
# lands in sender logs)
secrets = {}
if fc.token:
secrets["panel token"] = fc.token
st, body = fc.get("/system/camera-key")
if st == 200 and isinstance(body, dict) and body.get("key"):
secrets["camera key"] = body["key"]
ctx.check("camera key" in secrets, "GET /system/camera-key -> %s: no key to check for", st)
contents = [(m.name, tf.extractfile(m).read()) for m in tf.getmembers() if m.isfile()]
for what, value in secrets.items():
leaked = [name for name, content in contents if value.encode() in content]
ev[what.replace(" ", "_") + "_leaks"] = leaked
ctx.check(not leaked, "the sanitized bundle contains the %s: %s", what, leaked)
# The routing test proves the whole path every logger takes: emitter (or
forgetest/forgetest/suite/motion.py
+1 -1
View File
@@ -392,7 +392,7 @@ def _liveness_masked_restart(ctx, fc, ev):
ctx.check(lines and "MOTION OK" in lines[0],
"the first probe after the restart was not MOTION OK: %s", lines[:1])
ctx.check(len(lines) == 1, "the probe needed the recovery ladder (%d probes) - a false dead verdict", len(lines))
ctx.check(ctx.sysfs("cnc/motor_lock") == "8", "motor_lock reads %s after the controller start (expected 8)",
ctx.check(ctx.sysfs("cnc/motor_lock") == "0", "motor_lock reads %s after the controller start (expected 0)",
ctx.sysfs("cnc/motor_lock"))
ctx.check(fc.wait_idle(15, abort=ctx.aborted), "machine not idle after the probe")
x1 = _kernel_x_mm(ctx)
forgetest/tests/helpers.py
+13 -1
View File
@@ -108,6 +108,7 @@ class FakeForgectrl:
}
self.posts = []
self.on_post = None
self.on_get = None # (path, query) -> (status, body) | (status, bytes, ctype) | None
fake = self
class H(http.server.BaseHTTPRequestHandler):
@@ -123,7 +124,18 @@ class FakeForgectrl:
self.wfile.write(data)
def do_GET(self):
path = self.path.split("?", 1)[0]
path, _, query = self.path.partition("?")
if fake.on_get:
r = fake.on_get(path, dict(_up.parse_qsl(query, keep_blank_values=True)))
if r is not None:
if isinstance(r[1], (bytes, bytearray)):
self.send_response(r[0])
self.send_header("Content-Type", r[2] if len(r) > 2 else "application/octet-stream")
self.send_header("Content-Length", str(len(r[1])))
self.end_headers()
self.wfile.write(r[1])
return
return self._send(r[0], r[1])
key = {"/mode": "mode", "/status": "status", "/cool/status": "cool", "/cam/status": "cam",
"/diag/status": "diag", "/settings": "settings",
"/logs/tail": "logs_tail"}.get(path)
forgetest/tests/test_baseline.py
+19 -5
View File
@@ -82,11 +82,11 @@ class BaselineTests(unittest.TestCase):
self.assertEqual(set(items), {"cnc/motor_lock", "cnc/step_freq", "cnc/streaming"})
for x in left:
self.assertEqual(x.action, "restored", str(x))
self.assertEqual(self._read("cnc/motor_lock"), "8")
self.assertEqual(self._read("cnc/motor_lock"), "0")
self.assertEqual(self._read("cnc/step_freq"), "28160")
self.assertEqual(self._read("cnc/streaming"), "0")
self.assertEqual(items["cnc/motor_lock"].found, "15")
self.assertEqual(items["cnc/motor_lock"].expected, "8")
self.assertEqual(items["cnc/motor_lock"].expected, "0")
def test_unlocked_latch_is_relocked(self):
self._attr("cnc/interlock_circuit", "5") # bit 3 clear = unlocked
@@ -194,7 +194,7 @@ class BaselineTests(unittest.TestCase):
else:
# a young host: the reference is taken from the fake tree
self.assertIsNotNone(ref)
self.assertEqual(ref["sysfs"]["cnc/motor_lock"], "8")
self.assertEqual(ref["sysfs"]["cnc/motor_lock"], "0")
self.assertTrue(os.path.exists(os.path.join(self.tmp, "boot-test-boot.json")))
# and loaded back the second time
self.lines[:] = []
@@ -205,7 +205,7 @@ class BaselineTests(unittest.TestCase):
os.environ.pop("FORGETEST_BOOT_ID", None)
def test_fixed_constants_checked_against_a_dump(self):
ref = {"sysfs": {"cnc/motor_lock": "8", "cnc/step_freq": "10000"}}
ref = {"sysfs": {"cnc/motor_lock": "0", "cnc/step_freq": "10000"}}
diffs = baseline.check_fixed_against(ref, self.lines.append)
self.assertEqual(diffs, ["cnc/step_freq: boot=10000 constant=28160"])
@@ -286,7 +286,7 @@ class BaselineTests(unittest.TestCase):
self.assertEqual(ref["ts"], "old")
else:
self.assertTrue(any("retaking" in l for l in self.lines))
self.assertEqual(ref["sysfs"]["cnc/motor_lock"], "8")
self.assertEqual(ref["sysfs"]["cnc/motor_lock"], "0")
finally:
os.environ.pop("FORGETEST_BOOT_ID", None)
@@ -325,6 +325,7 @@ class BaselineModeTests(BaselineTests):
The fake client takes the marker down as gfcloud does, first thing
at its start; with refuse the POST is refused and no client starts."""
baseline.NOHUNT_MARKER = os.path.join(self.tmp, "nohunt-marker")
self.fc.state["settings"].setdefault("cloud_enabled", "1")
seen = []
def on_post(path, form):
@@ -355,6 +356,19 @@ class BaselineModeTests(BaselineTests):
self.assertEqual(seen, [("/mode", {"controller": "cloud"}, True)])
self.assertFalse(os.path.exists(baseline.NOHUNT_MARKER)) # one start, never left behind
def test_a_switch_to_cloud_turns_cloud_mode_on_when_it_is_off(self):
# A test that declares cloud mode gets it: cloud_enabled goes to 1
# first, with the typed phrase the daemon asks for and a log line,
# and only then the mode switch.
lines = []
seen = self.nohunt_marker()
self.fc.state["settings"]["cloud_enabled"] = "0"
ok, detail = baseline.Baseline(lines.append).switch_mode("cloud")
self.assertTrue(ok, detail)
self.assertEqual(seen[0], ("/settings", {"cloud_enabled": "1", "phrase": "I UNDERSTAND"}, False))
self.assertEqual(seen[1], ("/mode", {"controller": "cloud"}, True))
self.assertTrue(any("cloud mode turned on for the test" in ln for ln in lines))
def test_a_switch_to_grbl_sets_no_marker(self):
seen = self.nohunt_marker()
self.cloud()
forgetest/tests/test_cloud_suite.py
+7 -5
View File
@@ -650,11 +650,13 @@ class CloudSuiteTests(unittest.TestCase):
threading.Thread(target=self.wait_home_command, daemon=True).start()
self.assertFails(cloud.mode_switch, "no 'homing complete' line", hooks=hooks)
def test_mode_switch_precheck_needs_gfcloud_homing(self):
self.fc.state["settings"]["homing_mode"] = "switches"
self.assertIn("needs gfcloud", cloud.homing_mode_is_gfcloud())
self.fc.state["settings"]["homing_mode"] = "gfcloud"
self.assertIsNone(cloud.homing_mode_is_gfcloud())
def test_mode_switch_needs_no_setting_from_the_operator(self):
# The test sets the gfcloud homing itself and restores it: no precheck
# names a setting the operator should change first.
from forgetest import catalog
t = catalog.load_suite()["cloud.mode-switch"]
self.assertIsNone(getattr(t, "precheck", None))
self.assertNotIn("homing_mode_is_gfcloud", dir(cloud))
# -- pause / resume ---------------------------------------------------------
def replay_print(self, name, at_run, at_end, tail_delay=0.3):
forgetest/tests/test_commission_dark.py
+116
View File
@@ -0,0 +1,116 @@
"""The setup's checks in the catalog: registered as automatic tests with
no precheck, the takeover ones marked so, every covers map naming the
check runner; and the check driver itself against the fake daemon: a
start, a prompt answered, the result taken, the record read at version
1, and the settings a check may write put back as found."""
import os
import sys
import unittest
sys.path.insert(0, os.path.dirname(__file__))
import helpers # noqa: E402
from forgetest import catalog # noqa: E402
from forgetest.runner import Context, Run # noqa: E402
IDS = ["commission.check-switches", "commission.check-sensors", "commission.check-airflow",
"commission.check-cameras", "commission.check-motion", "commission.check-flow-verify"]
class RegistrationTests(unittest.TestCase):
def setUp(self):
self.reg = catalog.load_suite()
def test_every_check_runs_unattended_with_the_fixture(self):
# The lid and button checks name their machine actions, which the
# bench fixture performs; the rest need no hands at all.
for tid in IDS:
self.assertIn(tid, self.reg)
for tid in ("commission.check-switches", "commission.check-cameras"):
self.assertEqual(self.reg[tid].kind, "operator", tid)
self.assertTrue(self.reg[tid].fixture_runnable(("button", "lid", "interlock")), tid)
for tid in ("commission.check-sensors", "commission.check-airflow", "commission.check-motion",
"commission.check-flow-verify"):
self.assertEqual(self.reg[tid].kind, "auto", tid)
def test_no_check_asks_the_operator_to_prepare_anything(self):
for tid in IDS:
self.assertIsNone(getattr(self.reg[tid], "precheck", None), tid)
def test_the_checks_that_take_the_machine_say_so(self):
for tid in ("commission.check-airflow", "commission.check-motion", "commission.check-flow-verify"):
self.assertEqual(self.reg[tid].hardware, "takeover", tid)
def test_every_check_covers_the_runner_and_the_page(self):
for tid in IDS:
covers = set(self.reg[tid].covers)
self.assertIn(("forgectrl", "src/wizdark.*"), covers, tid)
self.assertIn(("forgectrl", "src/ui/wizard.*"), covers, tid)
class SensorsCheckTests(unittest.TestCase):
"""The check driver against a scripted daemon."""
RESULT = {"coolant_down_c": 22.4, "coolant_up_c": 21.9, "chassis_c": 28.0, "soc_c": 45.2,
"lid_ir_max": [110, 120, 105, 118], "accel_events": 0, "laser_pgood": 1,
"hv_current_max": 0, "exhaust_rpm_idle": 3400, "intake_rpm_idle": 2200}
def setUp(self):
self.fake = helpers.FakeForgectrl().start()
self.fake.state["settings"].update({"cool_temp_offset_c": "0.5"})
self.dark = {"id": "", "running": False, "phase": "", "progress": 0, "elapsed_s": 0,
"log": [], "prompt": None, "result": None, "error": "",
"shots": {"lid": False, "head": False}}
self.versions = {}
self.answers = []
def on_get(path, q):
if path == "/wiz/dark":
return 200, dict(self.dark)
if path == "/wiz":
return 200, {"versions": dict(self.versions), "required": [], "completed": True}
return None
def on_post(path, form):
s = self.fake.state["settings"]
if path == "/wiz/sensors/start":
self.dark.update({"id": "sensors", "running": True, "phase": "sampling",
"prompt": {"seq": 7, "id": "room-temp", "kind": "number",
"text": "Optional: the room temperature.",
"options": ["Set", "Skip"]}})
return 200, {"started": True, "id": "sensors"}
if path == "/wiz/sensors/answer":
self.answers.append(dict(form))
if form.get("seq") != "7":
return 409, {"error": "no such prompt is open"}
if form.get("value") == "Skip":
self.dark.update({"running": False, "prompt": None, "progress": 100,
"result": dict(self.RESULT)})
self.versions["sensors"] = 1
return 200, {"ok": True}
if path == "/settings":
for k, v in form.items():
if v == "":
s.pop(k, None)
else:
s[k] = v
return 200, dict(s)
return None
self.fake.on_get = on_get
self.fake.on_post = on_post
def tearDown(self):
self.fake.stop()
def test_start_prompt_answer_result_and_restore(self):
t = catalog.load_suite()["commission.check-sensors"]
run = Run("test", t.id, t.title)
t.fn(Context(run, None, t))
self.assertIn(("/wiz/sensors/start", {}), [(p, f) for p, f in self.fake.posts])
self.assertEqual(self.answers, [{"seq": "7", "value": "Skip"}])
self.assertEqual(run.evidence["result"]["laser_pgood"], 1)
self.assertEqual(run.evidence["settings_before"], {"cool_temp_offset_c": "0.5"})
self.assertEqual(self.fake.state["settings"]["cool_temp_offset_c"], "0.5")
if __name__ == "__main__":
unittest.main()
forgetest/tests/test_commission_sheet.py
+266
View File
@@ -0,0 +1,266 @@
"""The setup's sheet in the catalog: one live test on the button with a
piece of wood at hand, its covers map naming the live runner, the
renderer, the sender, the page, and the driver's laser module; and the
whole run against a scripted daemon: the placement's jogs and origin,
then each card's preview and program read, its start, the arm prompt and
the pick answered, the witnesses judged, the record read at version 1,
every card's evidence filed under its own key, and the settings the cards
wrote put back as found at the end."""
import os
import sys
import unittest
sys.path.insert(0, os.path.dirname(__file__))
import helpers # noqa: E402
from forgetest import catalog # noqa: E402
from forgetest.runner import Context, Run # noqa: E402
from forgetest.suite import commission_sheet # noqa: E402
LIVE = ["sheet.frame", "laser.focus", "laser.floor", "laser.dose-curve", "laser.corner", "cooling.flow-load"]
EMISSION = {"hv_max": 610, "laser_on_samples": 900, "thermopile_delta": 600, "lit_s": 30.0}
# Each wizard as the daemon runs it: its prompts in order (id, kind,
# options) and the result it ends with. A "jog" prompt repeats until the
# origin is set.
SCRIPT = {
"sheet.place": ([("place", "jog", ["X-10", "X+10", "Set origin"]),
("sheet-kind", "choice", ["Full sheet", "One card"]),
("thickness", "number", ["Set", "Skip"])],
{"origin_x": 100, "origin_y": 200, "origin_z": 0, "steps_per_mm": {"x": 40.0, "y": 40.0},
"z_referenced": True, "thickness_mm": 3.2, "alone": False}),
"sheet.frame": ([("frame-arm", "continue", ["Continue"]), ("frame-ok", "choice", ["Yes", "No", "Again"])],
{"mark_s": 400, "mark_feed": 3000, "emission": EMISSION}),
"laser.focus": ([("focus-arm", "continue", ["Continue"]),
("focus-pick", "multichoice", [str(i) for i in range(1, 13)]),
("thickness", "number", ["Keep", "Set"])],
{"pick": 11, "thickness_mm": 3.2, "pick_half_steps": -11, "edge_z_mm": 6.96,
"steps_per_mm": 2.922, "max_height_mm": 13.81,
"focus_range_mm": {"min": 2.17, "max": 13.81},
"stops": {"found": True, "below": 14, "above": 20, "contact_below": 16,
"contact_above": 22, "why": ""}, "emission": EMISSION}),
"laser.floor": ([("floor-arm", "continue", ["Continue"]),
("floor-pick", "choice", ["2", "4", "6", "8", "None"])],
{"faintest_density": 8.0, "floor_density": 10.0, "emission": EMISSION}),
"laser.dose-curve": ([("dose-arm", "continue", ["Continue"])],
{"points": [[10, 0.44], [100, 100]], "curve": "10:0.44,100:100", "emission": EMISSION}),
"laser.corner": ([("corner-arm", "continue", ["Continue"]),
("corner-pick", "choice", ["1.00", "1.25", "1.50", "1.75", "2.00"])],
{"gamma": 1.5, "emission": EMISSION}),
"cooling.flow-load": ([("load-arm", "continue", ["Continue"])],
{"lit_s": 30.0, "dose_raw_s": 12000, "peak_c": 0.72, "k_density": 2.14e-5,
"k_cw": 2.78e-5, "emission": EMISSION}),
}
# What each card writes when it ends.
WRITES = {
"laser.focus": {"lens_hall_edge_z_mm": "6.96", "lens_stop_below_steps": "14",
"lens_stop_above_steps": "20"},
"laser.floor": {"laser_floor_density": "10"},
"laser.dose-curve": {"laser_dose_curve": "10:0.44,100:100"},
"laser.corner": {"laser_corner_gamma": "1.5"},
"cooling.flow-load": {"cool_laser_heat_density": "2.14e-5", "cool_laser_heat_cw": "2.78e-5"},
}
FOUND = {"lens_hall_edge_z_mm": "3.35", "lens_stop_below_steps": "10", "lens_stop_above_steps": "12",
"laser_floor_density": "12",
"laser_dose_curve": "off", "laser_corner_gamma": "2", "cool_laser_heat_density": "1e-5",
"cool_laser_heat_cw": "1e-5"}
class RegistrationTests(unittest.TestCase):
def setUp(self):
self.t = catalog.load_suite()["commission.sheet"]
def test_one_live_test_on_the_button_with_the_sheet_at_hand(self):
t = self.t
self.assertEqual(t.kind, "live")
self.assertEqual(t.mode, "grbl")
self.assertEqual(t.hardware, "takeover")
self.assertIn("button", t.actions)
self.assertTrue(t.hands)
self.assertFalse(t.fixture_runnable(("button", "lid", "interlock")))
self.assertIsNone(getattr(t, "precheck", None))
for want in ("commission.check-motion", "laser.emission-witness", "cooling.flow-verify"):
self.assertIn(want, t.requires)
def test_no_other_sheet_test_remains(self):
ids = [i for i in catalog.load_suite() if i.startswith("commission.sheet")]
self.assertEqual(ids, ["commission.sheet"])
def test_covers_the_live_runner_the_renderer_the_sender_the_page_and_the_driver(self):
covers = set(self.t.covers)
for want in (("forgectrl", "src/wizlive.*"), ("forgectrl", "src/sheet.*"), ("forgectrl", "src/jobstream.*"),
("forgectrl", "src/curverec.*"), ("forgectrl", "src/font_hershey.*"),
("forgectrl", "src/ui/wizard.*"), ("forgectrl", "tools/hershey_gen.py"),
("grblhal-glowforge", "src/glowforge_laser.c")):
self.assertIn(want, covers, "lacks %s" % (want,))
def test_the_steps_ask_for_one_press_and_no_page(self):
text = " ".join(self.t.steps).lower()
self.assertIn("press the machine's button once", text)
self.assertNotIn("when the page", text)
self.assertEqual(len(commission_sheet.CARDS), 6)
class ScriptedDaemon:
"""The wizard runner as the suite sees it: /wiz/<id>/start opens the
first prompt, each answer opens the next or ends the wizard with its
result and its writes, /wiz/dark reports it, GET /wiz carries the
versions, the sheet preview and program are served per card."""
def __init__(self, fake):
self.fake = fake
fake.state["settings"].update(FOUND)
self.dark = {"id": "", "running": False, "phase": "", "progress": 0, "elapsed_s": 0,
"log": [], "prompt": None, "result": None, "error": "",
"shots": {"lid": False, "head": False}}
self.versions = {}
self.answers = {} # wid -> [values]
self.order = [] # the wizards as started
self.previews = []
self.queue = []
self.seq = 0
fake.on_get = self.on_get
fake.on_post = self.on_post
def open_next(self):
if not self.queue:
return False
pid, kind, options = self.queue[0]
self.seq += 1
self.dark["prompt"] = {"seq": self.seq, "id": pid, "kind": kind, "text": pid, "options": options}
return True
def on_get(self, path, q):
if path == "/wiz/dark":
return 200, dict(self.dark)
if path == "/wiz":
return 200, {"versions": dict(self.versions), "required": [], "completed": True}
if path == "/wiz/sheet.svg":
self.previews.append(q.get("card"))
if q.get("card") not in LIVE:
return 404, {"error": "no such card"}
return 200, b"<svg xmlns='http://www.w3.org/2000/svg'></svg>", "image/svg+xml"
if path == "/wiz/sheet.gcode":
# the frame and the text are M4 at the mark dose, as the daemon serves them
return 200, b"; mock\nM4 S400\nG0 X60 Y42\nG1 X120 Y42 F3000\nM5\n", "text/plain"
return None
def on_post(self, path, form):
s = self.fake.state["settings"]
if path.startswith("/wiz/") and path.endswith("/start"):
wid = path[len("/wiz/"):-len("/start")]
if wid not in SCRIPT:
return 404, {"error": "no such wizard"}
self.order.append(wid)
self.queue = list(SCRIPT[wid][0])
self.dark.update({"id": wid, "running": True, "phase": "starting", "result": None, "error": ""})
self.open_next()
return 200, {"started": True, "id": wid}
if path.startswith("/wiz/") and path.endswith("/answer"):
wid = path[len("/wiz/"):-len("/answer")]
p = self.dark.get("prompt")
if not p or form.get("seq") != str(p["seq"]):
return 409, {"error": "no such prompt is open"}
self.answers.setdefault(wid, []).append(form.get("value"))
if not (p["kind"] == "jog" and form.get("value") != "Set origin"):
self.queue.pop(0)
if not self.open_next():
s.update(WRITES.get(wid, {}))
self.dark.update({"running": False, "prompt": None, "progress": 100,
"result": dict(SCRIPT[wid][1])})
self.versions[wid] = 1
return 200, {"ok": True}
if path == "/settings":
for k, v in form.items():
if v == "":
s.pop(k, None)
else:
s[k] = v
return 200, dict(s)
return None
class SheetRunTests(unittest.TestCase):
def setUp(self):
self.fake = helpers.FakeForgectrl().start()
self.daemon = ScriptedDaemon(self.fake)
def tearDown(self):
self.fake.stop()
def test_the_whole_sheet_on_one_piece(self):
t = catalog.load_suite()["commission.sheet"]
run = Run("test", t.id, t.title)
run.unattended = True # the ready gate passes; the arm cue is a notice
t.fn(Context(run, None, t))
d = self.daemon
self.assertEqual(d.order, ["sheet.place"] + LIVE)
self.assertEqual(d.answers["sheet.place"], ["X+10", "X-10", "Set origin", "Full sheet", "3.2"])
self.assertEqual(d.answers["sheet.frame"], ["Continue", "Yes"])
self.assertEqual(d.answers["laser.focus"], ["Continue", "11", "Keep"])
self.assertEqual(d.answers["laser.floor"], ["Continue", "8"])
self.assertEqual(d.answers["laser.corner"], ["Continue", "1.50"])
self.assertEqual(d.previews, LIVE)
cards = run.evidence["cards"]
self.assertEqual(sorted(cards), sorted(["sheet.place"] + LIVE))
for wid in LIVE:
self.assertEqual(cards[wid]["emission"]["hv_max"], 610, wid)
self.assertEqual(cards[wid]["result"], SCRIPT[wid][1], wid)
self.assertNotIn("emission", run.evidence)
self.assertEqual(cards["sheet.place"]["result"]["origin_x"], 100)
self.assertEqual(run.evidence["settings_before"], FOUND)
for k, v in FOUND.items():
self.assertEqual(self.fake.state["settings"][k], v, k) # restored
def test_imperial_units_answer_the_thickness_in_inches(self):
# The number prompts read ui_units: on an imperial machine the
# placement's thickness is 0.125 in and the record carries 3.175 mm.
self.fake.state["settings"]["ui_units"] = "imperial"
place_prompts, place_result = SCRIPT["sheet.place"]
SCRIPT["sheet.place"] = (place_prompts, dict(place_result, thickness_mm=3.175))
try:
t = catalog.load_suite()["commission.sheet"]
run = Run("test", t.id, t.title)
run.unattended = True
t.fn(Context(run, None, t))
self.assertEqual(self.daemon.answers["sheet.place"][-1], "0.125")
self.assertEqual(run.evidence["thickness_answer"], "0.125")
finally:
SCRIPT["sheet.place"] = (place_prompts, place_result)
def test_a_metric_answer_on_an_imperial_machine_fails_the_placement(self):
# The defect the bench found: 3.2 read as inches is over the 30 mm
# cap and the wizard records 0; the test must not read that as a pass.
from forgetest.runner import Failed
place_prompts, place_result = SCRIPT["sheet.place"]
SCRIPT["sheet.place"] = (place_prompts, dict(place_result, thickness_mm=0.0))
try:
t = catalog.load_suite()["commission.sheet"]
run = Run("test", t.id, t.title)
run.unattended = True
with self.assertRaises(Failed) as cm:
t.fn(Context(run, None, t))
self.assertIn("thickness 0.0 mm", str(cm.exception))
self.assertEqual(self.daemon.order, ["sheet.place"]) # nothing burned
finally:
SCRIPT["sheet.place"] = (place_prompts, place_result)
def test_a_dark_burn_fails_the_run_and_still_restores(self):
from forgetest.runner import Failed
SCRIPT_DARK = dict(SCRIPT["laser.floor"][1])
SCRIPT_DARK["emission"] = {"hv_max": 0, "laser_on_samples": 0, "thermopile_delta": 0}
SCRIPT["laser.floor"] = (SCRIPT["laser.floor"][0], SCRIPT_DARK)
try:
t = catalog.load_suite()["commission.sheet"]
run = Run("test", t.id, t.title)
run.unattended = True
with self.assertRaises(Failed):
t.fn(Context(run, None, t))
self.assertEqual(self.daemon.order, ["sheet.place", "sheet.frame", "laser.focus", "laser.floor"])
for k, v in FOUND.items():
self.assertEqual(self.fake.state["settings"][k], v, k)
finally:
SCRIPT["laser.floor"] = (SCRIPT["laser.floor"][0], dict(SCRIPT_DARK, emission=EMISSION))
if __name__ == "__main__":
unittest.main()
forgetest/tests/test_commission_suite.py
+415
View File
@@ -0,0 +1,415 @@
"""The commission.* suite on the host: the registration (ids, kinds, the
takeover tests, the operator tests' hands), the record builders, the
mDNS packet code, the cookie parsing, the LED cue, the settle rule for a
gated supervisor, and the cloud-off surface test driven end to end
against the fake daemon."""
import json
import os
import shutil
import socket
import struct
import tempfile
import unittest
import helpers
from forgetest import baseline, catalog
from forgetest.runner import Context, Run
from forgetest.suite import commission
IDS = ("commission.gate-blocks-controllers", "commission.override-until-reboot",
"commission.advisories-rehash", "commission.account-login", "commission.https-only-writes",
"commission.ssh-until-reboot", "commission.cloud-disabled-surface",
"commission.factory-return", "commission.mdns-announce", "commission.first-run-flow",
"commission.first-run-page", "commission.what-changed", "commission.record-export",
"commission.mirror")
OPERATOR = ("commission.first-run-flow", "commission.first-run-page")
class RegistrationTests(unittest.TestCase):
def setUp(self):
self.reg = catalog.load_suite()
def test_every_test_is_registered_with_its_kind(self):
kinds = {tid: self.reg[tid].kind for tid in IDS if tid in self.reg}
self.assertEqual(sorted(kinds), sorted(IDS))
for tid in OPERATOR:
self.assertEqual(kinds[tid], "operator", tid)
for tid in IDS:
if tid not in OPERATOR:
self.assertEqual(kinds[tid], "auto", tid)
self.assertNotIn("commission.root-ssh-refused", self.reg) # folded into ssh-until-reboot
self.assertNotIn("commission.wizard-first-run", self.reg) # split into the flow and the page
def test_the_record_swapping_tests_are_takeovers(self):
for tid in ("commission.gate-blocks-controllers", "commission.advisories-rehash",
"commission.first-run-flow", "commission.first-run-page",
"commission.what-changed", "commission.mirror"):
self.assertEqual(self.reg[tid].hardware, "takeover", tid)
self.assertEqual(self.reg["commission.record-export"].hardware, "api")
def test_the_lifecycle_tests_cover_their_files(self):
changed = set(self.reg["commission.what-changed"].covers)
export = set(self.reg["commission.record-export"].covers)
mirror = set(self.reg["commission.mirror"].covers)
self.assertIn(("forgectrl", "src/ui/panel.js"), changed)
self.assertIn(("forgectrl", "src/recordhtml.*"), export)
self.assertIn(("forgectrl", "src/logs.*"), export)
self.assertIn(("forgectrl", "src/wizdark.*"), mirror)
self.assertIn("commission.check-sensors", self.reg["commission.mirror"].requires)
def test_flag_of_reads_the_status_document(self):
w = {"required": ["airflow", {"id": "laser.floor", "reason": "the tube was replaced"}],
"recommended": [{"id": "laser.focus", "reason": "the tray was replaced"}]}
self.assertEqual(commission.flag_of(w, "required", "laser.floor"), "the tube was replaced")
self.assertEqual(commission.flag_of(w, "recommended", "laser.focus"), "the tray was replaced")
self.assertIsNone(commission.flag_of(w, "required", "airflow")) # a plain id is the table, not a flag
self.assertIsNone(commission.flag_of(w, "required", "laser.focus"))
self.assertIsNone(commission.flag_of({}, "recommended", "x"))
def test_the_first_run_flow_runs_unattended_with_the_fixture_and_the_page_walk_does_not(self):
flow = self.reg["commission.first-run-flow"]
self.assertEqual(list(flow.actions), ["button"])
self.assertFalse(flow.hands)
self.assertTrue(flow.fixture_runnable(("button", "lid", "interlock")))
page = self.reg["commission.first-run-page"]
self.assertTrue(page.hands)
self.assertFalse(page.fixture_runnable(("button", "lid", "interlock")))
self.assertIn("commission.first-run-flow", page.requires)
def test_the_first_run_split_keeps_the_backend_and_the_page_apart(self):
flow = set(self.reg["commission.first-run-flow"].covers)
page = set(self.reg["commission.first-run-page"].covers)
for want in (("forgectrl", "src/wiz.*"), ("forgectrl", "src/commission.*"), ("forgectrl", "src/users.*"),
("forgectrl", "src/button.*"), ("forgectrl", "src/led.*"), ("forgectrl", "src/advisories.*")):
self.assertIn(want, flow, want)
self.assertNotIn(want, page, want)
for want in (("forgectrl", "src/ui/wizard.*"), ("forgectrl", "src/ui/md.js"), ("forgectrl", "src/tls.*")):
self.assertIn(want, page, want)
def test_the_only_attended_tests_need_a_workstation_or_the_sheet(self):
# With the bench actuator up, three commissioning tests keep a
# person: the page walk, the print from the Glowforge app, and
# the sheet. Everything else runs from the queue.
attended = sorted(tid for tid, t in self.reg.items() if tid.startswith("commission.")
and t.kind != "auto" and not t.fixture_runnable(("button", "lid", "interlock")))
self.assertEqual(attended, ["commission.cloud-header-capture", "commission.first-run-page",
"commission.sheet"])
def test_the_factory_return_never_runs_the_return(self):
import inspect
t = self.reg["commission.factory-return"]
src = inspect.getsource(t.fn)
self.assertNotIn('"confirm": "1"', src) # the only argument that starts the return
self.assertIn('"confirm": "0"', src)
self.assertEqual(t.kind, "auto") # the return itself is a bench drill, not a test
self.assertFalse(t.hands)
def test_mdns_covers_nothing_by_design(self):
self.assertEqual(self.reg["commission.mdns-announce"].covers, ())
def test_the_login_test_makes_its_own_account(self):
# No bench credentials, no precheck: the test installs a temporary
# account under a takeover and restores the real one.
t = self.reg["commission.account-login"]
self.assertIsNone(getattr(t, "precheck", None))
self.assertEqual(t.hardware, "takeover")
with open(commission.__file__) as f:
src = f.read()
for k in ("FORGETEST_LOGIN_NAME", "FORGETEST_LOGIN_PASSWORD"):
self.assertNotIn(k, src)
class RecordTests(unittest.TestCase):
STATUS = {"sheet_id": "ABCDE-FGHIJ",
"documents": [{"id": "safety-and-risk", "hash": "a" * 64, "consent": "typed"},
{"id": "licenses", "hash": "b" * 64, "consent": "check"}],
"wizards": [{"id": "advisories", "version": 1}, {"id": "account", "version": 1},
{"id": "machine", "version": 2}]}
def test_complete_record_counts_as_commissioned(self):
rec = commission.complete_record(self.STATUS)
self.assertEqual(rec["schema"], 1)
self.assertEqual(rec["advisories"]["safety-and-risk"]["hash"], "a" * 64)
self.assertEqual(rec["advisories"]["safety-and-risk"]["method"], "typed")
self.assertIn("pressed_at", rec["acceptance"])
self.assertEqual(rec["account"]["name"], "bench")
self.assertEqual(rec["wizards"]["machine"]["version"], 2)
self.assertEqual(rec["sheet_id"], "ABCDE-FGHIJ")
self.assertTrue(rec["completed"])
json.dumps(rec)
def test_complete_record_keeps_the_base_account_and_machine(self):
base = {"account": {"name": "owner", "uid": 1000, "created": "x"}, "machine": {"model": "pro"},
"sheet_id": "KKKKK-LLLLL", "created": "then"}
rec = commission.complete_record(self.STATUS, base)
self.assertEqual(rec["account"]["name"], "owner")
self.assertEqual(rec["machine"], {"model": "pro"})
self.assertEqual(rec["sheet_id"], "KKKKK-LLLLL")
self.assertEqual(rec["created"], "then")
def test_without_wizards_keeps_consent_and_account(self):
rec = commission.complete_record(self.STATUS)
rec["flags"] = {"machine": {"level": "required", "reason": "x"}}
out = commission.record_without_wizards(rec)
self.assertEqual(out["wizards"], {})
self.assertEqual(out["flags"], {})
self.assertEqual(out["advisories"], rec["advisories"])
self.assertEqual(out["account"], rec["account"])
self.assertEqual(rec["flags"], {"machine": {"level": "required", "reason": "x"}}) # a copy
def test_write_file_and_remove(self):
tmp = tempfile.mkdtemp(prefix="forgetest-comm-")
try:
p = os.path.join(tmp, "sub", "commissioning.json")
commission.write_file(p, b"{}\n")
self.assertEqual(commission.read_file(p), b"{}\n")
if os.name == "posix": # a mode means nothing on a Windows host
self.assertEqual(oct(os.stat(p).st_mode & 0o777), oct(0o600))
commission.write_file(p, None)
self.assertIsNone(commission.read_file(p))
commission.write_file(p, None) # a second remove is silent
finally:
shutil.rmtree(tmp, ignore_errors=True)
def test_paths_follow_the_daemon_environment(self):
os.environ["FORGECTRL_DATA_DIR"] = "/tmp/ffdata"
os.environ["GF_RUN_DIR"] = "/tmp/ffrun"
try:
self.assertEqual(commission.record_path(), "/tmp/ffdata/commissioning.json")
self.assertEqual(commission.users_path(), "/tmp/ffdata/users")
self.assertEqual(commission.override_path(), "/tmp/ffrun/commissioning-override")
self.assertEqual(commission.ssh_flag_path(), "/tmp/ffrun/ssh-enabled")
finally:
os.environ.pop("FORGECTRL_DATA_DIR", None)
os.environ.pop("GF_RUN_DIR", None)
class MdnsTests(unittest.TestCase):
def test_query_is_a_unicast_response_question(self):
q = commission.mdns_query("forgefirm.local", qid=0x1234)
qid, flags, qd, an, ns, ar = struct.unpack(">HHHHHH", q[:12])
self.assertEqual((qid, flags, qd, an, ns, ar), (0x1234, 0, 1, 0, 0, 0))
self.assertEqual(q[12:], b"\x09forgefirm\x05local\x00" + struct.pack(">HH", 1, 0x8001))
def _response(self, qid, name_bytes, addr, flags=0x8400, extra=b""):
q = b"\x09forgefirm\x05local\x00" + struct.pack(">HH", 1, 1)
rr = name_bytes + struct.pack(">HHIH", 1, 0x8001, 120, 4) + socket.inet_aton(addr)
return struct.pack(">HHHHHH", qid, flags, 1, 1, 0, 0) + q + rr + extra
def test_answers_follow_a_compression_pointer(self):
pkt = self._response(7, b"\xc0\x0c", "192.168.1.9")
self.assertEqual(commission.mdns_answers(pkt, 7), [("forgefirm.local", "192.168.1.9")])
def test_answers_with_the_name_spelled_out(self):
pkt = self._response(0, b"\x09forgefirm\x05local\x00", "10.0.0.5")
self.assertEqual(commission.mdns_answers(pkt), [("forgefirm.local", "10.0.0.5")])
def test_wrong_id_or_a_query_yields_nothing(self):
pkt = self._response(7, b"\xc0\x0c", "192.168.1.9")
self.assertEqual(commission.mdns_answers(pkt, 8), [])
self.assertEqual(commission.mdns_answers(self._response(7, b"\xc0\x0c", "1.2.3.4", flags=0), 7), [])
self.assertEqual(commission.mdns_answers(b"\x00" * 5), [])
def test_a_truncated_packet_yields_nothing(self):
pkt = self._response(7, b"\xc0\x0c", "192.168.1.9")
self.assertEqual(commission.mdns_answers(pkt[:20], 7), [])
class SmallHelpersTests(unittest.TestCase):
def test_cookie_parsing(self):
sid = "ab" * 32
value = "ffsid=%s; Path=/; HttpOnly; Secure; SameSite=Strict; Max-Age=43200" % sid
self.assertEqual(commission.session_from_cookie(value), sid)
self.assertEqual(commission.cookie_flags(value), {"path", "httponly", "secure", "samesite", "max-age"})
self.assertIsNone(commission.session_from_cookie("ffsid=; Path=/; Max-Age=0"))
self.assertIsNone(commission.session_from_cookie(""))
def test_the_teal_breathe_cue(self):
self.assertTrue(commission.breathes_teal({"target": [0, 180, 200], "pulse_on": [0, 1400, 1400],
"pulse_off": [0, 1400, 1400]}))
self.assertFalse(commission.breathes_teal({"target": [0, 255, 40], "pulse_on": [0, 0, 0],
"pulse_off": [0, 0, 0]})) # solid green
self.assertFalse(commission.breathes_teal({"target": [200, 200, 200], "pulse_on": [1800, 1800, 1800],
"pulse_off": [1800] * 3})) # white
self.assertFalse(commission.breathes_teal({"target": [None, None, None], "pulse_on": [None] * 3}))
self.assertFalse(commission.breathes_teal({}))
def test_decode(self):
self.assertEqual(commission.decode(b'{"a": 1}'), {"a": 1})
self.assertEqual(commission.decode(b"cloud mode is not enabled"), "cloud mode is not enabled")
class GatedSettleTests(unittest.TestCase):
"""A supervisor that reports the commissioning gate closed is settled:
it spawns nothing until the gate opens, so a takeover that installs
a gating record must not wait the whole settle timeout for it."""
def setUp(self):
self.fake = helpers.FakeForgectrl().start()
baseline.Baseline._unreachable_until = 0.0
def tearDown(self):
self.fake.stop()
def test_gated_returns_at_once(self):
self.fake.state["mode"] = {"mode": "grbl", "controller": "gated", "pid": 0, "motion": "unverified",
"gated": True, "why": "commissioning required: machine"}
lines = []
body = baseline.Baseline(lines.append).wait_settled(timeout=8)
self.assertEqual(body["controller"], "gated")
self.assertTrue(any("controller=gated" in ln for ln in lines))
class FirstRunSeedTests(unittest.TestCase):
"""The record the first-run tests install: a first run behind the
checks and the sheet."""
STATUS = {"wizards": [{"id": "advisories", "version": 1, "class": "form"},
{"id": "account", "version": 1, "class": "form"},
{"id": "cloud", "version": 1, "class": "form"},
{"id": "switches", "version": 1, "class": "dark"},
{"id": "motion", "version": 2, "class": "dark"},
{"id": "sheet.frame", "version": 1, "class": "live"}]}
def test_no_consent_no_account_not_complete_and_every_check_done(self):
base = {"created": "2026-09-01T00:00:00Z", "completed": "2026-09-02T00:00:00Z",
"advisories": {"safety-and-risk": {"hash": "x"}}, "acceptance": {"pressed_at": "t"},
"account": {"name": "scott", "uid": 1000}, "flags": {"flow_thin": True},
"wizards": {"advisories": {"version": 1}, "switches": {"version": 1, "result": {"lid": True}}}}
rec = commission.first_run_record(self.STATUS, base)
self.assertNotIn("completed", rec)
self.assertNotIn("acceptance", rec)
self.assertNotIn("account", rec)
self.assertEqual(rec["advisories"], {})
self.assertEqual(rec["created"], "2026-09-01T00:00:00Z")
self.assertEqual(rec["flags"], {"flow_thin": True})
self.assertEqual(sorted(rec["wizards"]), ["motion", "sheet.frame", "switches"])
self.assertEqual(rec["wizards"]["switches"]["result"], {"lid": True}) # kept from the base
self.assertEqual(rec["wizards"]["motion"]["version"], 2) # written at the catalog version
def test_without_a_base_record(self):
rec = commission.first_run_record(self.STATUS, None)
self.assertEqual(sorted(rec["wizards"]), ["motion", "sheet.frame", "switches"])
self.assertEqual(rec["wizards"]["sheet.frame"]["version"], 1)
class SshdPolicyTests(unittest.TestCase):
def test_the_three_keys_out_of_sshd_t(self):
from forgetest import hw
calls = []
def fake_run(cmd, timeout=60):
calls.append(cmd)
return 0, "port 22\npermitrootlogin no\npasswordauthentication yes\npermitemptypasswords no\nx11forwarding no\n"
real = hw.run
hw.run = fake_run
try:
self.assertEqual(commission.sshd_policy(), {"permitrootlogin": "no", "passwordauthentication": "yes",
"permitemptypasswords": "no"})
finally:
hw.run = real
self.assertEqual(calls, [["/usr/sbin/sshd", "-T"]])
def test_a_failing_sshd_reports_the_error(self):
from forgetest import hw
real = hw.run
hw.run = lambda cmd, timeout=60: (1, "sshd: no hostkeys available")
try:
self.assertEqual(commission.sshd_policy(), {"error": "sshd: no hostkeys available"})
finally:
hw.run = real
class CloudDisabledSurfaceTests(unittest.TestCase):
"""The cloud-off surface test against the fake daemon: cloud_enabled
is turned off with one write that sweeps the cloud choices, the three
cloud-pointing writes and the phrase-less cloud_enabled=1 are refused,
the settings are left alone, and the prior values come back."""
def setUp(self):
self.fake = helpers.FakeForgectrl().start()
self.fake.state["settings"].update({"controller_mode": "grbl", "homing_mode": "none",
"cloud_enabled": "1", "ui_units": "metric"})
def on_post(path, form):
s = self.fake.state["settings"]
if path == "/settings":
# as the daemon rules it: on from off takes the typed phrase
if form.get("cloud_enabled") == "1" and s.get("cloud_enabled") != "1" \
and form.get("phrase") != "I UNDERSTAND":
return 400, "type I UNDERSTAND to turn cloud mode on"
# the request's own cloud choices, as main.c checks them
enabled = form.get("cloud_enabled", s.get("cloud_enabled")) == "1"
if not enabled and form.get("controller_mode") == "cloud":
return 409, "cloud mode is not enabled on this machine"
if not enabled and form.get("homing_mode") == "gfcloud":
return 409, "cloud homing needs cloud mode enabled"
for k, v in form.items():
if k == "phrase":
continue
if v == "":
s.pop(k, None)
else:
s[k] = v
# off sweeps what pointed at the cloud, as the cloud step does
if form.get("cloud_enabled") == "0":
if "homing_mode" not in form and s.get("homing_mode") == "gfcloud":
s["homing_mode"] = "none"
if "controller_mode" not in form and s.get("controller_mode") == "cloud":
s["controller_mode"] = "grbl"
return 200, dict(s)
if path == "/mode" and form.get("controller") == "cloud" and s.get("cloud_enabled") != "1":
return 409, "cloud mode is not enabled on this machine"
return None
self.fake.on_post = on_post
def tearDown(self):
self.fake.stop()
def run_test(self):
t = catalog.load_suite()["commission.cloud-disabled-surface"]
run = Run("test", t.id, t.title)
t.fn(Context(run, None, t))
return run
def test_refusals_and_restore(self):
run = self.run_test()
s = self.fake.state["settings"]
self.assertEqual(s["cloud_enabled"], "1") # restored, with the phrase
self.assertEqual(s["controller_mode"], "grbl")
self.assertEqual(s["homing_mode"], "none")
self.assertNotIn("phrase", s)
posts = [(p, f) for p, f in self.fake.posts]
self.assertIn(("/settings", {"cloud_enabled": "0"}), posts)
self.assertIn(("/settings", {"cloud_enabled": "1"}), posts) # refused: no phrase
self.assertIn(("/settings", {"cloud_enabled": "1", "phrase": "I UNDERSTAND"}), posts)
self.assertIn(("/mode", {"controller": "cloud"}), posts)
self.assertEqual(run.evidence["mode_cloud"]["status"], 409)
self.assertEqual(run.evidence["settings cloud_enabled=1 no phrase"], 400)
self.assertEqual(run.evidence["found"]["cloud_enabled"], "1")
def test_cloud_homing_and_boot_mode_are_swept_by_the_one_write_and_go_back(self):
# As the daemon rules it: cloud_enabled=0 takes the gfcloud homing
# and the cloud boot mode down with it, so the test writes the one
# key, checks the sweep, and puts the three back, cloud mode first.
self.fake.state["settings"].update({"controller_mode": "cloud", "homing_mode": "gfcloud"})
run = self.run_test()
s = self.fake.state["settings"]
self.assertEqual((s["cloud_enabled"], s["homing_mode"], s["controller_mode"]), ("1", "gfcloud", "cloud"))
posts = [(p, f) for p, f in self.fake.posts if p == "/settings"]
order = [f for _, f in posts]
self.assertEqual(order[:1], [{"cloud_enabled": "0"}])
self.assertEqual(order[-3:], [{"cloud_enabled": "1", "phrase": "I UNDERSTAND"},
{"homing_mode": "gfcloud"}, {"controller_mode": "cloud"}])
self.assertEqual(run.evidence["found"], {"cloud_enabled": "1", "homing_mode": "gfcloud",
"controller_mode": "cloud"})
def test_an_unset_value_is_cleared_back(self):
self.fake.state["settings"].pop("cloud_enabled")
self.run_test()
self.assertNotIn("cloud_enabled", self.fake.state["settings"])
self.assertIn(("/settings", {"cloud_enabled": ""}), self.fake.posts)
if __name__ == "__main__":
unittest.main()
forgetest/tests/test_fixture.py
+27
View File
@@ -586,6 +586,33 @@ class RoutingTests(unittest.TestCase):
finally:
runner_mod.hw.button_lit = saved
def test_the_arm_press_waits_as_long_as_the_card_says_for_the_light(self):
"""A card that settles the coolant before it lights the button
takes minutes; the caller names the wait, and the actuator presses
when the light comes instead of handing the press to a person at
the default 60 s (found on the bench by the flow-load card)."""
run = Run("test", "r.live", "r.live")
ctx = Context(run, self.runner, self.reg["r.live"])
self.runner.probe_fixture(force=True)
run.fixture_takeover = True
lit_at = time.time() + 0.6
saved = runner_mod.hw.button_lit
runner_mod.hw.button_lit = lambda: time.time() >= lit_at
seen = []
run.set_notice = lambda text: seen.append(text)
try:
self.assertTrue(ctx.arm_press(lit_timeout=5))
deadline = time.time() + 5
while ("button", "press") not in self.stub.acts and time.time() < deadline:
time.sleep(0.05)
self.assertIn(("button", "press"), self.stub.acts)
self.assertEqual([n for n in seen if n], []) # nobody was asked
rec = run.evidence["actions"][-1]
self.assertEqual(rec["by"], "fixture")
self.assertGreaterEqual(rec["took_s"], 0.5)
finally:
runner_mod.hw.button_lit = saved
def test_an_actuator_lost_after_the_takeover_is_said_out_loud(self):
"""Falling back to the operator without a word is how one dropped
actuator becomes a press nobody can account for afterwards."""