forgetest: update.job-locks and homing.cloud-offsets

update.job-locks (suite/updlock.py, its own module): with a release
published on the releases API, POST /update/download starts the
download job (202), which fetches and checks the release's archive and
applies nothing. While it runs, /status names update:download, of kind
system, as the lease's holder, and a settings write and a posted job are
each refused in its name (409). 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.

homing.cloud-offsets (suite/homeoff.py, its own module): with cloud mode
on, homing_mode = gfcloud, and the camera-home offsets at (4.5, -3.25),
$H runs the web-service homing session, and when it ends the controller
declares (4.5, -3.25) on the step grid. The work envelope reaches back
to the home on the negative axis: a jog 2 mm out and 2 mm back in Y and
then in X is accepted and ends at the home. The homing mode and the
offsets are put back as found, each retried until the daemon takes it,
since a settings write is refused for a moment after the suite's Grbl
client closes.

Proof: on the bench reference, update.job-locks PASS after
update.release-check, and homing.cloud-offsets PASS in a campaign with
forgectrl.auth, kernel.latch-locked-idle, and motion.pacing, which also
PASS. forgetest's unit tests pass (452), and the coverage lint passes
with --enforce.

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

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")

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")