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

Catalog: eleven tests leave for their host proofs

Browse Source 
Each left after its host cases passed (forgectrl 1dc85f6) and with
the coverage lint clean at the pins, its only failure the known
setup.check-lens entry on src/lensfind.*:

- forgectrl.tokens: host_routes.sh's scoped tokens (a route refused to a
  token not given it, neither token in the list or the store), with
  tokens_test and auth_scoped_test.
- update.product-gate: host_update.sh's product gate on the upload, with
  fwproduct_test.
- setup.first-run-page: tests/wizard_walk.py.
- setup.check-sensors, setup.check-intakes, setup.check-cameras,
  setup.check-flow-verify, setup.check-envelope, setup.floor:
  host_setup.sh's checks and the floor burn, with wizcalc_test.
  envelope_test.py's cases are unchanged. setup.floor's emission stays
  with setup.focus and laser.emission-witness.
- motion.job: jobpost_test, with jobrun_test and lease_test; jobpost.* is
  covered by exthost.motion-job, which stays.
- exthost.motion-jog: host_routes.sh's credential case.

Two HOST_TESTED rows, each a binary forgectrl's build.yml runs at the pin
c274f19 that proves the path the lint named: src/fwproduct.* to
fwproduct_test, src/grblport.* to grblport_test.

exthost.motion-job now requires motion.jog-roundtrip in place of
motion.job. bedsize.py held setup.check-envelope alone and goes;
forgectrl.py loses _words_of and its hw import, motion.py the job
runner's helpers, setup_burns.py FLOOR_KEY and FLOOR_KEYS.

forgetest's own tests: test_setup_dark.py loses the sensors and intakes
judges and the departed ids; test_setup_burns.py loses the floor's cases,
and the guards burn() and preview_ok() hold for the focus burn (the
program's tail and reach, the thickness in inches and the one refused, a
prompt not answered, the witnesses) run on setup.focus.

Stay: forgectrl.auth (the fuse view's button refusal on the real
button); forgectrl.settings-bounds (the lid lamp: cam.c writes
/sys/glowforge/pic/lid_led, not under GF_SYSFS_ROOT); forgectrl.panel-serves
(src/ui.h, which no binary at the pin proves; Z inside the lens reach is
setup.check-lens's); logs.routing; setup.account-login (the account's
render and its home, image content); setup.ssh-until-reboot
(src/paths.h); setup.machine-name (udhcpc -x hostname and no mDNS
responder, image content); setup.first-run-flow (src/button.*, led.*,
advisories.*, hooks.h and the four advisory documents); exthost.motion-job
(jobpost.*, which no binary at the pin proves).

Proof: the coverage lint, 55 tests; pyflakes; test_setup_burns,
test_setup_dark, test_takeover_position and test_image (54 OK).

