diff --git a/forgetest/forgetest/suite/__init__.py b/forgetest/forgetest/suite/__init__.py index 9582dd4..d16b23b 100644 --- a/forgetest/forgetest/suite/__init__.py +++ b/forgetest/forgetest/suite/__init__.py @@ -29,3 +29,5 @@ from . import evmore # noqa: F401,E402 from . import extcat # noqa: F401,E402 from . import extmcode # noqa: F401,E402 from . import extwizard # noqa: F401,E402 +from . import updlock # noqa: F401,E402 +from . import homeoff # noqa: F401,E402 diff --git a/forgetest/forgetest/suite/homeoff.py b/forgetest/forgetest/suite/homeoff.py new file mode 100644 index 0000000..8597238 --- /dev/null +++ b/forgetest/forgetest/suite/homeoff.py @@ -0,0 +1,75 @@ +# Copyright 2026 514 LLC d/b/a OpenGlow +# Written by Scott Wiederhold +# https://community.openglow.org +# SPDX-License-Identifier: MIT + +"""A camera home's offset pair, on the machine. + +Its own module, so that no other test's fingerprint moves. The homing is +the web-service session cloud.mode-switch runs; the head ends parked at +the home, as it does there. +""" + +from ..catalog import test +from .cloud import _HOMING_PATH, gfhome_homing +from .motion import wait_state + +HOME_X, HOME_Y = 4.5, -3.25 + + +@test("homing.cloud-offsets", title="A camera home declares its offset pair, and the envelope reaches back to a " + "negative one", + subsystem="homing", kind="auto", mode="grbl", est_min=5, + covers=_HOMING_PATH + [("grblhal-glowforge", "src/glowforge_homing.*")], + requires=["forgectrl.auth", "motion.pacing"], + steps=["Bed clear, lid closed; cloud credentials configured; the machine on the network. The head ends " + "parked at the home, as it does after cloud.mode-switch."], + description="With cloud mode on, homing_mode = gfcloud and the camera-home offsets set to (4.5, -3.25), " + "$H runs the web-service homing session. When it ends, the controller declares the position " + "(4.5, -3.25): a camera-home offset may be negative, and the declared position is snapped to " + "the step grid. The work envelope reaches back to the home on the negative axis, since the head " + "stands there: a jog 2 mm out in +Y and 2 mm back ends at the home, accepted, and the same in " + "X. The homing mode and the offsets are put back as found. That an offset past the axis travel " + "is refused, and the envelope's edge on the grid, are the driver's unit and harness cases.") +def cloud_offsets(ctx): + fc = ctx.forgectrl + ev = ctx.evidence + s0 = fc.settings() or {} + ctx.check(s0.get("cloud_enabled") == "1", "cloud mode is off: a camera home needs it on") + found = {k: s0.get(k) or "" for k in ("homing_mode", "gfcloud_home_x", "gfcloud_home_y")} + ev["found"] = found + try: + for k, v in (("homing_mode", "gfcloud"), ("gfcloud_home_x", str(HOME_X)), ("gfcloud_home_y", str(HOME_Y))): + st, body = fc.post("/settings", data={k: v}) + ctx.check(st == 200, "%s=%s -> %s %s", k, v, st, body) + with ctx.grbl() as g: + gfhome_homing(ctx, ev, g) + mpos = g.status_report().get("MPos") or (None, None, None) + ev["declared"] = mpos + ctx.log("the camera home declared %s", mpos) + ctx.check(mpos[0] is not None and abs(mpos[0] - HOME_X) < 0.02 and abs(mpos[1] - HOME_Y) < 0.02, + "the camera home did not declare (%.2f, %.2f): %s", HOME_X, HOME_Y, mpos) + jogs = {} + for axis, home in (("Y", HOME_Y), ("X", HOME_X)): + out = g.command("$J=G91 G21 %s2 F1000" % axis, timeout=10) + wait_state(ctx, g, "Idle", 20) + back = g.command("$J=G91 G21 %s-2 F1000" % axis, timeout=10) + wait_state(ctx, g, "Idle", 20) + at = g.status_report().get("MPos") + jogs[axis] = {"out": out[-1:], "back": back[-1:], "at": at} + ctx.check(out[-1:] == ["ok"] and back[-1:] == ["ok"], + "a jog out and back to the home in %s was refused: %s", axis, jogs[axis]) + i = "XY".index(axis) + ctx.check(at and abs(at[i] - home) < 0.02, "the jog back in %s did not end at the home: %s", axis, at) + ev["jogs"] = jogs + ctx.log("out and back to the home: %s", jogs) + finally: + # The settings are refused for a moment after the suite's Grbl client closes; each is put back when taken. + for k, v in found.items(): + def put(): + st_, body_ = (fc.post("/settings", params={k: ""}) if v == "" else fc.post("/settings", data={k: v})) + if st_ != 200: + ctx.log("restore %s=%r -> %s %s", k, v, st_, body_) + return st_ == 200 or None + took = ctx.wait_for(put, 20, poll=0.5) + ctx.log("restore %s=%r -> %s", k, v, "taken" if took is not None else "not taken") diff --git a/forgetest/forgetest/suite/updlock.py b/forgetest/forgetest/suite/updlock.py new file mode 100644 index 0000000..d75586a --- /dev/null +++ b/forgetest/forgetest/suite/updlock.py @@ -0,0 +1,84 @@ +# Copyright 2026 514 LLC d/b/a OpenGlow +# Written by Scott Wiederhold +# https://community.openglow.org +# SPDX-License-Identifier: MIT + +"""An update job holds the machine, on the machine. + +Its own module, so that no other test's fingerprint moves. The job is the +download of the published release: it fetches the firmware archive and +checks its signature, and applies nothing. The archive it leaves is +removed when it was not there before. +""" + +import os + +from ..catalog import test +from .motion import _job_post, _words + +DL_FW = "/data/forgefirm/download.fw" + + +@test("update.job-locks", title="An update job holds the machine: the settings are locked while it runs", + subsystem="update", kind="auto", est_min=3, + covers=[("forgectrl", "src/update.*"), ("forgectrl", "src/lease.*")], + requires=["update.release-check"], + description="With a release published on the releases API, POST /update/download starts the download " + "job (202), which fetches the release's firmware archive and checks its signature and applies " + "nothing. While it runs, /status names the job (update:download, of kind system) as the " + "machine lease's holder, and a settings write and a posted job are each refused in its name " + "(409, the settings locked). When it ends, the lease is free and the same settings write is " + "taken. The downloaded archive is removed when it was not there before. That a log export " + "does not lock the settings while every other holder does is lease_test's.") +def job_locks(ctx): + fc = ctx.forgectrl + ev = ctx.evidence + ctx.check(fc.wait_idle(timeout=30, abort=ctx.aborted), "machine not idle") + st, rel = fc.get("/update/release") + ev["release"] = {k: (rel or {}).get(k) for k in ("available", "version", "bytes")} if isinstance(rel, dict) else rel + ctx.check(st == 200 and isinstance(rel, dict) and rel.get("available"), "no release is published to download: %s", rel) + had = os.path.exists(DL_FW) + units = fc.settings().get("ui_units") or "metric" + + def holder(): + return ((fc.status().get("lease") or {}).get("holder") or {}) + + def job(): + st_, j = fc.get("/update/status") + return j if isinstance(j, dict) else {} + + try: + st, body = fc.post("/update/download") + ctx.check(st == 202, "POST /update/download -> %s %s", st, _words(body)[:200]) + ctx.wait_for(lambda: holder().get("owner") == "update:download" or None, 20, poll=0.2) + held = holder() + ev["holder"] = held + ctx.log("the lease while the download runs: %s", held) + ctx.check(held.get("owner") == "update:download" and held.get("kind") == "system", + "the download job does not hold the machine: %s", held) + refused = {} + st, body = fc.post("/settings", params={"ui_units": units}) + refused["a settings write"] = [st, _words(body)[:200]] + st2, body2 = _job_post(fc, "G21\n", name="beside-an-update") + refused["a posted job"] = [st2, _words(body2)[:200]] + ev["refused"] = refused + ctx.log("beside the download: %s", refused) + ctx.check(job().get("running") is True, "the download ended before the refusals were read: %s", job()) + ctx.check(st == 409 and "settings are locked" in _words(body), "a settings write beside the download -> %s %s", + st, _words(body)[:200]) + ctx.check(st2 == 409 and "holds the machine" in _words(body2), "a job beside the download -> %s %s", + st2, _words(body2)[:200]) + ctx.wait_for(lambda: job().get("running") is False or None, 600, poll=2.0) + done = job() + ev["job"] = done + ctx.log("the download ended: %s", done) + ctx.check(done and (done.get("result") or {}).get("ok") is True, "the download did not end well: %s", done) + ctx.check(not holder(), "the lease was not given back: %s", holder()) + st, body = fc.post("/settings", params={"ui_units": units}) + ctx.check(st == 200, "the same settings write after the job -> %s %s", st, _words(body)[:200]) + finally: + # The job runs on after a check that failed: its archive lands when it ends. + ctx.wait_for(lambda: job().get("running") is False or None, 600, poll=2.0) + if not had and os.path.exists(DL_FW): + os.remove(DL_FW) + ctx.log("removed the downloaded archive")