Shared text moved the fingerprints of exthost.armed-freeze,
exthost.platform, exthost.service, forgectrl.auth,
forgectrl.panel-serves, forgectrl.settings-bounds, laser.armed-kill,
laser.emission-witness, laser.pause-resume-lid-cancel,
motion.cancel-abort, motion.deadman, motion.jog-roundtrip,
motion.liveness-probe, motion.microstep-modes, motion.pacing,
motion.release, motion.step-timing-under-load, setup.account-login,
setup.check-airflow, setup.check-lens, setup.check-motion,
setup.check-switches, setup.cloud-header-capture, setup.first-run-flow,
setup.focus, setup.machine-name, setup.ssh-until-reboot and
update.slots-and-signature; exthost.motion-job's own text moved with its
requires.
This commit is contained in:
ScottW514
2026-10-14 21:03:58 -04:00
parent 754c239e08
commit f9c298a347
13 changed files with 64 additions and 1474 deletions
Download Patch File Download Diff File Expand all files Collapse all files
forgetest/forgetest/manifest.py
+8
View File
@@ -165,6 +165,14 @@ HOST_TESTED = [
# The crumb tray's mode as the daemon keeps it: its offset and what it # The crumb tray's mode as the daemon keeps it: its offset and what it
# refuses (the route and the panel's switch are tests/host_routes.sh). # refuses (the route and the panel's switch are tests/host_routes.sh).
("forgectrl", "src/tray.*", "forgectrl:tests/tray_test.c", "forgectrl:.github/workflows/build.yml"), ("forgectrl", "src/tray.*", "forgectrl:tests/tray_test.c", "forgectrl:.github/workflows/build.yml"),
# The product gate on the firmware paths: firmware only, an extension
# package refused in words that say what it is (the upload route over
# a stand-in fwup is tests/host_update.sh).
("forgectrl", "src/fwproduct.*", "forgectrl:tests/fwproduct_test.c", "forgectrl:.github/workflows/build.yml"),
# The controller port's client and its two operation sets (the motion
# routes and a package's jog through the host's credential are
# tests/host_routes.sh).
("forgectrl", "src/grblport.*", "forgectrl:tests/grblport_test.c", "forgectrl:.github/workflows/build.yml"),
] ]
# This repository's root, when the package runs from a checkout of it; a # This repository's root, when the package runs from a checkout of it; a
forgetest/forgetest/suite/__init__.py
-1
View File
@@ -22,5 +22,4 @@ from . import cloud # noqa: F401,E402
# Last: it stands on the setup record's consent and on the cloud suite's dark print. # Last: it stands on the setup record's consent and on the cloud suite's dark print.
from . import exthost # noqa: F401,E402 from . import exthost # noqa: F401,E402
from . import homeoff # noqa: F401,E402 from . import homeoff # noqa: F401,E402
from . import bedsize # noqa: F401,E402
from . import tlsrec # noqa: F401,E402 from . import tlsrec # noqa: F401,E402
forgetest/forgetest/suite/bedsize.py
-287
View File
@@ -1,287 +0,0 @@
# Copyright 2026 514 LLC d/b/a OpenGlow
# Written by Scott Wiederhold
# https://community.openglow.org
# SPDX-License-Identifier: MIT
"""The Setup page's bed check, on the machine.
Its own module, so that no other test's fingerprint moves. The check is
driven from here inside the travel the machine already knows: a few jogs
out from the camera home, never toward an end, so it needs no operator.
Measuring the real ends is the operator's own run of the check. The
homing is the web-service session cloud.mode-switch runs.
"""
import json
import time
from ..catalog import test
from .cloud import _HOMING_PATH, gfhome_homing
from .homeoff import camera_home_return, judge_whole_motions, session_mark
from .setup import read_file, record_path, write_file
from .setup_dark import DARK_COVERS, answer, dark, start
WID = "motion.envelope"
END = "This is the end"
# The operator's answers, as the page sends them: out 60.1 mm in X and
# 61.1 mm in Y from the home, the fine steps included. A far edge is 50 mm
# at the least, so the ends are well past that and well inside the bed.
X_STEPS = ["X+10"] * 6 + ["X+1", "X+0.1", "X-1"]
Y_STEPS = ["Y+10"] * 6 + ["Y+1", "Y+0.1"]
X_OUT, Y_OUT = 60.1, 61.1
MARGIN_MM = 1.0
# The head is back on the home's step: half a step (213.3 steps/mm) and
# the port's three decimals.
HOME_TOL_MM = 0.003
KEYS =("envelope_x_mm", "envelope_y_mm", "homing_mode")
CLOSED = "closed the check's envelope"
CLOSED_MSG = "Bed check envelope closed: a sender line"
def _port(fc):
st, body = fc.get("/motion/state")
return body if st == 200 and isinstance(body, dict) else {}
def _drive(ctx, fc, x_steps, y_steps, hook=None, timeout=300):
"""Start the check and answer each prompt from the steps in turn, then
This is the end; hook(prompt_id, n) runs before the answer to the n-th
prompt. The final status document, and the answers given."""
# The daemon calls the machine busy for a moment after the last jog.
ctx.check(fc.wait_idle(timeout=30, abort=ctx.aborted), "the machine is not idle: %s", fc.status().get("state"))
start(ctx, fc, WID)
queue = {"x-end": list(x_steps) + [END], "y-end": list(y_steps) + [END]}
seen, given = set(), []
t0 = time.time()
try:
while time.time() - t0 < timeout:
ctx.checkpoint()
d = dark(fc)
p = d.get("prompt")
if d.get("running") and p and p.get("seq") not in seen:
seen.add(p["seq"])
pid = p.get("id")
ctx.check(pid in queue and queue[pid], "the check asked %s: %s (the lid must be closed)",
pid, p.get("text"))
ctx.check(p.get("kind") == "jog" and END in (p.get("options") or []),
"the %s prompt is not a jog with %r: %s", pid, END, p)
if hook:
hook(pid, len(given))
value = queue[pid].pop(0)
st, body = answer(fc, WID, p, value)
ctx.check(st == 200, "the answer %s to %s -> %s %s", value, pid, st, body)
given.append(value)
if not d.get("running"):
return d, given
time.sleep(0.3)
except BaseException:
fc.post("/wiz/%s/abort" % WID)
raise
fc.post("/wiz/%s/abort" % WID)
ctx.fail("the bed check did not end within %d s", timeout)
def _jog(fc, dx=0.0, dy=0.0):
params = {"feed": "1200"}
if dx:
params["x"] = "%.3f" % dx
if dy:
params["y"] = "%.3f" % dy
return fc.post("/motion/jog", params=params)
def _settled(ctx, fc):
"""After the suite's Grbl client closes, the daemon calls the machine
busy for a moment: the check and a settings write each wait for it."""
ctx.wait_for(lambda: _port(fc).get("sender") is False, 20, poll=0.5)
ctx.check(fc.wait_idle(timeout=30, abort=ctx.aborted), "the machine is not idle: %s", fc.status().get("state"))
def _put_back(ctx, fc, found):
for k in KEYS:
v = found.get(k) or ""
def put():
st_, body_ = fc.post("/settings", params={k: ""}) if v == "" else fc.post("/settings", data={k: v})
if st_ != 200:
ctx.log("restore %s=%r -> %s %s", k, v, st_, body_)
return st_ == 200 or None
took = ctx.wait_for(put, 20, poll=0.5)
ctx.log("restore %s=%r -> %s", k, v, "taken" if took is not None else "not taken")
def _back_home(ctx, fc, home):
"""The head back where the home put it: one jog to the home in machine
coordinates, from a Grbl client. A relative jog aimed at the home from
the port's position can be refused whole: the port reports the step
counters, the core adds a relative jog to its own position, which the
check's jogs leave up to half a step off the counters, and the envelope
begins at the home, so the target can land microns behind it (error:15).
Where the head stands after, from the port; None when unread."""
# 0.5 um inside an envelope that begins at the home (a camera home
# behind the origin), far under half a step: the head lands on the
# home's own step.
line = "$J=G90 G53 X%.4f Y%.4f F1200" % (home[0] + 0.0005, home[1] + 0.0005)
try:
with ctx.grbl() as g:
reply = g.command(line, timeout=10)
ctx.log("the jog back to the home: %s -> %s", line, " | ".join(reply))
if reply[-1:] == ["ok"]:
t0 = time.time()
ctx.wait_for(lambda: time.time() - t0 > 0.5 and g.status_report()["state"].startswith("Idle"),
60, poll=0.2)
except (OSError, ValueError) as e:
ctx.log("the jog back to the home: %s", e)
ctx.wait_for(lambda: _port(fc).get("sender") is False and _port(fc).get("state") == "Idle", 20, poll=0.5)
at = _port(fc).get("mpos")
ctx.log("the head after the jog back: %s (the home %s)", at, home)
return at
def _port_idle(ctx, fc):
ok = ctx.wait_for(lambda: (lambda s: s.get("state") == "Idle" and not s.get("port_jog"))(_port(fc)), 30,
poll=0.1)
ctx.check(ok is not None, "the port jog did not end: %s", _port(fc))
def _edge(ctx, fc, ev, axis, edge):
"""From where the head stands, a jog 0.5 mm short of the far edge runs
and comes back, and one 0.5 mm past it is refused with nothing moved."""
s = _port(fc)
at = s["mpos"]["XY".index(axis)]
short, past = edge - 0.5 - at, edge + 0.5 - at
kw = (lambda d: {"dx": d}) if axis == "X" else (lambda d: {"dy": d})
st1, b1 = _jog(fc, **kw(short))
_port_idle(ctx, fc)
st2, b2 = _jog(fc, **kw(-short))
_port_idle(ctx, fc)
st3, b3 = _jog(fc, **kw(past))
_port_idle(ctx, fc)
back = _port(fc)["mpos"]["XY".index(axis)]
ev["edge_" + axis] = {"edge": edge, "short": [st1, st2], "past": [st3, b3], "at": [at, back]}
ctx.check(st1 == 200 and st2 == 200, "a jog 0.5 mm short of the %s edge %.2f and back -> %s %s, %s %s",
axis, edge, st1, b1, st2, b2)
ctx.check(st3 == 409 and "envelope" in json.dumps(b3) and abs(back - at) < 0.02,
"a jog 0.5 mm past the %s edge %.2f -> %s %s, the head at %.3f from %.3f", axis, edge, st3, b3, back, at)
@test("setup.check-envelope", title="The bed check measures the far edges from the operator's jogs, and a Grbl "
"client's line ends it with nothing written",
subsystem="setup", kind="auto", mode="grbl", est_min=10,
covers=DARK_COVERS + _HOMING_PATH + [("forgectrl", "src/grblport.*"),
("grblhal-glowforge", "src/glowforge_homing.*"),
("grblhal-glowforge", "src/ctlport.*"),
("grblhal-glowforge", "src/driver.c")],
requires=["forgectrl.auth", "motion.pacing"],
steps=["Bed clear, lid closed; cloud credentials configured; the machine on the network. The head moves "
"at most about 62 mm out from the camera home in X and in Y, and ends where it was found."],
description="With cloud mode on, the machine is homed with the camera (the web-service session), every "
"service motion of it run whole. "
"POST /wiz/motion.envelope/start runs the Bed size check; its prompts are jogs with This is the "
"end among the options, and are answered as the page would, out 60.1 mm in X and 61.1 mm in Y "
"(steps of 10, 1, and 0.1 mm). The check ends complete: the ends it reports are the home plus "
"those, the far edges 1 mm short of them are written as envelope_x_mm and envelope_y_mm, and the "
"controller holds them at once, without a home: a panel jog 0.5 mm short of each edge runs, "
"and one 0.5 mm past it is refused with nothing moved. A second run is opened, and while its "
"first prompt waits a Grbl client sends one line, which draws the controller's word that it "
"closed the envelope and then ok; the port's state then says the envelope is closed, and the "
"next answer ends the check in words, with the keys as the "
"first run left them and the edges still in force. The keys, the homing mode, and the setup "
"record are put back as found, the record under a restart, and the head goes back to the "
"home by a Grbl client's jog in machine coordinates; the run fails unless the port reads "
"the head on the home's step before the restart. Then the camera home is dropped and the "
"head goes back to where the test found it, by the travel the session's motions logged.")
def check_envelope(ctx):
fc = ctx.forgectrl
ev = ctx.evidence
s0 = fc.settings() or {}
ctx.check(s0.get("cloud_enabled") == "1", "cloud mode is off: a camera home needs it on")
raw = read_file(record_path())
found = {k: s0.get(k) or "" for k in KEYS}
ev["found"] = found
home = back = session_at = None
try:
st, body = fc.post("/settings", data={"homing_mode": "gfcloud"})
ctx.check(st == 200, "homing_mode=gfcloud -> %s %s", st, body)
with ctx.grbl() as g:
session_at = session_mark()
gfhome_homing(ctx, ev, g)
home = g.status_report().get("MPos")
judge_whole_motions(ctx, ev, session_at)
ev["home"] = home
ctx.log("the camera home declared %s", home)
ctx.check(home and home[0] is not None, "no position after the home: %s", home)
_settled(ctx, fc)
last, given = _drive(ctx, fc, X_STEPS, Y_STEPS)
r = last.get("result") or {}
ev["run1"] = {"answers": given, "error": last.get("error"), "result": r, "log": last.get("log", [])[-8:]}
ctx.log("the bed check ended: %s", json.dumps(r))
ctx.check(not last.get("error"), "the bed check failed: %s", last.get("error"))
want_x, want_y = home[0] + X_OUT, home[1] + Y_OUT
ctx.check(abs((r.get("x_end_mm") or -1) - want_x) < 0.02 and abs((r.get("y_end_mm") or -1) - want_y) < 0.02,
"the ends are not the home plus the jogs (%.2f, %.2f): %s", want_x, want_y, r)
edge_x, edge_y = round(want_x - MARGIN_MM, 2), round(want_y - MARGIN_MM, 2)
s1 = fc.settings() or {}
ev["keys_after_run1"] = {k: s1.get(k) for k in ("envelope_x_mm", "envelope_y_mm")}
ctx.check(abs(float(s1.get("envelope_x_mm") or 0) - edge_x) < 0.011 and
abs(float(s1.get("envelope_y_mm") or 0) - edge_y) < 0.011,
"the keys are not the ends less %.1f mm (%.2f, %.2f): %s", MARGIN_MM, edge_x, edge_y,
ev["keys_after_run1"])
ctx.check(_port(fc).get("envelope_open") is False, "the envelope is still open after the check: %s",
_port(fc))
_edge(ctx, fc, ev, "X", edge_x)
_edge(ctx, fc, ev, "Y", edge_y)
# A Grbl client's line while the envelope is open ends the check.
said = {}
def client_line(pid, n):
if n:
return
ctx.check(_port(fc).get("envelope_open") is True, "the second run did not open the envelope: %s",
_port(fc))
with ctx.grbl() as g2:
said["reply"] = g2.command("G90", timeout=10)
said["text"] = g2.drain()
said["port"] = _port(fc)
last2, given2 = _drive(ctx, fc, ["X+1"], [], hook=client_line)
ev["run2"] = {"answers": given2, "said": said, "error": last2.get("error"), "log": last2.get("log", [])[-6:]}
ctx.log("the second run: %s", json.dumps(ev["run2"]))
ctx.check((said.get("reply") or [])[-1:] == ["ok"], "the client's line did not draw ok: %s", said)
ctx.check(any(CLOSED_MSG in l for l in (said.get("reply") or [])) or CLOSED_MSG in (said.get("text") or ""),
"the client was not told its line closed the envelope: %s", said)
ctx.check((said.get("port") or {}).get("envelope_open") is False,
"the client's line did not close the envelope: %s", said.get("port"))
ctx.check(CLOSED in (last2.get("error") or ""), "the second run did not end on the closed envelope: %s",
last2.get("error"))
s2 = fc.settings() or {}
ctx.check({k: s2.get(k) for k in ("envelope_x_mm", "envelope_y_mm")} == ev["keys_after_run1"],
"the second run wrote the keys: %s", s2)
_settled(ctx, fc)
_edge(ctx, fc, ev, "X", edge_x)
finally:
fc.post("/wiz/%s/abort" % WID)
if home and home[0] is not None:
ctx.wait_for(lambda: not dark(fc).get("running"), 30, poll=0.5)
back = _back_home(ctx, fc, home)
_put_back(ctx, fc, found)
try:
# Before the record's restart: the counters since the home say
# where the head stands until a controller start zeroes them.
if session_at is not None:
camera_home_return(ctx, ev, session_at)
finally:
# The check records itself when it completes; the record goes back
# as found under a restart, which also ends the envelope it set.
if read_file(record_path()) != raw:
with ctx.takeover():
write_file(record_path(), raw)
ctx.log("the previous setup record is back under a restart")
# Judged from the reading before the restart: the restart zeroes the
# counters where the head stands, so the baseline's position check
# after it cannot see a head left out.
ev["back"] = back
ctx.check(back and all(abs(back[i] - home[i]) < HOME_TOL_MM for i in (0, 1)),
"the head is not back at the home %s: %s", home, back)
forgetest/forgetest/suite/exthost.py
+1 -123
View File
@@ -1306,132 +1306,11 @@ def armed_freeze(ctx):
_as_found(ctx, fc, prior, raw, dir_mode, found_tree) _as_found(ctx, fc, prior, raw, dir_mode, found_tree)
@test("exthost.motion-jog", title="A package jogs the machine, bounded and dark",
subsystem="exthost", kind="auto", hardware="takeover", mode="grbl", est_min=7,
covers=[("forgeext", "src/api.*"), ("forgeext", "src/machine.*"), ("forgeext", "src/run.*"),
("forgectrl", "src/tokens.*"), ("forgectrl", "src/auth.*"), ("forgectrl", "src/main.c"),
("forgectrl", "src/grblport.*")],
requires=["exthost.service"],
description="The reference package is installed with motion.jog. Its jog reaches the machine "
"through the host's own credential, which forgectrl mints for each run of the daemon "
"into a file only root can read: the file is 0600 and root's, an extension account "
"cannot read it, and the credential holds motion.jog and not the routes it was not "
"given. The package's jog moves the head, and the head accelerometer - the "
"supervisor's own motion witness - sees it move; the laser stays dark throughout "
"(the latch locked, no fire seconds, no emission). A jog past the bounds is refused "
"by the host without the machine being asked, and a jog with the capability removed "
"is refused. Everything is put back as exthost.service puts it back, and the head "
"returns to where it started.")
def motion_jog(ctx):
import shutil
import tempfile
fc = ctx.forgectrl
ev = ctx.evidence
ctx.check(len(_host_pids()) == 1, "the extension host is not one running process: %s", _host_pids())
ctx.check(fc.wait_idle(timeout=30, abort=ctx.aborted), "machine not idle: settings are locked")
prior = fc.settings().get("ext_enabled") or ""
raw = read_file(record_path())
ctx.check(raw, "no setup record at %s", record_path())
ctx.check(not os.path.exists(SAFE_FILE), "%s exists: the machine is in safe mode", SAFE_FILE)
found_tree = _tree(EXT_ROOT)
dir_mode = os.stat(os.path.dirname(EXT_ROOT)).st_mode & 0o7777
st, body, hdrs = request(fc.base, "GET", "/advisories/extensions", headers={"Host": fc.host_header()})
etag = hdrs.get("etag")
work = tempfile.mkdtemp(prefix="forgetest-ffx.")
owner_key = os.path.join(EXT_ROOT, "keys", REF_KEY + ".pub")
# The host's credential: root's alone, and holding only what it relays.
cred = "/run/forgefirm/ext-host.token"
ctx.check(os.path.exists(cred), "forgectrl minted no host credential at %s", cred)
cst = os.stat(cred)
ev["host_credential"] = {"mode": "%04o" % (cst.st_mode & 0o7777), "uid": cst.st_uid, "gid": cst.st_gid}
ctx.check((cst.st_mode & 0o7777) == 0o600 and cst.st_uid == 0,
"the host credential is not root's alone: %s", ev["host_credential"])
token = _read(cred).strip()
ctx.check(token.startswith("fft_"), "the host credential is not a scoped credential")
# It holds motion.jog and nothing it was not given.
st_, why_ = fc.get("/cam/snapshot", params={"cam": "lid"}, headers={"X-ForgeFIRM-Token": token})
ev["credential_beyond_its_grant"] = [st_, why_ if st_ != 200 else "served"]
ctx.check(st_ != 200, "the host credential reached a camera it does not hold: %s", st_)
try:
archive, pub = _pack_reference(work, lan_ip(), more_caps=["motion.jog"])
shutil.copy(pub, owner_key)
os.chmod(owner_key, 0o644)
r = _forgeext("install", archive, "--consent-community", "--grant", "motion.jog")
if r.get("ok") is not True:
r = _forgeext("install", archive, "--consent-community")
ctx.check(r.get("ok") is True, "the install -> %s", r.get("error"))
st, reply = fc.post("/settings", data={"ext_enabled": "1", "advisory": etag, "phrase": SAFETY_PHRASE})
ctx.check(st == 200, "ext_enabled=1 over the advisory -> %s %r", st, reply)
x = _until(ctx, lambda: _svc(REF_ID) if _svc(REF_ID).get("state") == "running" else None, 90, poll=0.5)
ctx.check(x, "the service is not running: %s", _svc(REF_ID))
data_dir = os.path.join(EXT_ROOT, "data", REF_ID)
jogged = os.path.join(data_dir, "jogged")
def ask_jog(words, seconds=40):
if os.path.exists(jogged):
os.remove(jogged)
_write(os.path.join(data_dir, "jog"), json.dumps(words))
return _until(ctx, lambda: json.loads(_read(jogged)) if os.path.exists(jogged) else None,
seconds, poll=0.5)
# A jog the host refuses on its own: the machine is never asked.
far = ask_jog({"x": 500})
ev["jog_past_the_bounds"] = far
ctx.check(far and far[0] == 400 and "100 mm" in str(far[1]),
"a jog past the bounds was not refused by the host: %s", far)
# A real jog, watched by the accelerometer and by the laser's own witnesses.
accel = hw.AccelSampler()
ctx.check(accel.available, "no head accelerometer found: the motion witness is missing")
before = fc.get("/status")[1] or {}
with accel:
# The sampler stamps its samples with the wall clock, so the
# window is read from the same clock.
t0 = time.time()
got = ask_jog({"x": 20, "feed": 3000})
ctx.sleep(1.5)
t1 = time.time()
ev["jog"] = got
ctx.check(got and got[0] == 200, "the package's jog was refused: %s", got)
p2px, p2py, n = accel.p2p(t0, t1)
ev["accel"] = {"p2p_x": p2px, "p2p_y": p2py, "samples": n}
ctx.log("the package jogged X 20 mm: accel p2p x=%d y=%d over %d samples", p2px, p2py, n)
ctx.check(n > 0 and max(p2px, p2py) > 0,
"the accelerometer saw no motion for the package's jog: %s", ev["accel"])
after = fc.get("/status")[1] or {}
def laser_of(st_):
return {"locked": st_.get("laser_locked"),
"emission_samples": (st_.get("laser") or {}).get("emission_samples")}
ev["laser"] = {"before": laser_of(before), "after": laser_of(after)}
ctx.log("the laser across the jog: %s", ev["laser"])
# The names are read from the machine's own status, so a field
# that is not there fails rather than comparing None to None.
ctx.check(ev["laser"]["before"]["locked"] is True and ev["laser"]["after"]["locked"] is True,
"the laser latch was not locked across a package's jog: %s", ev["laser"])
ctx.check(ev["laser"]["after"]["emission_samples"] == 0
and ev["laser"]["before"]["emission_samples"] == 0,
"the emission witness is not zero across a package's jog: %s", ev["laser"])
# Put the head back where it started.
back = ask_jog({"x": -20, "feed": 3000})
ctx.check(back and back[0] == 200, "the head was not jogged back: %s", back)
ctx.sleep(1.5)
finally:
_put_back(ctx, fc, work, prior, etag, raw, dir_mode)
_as_found(ctx, fc, prior, raw, dir_mode, found_tree)
@test("exthost.motion-job", title="A package runs a program, under every gate a sender is under", @test("exthost.motion-job", title="A package runs a program, under every gate a sender is under",
subsystem="exthost", kind="auto", hardware="takeover", mode="grbl", est_min=9, subsystem="exthost", kind="auto", hardware="takeover", mode="grbl", est_min=9,
covers=[("forgeext", "src/api.*"), ("forgeext", "src/machine.*"), ("forgeext", "src/run.*"), covers=[("forgeext", "src/api.*"), ("forgeext", "src/machine.*"), ("forgeext", "src/run.*"),
("forgectrl", "src/jobpost.*"), ("forgectrl", "src/jobrun.*"), ("forgectrl", "src/tokens.*")], ("forgectrl", "src/jobpost.*"), ("forgectrl", "src/jobrun.*"), ("forgectrl", "src/tokens.*")],
requires=["exthost.service", "motion.job"], requires=["exthost.service", "motion.jog-roundtrip"],
steps=["Setup: lid closed, nothing in the bed."], steps=["Setup: lid closed, nothing in the bed."],
description="The reference package is installed with motion.job, which the operator grants. It " description="The reference package is installed with motion.job, which the operator grants. It "
"writes a program into its own data directory and asks the host to run it. The program " "writes a program into its own data directory and asks the host to run it. The program "
@@ -1564,4 +1443,3 @@ def motion_job(ctx):
finally: finally:
_put_back(ctx, fc, work, prior, etag, raw, dir_mode) _put_back(ctx, fc, work, prior, etag, raw, dir_mode)
_as_found(ctx, fc, prior, raw, dir_mode, found_tree) _as_found(ctx, fc, prior, raw, dir_mode, found_tree)
forgetest/forgetest/suite/forgectrl.py
-208
View File
@@ -10,7 +10,6 @@ import socket
import time import time
from ..catalog import test from ..catalog import test
from .. import hw
# The write guard reads the account and the session store (a machine with # 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 # an account needs a session from the LAN; this host and the dev image
@@ -452,210 +451,3 @@ def panel_serves(ctx):
st, text = fc.get("/curve/ladder.gcode", raw=True) st, text = fc.get("/curve/ladder.gcode", raw=True)
ctx.check(st == 200 and b"S1000" in (text or b"") and b"M5" in (text or b""), ctx.check(st == 200 and b"S1000" in (text or b"") and b"M5" in (text or b""),
"GET /curve/ladder.gcode -> %s without the ladder", st) "GET /curve/ladder.gcode -> %s without the ladder", st)
# ---------------------------------------------------------------- scoped tokens
def _words_of(body):
b = body if isinstance(body, (dict, str)) else ""
return b.get("error", "") if isinstance(b, dict) else b
@test("forgectrl.tokens", title="Scoped API tokens: a token reaches what it was granted, and nothing else",
subsystem="forgectrl", kind="auto", mode="grbl", est_min=2,
covers=_COVERS_AUTH + [("forgectrl", "src/tokens.*"), ("forgectrl", "src/grblport.*"),
("forgectrl", "src/settings.*")],
steps=["Bed clear; the head needs 2 mm of free travel toward +X."],
description="Three tokens are created through the panel's route: one holding machine.read "
"and camera.lid, one holding motion.jog, one holding camera.lid alone. Each is then "
"used the way a client on the network uses it, over HTTPS to the machine's own LAN "
"address, so the daemon sees a peer that is not this host. The first reads /status "
"as a bearer token, /mode in the panel token's header, and /cam/status as a bearer "
"token; the camera-only token reads /cam/status as ?key=. Refused, each with 403 "
"and the reason in words: the first token in a URL (it holds more than a camera), "
"the camera-only token on the head camera, and for the first token a jog, a "
"settings write, a mode switch, and the token routes. The second jogs the head 1 mm with no "
"session and no panel token (the kernel's counters are the witness) and is refused "
"/status. A token presented over plain HTTP from the LAN is refused, and the "
"daemon's log names it. A token that is not this machine's is refused. With "
"panel_open_reads=0 the LAN reads /status with machine.read and with nothing else. "
"The list names both tokens with a last-used time and carries neither a token nor "
"a hash, and the store is mode 0600 and holds neither token. A revoked token is "
"refused at once, with reads open again. Both tokens are revoked, the setting "
"restored, and the head returned on the way out. No token is written to the log "
"or the evidence.")
def scoped_tokens(ctx):
from .setup import request, decode
from .motion import kernel_xy_mm, machine_idle
ev = ctx.evidence
fc = hw.Forgectrl()
ip = lan_ip()
ctx.check(ip, "cannot determine the board's LAN address")
tls, plain = "https://%s" % ip, "http://%s" % ip
machine_idle(ctx)
x0 = kernel_xy_mm(ctx)[0]
reads_were = str(fc.settings().get("panel_open_reads", "1"))
mode = (fc.get("/mode")[1] or {}).get("mode")
units = fc.settings().get("ui_units") or "metric"
made = {}
def create(name, caps):
st, body = fc.post("/tokens", data={"name": name, "caps": caps})
ctx.check(st == 200 and isinstance(body, dict) and str(body.get("token", "")).startswith("fft_")
and len(body["token"]) == 36, "POST /tokens %s -> %s", name, st)
made[name] = body
return body["token"]
def lan(token, method, path, how="bearer", base=None, data=None):
headers = {}
if token and how == "bearer":
headers["Authorization"] = "Bearer " + token
elif token and how == "header":
headers["X-ForgeFIRM-Token"] = token
elif token and how == "key":
path += ("&" if "?" in path else "?") + "key=" + token
st, content, _h = request(base or tls, method, path, data=data, headers=headers)
return st, decode(content)
try:
# Whatever an earlier, interrupted run left behind.
st, listed = fc.get("/tokens")
ctx.check(st == 200 and isinstance(listed, dict), "GET /tokens -> %s", st)
for t in listed.get("tokens", []):
if t.get("name", "").startswith("forgetest "):
fc.post("/tokens/revoke", data={"id": t["id"]})
st, body = fc.post("/tokens", data={"name": "forgetest bad", "caps": "settings.write"})
ev["outside_the_list"] = [st, _words_of(body)[:120]]
ctx.check(st == 400 and "not a capability" in _words_of(body),
"a capability outside the list -> %s %s", st, _words_of(body)[:120])
hub = create("forgetest hub", "machine.read,camera.lid")
pend = create("forgetest jogger", "motion.jog")
view = create("forgetest viewer", "camera.lid")
passed = {}
for name, call in (("bearer GET /status", lambda: lan(hub, "GET", "/status")),
("header GET /mode", lambda: lan(hub, "GET", "/mode", how="header")),
("bearer GET /cam/status", lambda: lan(hub, "GET", "/cam/status")),
("camera-only ?key= GET /cam/status",
lambda: lan(view, "GET", "/cam/status", how="key"))):
st, body = call()
passed[name] = st
ctx.check(st == 200 and isinstance(body, dict), "the hub token, %s -> %s %s", name, st,
_words_of(body)[:120])
ev["hub_passed"] = passed
refused = {}
for name, want, call in (
("the head camera", "does not hold camera.head",
lambda: lan(view, "GET", "/cam/snapshot?cam=head", how="key")),
("a token with more than a camera, in a URL", "in a URL may hold camera capabilities only",
lambda: lan(hub, "GET", "/cam/status", how="key")),
("a jog", "does not hold motion.jog", lambda: lan(hub, "POST", "/motion/jog", data={"x": "1"})),
("a settings write", "no scoped token reaches", lambda: lan(hub, "POST", "/settings",
data={"ui_units": units})),
("a mode switch", "no scoped token reaches", lambda: lan(hub, "POST", "/mode",
data={"controller": mode})),
("the token list", "no scoped token reaches", lambda: lan(hub, "GET", "/tokens")),
("a token of its own", "no scoped token reaches",
lambda: lan(hub, "POST", "/tokens", data={"name": "forgetest child", "caps": "machine.read"})),
("the panel's ?token= form", "authentication required",
lambda: lan(None, "POST", "/settings?token=" + hub, data={"ui_units": units})),
("a stranger's token", "authentication required",
lambda: lan("fft_" + "0" * 32, "GET", "/status")),
("the jog token on /status", "does not hold machine.read", lambda: lan(pend, "GET", "/status"))):
st, body = call()
refused[name] = [st, _words_of(body)[:120]]
ctx.check(st == 403 and want in _words_of(body), "%s -> %s %s (wanted 403 %r)", name, st,
_words_of(body)[:120], want)
ev["refused"] = refused
ctx.check(abs(kernel_xy_mm(ctx)[0] - x0) < 0.01, "a refused jog moved the head")
# A write with a scoped token and nothing else: no session, no panel token.
st, body = lan(pend, "POST", "/motion/jog", data={"x": "1", "feed": "1200"})
ctx.check(st == 200, "the jog token's jog -> %s %s", st, _words_of(body)[:120])
ctx.wait_for(lambda: abs(kernel_xy_mm(ctx)[0] - x0 - 1.0) < 0.02, 10, poll=0.1)
machine_idle(ctx)
moved = kernel_xy_mm(ctx)[0] - x0
ev["jog_mm"] = round(moved, 3)
ctx.log("the jog token jogged the head %.3f mm from the LAN with no session", moved)
ctx.check(abs(moved - 1.0) < 0.02, "the jog moved %.3f mm, not 1", moved)
# Plain HTTP from the LAN: refused, and the log says which token crossed in the clear.
log_path = "/data/log/forgefirm/forgectrl/forgectrl.log"
try:
log_from = os.path.getsize(log_path)
except OSError:
log_from = 0
st, body = lan(hub, "GET", "/status", base=plain)
ev["plain_http"] = [st, _words_of(body)[:120]]
ctx.check(st == 403 and "HTTPS only" in _words_of(body), "a token over plain HTTP -> %s %s", st,
_words_of(body)[:120])
def logged():
try:
with open(log_path, "r", encoding="utf-8", errors="replace") as f:
f.seek(log_from)
return "scoped token %s was presented over plain HTTP" % made["forgetest hub"]["id"] in f.read()
except OSError:
return False
ctx.check(ctx.wait_for(logged, 10, poll=0.5) is not None, "the log does not name the token sent in the clear")
# Reads closed to the LAN: machine.read opens them, nothing else does.
st, body = fc.post("/settings", params={"panel_open_reads": "0"})
ctx.check(st == 200, "closing the reads -> %s", st)
closed = {"nobody": lan(None, "GET", "/status")[0], "machine.read": lan(hub, "GET", "/status")[0],
"motion.jog": lan(pend, "GET", "/status")[0]}
ev["reads_closed"] = closed
ctx.check(closed == {"nobody": 403, "machine.read": 200, "motion.jog": 403},
"with the reads closed: %s", closed)
st, body = fc.post("/settings", params={"panel_open_reads": reads_were})
ctx.check(st == 200, "restoring panel_open_reads=%s -> %s", reads_were, st)
# The list and the store: names and times, never a token or a hash.
st, listed = fc.get("/tokens")
mine = {t["name"]: t for t in listed.get("tokens", []) if t.get("name", "").startswith("forgetest ")}
ev["listed"] = {n: {k: t[k] for k in ("id", "caps", "last_used")} for n, t in mine.items()}
ctx.check(set(mine) == {"forgetest hub", "forgetest jogger", "forgetest viewer"},
"the list: %s", sorted(mine))
ctx.check(mine["forgetest hub"]["caps"] == ["machine.read", "camera.lid"] and
mine["forgetest jogger"]["caps"] == ["motion.jog"], "the capabilities listed: %s", ev["listed"])
ctx.check(all(t["last_used"] > 0 for t in mine.values()), "a used token reads as never used: %s", ev["listed"])
text = json.dumps(listed)
ctx.check(hub[4:] not in text and pend[4:] not in text and view[4:] not in text and "sha" not in text and
all(set(t) == {"id", "name", "caps", "created", "last_used"} for t in mine.values()),
"the list carries more than names, capabilities, and times")
store = "/data/forgefirm/tokens"
mode_bits = os.stat(store).st_mode & 0o777
with open(store, "r", encoding="utf-8", errors="replace") as f:
kept = f.read()
ev["store"] = {"mode": oct(mode_bits),
"holds_a_token": hub[4:] in kept or pend[4:] in kept or view[4:] in kept}
ctx.check(mode_bits == 0o600 and not ev["store"]["holds_a_token"], "the store: %s", ev["store"])
# A revoke ends it at once, and the open reads are no way around it.
st, body = fc.post("/tokens/revoke", data={"id": made["forgetest hub"]["id"]})
ctx.check(st == 200, "the revoke -> %s %s", st, _words_of(body)[:120])
st, body = lan(hub, "GET", "/status")
ev["after_revoke"] = [st, _words_of(body)[:120]]
ctx.check(st == 403 and "authentication required" in _words_of(body),
"a revoked token -> %s %s", st, _words_of(body)[:120])
ctx.check(lan(pend, "POST", "/motion/cancel")[0] == 200, "the revoke took the other token too")
finally:
fc.post("/settings", params={"panel_open_reads": reads_were})
for t in made.values():
fc.post("/tokens/revoke", data={"id": t.get("id", "")})
back = x0 - kernel_xy_mm(ctx)[0]
if abs(back) > 0.02:
machine_idle(ctx)
fc.post("/motion/jog", params={"x": "%.3f" % back, "feed": "1200"})
ctx.wait_for(lambda: abs(kernel_xy_mm(ctx)[0] - x0) < 0.02, 10, poll=0.1)
machine_idle(ctx)
ctx.check(abs(kernel_xy_mm(ctx)[0] - x0) < 0.05, "the head was not returned: %.3f mm off",
kernel_xy_mm(ctx)[0] - x0)
st, listed = fc.get("/tokens")
ctx.check(not [t for t in listed.get("tokens", []) if t.get("name", "").startswith("forgetest ")],
"a test token was left on the machine")
ctx.log("PASS: each token reached what it was granted from the LAN with no session, was refused "
"everything else in words, was refused over plain HTTP and once revoked, and the machine "
"keeps no token")
forgetest/forgetest/suite/motion.py
-195
View File
@@ -1346,198 +1346,3 @@ def _port_idle(ctx, fc, timeout=30.0):
timeout, poll=0.1) timeout, poll=0.1)
ctx.check(ok is not None, "the port jog did not end: %s", _port_state(fc)) ctx.check(ok is not None, "the port jog did not end: %s", _port_state(fc))
machine_idle(ctx) machine_idle(ctx)
def _words(body):
return body if isinstance(body, str) else json.dumps(body)
# ---------------------------------------------------------------- the job runner
def _job(fc):
st, body = fc.get("/job")
if st != 200 or not isinstance(body, dict):
raise hw.HwError("forgectrl /job -> %s %s" % (st, body))
return body
def _job_post(fc, program, name="forgetest", **fields):
"""POST /job the way a client does: a multipart form, the program as
the file part."""
mark = "forgetestJobBoundary7d1"
form = ([("name", name)] if name is not None else []) + sorted(fields.items())
body = "".join('--%s\r\nContent-Disposition: form-data; name="%s"\r\n\r\n%s\r\n' % (mark, k, v)
for k, v in form)
body += ('--%s\r\nContent-Disposition: form-data; name="program"; filename="job.gcode"\r\n'
'Content-Type: text/plain\r\n\r\n%s\r\n--%s--\r\n' % (mark, program, mark))
return fc.post("/job", data=body.encode("ascii"),
headers={"Content-Type": "multipart/form-data; boundary=%s" % mark})
def _job_wait(ctx, fc, timeout, x0=None):
"""Until the job is over; the record, and the farthest the kernel saw X go
from x0. A caller that reads the distance passes the X it took before it
posted the job: by the time this is called the head is already moving."""
if x0 is None:
x0 = kernel_xy_mm(ctx)[0]
far = 0.0
deadline = time.time() + timeout
while time.time() < deadline:
far = max(far, abs(kernel_xy_mm(ctx)[0] - x0))
if _job(fc)["state"] != "running":
break
ctx.sleep(0.1)
return _job(fc), far
def _lease_holder(fc):
return ((fc.status().get("lease") or {}).get("holder") or {})
@test("motion.job", title="The job runner: a posted program runs with the daemon as its sender",
subsystem="motion", kind="auto", mode="grbl", est_min=3,
covers=_MOTION_COVERS + [("forgectrl", "src/jobrun.*"), ("forgectrl", "src/jobpost.*"),
("forgectrl", "src/jobstream.*"), ("forgectrl", "src/lease.*"),
("forgectrl", "src/main.c"), ("forgectrl", "src/status.*")],
requires=["motion.jog-roundtrip"],
steps=["Bed clear; the head needs 60 mm of free travel toward +X. Nobody touches the gantry."],
description="POST /job takes a G-code program as a multipart form and runs it with the daemon "
"as the machine's one sender. A program with a $ line, and one with no name, are "
"refused with 400 and nothing moved. With the suite connected as a Grbl client the "
"job is refused with 409 (a sender is connected) and nothing moved. A dark program "
"(out 20 mm, a dwell, back) then plays: while it does, /status names job:forgetest "
"as the lease holder of kind sender, GET /job reports it running, and a second job, "
"a port jog, and a settings write are each refused in the holder's name. At its "
"end the record says done with every line and the runner's own M2 acknowledged, no "
"discharge and no LASER_ON sample, the kernel's counters saw the 20 mm and are back "
"where they began, and the lease is free. POST /job/abort then stops a long move "
"short: the record says failed (aborted), the controller is in its alarm state, and "
"the lease is free. A job sent into the alarm without unlock fails at its first "
"line with nothing moved; the same job with unlock=1 clears the alarm with the "
"runner's own $X and returns the head to where it was found.")
def job_runner(ctx):
ev = ctx.evidence
fc = hw.Forgectrl()
machine_idle(ctx)
x0, y0 = kernel_start(ctx)
rest = _job(fc)
ev["record_keys"] = sorted(rest)
ctx.check({"state", "owner", "program", "lines", "sent", "acked", "elapsed_s", "lit", "emission",
"reason"} <= set(rest), "GET /job lacks keys: %s", sorted(rest))
ctx.check(rest["state"] != "running", "a job is already running: %s", rest)
ctx.check(not _lease_holder(fc), "the machine is not free at the start: %s", _lease_holder(fc))
# Refused at the door, nothing moved.
refused = {}
st, body = _job_post(fc, "G21\n$X\nG91\nG1 X5 F1200\n")
refused["a $ line"] = [st, _words(body)[:160]]
ctx.check(st == 400 and "line 2" in _words(body) and "$ command" in _words(body),
"a program with a $ line -> %s %s", st, _words(body)[:160])
st, body = _job_post(fc, "G21\nG91\nG1 X5 F1200\n", name=None)
refused["no name"] = [st, _words(body)[:160]]
ctx.check(st == 400 and "name" in _words(body), "a job with no name -> %s %s", st, _words(body)[:160])
with ctx.grbl() as g:
clean_slate(ctx, g)
# The lease asks the controller's published state, which follows the connection.
ok = ctx.wait_for(lambda: ((fc.status().get("lease") or {}).get("observed") or {}).get("sender"),
10, poll=0.2)
ctx.check(ok is not None, "the controller never reported the suite's client as connected")
st, body = _job_post(fc, "G21\nG91\nG1 X5 F1200\n")
refused["a connected client"] = [st, _words(body)[:160]]
ctx.check(st == 409 and "sender is connected" in _words(body),
"a job beside a connected Grbl client -> %s %s", st, _words(body)[:160])
ev["refused_at_the_door"] = refused
ctx.check(abs(kernel_xy_mm(ctx)[0] - x0) < 0.01, "a refused job moved the head")
ok = ctx.wait_for(lambda: not ((fc.status().get("lease") or {}).get("observed") or {}).get("sender"),
10, poll=0.2)
ctx.check(ok is not None, "the controller still reports the suite's client as connected")
moved = 0.0
try:
# A dark program plays, and holds the machine while it does.
st, body = _job_post(fc, "(forgetest motion.job)\nG21\nG91\nG1 X20 F1200 ; out\nG4 P2\n"
"G1 X-20 F1200\nG90\n")
ctx.check(st == 200 and isinstance(body, dict) and body.get("state") == "running",
"POST /job -> %s %s", st, _words(body)[:200])
ok = ctx.wait_for(lambda: _lease_holder(fc).get("owner") == "job:forgetest", 5, poll=0.1)
held = _lease_holder(fc)
ev["holder"] = held
ctx.log("the lease while the job plays: %s", held)
ctx.check(ok is not None and held.get("kind") == "sender" and "forgetest" in (held.get("words") or ""),
"the lease does not name the job: %s", held)
rec = _job(fc)
ctx.check(rec["state"] == "running" and rec["owner"] == "job:forgetest" and rec["program"] is True
and rec["lines"] == 6, "the record while it plays: %s", rec)
beside = {}
for name, call in (("a second job", lambda: _job_post(fc, "G21\n", name="second")),
("a port jog", lambda: fc.post("/motion/jog", params={"x": "1"})),
("a settings write", lambda: fc.post("/settings", params={
"ui_units": fc.settings().get("ui_units") or "metric"}))):
st, body = call()
beside[name] = [st, _words(body)[:160]]
ctx.check(st == 409 and "a job (forgetest) holds the machine" in _words(body),
"%s beside the job -> %s %s", name, st, _words(body)[:160])
ev["refused_beside_the_job"] = beside
rec, far = _job_wait(ctx, fc, 40, x0)
x1 = kernel_xy_mm(ctx)[0]
ev["dark_job"] = {"record": rec, "farthest_mm": round(far, 3), "kernel": [x0, x1]}
ctx.log("the dark job: %s; the kernel saw %.3f mm out and ended %.3f from the start",
rec, far, x1 - x0)
ctx.check(rec["state"] == "done" and rec["reason"] == "", "the job did not end well: %s", rec)
ctx.check(rec["sent"] == rec["acked"] == rec["lines"] + 1,
"lines %s, sent %s, acked %s (the runner's M2 is the one more)",
rec["lines"], rec["sent"], rec["acked"])
ctx.check(rec["lit"] is False and rec["emission"]["laser_on_samples"] == 0,
"a dark job's witnesses: %s", rec["emission"])
ctx.check(19.0 < far < 21.0, "the kernel saw %.3f mm of the 20", far)
ctx.check(abs(x1 - x0) < 0.1, "the job did not end where it began: %.3f mm off", x1 - x0)
ctx.check(not _lease_holder(fc), "the lease was not given back: %s", _lease_holder(fc))
# The abort stops a long move short.
st, body = _job_post(fc, "G21\nG91\nG1 X40 F300\nG90\n")
ctx.check(st == 200, "the long job -> %s %s", st, _words(body)[:200])
ctx.sleep(2.0)
st, body = fc.post("/job/abort")
ctx.check(st == 200 and isinstance(body, dict) and body.get("state") == "failed"
and "aborted" in body.get("reason", ""), "POST /job/abort -> %s %s", st, _words(body)[:200])
ctx.sleep(1.0)
moved = kernel_xy_mm(ctx)[0] - x0
after = _port_state(fc)
ev["abort"] = {"record": body, "moved_mm": round(moved, 3), "controller": after.get("state")}
ctx.log("aborted a 40 mm move at %.3f mm; the controller is in %s", moved, after.get("state"))
ctx.check(1.0 < moved < 35.0, "the abort did not stop the move short: moved %.3f of 40", moved)
ctx.check(str(after.get("state", "")).startswith("Alarm"),
"the controller is in %s after the abort, not its alarm state", after.get("state"))
ctx.check(not _lease_holder(fc), "the lease was not given back after the abort: %s", _lease_holder(fc))
# Into the alarm: refused by the controller without the unlock, and
# with it the head goes back where it was found.
back = "G21\nG91\nG1 X%.3f F1200\nG90\n" % -moved
st, body = _job_post(fc, back)
ctx.check(st == 200, "the job into the alarm -> %s %s", st, _words(body)[:200])
rec, far = _job_wait(ctx, fc, 20)
ev["into_the_alarm"] = rec
ctx.check(rec["state"] == "failed" and "line 1" in rec["reason"],
"a job into the alarm without unlock: %s", rec)
ctx.check(abs((kernel_xy_mm(ctx)[0] - x0) - moved) < 0.05, "a refused job moved the head")
st, body = _job_post(fc, back, unlock="1")
ctx.check(st == 200, "the unlocking job -> %s %s", st, _words(body)[:200])
rec, far = _job_wait(ctx, fc, 40)
ev["unlocked"] = rec
ctx.log("the unlocking job: %s", rec)
ctx.check(rec["state"] == "done" and rec["sent"] == rec["acked"] == rec["lines"] + 2,
"the unlocking job (its $X and its M2 are the two more): %s", rec)
moved = kernel_xy_mm(ctx)[0] - x0
finally:
if _job(fc)["state"] == "running":
fc.post("/job/abort")
ctx.sleep(1.0)
left = kernel_xy_mm(ctx)[0] - x0
if abs(left) > 0.05:
ctx.log("returning the head %.3f mm through a Grbl client", -left)
_return_x(ctx, left)
check_kernel_returned(ctx, ev, (x0, y0), tag="job")
machine_idle(ctx)
ctx.log("PASS: a posted program ran as job:forgetest with the lease held as a sender, everything "
"beside it was refused in its name, the abort stopped a move short and freed the machine, "
"and the unlocking job cleared the alarm and returned the head")
forgetest/forgetest/suite/setup.py
-69
View File
@@ -999,72 +999,3 @@ def first_run_flow(ctx):
ctx.check(st == 200, "POST /wiz/complete -> %s %s", st, body if isinstance(body, dict) else "") ctx.check(st == 200, "POST /wiz/complete -> %s %s", st, body if isinstance(body, dict) else "")
finished_first_run(ctx) finished_first_run(ctx)
ctx.log("the run completed: gate open, the controller running") ctx.log("the run completed: gate open, the controller running")
@test("setup.first-run-page", title="The first-run page, walked once", subsystem="setup",
kind="operator", hardware="takeover", est_min=5,
covers=[("forgectrl", "src/ui/wizard.*"), ("forgectrl", "src/ui/forms.js"), ("forgectrl", "src/ui/md.js"),
("forgectrl", "src/ui/login.*"), ("forgectrl", "src/ui/embed_docs.cmake"), ("forgectrl", "src/tls.*")],
requires=["setup.first-run-flow"],
actions=["button"], hands=["workstation"],
steps=["From the workstation open https://<address>/ and accept the self-signed certificate.",
"Click through the setup: scroll each document to its end and accept it (type I "
"UNDERSTAND for the safety document); press the machine's button when it breathes "
"teal; create a temporary account (any name and password: it is removed at the end); "
"take the preferences, the machine, and the cloud step as offered. The intake fans "
"show as counted already. The page then offers to continue the setup or to start "
"using the machine: click Start using the machine.",
"Confirm that the control panel opened."],
description="The page against the real daemon over HTTPS, the one place a browser is the "
"client: the record seeded as a first run behind the checks and no account, "
"'/' serves the wizard with its choice after the cloud step and the intake fans, "
"and the test follows the record while the operator clicks through (the "
"documents accepted, the press step with the LEDs breathing teal, the account "
"created, preferences, machine, and cloud done and the gate open with the "
"intake fans counted, then Start using the machine: the run complete, the gate "
"open, the controller running). The operator confirms the panel opened. "
"Everything the run wrote is restored as setup.first-run-flow restores it.")
def first_run_page(ctx):
fc = ctx.forgectrl
ev = ctx.evidence
ip = lan_ip()
ctx.check(ip, "cannot determine the board's LAN address")
with first_run(ctx) as state:
st, page, hdrs = request("https://%s" % ip, "GET", "/")
ev["root_page"] = st
ctx.check(st == 200 and b'id="exitlink"' in page, "'/' over HTTPS does not serve the wizard (%s)", st)
ctx.check(b'id="st-choice"' in page and b"Start using the machine" in page,
"the wizard carries no choice after the required steps")
ctx.notice("Open https://%s/ from the workstation, accept the certificate, and click through "
"the setup as the steps describe. The test follows the record." % ip)
took = ctx.wait_for(lambda: wiz(fc).get("advisories_complete") is True, 900)
ctx.check(took is not None, "the four documents were not accepted within 15 min")
ctx.log("the documents are accepted (%.0f s)", took)
took = ctx.wait_for(lambda: wiz(fc).get("button") == "waiting", 300)
ctx.check(took is not None, "the press step did not start within 5 min")
took = ctx.wait_for(lambda: breathes_teal(led_pattern()), 10)
pattern = led_pattern()
ev["led_pattern"] = pattern
ctx.check(took is not None, "the button LEDs do not breathe teal at the press step: %s", pattern)
ctx.act("button", "press", until=lambda: wiz(fc).get("acceptance_done") is True, timeout=600,
text="It breathes teal: this press is the acceptance. The page must be at the press step.")
ctx.notice("Continue on the page: the account, the preferences, the machine, the cloud step, "
"then Start using the machine.")
took = ctx.wait_for(lambda: (wiz(fc).get("users") or {}).get("exists") is True, 600)
ctx.check(took is not None, "no account was created within 10 min")
state["temp_account"] = (wiz(fc).get("users") or {}).get("name")
ev["temp_account"] = state["temp_account"]
ctx.log("account %r created (%.0f s)", state["temp_account"], took)
took = ctx.wait_for(lambda: wiz(fc).get("gate") == "open", 900)
w = wiz(fc)
ev["after_required"] = wiz_summary(w)
ctx.check(took is not None, "the gate did not open within 15 min of the account: %s", w.get("why"))
done = w.get("versions") or {}
ctx.check(all(done.get(wid) == 1 for wid in REQUIRED_STEPS) and w.get("fans_counted") is True,
"the gate opened without the required steps or the intake fans: %s", wiz_summary(w))
ctx.log("preferences, machine, and cloud done; the gate is open (%.0f s)", took)
took = ctx.wait_for(lambda: wiz(fc).get("completed") is True, 900)
ctx.check(took is not None, "the setup did not complete within 15 min")
finished_first_run(ctx)
ctx.clear_notice()
ctx.confirm("Did the control panel open after Start using the machine?")
forgetest/forgetest/suite/setup_burns.py
-74
View File
@@ -48,10 +48,8 @@ BURN_COVERS = [("forgectrl", "src/wizlive.*"), ("forgectrl", "src/wizrun.h"),
# burn's tail brings it back. # burn's tail brings it back.
HOME = "The head is home" HOME = "The head is home"
EDGE_KEY = "lens_hall_edge_z_mm" EDGE_KEY = "lens_hall_edge_z_mm"
FLOOR_KEY = "laser_floor_density"
FOCUS_KEYS = ["thickness_mm", "z_passes", "ladder_z", "window", "emission", "pick", "picked_lines", FOCUS_KEYS = ["thickness_mm", "z_passes", "ladder_z", "window", "emission", "pick", "picked_lines",
"pick_half_steps", "edge_z_mm", "steps_per_mm", "max_height_mm", "focus_range_mm", "summary"] "pick_half_steps", "edge_z_mm", "steps_per_mm", "max_height_mm", "focus_range_mm", "summary"]
FLOOR_KEYS = ["thickness_mm", "z_passes", "emission", "faintest_density", "floor_density", "summary"]
def preview_ok(ctx, wid): def preview_ok(ctx, wid):
@@ -319,75 +317,3 @@ def focus(ctx):
ctx.check(ok is not None, "the controller did not come back on the settings as found") ctx.check(ok is not None, "the controller did not come back on the settings as found")
ctx.log("PASS: the focus burn ran in the window the machine holds and wrote the edge alone; " ctx.log("PASS: the focus burn ran in the window the machine holds and wrote the edge alone; "
"settings restored and the controller restarted on them") "settings restored and the controller restarted on them")
@test("setup.floor", title="The laser floor burn: the rungs and the floor it writes",
subsystem="setup", kind="live", hardware="takeover", mode="grbl", est_min=10,
covers=BURN_COVERS,
requires=["forgectrl.auth", "setup.check-motion", "laser.emission-witness", "setup.focus"],
actions=["button"], hands=["scrap"],
steps=["The gantry stops installed and the head at its home against them. A fresh flat scrap "
"of wood about 3 mm (1/8 in) thick and at least 86 x 68 mm (3.4 x 2.7 in), pushed into "
"the back-left corner of the cut area: as far left as it goes, its top edge at the top "
"of the cut area; 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 the arm press after it, when the burn itself "
"asks for it. With no actuator the start is the Ready answer instead, and you press "
"the button again at the arm, as the page asks.",
"Nothing else: the test answers every prompt itself and puts back the setting the "
"burn writes."],
description="The floor burns at the focus the focus burn measured, and the wizard refuses "
"to start without Focus on the record, so the record must carry laser.focus "
"(setup.focus, ahead of this test, leaves it there), and the lens reach /status "
"reports must include the scrap's thickness (the wizard refuses a surface the "
"lens cannot focus on); both are read before the scrap is asked for. Then POST "
"/wiz/laser.floor/start on a fresh scrap, set up as the focus burn is: "
"answered that the gantry stops are installed, the lens referenced, the "
"controller in loopback posture, answered that the head is home and that the "
"scrap is in the corner, the thickness in the machine's units (3.2 mm or 0.125 "
"in); its program declares X0 Y0 at the head, and its tail puts the lens back "
"on the reference, returns the head there, clears the G92 origin (G92.1), and "
"ends (M2). With its preview and program served, every Z the program commands "
"inside the reach /status reports, and the program ending in that tail, it "
"burns twelve rungs from 2 to 24 percent density at "
"the focal point on the scrap's surface, the floor and the curve off for the "
"job, after the press, seen by the two witnesses. The pick is 8, so the floor "
"written is 10 (two above it) and the setting reads it; the result carries its "
"keys; the record carries laser.floor at version 1. laser_floor_density is "
"restored at the end (the controller loads it at every arm).")
def floor(ctx):
fc = ctx.forgectrl
ev = ctx.evidence
# The record and the reach, before the scrap is asked for: the wizard
# refuses the burn without Focus on the record, and it refuses a scrap
# the lens cannot focus on, so the operator's setup would be for
# nothing.
st, w = fc.get("/wiz")
focus_v = ((w if isinstance(w, dict) else {}).get("versions") or {}).get("laser.focus")
ev["focus_on_record"] = focus_v
ctx.check(st == 200 and isinstance(focus_v, int) and focus_v >= 1,
"the record carries no Focus (laser.focus, %s): the floor burns at the focus it measures, "
"and the wizard refuses to start without it", focus_v)
_answer, want_mm = thickness_answer(ctx)
lens = (fc.status() or {}).get("lens") or {}
lo, hi = lens.get("reach_min"), lens.get("reach_max")
ev["floor_reach"] = [lo, hi]
ctx.check(isinstance(lo, (int, float)) and isinstance(hi, (int, float))
and lo - 0.01 <= want_mm <= hi + 0.01,
"the lens focuses %s to %s mm above the tray, which leaves out the %s mm scrap: the floor "
"refuses a surface the lens cannot focus on", lo, hi, want_mm)
ctx.ready("The laser floor burn: twelve rungs on the scrap after a press. The bench presses when "
"the button lights.")
with Restore(ctx, [FLOOR_KEY]):
r = burn(ctx, "laser.floor", "floor-arm", {"floor-pick": "8"}, FLOOR_KEYS)
ctx.check(r.get("faintest_density") == 8 and r.get("floor_density") == 10,
"the floor reads %s from the faintest rung %s, answered 8", r.get("floor_density"),
r.get("faintest_density"))
held = (fc.settings() or {}).get(FLOOR_KEY)
try:
held_ok = float(held) == r.get("floor_density")
except (TypeError, ValueError):
held_ok = False
ctx.check(held_ok, "%s reads %r, the burn wrote %s", FLOOR_KEY, held, r.get("floor_density"))
ctx.log("floor: the faintest rung %s, the floor %s", r.get("faintest_density"), r.get("floor_density"))
ctx.log("PASS: the floor burn wrote two above the faintest rung; settings restored")
forgetest/forgetest/suite/setup_dark.py
-130
View File
@@ -174,38 +174,6 @@ def check_switches(ctx):
ctx.log("switches: %s", json.dumps(r)) ctx.log("switches: %s", json.dumps(r))
@test("setup.check-sensors", title="The sensors check reads a plausible machine at rest",
subsystem="setup", 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 and no question asked. 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 the "
"settings read as before (the check writes none).")
def check_sensors(ctx):
fc = ctx.forgectrl
ev = ctx.evidence
before = fc.settings() or {}
ev["settings_before"] = dict(before)
def on_prompt(p):
ctx.fail("the sensors check asked a question: %s", p)
last = run_check(ctx, "sensors", on_prompt, 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"))
after = fc.settings() or {}
changed = {k: (before.get(k), after.get(k)) for k in set(before) | set(after)
if before.get(k) != after.get(k)}
ctx.check(not changed, "the sensors check wrote a setting: %s", changed)
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("setup.check-airflow", title="The airflow check measures the fans and sets the floors", @test("setup.check-airflow", title="The airflow check measures the fans and sets the floors",
subsystem="setup", kind="auto", hardware="takeover", est_min=3, subsystem="setup", kind="auto", hardware="takeover", est_min=3,
covers=DARK_COVERS + [("forgectrl", "src/cool.c"), ("forgectrl", "src/airflow.*"), covers=DARK_COVERS + [("forgectrl", "src/cool.c"), ("forgectrl", "src/airflow.*"),
@@ -293,86 +261,6 @@ def check_airflow(ctx):
ctx.check(ok is not None, "the controller did not come back after the check") ctx.check(ok is not None, "the controller did not come back after the check")
@test("setup.check-intakes", title="The intake fans step counts the machine's intake fans",
subsystem="setup", kind="auto", hardware="takeover", est_min=2,
covers=DARK_COVERS + [("forgectrl", "src/cool.c"), ("forgectrl", "src/airflow.*")],
requires=["forgectrl.auth"], precheck=intakes_counted,
description="POST /wiz/airflow.intakes/start with the intake fan count set (the precheck "
"refuses the start without it: the gate would hold every test after this one). "
"The controller stops, the fans run at the run profile for 20 s, and the tachs "
"count the intakes: both turning is two. The step asks the operator to count "
"when the tachs cannot (intake 1 not up to speed, or intake 2 giving no "
"reading, which reads the same as a machine built with one intake fan); both "
"intakes turn on the bench reference, so that question is a fan fault, and it "
"fails the test unanswered. The result carries both intakes' speeds, whether "
"intake 2 was silent, and the count by the tachs, which must be the count the "
"machine held. The setting reads the count, the record carries airflow.intakes "
"at version 1, the count is put back as found, and the controller must be "
"running again.")
def check_intakes(ctx):
fc = ctx.forgectrl
ev = ctx.evidence
def on_prompt(p):
if p.get("id") == "intake-count":
# The tachs could not count, and the step's text says why.
ctx.fail("the intake fans step asked for the count, which the tachs could not make; both "
"intakes turn on the bench reference, so a fan needs attention: %s", p.get("text"))
ctx.fail("the intake fans step asked a question: %s", p)
with Restore(ctx, [INTAKE_KEY]) as held:
found = held.before.get(INTAKE_KEY)
ctx.check(found in ("1", "2"), "%s reads %r, not a count", INTAKE_KEY, found)
last = run_check(ctx, "airflow.intakes", on_prompt, 180)
r = last.get("result") or {}
n = r.get("intake_fans")
ev["count"] = {"found": found, "counted": n, "by": r.get("counted_by"),
"intake_1_rpm": r.get("intake_1_rpm"), "intake_2_rpm": r.get("intake_2_rpm"),
"intake_2_silent": r.get("intake_2_silent")}
ctx.check(r.get("counted_by") == "tachs", "the fans were not counted by the tachs: %s", r)
ctx.check(str(n) == found, "the tachs count %r intake fans, the machine held %s", n, found)
s = fc.settings() or {}
ctx.check(s.get(INTAKE_KEY) == str(n), "%s reads %r, the step counted %r", INTAKE_KEY, s.get(INTAKE_KEY), n)
ctx.log("intake fans: %s by the tachs (intake 1 %s rpm, intake 2 %s rpm)", n, r.get("intake_1_rpm"),
r.get("intake_2_rpm"))
ok = ctx.wait_for(lambda: (fc.get("/mode")[1] or {}).get("controller") == "running", 60)
ctx.check(ok is not None, "the controller did not come back after the step")
@test("setup.check-cameras", title="The cameras check captures both cameras",
subsystem="setup", 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("setup.check-motion", title="The motion check proves the rail, the lens reference, and the jogs", @test("setup.check-motion", title="The motion check proves the rail, the lens reference, and the jogs",
subsystem="setup", kind="auto", hardware="takeover", est_min=5, subsystem="setup", kind="auto", hardware="takeover", est_min=5,
covers=DARK_COVERS + [("forgectrl", "src/super.c"), ("forgectrl", "src/liveness.c"), covers=DARK_COVERS + [("forgectrl", "src/super.c"), ("forgectrl", "src/liveness.c"),
@@ -513,24 +401,6 @@ def check_lens(ctx):
ctx.check(ok is not None, "the controller did not come back after the check") ctx.check(ok is not None, "the controller did not come back after the check")
@test("setup.check-flow-verify", title="The flow check runs as a setup check",
subsystem="setup", 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("setup.cloud-header-capture", title="The cloud header check takes one print's envelope", @test("setup.cloud-header-capture", title="The cloud header check takes one print's envelope",
subsystem="setup", kind="operator", hardware="takeover", mode="grbl", est_min=8, subsystem="setup", kind="operator", hardware="takeover", mode="grbl", est_min=8,
covers=DARK_COVERS + [("forgectrl", "src/super.c"), covers=DARK_COVERS + [("forgectrl", "src/super.c"),
forgetest/forgetest/suite/update.py
-82
View File
@@ -147,85 +147,3 @@ def slots_and_signature(ctx):
except OSError: except OSError:
pass pass
shutil.rmtree(work, ignore_errors=True) shutil.rmtree(work, ignore_errors=True)
@test("update.product-gate", title="The firmware upload takes firmware only: an extension package is refused by name",
subsystem="update", kind="auto", est_min=1,
covers=_UPDATE_COVERS + [("forgectrl", "src/fwproduct.c"), ("forgectrl", "src/fwproduct.h")],
requires=["forgectrl.auth"],
description="Firmware and extension packages are the same container, and a signature says who "
"made an archive, never what it is. Archives made on the spot with the machine's own "
"fwup go to POST /update/upload: an extension package unsigned, one signed with a "
"throwaway key, and one that carries a task are each refused with 400 in words that "
"say what they are, and nothing is staged; an archive of some other product is "
"refused as not firmware; and one whose product is ForgeFIRM firmware is taken as an "
"unsigned upload, which is what it is. Nothing is applied: the apply calls the same "
"gate before fwup -a, and its proof is the daemon's host test, because a catalog test "
"that could fail there would write a slot.")
def product_gate(ctx):
fc = ctx.forgectrl
ev = ctx.evidence
staged = "/data/forgefirm/upload.fw"
ctx.check(not os.path.exists(staged), "an upload is already staged: %s", staged)
work = tempfile.mkdtemp(prefix="forgetest-gate-")
def archive(name, product, task=False, sign=False):
conf = os.path.join(work, name + ".conf")
with open(conf, "w") as f:
f.write('meta-product = "%s"\nmeta-version = "0.0.0-test"\n'
'file-resource payload.tar.gz { host-path = "payload.tar.gz" }\n' % product)
if task:
f.write('task upgrade.a { on-resource payload.tar.gz { raw_write(0) } }\n')
out = name + ".fw"
cmd = "cd %s && fwup -c -f %s.conf -o %s" % (work, name, out)
if sign:
cmd += " && fwup -S -s fwup-key.priv -i %s -o %s.signed && mv %s.signed %s" % (out, out, out, out)
rc, text = hw.run(["sh", "-c", cmd], timeout=60)
ctx.check(rc == 0, "could not make %s (rc %s): %s", name, rc, text.strip()[:200])
return os.path.join(work, out)
def upload(path):
mark = "forgetestGateBoundary9c2"
body = ('--%s\r\nContent-Disposition: form-data; name="file"; filename="%s"\r\n'
'Content-Type: application/octet-stream\r\n\r\n' % (mark, os.path.basename(path))).encode()
with open(path, "rb") as f:
body += f.read() + ("\r\n--%s--\r\n" % mark).encode()
return fc.post("/update/upload", data=body,
headers={"Content-Type": "multipart/form-data; boundary=%s" % mark})
try:
with open(os.path.join(work, "payload.tar.gz"), "wb") as f:
f.write(b"forgetest product-gate drill\n" * 8)
rc, text = hw.run(["sh", "-c", "cd %s && fwup -g" % work], timeout=60)
ctx.check(rc == 0, "fwup -g did not make a key pair (rc %s): %s", rc, text.strip()[:200])
seen = {}
for name, product, task, sign in (("extension-unsigned", "ForgeFIRM extension", False, False),
("extension-signed", "ForgeFIRM extension", False, True),
("extension-with-task", "ForgeFIRM extension", True, False)):
st, body = upload(archive(name, product, task, sign))
words = body.get("error", "") if isinstance(body, dict) else str(body)
seen[name] = [st, words]
ctx.log("upload %s -> %s %s", name, st, words)
ctx.check(st == 400 and "an extension package, not firmware" in words,
"%s at the firmware upload -> %s %s", name, st, words)
ctx.check(not os.path.exists(staged), "%s was refused and is staged all the same", name)
st, body = upload(archive("another-product", "forgetest"))
words = body.get("error", "") if isinstance(body, dict) else str(body)
seen["another-product"] = [st, words]
ctx.log("upload another-product -> %s %s", st, words)
ctx.check(st == 400 and "not firmware" in words and not os.path.exists(staged),
"an archive of another product -> %s %s", st, words)
st, body = upload(archive("firmware-unsigned", "ForgeFIRM firmware"))
seen["firmware-unsigned"] = [st, body]
ctx.log("upload firmware-unsigned -> %s %s", st, body)
ctx.check(st == 200 and isinstance(body, dict) and body.get("signature") == "unsigned"
and os.path.exists(staged), "an unsigned archive that is firmware -> %s %s", st, body)
ev["uploads"] = seen
finally:
try:
os.unlink(staged)
except OSError:
pass
shutil.rmtree(work, ignore_errors=True)
ctx.log("PASS: three extension packages and one foreign product refused at the firmware upload with "
"nothing staged; firmware taken as the unsigned upload it is")
forgetest/tests/test_setup_burns.py
+44 -116
View File
@@ -3,16 +3,15 @@
# https://community.openglow.org # https://community.openglow.org
# SPDX-License-Identifier: MIT # SPDX-License-Identifier: MIT
"""The setup's two burns in the catalog: two live tests on the button with """The setup's focus burn in the catalog: a live test on the button with
a scrap at hand, their covers naming the live runner, the renderer, the a scrap at hand, its covers naming the live runner, the renderer, the
sender, the page, and the driver's laser module; and each burn against a sender, the page, and the driver's laser module; and the burn against a
scripted daemon: the preview and the program read, the start, the scripted daemon: the preview and the program read, the start, the
prompts answered in order (the gantry stops, the head at home, the scrap, prompts answered in order (the gantry stops, the head at home, the scrap,
its thickness in the machine's units, the arm, the machine's own press its thickness in the machine's units, the arm, the machine's own press
prompt, the pick), the witnesses judged, the record read at version 1, prompt, the pick), the witnesses judged, the record read at version 1,
the setting the burn wrote put back as found, and for the focus burn the the setting the burn wrote put back as found, and the controller started
controller started again on the edge and the window put back. The floor again on the edge and the window put back."""
starts only with Focus on the record and a scrap the lens reaches."""
import os import os
import sys import sys
import unittest import unittest
@@ -23,7 +22,7 @@ from forgetest import catalog # noqa: E402
from forgetest.runner import Aborted, Context, Failed, Run # noqa: E402 from forgetest.runner import Aborted, Context, Failed, Run # noqa: E402
from forgetest.suite import setup_burns, setup_dark # noqa: E402 from forgetest.suite import setup_burns, setup_dark # noqa: E402
LIVE = ["laser.focus", "laser.floor"] LIVE = ["laser.focus"]
EMISSION = {"hv_max": 610, "laser_on_samples": 900, "thermopile_delta": 600, "beam_sensor_seen": True, EMISSION = {"hv_max": 610, "laser_on_samples": 900, "thermopile_delta": 600, "beam_sensor_seen": True,
"lit_s": 30.0} "lit_s": 30.0}
SPM = 2.922 # the lens screw, half-steps per mm ($102) SPM = 2.922 # the lens screw, half-steps per mm ($102)
@@ -55,7 +54,7 @@ def ladder_z(edge, below, above):
return [round(edge + round(above - (above + below) * i / 11) / SPM, 2) for i in range(12)] return [round(edge + round(above - (above + below) * i / 11) / SPM, 2) for i in range(12)]
# Each burn as the daemon runs it: its prompts in order (id, kind, # The burn as the daemon runs it: its prompts in order (id, kind,
# options) and the result it ends with. The press prompt is the # options) and the result it ends with. The press prompt is the
# machine's own: nothing answers it, the discharge closes it. The focus # machine's own: nothing answers it, the discharge closes it. The focus
# burn's window and ladder are the ones the settings held at its start. # burn's window and ladder are the ones the settings held at its start.
@@ -69,14 +68,10 @@ SCRIPT = {
{"z_passes": [100, 101], "emission": EMISSION, "pick": 11, "picked_lines": [11, 11], {"z_passes": [100, 101], "emission": EMISSION, "pick": 11, "picked_lines": [11, 11],
"pick_half_steps": -8, "edge_z_mm": 6.96, "steps_per_mm": SPM, "max_height_mm": 13.81, "pick_half_steps": -8, "edge_z_mm": 6.96, "steps_per_mm": SPM, "max_height_mm": 13.81,
"focus_range_mm": {"min": 2.17, "max": 13.81}, "summary": "Line 11."}), "focus_range_mm": {"min": 2.17, "max": 13.81}, "summary": "Line 11."}),
"laser.floor": (SETUP + [("floor-arm", "continue", ["Continue"]), PRESS,
("floor-pick", "choice", [str(2 * i) for i in range(1, 13)] + ["None"])],
{"z_passes": [100, 101], "emission": EMISSION, "faintest_density": 8.0,
"floor_density": 10.0, "summary": "The floor is 10."}),
} }
# What each burn writes when it ends. # What the burn writes when it ends.
WRITES = {"laser.focus": {"lens_hall_edge_z_mm": "6.96"}, "laser.floor": {"laser_floor_density": "10"}} WRITES = {"laser.focus": {"lens_hall_edge_z_mm": "6.96"}}
# The machine as the burns find it: its own edge, the window the Lens # The machine as the burn finds it: its own edge, the window the Lens
# travel check wrote, its floor. # travel check wrote, its floor.
FOUND = {"lens_hall_edge_z_mm": "4.10", "lens_stop_below_steps": "14", "lens_stop_above_steps": "20", FOUND = {"lens_hall_edge_z_mm": "4.10", "lens_stop_below_steps": "14", "lens_stop_above_steps": "20",
"laser_floor_density": "12", "ui_units": "metric"} "laser_floor_density": "12", "ui_units": "metric"}
@@ -89,27 +84,23 @@ class RegistrationTests(unittest.TestCase):
def setUp(self): def setUp(self):
self.reg = catalog.load_suite() self.reg = catalog.load_suite()
def test_two_live_tests_on_the_button_with_a_scrap_at_hand(self): def test_a_live_test_on_the_button_with_a_scrap_at_hand(self):
for tid in ("setup.focus", "setup.floor"): t = self.reg["setup.focus"]
t = self.reg[tid] self.assertEqual(t.kind, "live")
self.assertEqual(t.kind, "live", tid) self.assertEqual(t.mode, "grbl")
self.assertEqual(t.mode, "grbl", tid) self.assertEqual(t.hardware, "takeover")
self.assertEqual(t.hardware, "takeover", tid) self.assertEqual(list(t.actions), ["button"])
self.assertEqual(list(t.actions), ["button"], tid) self.assertEqual(list(t.hands), ["scrap"])
self.assertEqual(list(t.hands), ["scrap"], tid) self.assertFalse(t.fixture_runnable(("button", "lid", "interlock")))
self.assertFalse(t.fixture_runnable(("button", "lid", "interlock")), tid) self.assertIsNone(getattr(t, "precheck", None))
self.assertIsNone(getattr(t, "precheck", None), tid) self.assertEqual(list(t.requires),
self.assertEqual(list(self.reg["setup.focus"].requires),
["forgectrl.auth", "setup.check-motion", "laser.emission-witness", "setup.check-lens"]) ["forgectrl.auth", "setup.check-motion", "laser.emission-witness", "setup.check-lens"])
# the floor burns at the focus Focus measured, and refuses without it
self.assertEqual(list(self.reg["setup.floor"].requires),
["forgectrl.auth", "setup.check-motion", "laser.emission-witness", "setup.focus"])
def test_the_sheet_is_gone(self): def test_the_sheet_is_gone(self):
self.assertFalse([i for i in self.reg if i.startswith("setup.sheet")]) self.assertFalse([i for i in self.reg if i.startswith("setup.sheet")])
def test_covers_the_live_runner_the_renderer_the_sender_the_page_and_the_driver(self): def test_covers_the_live_runner_the_renderer_the_sender_the_page_and_the_driver(self):
for tid in ("setup.focus", "setup.floor"): for tid in ("setup.focus",):
covers = set(self.reg[tid].covers) covers = set(self.reg[tid].covers)
for want in (("forgectrl", "src/wizlive.*"), ("forgectrl", "src/sheet.*"), for want in (("forgectrl", "src/wizlive.*"), ("forgectrl", "src/sheet.*"),
("forgectrl", "src/jobstream.*"), ("forgectrl", "src/curverec.*"), ("forgectrl", "src/jobstream.*"), ("forgectrl", "src/curverec.*"),
@@ -120,7 +111,7 @@ class RegistrationTests(unittest.TestCase):
self.assertIn(want, covers, "%s lacks %s" % (tid, want)) self.assertIn(want, covers, "%s lacks %s" % (tid, want))
def test_the_steps_ask_for_a_scrap_in_the_corner_and_one_press(self): def test_the_steps_ask_for_a_scrap_in_the_corner_and_one_press(self):
for tid, size in (("setup.focus", "78 x 68 mm"), ("setup.floor", "86 x 68 mm")): for tid, size in (("setup.focus", "78 x 68 mm"),):
text = " ".join(self.reg[tid].steps) text = " ".join(self.reg[tid].steps)
self.assertIn(size, text, tid) self.assertIn(size, text, tid)
self.assertIn("back-left corner", text, tid) self.assertIn("back-left corner", text, tid)
@@ -130,10 +121,9 @@ class RegistrationTests(unittest.TestCase):
class ScriptedDaemon: class ScriptedDaemon:
"""The wizard runner as the suite sees it: /wiz/<id>/start opens the """The wizard runner as the suite sees it: /wiz/<id>/start opens the
first prompt (the floor ends at once, with the wizard's error, when the first prompt, each answer opens the next (a thickness that is not a
record carries no Focus), each answer opens the next (a thickness that plain number or out of range in the machine's units opens it again),
is not a plain number, out of range in the machine's units, or, for the press prompt
the floor, outside the lens reach opens it again), the press prompt
closes on the poll after the one that served it, the last answer ends closes on the poll after the one that served it, the last answer ends
the burn with its result and its writes, /wiz/dark reports it, GET /wiz the burn with its result and its writes, /wiz/dark reports it, GET /wiz
carries the versions (the record carries Focus already, as the bench's carries the versions (the record carries Focus already, as the bench's
@@ -234,11 +224,6 @@ class ScriptedDaemon:
return 404, {"error": "no such wizard"} return 404, {"error": "no such wizard"}
self.order.append(wid) self.order.append(wid)
self.settings_at_start[wid] = dict(s) self.settings_at_start[wid] = dict(s)
if wid == "laser.floor" and not self.versions.get("laser.focus"):
# the floor burns at the focus Focus measured: no prompt, no motion
self.dark.update({"id": wid, "running": False, "phase": "", "prompt": None, "result": None,
"error": "run Focus first: the floor burns at the focus it measures"})
return 200, {"started": True, "id": wid}
s.update(self.start_writes.get(wid, {})) s.update(self.start_writes.get(wid, {}))
self.lens_at_start = lens_block(s) self.lens_at_start = lens_block(s)
self.queue = list(SCRIPT[wid][0]) self.queue = list(SCRIPT[wid][0])
@@ -259,9 +244,7 @@ class ScriptedDaemon:
mm = float(value.strip()) * (25.4 if imperial else 1.0) mm = float(value.strip()) * (25.4 if imperial else 1.0)
except ValueError: except ValueError:
mm = -1 # "1/8" or "3 mm": not a plain number mm = -1 # "1/8" or "3 mm": not a plain number
lens = self.lens_at_start if not 0 < mm <= 30:
if not 0 < mm <= 30 or (wid == "laser.floor"
and not lens["reach_min"] <= mm <= lens["reach_max"]):
self.open_next() # the same prompt again, a new sequence number self.open_next() # the same prompt again, a new sequence number
return 200, {"ok": True} return 200, {"ok": True}
self.thickness = mm self.thickness = mm
@@ -370,66 +353,12 @@ class BurnRunTests(unittest.TestCase):
self.assert_restarted_on_found({"lens_hall_edge_z_mm": "4.10", "lens_stop_below_steps": None, self.assert_restarted_on_found({"lens_hall_edge_z_mm": "4.10", "lens_stop_below_steps": None,
"lens_stop_above_steps": None}) "lens_stop_above_steps": None})
def test_the_floor_burn(self):
run = self.run_test("setup.floor")
d = self.daemon
self.assertEqual(d.order, ["laser.floor"])
self.assertEqual(d.answers["laser.floor"], ["Yes", HOME, "Continue", "3.2", "Continue", "8"])
self.assertEqual(d.previews, ["laser.floor"])
self.assertEqual(run.evidence["result"]["floor_density"], 10.0)
self.assertEqual(run.evidence["settings_before"], {"laser_floor_density": "12"})
self.assertEqual(run.evidence["focus_on_record"], 1)
self.assertEqual(run.evidence["program_tail"]["laser.floor"], ["G0 Z4.10", "G0 X0 Y0", "G92.1", "M2"])
self.assert_found()
# the controller loads the floor at every arm: nothing to restart
self.assertEqual(d.lever_posts(), [])
def test_the_floor_needs_focus_on_the_record(self):
# The floor burns at the focus Focus measured, and the wizard
# refuses without it: the test says so before the scrap is asked
# for, and nothing starts.
self.daemon.versions = {}
with self.assertRaises(Failed) as cm:
self.run_test("setup.floor")
self.assertIn("the record carries no Focus", str(cm.exception))
self.assertEqual(self.daemon.order, [])
self.assertEqual(self.daemon.previews, [])
self.assert_found()
def test_the_floors_own_refusal_without_focus_ends_the_run(self):
# Past the test's own reading (a record that lost Focus in
# between): the wizard's refusal ends the run with its word, before
# any prompt.
real = self.daemon.on_get
def on_get(path, q):
if path == "/wiz" and not self.daemon.order:
return 200, {"versions": {"laser.focus": 1}, "recommended": [], "completed": True}
return real(path, q)
self.fake.on_get = on_get
self.daemon.versions = {}
with self.assertRaises(Failed) as cm:
self.run_test("setup.floor")
self.assertIn("run Focus first", str(cm.exception))
self.assertEqual(self.daemon.order, ["laser.floor"])
self.assertEqual(self.daemon.answers, {})
self.assert_found()
def test_a_scrap_the_lens_does_not_reach_fails_before_the_floor(self):
# An edge high enough that the lens cannot focus 3.2 mm above the
# tray: the floor would refuse the thickness, so nothing starts.
self.fake.state["settings"]["lens_hall_edge_z_mm"] = "9.50"
with self.assertRaises(Failed) as cm:
self.run_test("setup.floor")
self.assertIn("leaves out the 3.2 mm scrap", str(cm.exception))
self.assertEqual(self.daemon.order, [])
def test_a_program_that_keeps_its_g92_origin_fails_before_the_burn(self): def test_a_program_that_keeps_its_g92_origin_fails_before_the_burn(self):
# The tail clears the G92 origin, which the controller would # The tail clears the G92 origin, which the controller would
# otherwise carry to its next start under a sender's own offsets. # otherwise carry to its next start under a sender's own offsets.
self.daemon.tail = ["G0 X0 Y0", "M2"] self.daemon.tail = ["G0 X0 Y0", "M2"]
with self.assertRaises(Failed) as cm: with self.assertRaises(Failed) as cm:
self.run_test("setup.floor") self.run_test("setup.focus")
self.assertIn("not the tail", str(cm.exception)) self.assertIn("not the tail", str(cm.exception))
self.assertIn("G92.1", str(cm.exception)) self.assertIn("G92.1", str(cm.exception))
self.assertEqual(self.daemon.order, []) self.assertEqual(self.daemon.order, [])
@@ -439,8 +368,8 @@ class BurnRunTests(unittest.TestCase):
# The number prompts read ui_units: on an imperial machine the # The number prompts read ui_units: on an imperial machine the
# thickness is 0.125 in and the record carries 3.175 mm. # thickness is 0.125 in and the record carries 3.175 mm.
self.fake.state["settings"]["ui_units"] = "imperial" self.fake.state["settings"]["ui_units"] = "imperial"
run = self.run_test("setup.floor") run = self.run_test("setup.focus")
self.assertEqual(self.daemon.answers["laser.floor"][3], "0.125") self.assertEqual(self.daemon.answers["laser.focus"][3], "0.125")
self.assertEqual(run.evidence["thickness_answer"], "0.125") self.assertEqual(run.evidence["thickness_answer"], "0.125")
self.assertEqual(run.evidence["result"]["thickness_mm"], 3.175) self.assertEqual(run.evidence["result"]["thickness_mm"], 3.175)
@@ -449,44 +378,44 @@ class BurnRunTests(unittest.TestCase):
# the test ends the burn rather than answer the same thing forever. # the test ends the burn rather than answer the same thing forever.
self.daemon.units = "imperial" self.daemon.units = "imperial"
with self.assertRaises(Failed) as cm: with self.assertRaises(Failed) as cm:
self.run_test("setup.floor") self.run_test("setup.focus")
self.assertIn("asked thickness again", str(cm.exception)) self.assertIn("asked thickness again", str(cm.exception))
self.assertEqual(self.daemon.answers["laser.floor"], ["Yes", HOME, "Continue", "3.2"]) self.assertEqual(self.daemon.answers["laser.focus"], ["Yes", HOME, "Continue", "3.2"])
self.assertEqual(self.daemon.aborted, ["laser.floor"]) self.assertEqual(self.daemon.aborted, ["laser.focus"])
self.assert_found() self.assert_found()
def test_a_prompt_the_test_does_not_answer_ends_the_burn(self): def test_a_prompt_the_test_does_not_answer_ends_the_burn(self):
# The motors found released: the burn asks for the head pushed home # The motors found released: the burn asks for the head pushed home
# by hand, which the bench never needs. # by hand, which the bench never needs.
SCRIPT["laser.floor"][0].insert(1, ("home-push", "continue", ["Continue"])) SCRIPT["laser.focus"][0].insert(1, ("home-push", "continue", ["Continue"]))
try: try:
with self.assertRaises(Failed) as cm: with self.assertRaises(Failed) as cm:
self.run_test("setup.floor") self.run_test("setup.focus")
finally: finally:
SCRIPT["laser.floor"][0].pop(1) SCRIPT["laser.focus"][0].pop(1)
self.assertIn("asked home-push, which the test does not answer", str(cm.exception)) self.assertIn("asked home-push, which the test does not answer", str(cm.exception))
self.assertEqual(self.daemon.aborted, ["laser.floor"]) self.assertEqual(self.daemon.aborted, ["laser.focus"])
self.assert_found() self.assert_found()
def test_a_dark_burn_fails_and_still_restores(self): def test_a_dark_burn_fails_and_still_restores(self):
self.daemon.patch["laser.floor"] = {"emission": {"hv_max": 0, "laser_on_samples": 0, "thermopile_delta": 0}} self.daemon.patch["laser.focus"] = {"emission": {"hv_max": 0, "laser_on_samples": 0, "thermopile_delta": 0}}
with self.assertRaises(Failed) as cm: with self.assertRaises(Failed) as cm:
self.run_test("setup.floor") self.run_test("setup.focus")
self.assertIn("the tube current never rose", str(cm.exception)) self.assertIn("the tube current never rose", str(cm.exception))
self.assert_found() self.assert_found()
def test_a_burn_with_a_silent_beam_sensor_passes(self): def test_a_burn_with_a_silent_beam_sensor_passes(self):
# A head whose beam sensor gives no output: the tube current and # A head whose beam sensor gives no output: the tube current and
# LASER_ON saw the beam, so the burn counts. # LASER_ON saw the beam, so the burn counts.
self.daemon.patch["laser.floor"] = {"emission": dict(EMISSION, thermopile_delta=0, self.daemon.patch["laser.focus"] = {"emission": dict(EMISSION, thermopile_delta=0,
beam_sensor_seen=False)} beam_sensor_seen=False)}
self.run_test("setup.floor") self.run_test("setup.focus")
self.assert_found() self.assert_found()
def test_a_burn_that_does_not_record_the_beam_sensor_fails(self): def test_a_burn_that_does_not_record_the_beam_sensor_fails(self):
self.daemon.patch["laser.floor"] = {"emission": {"hv_max": 610, "laser_on_samples": 900}} self.daemon.patch["laser.focus"] = {"emission": {"hv_max": 610, "laser_on_samples": 900}}
with self.assertRaises(Failed) as cm: with self.assertRaises(Failed) as cm:
self.run_test("setup.floor") self.run_test("setup.focus")
self.assertIn("the result does not record the beam sensor", str(cm.exception)) self.assertIn("the result does not record the beam sensor", str(cm.exception))
self.assert_found() self.assert_found()
@@ -540,15 +469,14 @@ class BurnRunTests(unittest.TestCase):
# refused before the burn starts: nothing burns. # refused before the burn starts: nothing burns.
self.daemon.stray_z = 12.6 self.daemon.stray_z = 12.6
with self.assertRaises(Failed) as cm: with self.assertRaises(Failed) as cm:
self.run_test("setup.floor") self.run_test("setup.focus")
self.assertIn("outside the lens window", str(cm.exception)) self.assertIn("outside the lens window", str(cm.exception))
self.assertIn("12.6", str(cm.exception)) self.assertIn("12.6", str(cm.exception))
self.assertEqual(self.daemon.order, []) self.assertEqual(self.daemon.order, [])
self.assert_found() self.assert_found()
def test_the_result_keys_the_tests_require(self): def test_the_result_keys_the_tests_require(self):
# every key either test reads is one the scripted burn reports # every key the test reads is one the scripted burn reports
self.assertTrue(set(setup_burns.FLOOR_KEYS) <= set(SCRIPT["laser.floor"][1]) | {"thickness_mm"})
self.assertTrue(set(setup_burns.FOCUS_KEYS) self.assertTrue(set(setup_burns.FOCUS_KEYS)
<= set(SCRIPT["laser.focus"][1]) | {"thickness_mm", "window", "ladder_z"}) <= set(SCRIPT["laser.focus"][1]) | {"thickness_mm", "window", "ladder_z"})
forgetest/tests/test_setup_dark.py
+10 -188
View File
@@ -5,8 +5,8 @@
"""The setup's checks in the catalog: registered as automatic tests, the """The setup's checks in the catalog: registered as automatic tests, the
takeover ones marked so, every covers map naming the check runner, and takeover ones marked so, every covers map naming the check runner, and
the two that put the intake fan count back as found refused unless it is the one that puts the intake fan count back as found refused unless it
set; and the check driver itself against the fake daemon: a start, the is set; and the check driver itself against the fake daemon: a start, the
run polled to its end with no question asked, the result taken, the run polled to its end with no question asked, the result taken, the
record read at version 1, the settings read as before or put back as record read at version 1, the settings read as before or put back as
found, and the controller started again on a lens window put back.""" found, and the controller started again on a lens window put back."""
@@ -19,11 +19,10 @@ import helpers # noqa: E402
from forgetest import catalog # noqa: E402 from forgetest import catalog # noqa: E402
from forgetest.runner import Context, Run # noqa: E402 from forgetest.runner import Context, Run # noqa: E402
IDS = ["setup.check-switches", "setup.check-sensors", "setup.check-airflow", "setup.check-intakes", IDS = ["setup.check-switches", "setup.check-airflow", "setup.check-motion", "setup.check-lens"]
"setup.check-cameras", "setup.check-motion", "setup.check-lens", "setup.check-flow-verify"] # The check that puts the intake fan count back as found: it starts only
# The checks that put the intake fan count back as found: they start only # with it set, so a put-back never leaves the gate closed behind it.
# with it set, so a put-back never leaves the gate closed behind them. COUNTED = ("setup.check-airflow",)
COUNTED = ("setup.check-airflow", "setup.check-intakes")
class RegistrationTests(unittest.TestCase): class RegistrationTests(unittest.TestCase):
@@ -35,11 +34,9 @@ class RegistrationTests(unittest.TestCase):
# bench fixture performs; the rest need no hands at all. # bench fixture performs; the rest need no hands at all.
for tid in IDS: for tid in IDS:
self.assertIn(tid, self.reg) self.assertIn(tid, self.reg)
for tid in ("setup.check-switches", "setup.check-cameras"): self.assertEqual(self.reg["setup.check-switches"].kind, "operator")
self.assertEqual(self.reg[tid].kind, "operator", tid) self.assertTrue(self.reg["setup.check-switches"].fixture_runnable(("button", "lid", "interlock")))
self.assertTrue(self.reg[tid].fixture_runnable(("button", "lid", "interlock")), tid) for tid in ("setup.check-airflow", "setup.check-motion", "setup.check-lens"):
for tid in ("setup.check-sensors", "setup.check-airflow", "setup.check-intakes",
"setup.check-motion", "setup.check-lens", "setup.check-flow-verify"):
self.assertEqual(self.reg[tid].kind, "auto", tid) self.assertEqual(self.reg[tid].kind, "auto", tid)
def test_no_check_asks_the_operator_to_prepare_anything(self): def test_no_check_asks_the_operator_to_prepare_anything(self):
@@ -65,8 +62,7 @@ class RegistrationTests(unittest.TestCase):
fake.stop() fake.stop()
def test_the_checks_that_take_the_machine_say_so(self): def test_the_checks_that_take_the_machine_say_so(self):
for tid in ("setup.check-airflow", "setup.check-intakes", "setup.check-motion", "setup.check-lens", for tid in ("setup.check-airflow", "setup.check-motion", "setup.check-lens"):
"setup.check-flow-verify"):
self.assertEqual(self.reg[tid].hardware, "takeover", tid) self.assertEqual(self.reg[tid].hardware, "takeover", tid)
def test_the_lens_check_covers_the_finder_and_the_frame(self): def test_the_lens_check_covers_the_finder_and_the_frame(self):
@@ -74,9 +70,6 @@ class RegistrationTests(unittest.TestCase):
for want in (("forgectrl", "src/lensfind.*"), ("forgectrl", "src/lens.*")): for want in (("forgectrl", "src/lensfind.*"), ("forgectrl", "src/lens.*")):
self.assertIn(want, covers) self.assertIn(want, covers)
self.assertIn("setup.check-motion", self.reg["setup.check-lens"].requires) self.assertIn("setup.check-motion", self.reg["setup.check-lens"].requires)
covers = set(self.reg["setup.check-intakes"].covers)
for want in (("forgectrl", "src/cool.c"), ("forgectrl", "src/airflow.*")):
self.assertIn(want, covers)
def test_every_check_covers_the_runner_and_the_page(self): def test_every_check_covers_the_runner_and_the_page(self):
for tid in IDS: for tid in IDS:
@@ -85,91 +78,6 @@ class RegistrationTests(unittest.TestCase):
self.assertIn(("forgectrl", "src/ui/wizard.*"), 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 = []
self.polls = 0
def on_get(path, q):
if path == "/wiz/dark":
# One poll sees the sampling; the next sees the result.
if self.dark["running"]:
self.polls += 1
if self.polls >= 2:
self.dark.update({"running": False, "prompt": None, "progress": 100,
"result": dict(self.RESULT)})
self.versions["sensors"] = 1
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": None})
return 200, {"started": True, "id": "sensors"}
if path == "/wiz/sensors/answer":
self.answers.append(dict(form))
return 409, {"error": "no such prompt is open"}
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_no_question_result_and_settings_untouched(self):
t = catalog.load_suite()["setup.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, [])
self.assertEqual(run.evidence["result"]["laser_pgood"], 1)
self.assertEqual(run.evidence["settings_before"].get("cool_temp_offset_c"), "0.5")
self.assertEqual(self.fake.state["settings"]["cool_temp_offset_c"], "0.5")
self.assertNotIn("/settings", [p for p, _ in self.fake.posts])
def test_a_question_fails_the_check(self):
t = catalog.load_suite()["setup.check-sensors"]
run = Run("test", t.id, t.title)
real_get = self.fake.on_get
def on_get(path, q):
if path == "/wiz/dark" and self.dark["running"]:
d = dict(self.dark)
d["prompt"] = {"seq": 3, "id": "room-temp", "kind": "number", "text": "?",
"options": ["Set", "Skip"]}
return 200, d
return real_get(path, q)
self.fake.on_get = on_get
with self.assertRaises(Exception) as cm:
t.fn(Context(run, None, t))
self.assertIn("asked a question", str(cm.exception))
self.assertEqual(self.answers, [])
self.assertIn(("/wiz/sensors/abort", {}), [(p, f) for p, f in self.fake.posts])
class AirflowCheckTests(unittest.TestCase): class AirflowCheckTests(unittest.TestCase):
"""setup.check-airflow's judge against a scripted daemon: the check """setup.check-airflow's judge against a scripted daemon: the check
writes its floors, and the intake fan count is the machine's. Intake writes its floors, and the intake fan count is the machine's. Intake
@@ -522,91 +430,5 @@ class LensCheckTests(ScriptedCheck):
self.assertEqual(run.evidence["restart"], {"mode": "cloud", "pid": 100}) self.assertEqual(run.evidence["restart"], {"mode": "cloud", "pid": 100})
class IntakesCheckTests(ScriptedCheck):
"""setup.check-intakes's judge: the tachs count the intake fans the
machine held, the step writes the count, and it goes back as found.
The tachs count only two turning fans: intake 2 giving no reading
reads the same as a machine built with one, so the step asks the
operator, and so it does when intake 1 does not come up. Both intakes
turn on the bench reference, so either question fails the test
unanswered."""
WID = "airflow.intakes"
# The step's two questions, as the daemon asks them.
SILENT_2 = ("Intake 1 turns; intake 2 gives no reading. Look at the back of the machine: how many "
"intake fans does it have? If it has two, the second one needs attention before the "
"machine cuts.")
SLOW_1 = ("The intake fans could not be counted: intake 1 did not come up to speed. Look at the back "
"of the machine: how many intake fans does it have? A fan that does not turn needs "
"attention before the machine cuts.")
def setUp(self):
super().setUp()
self.fake.state["settings"]["cool_intake_fans"] = "2"
def finish(self, s):
# both intakes turned: two, by the tachs
s["cool_intake_fans"] = "2"
return {"intake_fans": 2, "intake_1_rpm": 4150.0, "intake_2_rpm": 4100.0, "intake_2_silent": False,
"counted_by": "tachs", "summary": "x"}
def ask(self, text):
self.prompt = {"id": "intake-count", "kind": "choice", "text": text, "options": ["One", "Two"]}
def test_two_intake_fans_counted_and_put_back(self):
run = self.run_check("setup.check-intakes")
self.assertEqual(run.evidence["count"]["counted"], 2)
self.assertEqual(run.evidence["count"]["by"], "tachs")
self.assertEqual(self.fake.state["settings"]["cool_intake_fans"], "2")
self.assertIn(("/settings", {"cool_intake_fans": "2"}), self.fake.posts) # the put-back
def test_a_count_other_than_the_machine_held_fails_and_puts_it_back(self):
# a machine that held one intake fan, whose tachs saw two turn
self.fake.state["settings"]["cool_intake_fans"] = "1"
with self.assertRaises(Exception) as cm:
self.run_check("setup.check-intakes")
self.assertIn("the tachs count 2 intake fans, the machine held 1", str(cm.exception))
self.assertEqual(self.fake.state["settings"]["cool_intake_fans"], "1")
def test_a_silent_intake_2_is_a_fault_the_test_does_not_answer(self):
# intake 1 turns, intake 2 gives no reading: the step asks the
# operator, and on the bench reference that is a fan needing attention
self.ask(self.SILENT_2)
with self.assertRaises(Exception) as cm:
self.run_check("setup.check-intakes")
self.assertIn("asked for the count, which the tachs could not make", str(cm.exception))
self.assertIn("intake 2 gives no reading", str(cm.exception))
self.assertEqual(self.answers, [])
self.assertIn(("/wiz/airflow.intakes/abort", {}), self.fake.posts)
self.assertEqual(self.fake.state["settings"]["cool_intake_fans"], "2")
def test_a_silent_intake_2_on_a_machine_that_held_one_still_fails(self):
# the test never answers the operator's question, whatever was held
self.fake.state["settings"]["cool_intake_fans"] = "1"
self.ask(self.SILENT_2)
with self.assertRaises(Exception) as cm:
self.run_check("setup.check-intakes")
self.assertIn("asked for the count", str(cm.exception))
self.assertEqual(self.answers, [])
self.assertEqual(self.fake.state["settings"]["cool_intake_fans"], "1")
def test_intake_1_not_up_to_speed_fails_the_check(self):
self.ask(self.SLOW_1)
with self.assertRaises(Exception) as cm:
self.run_check("setup.check-intakes")
self.assertIn("asked for the count", str(cm.exception))
self.assertIn("intake 1 did not come up to speed", str(cm.exception))
self.assertEqual(self.answers, [])
self.assertEqual(self.fake.state["settings"]["cool_intake_fans"], "2")
def test_the_tachs_count_is_judged_by_the_count_alone(self):
# counted by the tachs, the count is the whole judgment: the speeds
# are recorded, not judged again
real = self.finish
self.finish = lambda s: dict(real(s), intake_1_rpm=None)
run = self.run_check("setup.check-intakes")
self.assertIsNone(run.evidence["count"]["intake_1_rpm"])
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
forgetest/tests/test_takeover_position.py
+1 -1
View File
@@ -137,7 +137,7 @@ class TakeoverPositionTests(unittest.TestCase):
self.assertNotIn("restart_positions", cap) self.assertNotIn("restart_positions", cap)
def test_the_post_pass_fails_the_run_and_moves_nothing(self): def test_the_post_pass_fails_the_run_and_moves_nothing(self):
# setup.check-envelope's case: a declared camera home, the head # a check's case: a declared camera home, the head
# left out, the record put back under a takeover, whose restart # left out, the record put back under a takeover, whose restart
# zeroes the counters where the head stands # zeroes the counters where the head stands
cap = self.cap() cap = self.cap()