diff --git a/forgetest/forgetest.init b/forgetest/forgetest.init index a582981..a02489e 100644 --- a/forgetest/forgetest.init +++ b/forgetest/forgetest.init @@ -26,13 +26,6 @@ case "$1" in start) echo "Starting forgetest" mkdir -p "$DATA" - # The bench seed: the setup override lifts the wizard part of - # the controller gate until the next reboot (/run is tmpfs), so a - # campaign runs on a machine whose required wizards are not complete. - # Nothing lifts the agreements or the account: the bench accepts them - # once, and that stays in /data. - mkdir -p /run/forgefirm - : > /run/forgefirm/setup-override start-stop-daemon -S -q -p $PIDFILE -m -b -x /bin/sh -- -c \ "exec /usr/bin/python3 -m forgetest >> $LOG 2>&1" ;; diff --git a/forgetest/forgetest/suite/__init__.py b/forgetest/forgetest/suite/__init__.py index 3ab4eda..01441e3 100644 --- a/forgetest/forgetest/suite/__init__.py +++ b/forgetest/forgetest/suite/__init__.py @@ -11,7 +11,7 @@ from . import kernel # noqa: F401,E402 from . import forgectrl # noqa: F401,E402 from . import setup # noqa: F401,E402 from . import setup_dark # noqa: F401,E402 -from . import setup_sheet # noqa: F401,E402 +from . import setup_burns # noqa: F401,E402 from . import logs # noqa: F401,E402 from . import motion # noqa: F401,E402 from . import cooling # noqa: F401,E402 diff --git a/forgetest/forgetest/suite/setup.py b/forgetest/forgetest/suite/setup.py index 64ffed9..b5f20a3 100644 --- a/forgetest/forgetest/suite/setup.py +++ b/forgetest/forgetest/suite/setup.py @@ -41,12 +41,25 @@ from .forgectrl import lan_ip RECORD = "setup.json" USERS = "users" -OVERRIDE = "setup-override" SSH_FLAG = "ssh-enabled" TLS_BASE = "https://127.0.0.1" # FORGECTRL_TLS_URL overrides (host tests) -# The image's required wizards, all at version 1 (setup.c). -REQUIRED_WIZARDS = ("advisories", "account", "preferences", "machine", "cloud") +# The first run's form wizards, all at version 1 (wiz.c). +FORM_WIZARDS = ("advisories", "account", "preferences", "machine", "cloud") +# The steps the gate requires after the account, all at version 1, in +# the order the gate's reason names them (setup.c). +REQUIRED_STEPS = ("preferences", "machine", "cloud") +# What the image's step table recommends on a record with the required +# steps done and nothing else (setup.c): every step after them. The +# cloud header joins on a machine with cloud mode on. The gate's own +# parts (the advisories, the account, the required steps, the intake fan +# count) are never recommended as not run yet, and the laser floor and +# the bed size are offered, never recommended. +RECOMMENDED_STEPS = ("switches", "sensors", "airflow", "cooling.aa-offset", "cooling.flow", "motion", + "cameras", "lens.travel", "laser.focus") +NEVER_RECOMMENDED = ("advisories", "account", "airflow.intakes", "laser.floor", "motion.envelope") +# The setting that counts the intake fans: the gate holds without it. +INTAKE_KEY = "cool_intake_fans" ADVISORY_DOCS = ("safety-and-risk", "licenses", "privacy", "cloud-service") SAFETY_PHRASE = "I UNDERSTAND" LOGIN_FAILS = 5 @@ -72,10 +85,6 @@ def users_path(): return posixpath.join(data_dir(), USERS) -def override_path(): - return posixpath.join(run_dir(), OVERRIDE) - - def ssh_flag_path(): return posixpath.join(run_dir(), SSH_FLAG) @@ -110,33 +119,6 @@ def write_file(path, data): os.replace(tmp, path) -def override_present(): - return os.path.exists(override_path()) - - -def set_override(on): - write_file(override_path(), b"" if on else None) - - -def tmpfs_mount(path): - """The (mount point, fstype) that holds path, from /proc/mounts; - (None, None) when unreadable.""" - best = (None, None) - try: - with open("/proc/mounts", "r", encoding="utf-8", errors="replace") as f: - for line in f: - parts = line.split() - if len(parts) < 3: - continue - mp = parts[1] - if path == mp or path.startswith(mp.rstrip("/") + "/"): - if best[0] is None or len(mp) > len(best[0]): - best = (mp, parts[2]) - except OSError: - pass - return best - - # --------------------------------------------------------------- records class Restore: @@ -172,14 +154,44 @@ def now_iso(): def record_without_wizards(record): """A copy of a record with no completed wizard and no flag: the - consent and the account stay, so the gate closes on the wizard part - alone, which is the part the override lifts.""" + consent and the account stay, so the gate closes on the required + steps.""" out = json.loads(json.dumps(record)) out["wizards"] = {} out["flags"] = {} return out +def record_with_required_steps(record): + """A copy of a record with the required steps done (kept from the + record, or written at version 1) and nothing else, and no flag: on a + machine whose intake fans are counted the gate is open and every + other step is recommended as not run yet.""" + out = record_without_wizards(record) + have = record.get("wizards") or {} + for wid in REQUIRED_STEPS: + out["wizards"][wid] = have.get(wid) or {"version": 1, "completed": now_iso(), "result": {}, + "applied": {}} + return out + + +def record_without_consent(record): + """A copy of a record with no advisory accepted and no press: the + gate closes on the consent, whatever else the record holds.""" + out = json.loads(json.dumps(record)) + out["advisories"] = {} + out.pop("acceptance", None) + return out + + +def record_without_account(record): + """A copy of a record with the consent and no account: the gate + closes on the account.""" + out = json.loads(json.dumps(record)) + out.pop("account", None) + return out + + def complete_record(status, base=None, account_name="bench"): """A complete setup record for a machine that must count as set up: every advisory at the hash GET /wiz reports, the @@ -203,7 +215,7 @@ def complete_record(status, base=None, account_name="bench"): rec["account"] = acct or {"name": account_name, "uid": 1000, "created": ts} if isinstance(base.get("machine"), dict): rec["machine"] = base["machine"] - wizards = status.get("wizards") or [{"id": w, "version": 1} for w in REQUIRED_WIZARDS] + wizards = status.get("wizards") or [{"id": w, "version": 1} for w in FORM_WIZARDS] for w in wizards: rec["wizards"][w["id"]] = {"version": int(w.get("version") or 1), "completed": ts, "result": {}, "applied": {}} @@ -330,9 +342,14 @@ def press_status(fc): def wiz_summary(w): - return {k: w.get(k) for k in ("first_run", "gate", "override", "why", "required", + return {k: w.get(k) for k in ("first_run", "gate", "why", "recommended", "advisories_complete", "acceptance_done", "account", - "completed", "versions")} + "fans_counted", "completed", "versions")} + + +def recommended(w): + """The steps a GET /wiz body recommends: {id: reason}.""" + return {x.get("id"): x.get("reason") or "" for x in w.get("recommended") or [] if isinstance(x, dict)} def wait_settled(ctx): @@ -354,145 +371,140 @@ def gate_is_open(): return None - # -------------------------------------------------------------- the gate _GATE_COVERS = [("forgectrl", "src/setup.*"), ("forgectrl", "src/super.c"), ("forgectrl", "src/main.c"), ("forgectrl", "src/paths.h")] -@test("setup.gate-blocks-controllers", title="The setup gate spawns no controller", - subsystem="setup", kind="auto", hardware="takeover", mode="grbl", est_min=5, +@test("setup.gate-blocks-controllers", + title="The setup gate holds the controller until the advisories, the account, the required " + "steps, and the intake fans", + subsystem="setup", kind="auto", hardware="takeover", mode="grbl", est_min=8, covers=_GATE_COVERS, requires=["forgectrl.auth"], precheck=gate_is_open, - description="With the override removed and a record that lacks every required wizard " - "(installed under a forgectrl restart, the consent and the account kept), " - "GET /mode reports the controller gated with the reason naming the wizards, " - "GET /wiz lists them as required, no controller process runs and the Grbl port " - "is closed. The override file then opens the gate live: the supervisor spawns " - "the controller and it comes back running. The real record and the override " - "state are restored under another restart and the machine reads as before.") + description="Four records in turn, each installed under a forgectrl restart. Without the " + "consent (no advisory accepted, no press), GET /mode reports the controller " + "gated with the reason naming the advisories, GET /wiz agrees, no controller " + "process runs and the Grbl port is closed; with the consent and without the " + "account, the same with the reason naming the account; with the consent and " + "the account and no step done, the same with the reason 'setup required: " + "preferences, machine, cloud'. With those three done as well and no flag, the " + "gate is open with no reason: the supervisor spawns the controller, the Grbl " + "port accepts, and GET /wiz recommends every other step as not run yet (never " + "the advisories, the account, the three required steps, or the intake fan " + "count; not the optional laser floor or bed size). On that record the intake " + "fan count is cleared: the gate closes live with the reason naming the intake " + "fans, GET /wiz reports them not counted, and the supervisor stops the " + "controller; the count written back as found opens the gate again and the " + "controller comes back. The real record is restored under another restart and " + "the machine reads as before.") def gate_blocks_controllers(ctx): fc = ctx.forgectrl ev = ctx.evidence before = wiz(fc) ev["before"] = wiz_summary(before) - ctx.log("before: gate %s, override %s, required %s", before.get("gate"), before.get("override"), - before.get("required")) + ctx.log("before: gate %s, %d steps recommended", before.get("gate"), len(recommended(before))) raw = read_file(record_path()) ctx.check(raw, "no setup record at %s", record_path()) try: record = json.loads(raw.decode("utf-8")) except ValueError as e: ctx.fail("the setup record is not JSON: %s", e) - lacking = json.dumps(record_without_wizards(record), indent=1, sort_keys=True).encode() + b"\n" - had_override = override_present() - ev["override_before"] = had_override - with installed(ctx, {record_path(): lacking, override_path(): None}): - # the daemon is up again with the record that lacks the wizards + def encoded(rec): + return json.dumps(rec, indent=1, sort_keys=True).encode() + b"\n" + + def gated(part, words): + """The gate closed on `part`: the supervisor and GET /wiz say so + in `words`, and no controller runs.""" took = ctx.wait_for(lambda: mode(fc).get("controller") == "gated", 60) m = mode(fc) - ev["gated_mode"] = m - ctx.log("/mode after the restart: %s (after %s s)", m, took) + ev["gated_mode " + part] = m + ctx.log("/mode without the %s: %s (after %s s)", part, m, took) ctx.check(m.get("controller") == "gated" and m.get("gated") is True, - "the supervisor did not report the gate: %s", m) - ctx.check("setup required" in (m.get("why") or ""), - "the reason does not name the missing wizards: %r", m.get("why")) + "without the %s the supervisor did not report the gate: %s", part, m) + ctx.check(words in (m.get("why") or ""), "the reason does not name the %s: %r", part, m.get("why")) w = wiz(fc) - ev["gated_wiz"] = wiz_summary(w) - ctx.check(w.get("gate") == "closed" and w.get("override") is False, + ev["gated_wiz " + part] = wiz_summary(w) + ctx.check(w.get("gate") == "closed" and words in (w.get("why") or ""), "GET /wiz does not agree with the closed gate: %s", wiz_summary(w)) - ctx.check(set(w.get("required") or []) >= set(REQUIRED_WIZARDS), - "GET /wiz does not list every required wizard: %s", w.get("required")) ctx.check(not hw.grbl_port_open(timeout=2), "the Grbl port answers while the gate is closed") pids = hw.pidof("grblHAL_glowfor") + hw.pgrep_f("gfcloud.py") - ev["controller_pids_gated"] = pids + ev["controller_pids " + part] = pids ctx.check(not pids, "a controller process runs while the gate is closed: %s", pids) - ctx.log("gated: no controller process, the Grbl port is closed") + ctx.log("gated on the %s: no controller process, the Grbl port is closed", part) - # the override lifts the wizard part live: no restart - set_override(True) - ctx.wait_for(lambda: wiz(fc).get("gate") == "open", 15) - w = wiz(fc) - ev["override_wiz"] = wiz_summary(w) - ctx.check(w.get("gate") == "open" and w.get("override") is True, - "the override did not open the gate: %s", wiz_summary(w)) - ctx.check((w.get("why") or "").startswith("override active"), - "the reason does not say the override stands: %r", w.get("why")) + def running(what): + """The gate open: the supervisor spawns the controller and its + port accepts.""" # The supervisor spawns on its next pass; a gated machine counts # as settled, so wait for the spawn before the settle wait. took = ctx.wait_for(lambda: mode(fc).get("controller") == "running", 60) - ctx.log("the controller spawned under the override (%s s)", took) m = wait_settled(ctx) - ev["override_mode"] = m + ev["open_mode " + what] = m ctx.check(took is not None and isinstance(m, dict) and m.get("controller") == "running", - "the controller did not come back under the override: %s", m) + "the controller did not come back %s: %s", what, m) # /mode reads running from the fork; the controller binds its port # a few tens of milliseconds later opened = ctx.wait_for(lambda: hw.grbl_port_open(timeout=1), 10, poll=0.05) - ev["port_open_after_s"] = opened + ev["port_open_after_s " + what] = opened ctx.check(opened is not None, "the Grbl port is closed with the controller running") - ctx.log("override: the controller is back (%s), the Grbl port accepts after %.3f s", m, opened) + ctx.log("%s: the controller is running, the Grbl port accepts after %.3f s", what, opened) - # the real record is back under a restart; the override reads as found + with installed(ctx, {record_path(): encoded(record_without_consent(record))}): + gated("consent", "the advisories are not accepted") + with ctx.takeover(): + write_file(record_path(), encoded(record_without_account(record))) + gated("account", "no account exists") + with ctx.takeover(): + write_file(record_path(), encoded(record_without_wizards(record))) + gated("required steps", "setup required: " + ", ".join(REQUIRED_STEPS)) + with ctx.takeover(): + write_file(record_path(), encoded(record_with_required_steps(record))) + + # the consent, the account, the required steps, the intake fans + # counted: nothing else closes the gate + w = wiz(fc) + ev["open_wiz"] = wiz_summary(w) + ctx.check(w.get("gate") == "open" and not w.get("why") and w.get("fans_counted") is True, + "the consent, the account, and the required steps do not open the gate: %s", wiz_summary(w)) + running("with the required steps done") + recm = recommended(w) + missing = [s for s in RECOMMENDED_STEPS if s not in recm] + ctx.check(not missing, "GET /wiz does not recommend %s: %s", missing, recm) + stray = [s for s in recm if s in NEVER_RECOMMENDED + REQUIRED_STEPS] + ctx.check(not stray, "GET /wiz recommends %s", stray) + other = {k: v for k, v in recm.items() if v != "not run yet"} + ctx.check(not other, "with nothing else done, steps are recommended for another reason: %s", other) + ctx.log("recommended, each not run yet: %s", sorted(recm)) + + # the intake fan count holds the gate as the account does + with Restore(ctx, [INTAKE_KEY]) as count: + ctx.check(count.before.get(INTAKE_KEY), "%s is unset with the gate open", INTAKE_KEY) + st, body = fc.post("/settings", params={INTAKE_KEY: ""}) + ctx.check(st == 200, "clearing %s -> %s %s", INTAKE_KEY, st, body) + ctx.wait_for(lambda: wiz(fc).get("fans_counted") is False, 15) + ctx.check(wiz(fc).get("fans_counted") is False, + "GET /wiz reports the intake fans counted with %s unset", INTAKE_KEY) + gated("intake fan count", "the intake fans are not counted") + running("with the intake fans counted again") + ctx.check((fc.settings() or {}).get(INTAKE_KEY) == count.before.get(INTAKE_KEY), + "%s reads %r, was %r", INTAKE_KEY, (fc.settings() or {}).get(INTAKE_KEY), + count.before.get(INTAKE_KEY)) + + # the real record is back under a restart after = wiz(fc) ev["after"] = wiz_summary(after) ctx.check(after.get("gate") == before.get("gate"), "the gate reads %s after the restore, was %s", after.get("gate"), before.get("gate")) - ctx.check(sorted(after.get("required") or []) == sorted(before.get("required") or []), - "the required list changed: %s -> %s", before.get("required"), after.get("required")) - ctx.check(override_present() == had_override, "the override file does not read as found") + ctx.check(recommended(after) == recommended(before), + "the recommended list changed: %s -> %s", recommended(before), recommended(after)) m = mode(fc) ev["after_mode"] = m ctx.check(m.get("controller") == "running", "the controller is %s after the restore", m.get("controller")) ctx.log("restored: gate %s, controller %s", after.get("gate"), m.get("controller")) -@test("setup.override-until-reboot", title="The override stands until the next reboot", - subsystem="setup", kind="auto", est_min=1, - covers=_GATE_COVERS, requires=["forgectrl.auth"], - description="The bench seed: the dev image's forgetest init script writes " - "/run/forgefirm/setup-override at boot. With it present, GET /wiz " - "reports the override and an open gate, the reason names the wizards it lifts " - "when any is required, and GET /mode reports the controller running and not " - "gated. The file lives on tmpfs (/proc/mounts), so a reboot removes it by " - "construction and the gate is then evaluated from the record alone. The " - "override never lifts the advisories or the account.") -def override_until_reboot(ctx): - fc = ctx.forgectrl - ev = ctx.evidence - path = override_path() - ctx.check(os.path.exists(path), - "no override at %s: the dev image's forgetest init script creates it at boot", path) - mp, fstype = tmpfs_mount(os.path.dirname(path)) - ev["mount"] = {"point": mp, "fstype": fstype} - ctx.log("%s is on %s (%s)", path, mp, fstype) - ctx.check(fstype == "tmpfs", "%s is not on tmpfs (%s on %s): a reboot would keep the override", - path, fstype, mp) - - w = wiz(fc) - ev["wiz"] = wiz_summary(w) - ctx.log("GET /wiz: %s", ev["wiz"]) - ctx.check(w.get("override") is True, "GET /wiz does not see the override") - ctx.check(w.get("gate") == "open", "the gate is closed under the override: %s", w.get("why")) - ctx.check(w.get("advisories_complete") is True and w.get("acceptance_done") is True - and w.get("account") is True, - "the consent or the account is missing (%s): nothing lifts those", wiz_summary(w)) - required = [r if isinstance(r, str) else (r or {}).get("id") for r in w.get("required") or []] - why = w.get("why") or "" - if required: - ctx.check(why.startswith("override active") and all(str(r) in why for r in required), - "the reason does not name the lifted wizards %s: %r", required, why) - ctx.log("the override lifts: %s", required) - else: - ctx.check(why == "", "every wizard is complete but the reason reads %r", why) - ctx.log("every required wizard is complete; the override lifts nothing today") - m = mode(fc) - ev["mode"] = m - ctx.check(m.get("gated") is False and m.get("controller") == "running", - "the supervisor is gated or the controller is down: %s", m) - - # ------------------------------------------------------------ advisories @test("setup.advisories-rehash", title="An advisory is accepted at its current hash only", @@ -508,9 +520,8 @@ def override_until_reboot(ctx): "document is refused (400); an accept with a stale hash is refused (409); the " "typed document without its phrase is refused (400); the accept at the current " "hash with the phrase is recorded (200), and any new acceptance clears the " - "press (acceptance_done false, the gate closed on the consent, which the " - "override never lifts). The previous record is put back under a forgectrl " - "restart and the machine reads as before.") + "press (acceptance_done false, the gate closed on the consent). The previous " + "record is put back under a forgectrl restart and the machine reads as before.") def advisories_rehash(ctx): fc = ctx.forgectrl ev = ctx.evidence @@ -1379,13 +1390,13 @@ FIRST_RUN_SETTINGS = ["cloud_enabled", "homing_mode", "controller_mode", "gfclou def first_run_record(status, base): - """A record that reads as a first run with the checks and the sheet + """A record that reads as a first run with the checks and the burns behind it: no consent, no press, no account, no form wizard, not - complete; every check and live wizard of the catalog kept from - `base` (the real record) or written at its catalog version. It is - the machine after the button-hold reset of the consent and the - account, and it lets the first run reach Finish without the hour of - checks and burns that a fresh machine owes.""" + complete; every check and burn of the catalog kept from `base` (the + real record) or written at its catalog version. It is the machine + after the button-hold reset of the consent and the account, so the + run walks the form steps and not the hour of checks and burns a + fresh machine is offered.""" base = base or {} ts = now_iso() rec = {"schema": 1, "created": base.get("created") or ts, "advisories": {}, "wizards": {}, @@ -1403,10 +1414,10 @@ def first_run_record(status, base): def first_run(ctx): """The machine as a first run for the block: the record seeded by first_run_record and the account record removed, under a forgectrl - restart, with the override kept; the settings the form steps write - put back as found; the real records restored under another restart, - the system accounts replayed from them, and the temporary home - directory removed. Yields a dict the block fills with `temp_account`.""" + restart; the settings the form steps write put back as found; the + real records restored under another restart, the system accounts + replayed from them, and the temporary home directory removed. Yields + a dict the block fills with `temp_account`.""" fc = ctx.forgectrl ev = ctx.evidence before = wiz(fc) @@ -1425,7 +1436,7 @@ def first_run(ctx): state = {"temp_account": None} try: with Restore(ctx, FIRST_RUN_SETTINGS): - with installed(ctx, {record_path(): seed, users_path(): None, override_path(): b""}): + with installed(ctx, {record_path(): seed, users_path(): None}): w = wiz(fc) ev["fresh"] = wiz_summary(w) ctx.check(w.get("first_run") is True and not (w.get("users") or {}).get("exists") @@ -1458,12 +1469,12 @@ def first_run(ctx): def finished_first_run(ctx): """The end of a first run: the record complete, the gate open, the - controller back.""" + controller running.""" fc = ctx.forgectrl ev = ctx.evidence w = wiz(fc) ev["completed"] = wiz_summary(w) - ctx.check(w.get("completed") is True and w.get("gate") == "open" and not w.get("required"), + ctx.check(w.get("completed") is True and w.get("gate") == "open", "the run did not complete with the gate open: %s", wiz_summary(w)) m = wait_settled(ctx) ev["mode_after_setup"] = m @@ -1487,13 +1498,19 @@ _FIRST_RUN_REQUIRES = ["setup.gate-blocks-controllers", "setup.advisories-rehash "advisories_complete; POST /wiz/advisories/press, the button state waiting and " "the button LEDs breathing teal, the press (the bench actuator's), " "acceptance_done; POST /wiz/account with a temporary name and password, the " - "account exists and the response sets the session cookie; POST /wiz/preferences " - "with the units as found; POST /wiz/machine with the model as found (basic when " - "unknown) and no TEC; POST /wiz/cloud as found (enabled with the phrase and the " - "homing as found, or off); POST /wiz/complete, the record complete, the gate " - "open, the controller back. The settings the steps write, the real record and " - "the account are restored, the system accounts replayed, the temporary home " - "removed. Nothing moves and the laser is not involved.") + "account exists and the response sets the session cookie, and the gate holds " + "with the reason 'setup required: preferences, machine, cloud' and no " + "controller. Then the required steps, the gate's reason naming what is left " + "after each: POST /wiz/preferences with the units as found; POST /wiz/machine " + "with the model as found (basic when unknown) and no TEC; POST /wiz/cloud as " + "found (enabled with the phrase and the homing as found, or off). With the " + "intake fans counted (the machine's setting, which the seed leaves alone) the " + "gate opens and the supervisor starts the controller. POST /wiz/complete, which " + "Start using the machine and Finish later both call: the record complete, the " + "gate open, the controller running. The settings " + "the steps write, the real record and the account are restored, the system " + "accounts replayed, the temporary home removed. Nothing moves and the laser is " + "not involved.") def first_run_flow(ctx): fc = ctx.forgectrl ev = ctx.evidence @@ -1541,12 +1558,29 @@ def first_run_flow(ctx): "the account does not read created: %s", w.get("users")) ctx.log("account %r created, the session cookie set", name) + # After the account the gate waits for the required steps: no + # controller yet, and the reason names what is left. + def still_gated(left): + w = wiz(fc) + m = mode(fc) + why = "setup required: " + ", ".join(left) + ev["gated " + why] = {"wiz": wiz_summary(w), "mode": m} + ctx.check(w.get("gate") == "closed" and w.get("why") == why, + "GET /wiz does not hold the gate on %s: %s", ", ".join(left), wiz_summary(w)) + ctx.check(m.get("controller") == "gated", "the controller is %s with %s left", m.get("controller"), + ", ".join(left)) + + still_gated(REQUIRED_STEPS) + ctx.log("the gate holds after the account: %s", wiz(fc).get("why")) + st, body = fc.post("/wiz/preferences", data={"ui_units": settings.get("ui_units") or "metric"}) ctx.check(st == 200, "POST /wiz/preferences -> %s %s", st, body if isinstance(body, dict) else "") + still_gated(REQUIRED_STEPS[1:]) machine = (w.get("machine") or {}) model = machine.get("model") if machine.get("model") in ("basic", "plus", "pro") else "basic" st, body = fc.post("/wiz/machine", data={"model": model, "tec": "0"}) ctx.check(st == 200, "POST /wiz/machine -> %s %s", st, body if isinstance(body, dict) else "") + still_gated(REQUIRED_STEPS[2:]) if settings.get("cloud_enabled") == "1": data = {"enabled": "1", "phrase": SAFETY_PHRASE, "homing_mode": settings.get("homing_mode") or "gfcloud"} @@ -1557,18 +1591,32 @@ def first_run_flow(ctx): ctx.check(st == 200, "POST /wiz/cloud %s -> %s %s", data, st, body if isinstance(body, dict) else "") w = wiz(fc) done = w.get("versions") or {} - for wid in REQUIRED_WIZARDS: + for wid in FORM_WIZARDS: ctx.check(done.get(wid) == 1, "the record does not carry %s at version 1: %s", wid, done) ctx.log("preferences, machine (%s), cloud (%s) written", model, data.get("enabled")) + # The required steps done and the intake fans counted (the + # machine's setting, which the seed leaves alone, so the page shows + # the Intake fans step as done): the gate opens and the controller + # runs, whatever the setup does next. + ev["after_required"] = wiz_summary(w) + ctx.check(w.get("gate") == "open" and not w.get("why") and w.get("fans_counted") is True, + "the gate does not open with the required steps: %s", wiz_summary(w)) + took = ctx.wait_for(lambda: mode(fc).get("controller") == "running", 60) + m = wait_settled(ctx) + ev["mode_after_required"] = m + ctx.check(took is not None and isinstance(m, dict) and m.get("controller") == "running", + "the controller did not start with the required steps: %s", m) + ctx.log("the gate is open; the controller runs (%.1f s)", took) + st, body = fc.post("/wiz/complete") ctx.check(st == 200, "POST /wiz/complete -> %s %s", st, body if isinstance(body, dict) else "") finished_first_run(ctx) - ctx.log("the run completed: gate open, the controller back") + 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=6, + 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=_FIRST_RUN_REQUIRES + ["setup.first-run-flow"], @@ -1577,16 +1625,19 @@ def first_run_flow(ctx): "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; click Finish. The " - "checks and the sheet are behind you already.", + "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, 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, the run complete, the gate open, the " - "controller back). The operator confirms the panel opened. Everything the run " - "wrote is restored as setup.first-run-flow restores it.") + "'/' 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 @@ -1596,6 +1647,8 @@ def first_run_page(ctx): 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) @@ -1609,17 +1662,26 @@ def first_run_page(ctx): 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, Finish.") + 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 Finish?") + ctx.confirm("Did the control panel open after Start using the machine?") @test("setup.cert-page", title="The certificate is checkable before the warning", @@ -1672,43 +1734,46 @@ def preserved(ctx, *paths): ctx.log("restored under a restart: %s", ", ".join(os.path.basename(p) for p in saved)) -def flag_of(w, level, wid): - """The reason a GET /wiz body gives for wid at level ("required" or - "recommended"), or None when the flag is not raised.""" - for x in w.get(level) or []: - if isinstance(x, dict) and x.get("id") == wid: - return x.get("reason") or "" - return None +def flag_of(w, wid): + """The reason a GET /wiz body gives for recommending wid, or None when + it does not recommend it.""" + return recommended(w).get(wid) -@test("setup.what-changed", title="A replaced part asks for its checks again", +@test("setup.what-changed", title="A replaced part recommends its steps again", subsystem="setup", kind="auto", est_min=2, hardware="takeover", covers=[("forgectrl", "src/setup.*"), ("forgectrl", "src/wiz.*"), ("forgectrl", "src/main.c"), ("forgectrl", "src/ui/panel.js"), ("forgectrl", "src/ui/index.html"), ("forgectrl", "src/ui/help.js")], requires=["forgectrl.auth"], - description="GET /wiz carries the what-changed menu (every change with the wizards it maps " - "to, all of them in the catalog). POST /wiz/changed what=tray flags the focus " - "card as recommended with the tray as the reason and leaves the gate as it was; " - "what=fan flags the airflow check as required, and the gate closes unless the " - "override stands; an unknown change is refused (400). The record is put back " - "as found under a forgectrl restart.") + description="GET /wiz carries the what-changed menu: every change with the steps it " + "recommends, all of them in the catalog (the tube the laser floor; the pump and " + "the coolant the coolant flow; a fan the airflow check; the head the machine, " + "the lens travel, the focus, motion, and the cameras; the tray the focus; a " + "service the switches and motion). POST /wiz/changed what=tray recommends the " + "focus burn with the tray as the reason, and what=fan the airflow check with the " + "fan as the reason; neither moves the gate, which only the advisories, the " + "account, the required steps, and the intake fan count close. An unknown change " + "is refused (400). " + "The record is put back as found under a forgectrl restart.") def what_changed(ctx): fc = ctx.forgectrl ev = ctx.evidence + want = {"tube": ["laser.floor"], "pump": ["cooling.flow"], "coolant": ["cooling.flow"], + "fan": ["airflow"], "head": ["machine", "lens.travel", "laser.focus", "motion", "cameras"], + "tray": ["laser.focus"], "service": ["switches", "motion"]} before = wiz(fc) ev["before"] = wiz_summary(before) menu = before.get("changes") or [] catalog_ids = {z["id"] for z in before.get("wizards") or []} - ev["menu"] = [c.get("id") for c in menu] - ctx.check(len(menu) >= 7, "the menu lists %d changes, expected at least 7", len(menu)) + ev["menu"] = {c.get("id"): [x.get("id") for x in c.get("wizards") or []] for c in menu} + ctx.check(ev["menu"] == want, "the menu reads %s, expected %s", ev["menu"], want) unknown = [x["id"] for c in menu for x in c.get("wizards") or [] if x.get("id") not in catalog_ids] ctx.check(not unknown, "the menu names wizards outside the catalog: %s", unknown) ctx.check(all(c.get("title") for c in menu), "a change has no title") + gate_before = before.get("gate") gated_before = bool(mode(fc).get("gated")) - override = bool(before.get("override")) ev["gated_before"] = gated_before - ev["override"] = override with preserved(ctx, record_path()): st, body = fc.post("/wiz/changed", data={"what": "nothing-like-this"}) @@ -1719,35 +1784,32 @@ def what_changed(ctx): ctx.log("POST /wiz/changed what=tray -> %s", st) ctx.check(st == 200 and isinstance(body, dict), "what=tray -> %s", st) w = body if isinstance(body, dict) else wiz(fc) - reason = flag_of(w, "recommended", "laser.focus") + reason = flag_of(w, "laser.focus") ev["tray_reason"] = reason ctx.check(reason is not None and "tray" in reason, - "the tray does not recommend the focus card: %s", w.get("recommended")) - ctx.check(flag_of(w, "required", "laser.focus") is None, "the tray made the focus card required") - ctx.check(bool(mode(fc).get("gated")) == gated_before, "a recommendation moved the gate") + "the tray does not recommend the focus burn: %s", w.get("recommended")) + ctx.check(w.get("gate") == gate_before and bool(mode(fc).get("gated")) == gated_before, + "a recommendation moved the gate: %s", wiz_summary(w)) st, body = fc.post("/wiz/changed", data={"what": "fan"}) ctx.log("POST /wiz/changed what=fan -> %s", st) ctx.check(st == 200, "what=fan -> %s", st) w = wiz(fc) - reason = flag_of(w, "required", "airflow") + reason = flag_of(w, "airflow") ev["fan_reason"] = reason ctx.check(reason is not None and "fan" in reason, - "a fan does not require the airflow check: %s", w.get("required")) + "a fan does not recommend the airflow check: %s", w.get("recommended")) m = mode(fc) ev["gated_after_fan"] = m.get("gated") - ctx.check("airflow" in (w.get("why") or ""), "the gate's reason does not name airflow: %r", w.get("why")) - if override: - ctx.check(m.get("gated") is False, "the override stands but the gate closed: %s", m) - else: - ctx.check(m.get("gated") is True, "a required check left the gate open: %s", m) - ctx.log("tray -> focus recommended; fan -> airflow required; gate %s (override %s)", - "closed" if m.get("gated") else "open", override) + ctx.check(w.get("gate") == gate_before and bool(m.get("gated")) == gated_before, + "a recommendation moved the gate: %s, %s", wiz_summary(w), m) + ctx.log("tray -> focus recommended (%s); fan -> airflow recommended (%s); gate %s as found", + ev["tray_reason"], reason, gate_before) after = wiz(fc) ev["after"] = wiz_summary(after) - ctx.check(flag_of(after, "required", "airflow") is None and flag_of(after, "recommended", "laser.focus") is None, - "the flags survive the restore: %s / %s", after.get("required"), after.get("recommended")) + ctx.check(recommended(after) == recommended(before), + "the recommendations do not read as found: %s, was %s", recommended(after), recommended(before)) ctx.check(bool(mode(fc).get("gated")) == gated_before, "the gate does not read as before") diff --git a/forgetest/forgetest/suite/setup_burns.py b/forgetest/forgetest/suite/setup_burns.py new file mode 100644 index 0000000..68dc775 --- /dev/null +++ b/forgetest/forgetest/suite/setup_burns.py @@ -0,0 +1,296 @@ +# Copyright 2026 514 LLC d/b/a OpenGlow +# Written by Scott Wiederhold +# https://community.openglow.org +# SPDX-License-Identifier: MIT + +"""setup.focus, setup.floor - the setup's two burns (the live wizards), +driven the way the page drives them: POST /wiz//start, GET /wiz/dark +polled, the prompts answered from here, the burn armed by a press, the +result judged, and the setting the burn wrote put back as found. Each +burn stands alone on its own scrap of wood in the back-left corner, +drawn from the head's home against the gantry stops, and re-runs on the +real record, so a completed burn completes again at the same version. + +The operator's part is the scrap and one press on the machine at the +start (the presence check of the ready gate; with the bench actuator up +it then makes the arm press). The judgment a person makes by eye on the +scrap (the narrowest line, the faintest rung) is answered here with a +fixed pick: the test proves each wizard's mechanics and the emission +witnesses, not the operator's eye, and the setting it writes is +restored. The page is where the real numbers go in. +""" +import re + +from ..catalog import test +from .setup_dark import run_check, Restore, FALLBACK_WINDOW, LENS_WINDOW_KEYS + +BURN_COVERS = [("forgectrl", "src/wizlive.*"), ("forgectrl", "src/wizrun.h"), + ("forgectrl", "src/lens.*"), ("forgectrl", "src/sheet.*"), + ("forgectrl", "src/font_hershey.*"), ("forgectrl", "src/jobstream.*"), + ("forgectrl", "src/jobrun.*"), ("forgectrl", "src/lease.*"), + ("forgectrl", "src/curverec.*"), ("forgectrl", "src/wizdark.*"), + ("forgectrl", "src/wiz.*"), ("forgectrl", "src/setup.*"), + ("forgectrl", "src/main.c"), ("forgectrl", "src/ui/wizard.*"), + ("forgectrl", "src/super.c"), + ("forgectrl", "src/status.c"), ("forgectrl", "src/cool.c"), + ("forgectrl", "src/accel.*"), ("forgectrl", "src/wizcalc.*"), + ("grblhal-glowforge", "src/glowforge_laser.c"), + ("grblhal-glowforge", "src/glowforge_status.c"), + ("grblhal-glowforge", "src/glowforge_homing.*"), + ("grblhal-glowforge", "src/glowforge_io.*")] + +# Every burn asks first that the gantry stops are installed and the head +# is at its home against them; the bench's head stands there, and every +# burn's tail brings it back. +HOME = "The head is home" +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", + "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): + """The page's preview and the program for the burn are served.""" + fc = ctx.forgectrl + st, body = fc.get("/wiz/sheet.svg?card=%s" % wid, raw=True) + ctx.check(st == 200 and body.lstrip().startswith(b" hi + 0.01] + ctx.check(not out, "%s commands Z outside the lens window %.2f..%.2f: %s", + wid, lo, hi, out) + + +def emission_ok(ctx, last): + """The three witnesses of the burn, from the wizard's result.""" + e = (last.get("result") or {}).get("emission") or {} + ctx.evidence["emission"] = e + ctx.check(e.get("hv_max", 0) > 30, "the tube current never rose (%s)", e.get("hv_max")) + ctx.check(e.get("laser_on_samples", 0) > 0, "the kernel never sampled LASER_ON") + ctx.check(e.get("thermopile_delta", 0) >= 100, "the thermopile did not see the beam (%s)", + e.get("thermopile_delta")) + + +def thickness_answer(ctx): + """The thickness the burn is answered with, in the machine's units + (the number prompts read ui_units), and the millimeters the record + must carry for it.""" + imperial = (ctx.forgectrl.settings() or {}).get("ui_units") == "imperial" + return ("0.125", 3.175) if imperial else ("3.2", 3.2) + + +def burn(ctx, wid, arm, pick, want): + """One burn: the preview, then the burn after the press with every + prompt answered (the gantry stops installed, the head at home, the + scrap in the corner, its thickness in the machine's units, the arm, + and the pick in `pick`), the witnesses, and the keys of the result in + `want`. run_check makes the arm press when the burn asks for it. A + prompt the test does not answer, or one asked again because its + answer was refused, ends the burn. Returns the result.""" + answer, want_mm = thickness_answer(ctx) + ctx.evidence["thickness_answer"] = answer + answers = {"gantry-stops": "Yes", "home": HOME, "scrap": "Continue", "thickness": answer, + arm: "Continue"} + answers.update(pick) + preview_ok(ctx, wid) + asked = set() + + def on_prompt(p): + pid = p.get("id") + if pid == "press": + return None + if pid not in answers: + # the motors found released ask for the head pushed home by + # hand, which the bench's head never needs + ctx.fail("the %s burn asked %s, which the test does not answer: %s", wid, pid, p.get("text")) + if pid in asked: + ctx.fail("the %s burn asked %s again: it refused %r", wid, pid, answers[pid]) + asked.add(pid) + return answers[pid] + + try: + last = run_check(ctx, wid, on_prompt, 900) + finally: + ctx.clear_notice() + emission_ok(ctx, last) + r = last.get("result") or {} + missing = [k for k in want if k not in r] + ctx.check(not missing, "the result lacks %s: %s", missing, sorted(r)) + ctx.check(abs((r.get("thickness_mm") or 0) - want_mm) < 0.01, + "thickness %s mm, expected %s (answered %s)", r.get("thickness_mm"), want_mm, answer) + ctx.check(len(r.get("z_passes") or []) >= 1, "the lens was not referenced: %s", r.get("z_passes")) + return r + + +def window_held(ctx): + """The lens window the machine holds: the two settings, and the lens + block of /status, which reads them (the fallback where they are + unset).""" + fc = ctx.forgectrl + s = fc.settings() or {} + lens = (fc.status() or {}).get("lens") or {} + return {"settings": {k: s.get(k) for k in LENS_WINDOW_KEYS}, + "window": (lens.get("below"), lens.get("above")), "stops_found": lens.get("stops_found")} + + +def focus_window_ok(ctx, r, before): + """The ladder burned in the window the machine held before the burn + (`before`, from window_held): the Lens travel check's, or the fallback + while the settings are unset. The burn left that window as it found + it and wrote the edge alone, the one its result reports; every ladder + height lies in the reach, and the program served now agrees with + /status.""" + fc = ctx.forgectrl + ev = ctx.evidence + w = r.get("window") or {} + window = (w.get("below"), w.get("above")) + after = window_held(ctx) + ctx.check(after["settings"] == before["settings"], "the focus burn wrote the lens window: %s, was %s", + after["settings"], before["settings"]) + ctx.check(window == before["window"], "the ladder ran in %s/%s, the machine holds %s", + window[0], window[1], before["window"]) + held = before["settings"] + if before["stops_found"]: + ctx.check(tuple(str(held[k]) for k in LENS_WINDOW_KEYS) == tuple(str(v) for v in window), + "the ladder ran in %s/%s, the settings hold %s", window[0], window[1], held) + else: + ctx.check(window == FALLBACK_WINDOW, "the settings hold no window (%s), but the ladder ran in %s/%s", + held, window[0], window[1]) + edge = (fc.settings() or {}).get(EDGE_KEY) + try: + edge_ok = abs(float(edge) - float(r.get("edge_z_mm"))) < 0.006 + except (TypeError, ValueError): + edge_ok = False + ctx.check(edge_ok, "%s reads %r, the burn reports %s", EDGE_KEY, edge, r.get("edge_z_mm")) + zs = r.get("ladder_z") or [] + lo, hi = w.get("reach_min"), w.get("reach_max") + out = [z for z in zs if lo is None or hi is None or z < lo - 0.01 or z > hi + 0.01] + ctx.check(len(zs) == 12 and not out, "ladder heights outside the reach %s..%s: %s", lo, hi, out or zs) + first = dict(ev.get("program_z") or {}).get("laser.focus") + preview_ok(ctx, "laser.focus") + ev["focus_window"] = {"window": w, "ladder_z": zs, "held": before, "edge": edge, + "program_z_before": first, + "program_z_after": (ev.get("program_z") or {}).get("laser.focus"), + "preview_bytes": ev.pop("preview_bytes", None), + "program_lines": ev.pop("program_lines", None)} + + +@test("setup.focus", title="The focus burn: the ladder in the lens window, and the edge 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.check-lens"], + 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 78 x 68 mm (3.1 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 focus burn as a fresh head runs it: the edge setting is cleared first, so " + "the burn starts from the bench reference's edge, while the lens window stays " + "the one the Lens travel check wrote (or the fallback). POST " + "/wiz/laser.focus/start is answered that the gantry stops are installed, " + "references the lens, takes the controller in loopback posture, is answered that " + "the head is home and that the scrap is in the corner, and takes 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 returns there. With its preview and program served by GET " + "/wiz/sheet.svg and /wiz/sheet.gcode and every Z the program commands inside the " + "reach /status reports, it burns twelve lines after the press, from the top of " + "the window to the bottom, seen by the three witnesses (the tube current, the " + "kernel's LASER_ON samples, the thermopile rise). The pick is line 11. The " + "result carries its keys; the ladder ran in the window the machine held, which " + "the burn leaves as found (it writes the edge alone, the one its result " + "reports); every ladder height lies in the reach; the program served afterward " + "agrees with /status; the record carries laser.focus at version 1. The edge is " + "restored at the end.") +def focus(ctx): + fc = ctx.forgectrl + ev = ctx.evidence + ctx.ready("The focus burn: twelve lines on the scrap after a press. The bench presses when the " + "button lights.") + with Restore(ctx, [EDGE_KEY] + LENS_WINDOW_KEYS): + # A fresh head's focus: the edge unknown, so the burn runs from the + # bench reference's number and writes the head's own. The window + # stays: it is the Lens travel check's. + st, _ = fc.post("/settings", params={EDGE_KEY: ""}) + ctx.check(st == 200, "clearing %s -> %s", EDGE_KEY, st) + before = window_held(ctx) + ev["lens_fresh"] = (fc.status() or {}).get("lens") or {} + r = burn(ctx, "laser.focus", "focus-arm", {"focus-pick": "11"}, FOCUS_KEYS) + ctx.check(r.get("pick") == 11 and r.get("picked_lines") == [11, 11], + "the pick reads %s (lines %s), answered 11", r.get("pick"), r.get("picked_lines")) + ctx.check("stops" not in r, "the focus burn searched for the stops: %s", r.get("stops")) + focus_window_ok(ctx, r, before) + ctx.log("focus: line %s, the edge at %s mm, the window %s/%s", r.get("pick"), r.get("edge_z_mm"), + (r.get("window") or {}).get("below"), (r.get("window") or {}).get("above")) + ctx.log("PASS: the focus burn ran in the window the machine holds and wrote the edge alone; " + "settings restored") + + +@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"], + 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="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 returns there. With " + "its preview and program served and every Z the program commands inside the " + "reach /status reports, 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 three 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.") +def floor(ctx): + fc = ctx.forgectrl + 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") diff --git a/forgetest/forgetest/suite/setup_dark.py b/forgetest/forgetest/suite/setup_dark.py index f2a4b55..3aea767 100644 --- a/forgetest/forgetest/suite/setup_dark.py +++ b/forgetest/forgetest/suite/setup_dark.py @@ -17,7 +17,7 @@ import time from ..catalog import test from .. import hw from ..baseline import read_position -from .setup import wiz, Restore # noqa: F401 - Restore is re-exported for the sheet +from .setup import wiz, Restore, INTAKE_KEY POLL_S = 1.0 # What the machine's own press prompt says, used when it sends no text. @@ -29,6 +29,26 @@ PURGE_SPINUP_S = 30 DARK_COVERS = [("forgectrl", "src/wizdark.*"), ("forgectrl", "src/wizcalc.*"), ("forgectrl", "src/wiz.*"), ("forgectrl", "src/setup.*"), ("forgectrl", "src/main.c"), ("forgectrl", "src/ui/wizard.*")] +# The lens window: the free half-steps below and above the hall edge the +# Lens travel check writes, and the fallback the lens frame holds while +# they are unset (lens.h). +LENS_WINDOW_KEYS = ["lens_stop_below_steps", "lens_stop_above_steps"] +FALLBACK_WINDOW = (10, 12) + + +def intakes_counted(): + """Precheck: the intake fan count is set. The setup gate holds every + controller without it, and a test that puts the count back as found + would put an unset count back and close the gate on the tests after + it.""" + try: + s = hw.Forgectrl().settings() or {} + except hw.HwError as e: + return "forgectrl unreachable: %s" % e + if not s.get(INTAKE_KEY): + return ("%s is unset, so the setup gate holds every controller: count the intake fans " + "first (the Intake fans step of the setup page)" % INTAKE_KEY) + return None def dark(fc): @@ -189,7 +209,7 @@ def check_sensors(ctx): subsystem="setup", kind="auto", hardware="takeover", est_min=3, covers=DARK_COVERS + [("forgectrl", "src/cool.c"), ("forgectrl", "src/airflow.*"), ("forgectrl", "src/gates.c"), ("forgectrl", "src/super.c")], - requires=["forgectrl.auth", "cooling.fan-gate-trips"], + requires=["forgectrl.auth", "cooling.fan-gate-trips"], precheck=intakes_counted, description="POST /wiz/airflow/start: the controller stops, the fans run at the cut profile " "for 35 s, the purge is measured on and off, and six settings are written: the " "four floors at 55 percent of steady, the grace from the slowest spin-up, and the " @@ -254,6 +274,52 @@ def check_airflow(ctx): 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", "cooling.fan-gate-trips"], 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: intake 1 turning and intake 2 never reading a pulse is one, " + "anything else two. No question is asked; the one the step asks when intake 1 " + "does not come up (the operator's count) fails the test. 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: intake 1 at 1000 rpm or more, intake " + "2 silent exactly when the count is one. 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": + ctx.fail("intake 1 did not come up to speed, so the tachs could not count the fans: %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) + ctx.check((r.get("intake_1_rpm") or 0) >= 1000, "intake 1 turned at %s rpm", r.get("intake_1_rpm")) + ctx.check(r.get("intake_2_silent") is (n == 1), + "intake 2 %s with a count of %s", "silent" if r.get("intake_2_silent") else "turning", n) + 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%s)", n, r.get("intake_1_rpm"), + r.get("intake_2_rpm"), ", silent" if r.get("intake_2_silent") else "") + 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")], @@ -330,6 +396,55 @@ def check_motion(ctx): r.get("rest"), {k: (v.get("p2p_lp_x"), v.get("p2p_lp_y")) for k, v in moves.items()}) +@test("setup.check-lens", title="The lens travel check finds the stops and writes the window", + subsystem="setup", kind="auto", hardware="takeover", est_min=4, + covers=DARK_COVERS + [("forgectrl", "src/lensfind.*"), ("forgectrl", "src/lens.*"), + ("forgectrl", "src/lenshome.*"), ("forgectrl", "src/accel.c"), + ("forgectrl", "src/cool.c"), ("forgectrl", "src/super.c")], + requires=["forgectrl.auth", "setup.check-motion"], + description="POST /wiz/lens.travel/start: the controller stops, the lens takes its hall-edge " + "reference, the fans stop and the head accelerometer listens while the lens " + "steps to each stop and back (nothing fires), the two window settings are " + "written, and the controller comes back. No question is asked. The result " + "carries the reference passes, the stops (found, the free half-steps each way, " + "the contact counts, or why not), and the window: the stops found, or the " + "fallback (10 half-steps below, 12 above) when they were not. The settings and " + "the lens block of /status hold that window, the record carries lens.travel at " + "version 1, the two settings are put back as found, and the controller must be " + "running again.") +def check_lens(ctx): + fc = ctx.forgectrl + ev = ctx.evidence + + def on_prompt(p): + ctx.fail("the lens travel check asked a question: %s", p) + with Restore(ctx, LENS_WINDOW_KEYS): + last = run_check(ctx, "lens.travel", on_prompt, 300) + r = last.get("result") or {} + ctx.check(len(r.get("z_passes") or []) >= 1, "the lens was not referenced: %s", r.get("z_passes")) + stops, w = r.get("stops") or {}, r.get("window") or {} + ctx.check(isinstance(stops.get("found"), bool), "the result files no stops: %s", r) + window = (w.get("below"), w.get("above")) + if stops.get("found"): + ctx.check(window == (stops.get("below"), stops.get("above")) and w.get("found") is True, + "the window %s is not the stops found %s", w, stops) + else: + ctx.check(window == FALLBACK_WINDOW and w.get("found") is False, + "the stops were not found (%s), but the window is %s", stops.get("why"), w) + s = fc.settings() or {} + held = tuple(str(s.get(k)) for k in LENS_WINDOW_KEYS) + lens = (fc.status() or {}).get("lens") or {} + ev["window"] = {"stops": stops, "window": w, "settings": held, "status": lens} + ctx.check(held == tuple(str(v) for v in window), "the settings hold %s, the check wrote %s", held, window) + ctx.check((lens.get("below"), lens.get("above")) == window, + "/status reads the window %s/%s, the check wrote %s", lens.get("below"), lens.get("above"), window) + ctx.log("lens travel: %s half-steps below the edge, %s above (%s); contact %s/%s", + window[0], window[1], "the stops found" if stops.get("found") else "the fallback: %s" % stops.get("why"), + stops.get("contact_below"), stops.get("contact_above")) + 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 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")], diff --git a/forgetest/forgetest/suite/setup_sheet.py b/forgetest/forgetest/suite/setup_sheet.py deleted file mode 100644 index c0358d3..0000000 --- a/forgetest/forgetest/suite/setup_sheet.py +++ /dev/null @@ -1,283 +0,0 @@ -# Copyright 2026 514 LLC d/b/a OpenGlow -# Written by Scott Wiederhold -# https://community.openglow.org -# SPDX-License-Identifier: MIT - -"""setup.sheet - the setup's sheet (the live wizards) in one run, -driven the way the page drives them: POST /wiz//start, GET /wiz/dark -polled, the prompts answered from here, each burn armed by a press, the -result judged, and every setting a wizard wrote put back as found. One -piece of wood at least 200 x 150 mm (8 x 6 in) carries the placement, -the frame, and the five cards. Each wizard re-runs on the real record, -so a completed one completes again at the same version. - -The operator's part is one press on the machine at the start (the -presence check of the ready gate; with the bench actuator up it then -makes every arm press) and nothing else. The judgments a person makes -by eye on the sheet (the thinnest line, the faintest rung, the best -corner, the frame's fit) are answered here with a middle option or the -witnesses: the test proves each wizard's mechanics and the emission -witnesses, not the operator's eye, and the settings it writes are -restored. The page is where the real numbers go in. -""" -import re - -from ..catalog import test -from .setup_dark import run_check, Restore - -SHEET_COVERS = [("forgectrl", "src/wizlive.*"), ("forgectrl", "src/wizrun.h"), - ("forgectrl", "src/lens.*"), ("forgectrl", "src/sheet.*"), - ("forgectrl", "src/font_hershey.*"), ("forgectrl", "src/jobstream.*"), - ("forgectrl", "src/jobrun.*"), ("forgectrl", "src/lease.*"), - ("forgectrl", "src/curverec.*"), ("forgectrl", "src/wizdark.*"), - ("forgectrl", "src/wiz.*"), ("forgectrl", "src/setup.*"), - ("forgectrl", "src/main.c"), ("forgectrl", "src/ui/wizard.*"), - ("forgectrl", "src/super.c"), - ("forgectrl", "src/status.c"), ("forgectrl", "src/cool.c"), - ("forgectrl", "src/accel.*"), ("forgectrl", "src/wizcalc.*"), - ("grblhal-glowforge", "src/glowforge_laser.c"), - ("grblhal-glowforge", "src/glowforge_status.c"), - ("grblhal-glowforge", "src/glowforge_homing.*"), - ("grblhal-glowforge", "src/glowforge_io.*")] - -# The live wizards in the order the sheet burns them: the wizard id, the -# answers to its prompts (a middle pick where a person would look at the -# sheet), and the keys its result must carry. The arm press is not timed -# from here: each card opens a press prompt of its own when the button is -# lit and the tube still dark, and run_check presses on that. -CARDS = [ - ("sheet.frame", {"frame-arm": "Continue", "frame-ok": "Yes"}, ["mark_s", "mark_feed"]), - ("laser.focus", {"focus-arm": "Continue", "focus-pick": "11", "thickness": "Keep"}, - ["pick", "thickness_mm", "pick_half_steps", "edge_z_mm", "steps_per_mm", "max_height_mm", - "focus_range_mm", "stops"]), - ("laser.floor", {"floor-arm": "Continue", "floor-pick": "8"}, ["faintest_density", "floor_density"]), - ("laser.dose-curve", {"dose-arm": "Continue"}, ["points", "curve"]), - ("laser.corner", {"corner-arm": "Continue", "corner-pick": "1.50"}, ["gamma"]), - ("cooling.flow-load", {"load-arm": "Continue"}, ["lit_s", "dose_raw_s", "peak_c", "k_density", "k_cw"]), -] -# Every setting a card writes, restored as found when the run ends. -CARD_SETTINGS = ["lens_hall_edge_z_mm", "lens_stop_below_steps", "lens_stop_above_steps", - "laser_floor_density", "laser_dose_curve", "laser_corner_gamma", - "cool_laser_heat_density", "cool_laser_heat_cw"] -# The evidence run_check and the witnesses write at the top level, moved -# under the card's own key once it is done. -CARD_EVIDENCE = ("started", "elapsed_s", "phases", "log", "error", "result", - "preview_bytes", "program_lines", "emission") - - -def preview_ok(ctx, wid): - """The page's preview and the program for the wizard are served.""" - fc = ctx.forgectrl - st, body = fc.get("/wiz/sheet.svg?card=%s" % wid, raw=True) - ctx.check(st == 200 and body.lstrip().startswith(b" hi + 0.01] - ctx.check(not out, "%s commands Z outside the lens window %.2f..%.2f: %s", - wid, lo, hi, out) - - -def emission_ok(ctx, last): - """The three witnesses of the burn, from the wizard's result.""" - e = (last.get("result") or {}).get("emission") or {} - ctx.evidence["emission"] = e - ctx.check(e.get("hv_max", 0) > 30, "the tube current never rose (%s)", e.get("hv_max")) - ctx.check(e.get("laser_on_samples", 0) > 0, "the kernel never sampled LASER_ON") - ctx.check(e.get("thermopile_delta", 0) >= 100, "the thermopile did not see the beam (%s)", - e.get("thermopile_delta")) - - -def file_card(ctx, wid): - """Move the card's evidence under its own key.""" - ev = ctx.evidence - ev.setdefault("cards", {})[wid] = {k: ev.pop(k) for k in CARD_EVIDENCE if k in ev} - - -# The lens settings: the head's numbers, which a fresh machine does not -# have. Cleared before the cards so the run is a fresh machine's. -LENS_SETTINGS = CARD_SETTINGS[:3] -FALLBACK_WINDOW = (10, 12) - - -def fresh_lens(ctx): - """The lens as a fresh machine has it: the head's numbers unknown, so - the frame runs in the fallback window and the focus card has to find - the stops and write them before its ladder burns. The machine's own - numbers come back with the restore at the end of the run.""" - fc = ctx.forgectrl - for k in LENS_SETTINGS: - st, _ = fc.post("/settings", params={k: ""}) - ctx.check(st == 200, "clearing %s -> %s", k, st) - lens = (fc.status() or {}).get("lens") or {} - ctx.evidence["lens_fresh"] = lens - ctx.check(lens.get("stops_found") is False and (lens.get("below"), lens.get("above")) == FALLBACK_WINDOW, - "the lens window is not the fallback after the clear: %s", lens) - - -def focus_window_ok(ctx): - """The window the focus card burned its ladder in is the one it wrote - before its controller started, and the one the machine holds now: the - result's window is the stops it found (or the fallback), the settings - hold that window, every ladder height lies in its reach, and the - program served now agrees with /status.""" - fc = ctx.forgectrl - r = ((ctx.evidence.get("cards") or {}).get("laser.focus") or {}).get("result") or {} - w, stops = r.get("window") or {}, r.get("stops") or {} - s = fc.settings() or {} - held = (str(s.get("lens_stop_below_steps")), str(s.get("lens_stop_above_steps"))) - ctx.check(held == (str(w.get("below")), str(w.get("above"))), - "the settings hold the window %s, the ladder ran in %s/%s", held, w.get("below"), w.get("above")) - if stops.get("found"): - ctx.check((w.get("below"), w.get("above")) == (stops.get("below"), stops.get("above")), - "the ladder's window %s is not the stops found %s", w, stops) - else: - ctx.check((w.get("below"), w.get("above")) == FALLBACK_WINDOW, - "stops not found (%s), but the ladder's window is %s", stops.get("why"), w) - zs = r.get("ladder_z") or [] - lo, hi = w.get("reach_min"), w.get("reach_max") - out = [z for z in zs if lo is None or hi is None or z < lo - 0.01 or z > hi + 0.01] - ctx.check(len(zs) == 12 and not out, "ladder heights outside the reach %s..%s: %s", lo, hi, out or zs) - ev = ctx.evidence - before = dict(ev.get("program_z") or {}).get("laser.focus") - preview_ok(ctx, "laser.focus") - ev["focus_window"] = {"window": w, "stops": stops, "ladder_z": zs, - "settings": {k: s.get(k) for k in LENS_SETTINGS}, - "program_z_before": before, - "program_z_after": (ev.get("program_z") or {}).get("laser.focus"), - "preview_bytes": ev.pop("preview_bytes", None), - "program_lines": ev.pop("program_lines", None)} - - -def thickness_answer(ctx): - """The thickness the placement is answered with, in the machine's - units (the number prompts read ui_units), and the millimeters the - record must carry for it.""" - imperial = (ctx.forgectrl.settings() or {}).get("ui_units") == "imperial" - return ("0.125", 3.175) if imperial else ("3.2", 3.2) - - -def place(ctx): - """The placement: the lens reference, two jogs, the origin at the - head's position, the full sheet, a thickness. Nothing fires.""" - jogs = ["X+10", "X-10"] - answer, want_mm = thickness_answer(ctx) - ctx.evidence["thickness_answer"] = answer - - def on_prompt(p): - # Each jog answer closes the prompt and the wizard asks again with - # the next sequence number, so the pad is answered one move at a - # time until the origin is set. - if p.get("id") == "place": - return jogs.pop(0) if jogs else "Set origin" - if p.get("id") == "sheet-kind": - return "Full sheet" - if p.get("id") == "thickness": - return answer - return None - - last = run_check(ctx, "sheet.place", on_prompt, 300) - r = last.get("result") or {} - ctx.check("origin_x" in r and "origin_y" in r, "no origin in the result: %s", r) - ctx.check((r.get("steps_per_mm") or {}).get("x", 0) > 0, "no steps per mm: %s", r) - ctx.check(r.get("z_referenced") is True, "the lens was not referenced") - ctx.check(abs((r.get("thickness_mm") or 0) - want_mm) < 0.01, - "thickness %s mm, expected %s (answered %s)", r.get("thickness_mm"), want_mm, answer) - ctx.check(r.get("alone") is False, "alone %s", r.get("alone")) - ctx.log("placed: datum at %s, %s; steps per mm %s", r.get("origin_x"), r.get("origin_y"), - r.get("steps_per_mm")) - file_card(ctx, "sheet.place") - - -def burn(ctx, wid, answers, want): - """One live card: the preview, the burn after the press with the - prompts answered from `answers`, the witnesses, the keys of the - result in `want`. run_check makes the arm press when the card asks - for it.""" - preview_ok(ctx, wid) - - def on_prompt(p): - return answers.get(p.get("id")) - - try: - last = run_check(ctx, wid, on_prompt, 900) - finally: - ctx.clear_notice() - emission_ok(ctx, last) - r = last.get("result") or {} - missing = [k for k in want if k not in r] - ctx.check(not missing, "the result lacks %s: %s", missing, sorted(r)) - ctx.log("%s: %s, emission %s", wid, {k: r.get(k) for k in want}, ctx.evidence.get("emission")) - file_card(ctx, wid) - - -@test("setup.sheet", title="The setup sheet: the placement, the frame, and the five cards", - subsystem="setup", kind="live", hardware="takeover", mode="grbl", est_min=35, - covers=SHEET_COVERS, - requires=["forgectrl.auth", "setup.check-motion", "laser.emission-witness", "cooling.flow-verify"], - actions=["button"], hands=["scrap"], - steps=["A piece of wood at least 200 x 150 mm (8 x 6 in) on the bed, pushed as far left as " - "it goes with its top edge at the top of the cut area (the head's home corner); " - "lid closed. Eye protection on, exhaust on, extinguisher in reach.", - "Start the test and press the machine's button once. That press says you are at the " - "machine, and the bench actuator makes every arm press after it, each one when the " - "card itself asks for it. With no actuator the start is the Ready answer instead, " - "and you press the button each time the page asks: six times, the frame and the " - "five cards.", - "Nothing else: the test answers every prompt itself and puts back every setting the " - "cards write. It runs about half an hour, mostly the coolant settle and the dark " - "tails."], - description="One run over the sheet's seven wizards on one piece of wood, as a fresh " - "machine runs them: the three lens settings are cleared first, so the frame " - "runs in the fallback window and the focus card has to find the stops and write " - "them before its ladder burns. " - "POST /wiz/sheet.place/start references the lens, takes the controller in " - "loopback posture, jogs +X 10 and -X 10, and sets the origin at the head " - "(Full sheet; the thickness answered in the machine's units, 3.2 mm or " - "0.125 in). Then, each after the press and each with its preview " - "and program served by GET /wiz/sheet.svg and /wiz/sheet.gcode, every Z the " - "program commands inside the reach /status reports: the frame " - "(180 x 130 mm) and the header band at the mark dose; the focus card (the " - "lens on its hall edge, the stops found by the head accelerometer and written " - "as the window before the controller starts, twelve lines from the top of the " - "window to the bottom; the pick is line 11, the thickness kept; after it the " - "settings, the result's window, and the served program must hold the window " - "the ladder ran in); the floor card (twelve " - "rungs from 2 to 24 percent with the floor and the curve off for the job; the " - "pick is 8, the floor written is 10); the dose card (seven rungs sampled at " - "25 Hz, the curve fit on 20 s of dark); the corner card (five gammas reloaded " - "through M102 inside one armed job; the pick is 1.50); and the flow-load card " - "(the coolant loop settled, the box and the patch at 60 percent, the " - "downstream peak over 100 s of dark as the coefficient). Every burn must be " - "seen by the three witnesses (the tube current, the kernel's LASER_ON samples, " - "the thermopile rise); every result must carry its keys; the record must " - "carry each wizard at version 1; the seven settings the cards write are " - "restored at the end.") -def sheet(ctx): - ctx.ready("The sheet: the placement, then the frame and the five cards burn after a press " - "each. The bench presses when the button lights.") - place(ctx) - with Restore(ctx, CARD_SETTINGS): - fresh_lens(ctx) - for wid, answers, want in CARDS: - burn(ctx, wid, answers, want) - if wid == "laser.focus": - focus_window_ok(ctx) - ctx.log("PASS: the sheet's seven wizards ran on one piece as a fresh machine's; settings restored") diff --git a/forgetest/forgetest/suite/tray.py b/forgetest/forgetest/suite/tray.py index 854bf1e..6827240 100644 --- a/forgetest/forgetest/suite/tray.py +++ b/forgetest/forgetest/suite/tray.py @@ -25,7 +25,7 @@ from .motion import (_kernel_position, _port_state, _words, clean_slate, drain_t TRAY_MARKER = "/data/forgefirm/tray.out" LENS_SPM = 2.922 # $102: the lens screw's half-steps per mm -SETUP_CARD = "sheet.place" # the first setup card: it asks before it moves anything +SETUP_CARD = "laser.focus" # a setup burn: it asks before it moves anything def _grid_mm(offset_mm): @@ -70,8 +70,8 @@ def _tray_route(fc, mode): def _card_refused(ctx, fc, ev): """A setup card is refused while the tray is out. The machine is idle a moment after the suite's Grbl client closes; until then the refusal is - the idle rule's. A card that started anyway is ended at once: the first - one asks before it moves anything.""" + the idle rule's. A card that started anyway is ended at once: it asks + before it moves anything.""" path = "/wiz/%s/start" % SETUP_CARD deadline = time.time() + 20 while True: diff --git a/forgetest/tests/test_setup_burns.py b/forgetest/tests/test_setup_burns.py new file mode 100644 index 0000000..fe906b1 --- /dev/null +++ b/forgetest/tests/test_setup_burns.py @@ -0,0 +1,414 @@ +# Copyright 2026 514 LLC d/b/a OpenGlow +# Written by Scott Wiederhold +# https://community.openglow.org +# SPDX-License-Identifier: MIT + +"""The setup's two burns in the catalog: two live tests on the button with +a scrap at hand, their covers naming the live runner, the renderer, the +sender, the page, and the driver's laser module; and each burn against a +scripted daemon: the preview and the program read, the start, the +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 +prompt, the pick), the witnesses judged, the record read at version 1, +and the setting the burn wrote put back as found.""" +import os +import sys +import unittest + +sys.path.insert(0, os.path.dirname(__file__)) +import helpers # noqa: E402 +from forgetest import catalog # noqa: E402 +from forgetest.runner import Context, Failed, Run # noqa: E402 +from forgetest.suite import setup_burns, setup_dark # noqa: E402 + +LIVE = ["laser.focus", "laser.floor"] +EMISSION = {"hv_max": 610, "laser_on_samples": 900, "thermopile_delta": 600, "lit_s": 30.0} +SPM = 2.922 # the lens screw, half-steps per mm ($102) +EDGE_DEFAULT = 3.35 # the edge's height while the setting is unset (lens.h) + + +def lens_block(s): + """The /status lens block as the daemon builds it from the settings: a + stop count from 1 to 40 stands, anything else is the fallback.""" + def count(key, fallback): + try: + n = float(s.get(key, "")) + except ValueError: + return fallback, False + return (int(n), True) if 1 <= n <= 40 else (fallback, False) + try: + edge = float(s.get("lens_hall_edge_z_mm", "")) + except ValueError: + edge = EDGE_DEFAULT + below, a = count("lens_stop_below_steps", 10) + above, b = count("lens_stop_above_steps", 12) + return {"edge_z": round(edge, 2), "below": below, "above": above, "stops_found": a and b, + "reach_min": round(edge - below / SPM, 2), "reach_max": round(edge + above / SPM, 2)} + + +def ladder_z(edge, below, above): + """The focus ladder the daemon builds: twelve whole half-steps from + the top of the window to the bottom.""" + 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, +# options) and the result it ends with. The press prompt is the +# 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. +SETUP = [("gantry-stops", "confirm", ["Yes", "No"]), + ("home", "choice", ["The head is home", "Release the motors"]), + ("scrap", "continue", ["Continue"]), ("thickness", "number", ["Set"])] +PRESS = ("press", "wait", []) +SCRIPT = { + "laser.focus": (SETUP + [("focus-arm", "continue", ["Continue"]), PRESS, + ("focus-pick", "multichoice", [str(i) for i in range(1, 13)])], + {"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, + "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. +WRITES = {"laser.focus": {"lens_hall_edge_z_mm": "6.96"}, "laser.floor": {"laser_floor_density": "10"}} +# The machine as the burns find it: its own edge, the window the Lens +# travel check wrote, its floor. +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"} +HOME = "The head is home" + + +class RegistrationTests(unittest.TestCase): + def setUp(self): + self.reg = catalog.load_suite() + + def test_two_live_tests_on_the_button_with_a_scrap_at_hand(self): + for tid in ("setup.focus", "setup.floor"): + t = self.reg[tid] + self.assertEqual(t.kind, "live", tid) + self.assertEqual(t.mode, "grbl", tid) + self.assertEqual(t.hardware, "takeover", tid) + self.assertEqual(list(t.actions), ["button"], tid) + self.assertEqual(list(t.hands), ["scrap"], tid) + self.assertFalse(t.fixture_runnable(("button", "lid", "interlock")), tid) + self.assertIsNone(getattr(t, "precheck", None), tid) + self.assertEqual(list(self.reg["setup.focus"].requires), + ["forgectrl.auth", "setup.check-motion", "laser.emission-witness", "setup.check-lens"]) + self.assertEqual(list(self.reg["setup.floor"].requires), + ["forgectrl.auth", "setup.check-motion", "laser.emission-witness"]) + + def test_the_sheet_is_gone(self): + 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): + for tid in ("setup.focus", "setup.floor"): + covers = set(self.reg[tid].covers) + for want in (("forgectrl", "src/wizlive.*"), ("forgectrl", "src/sheet.*"), + ("forgectrl", "src/jobstream.*"), ("forgectrl", "src/curverec.*"), + ("forgectrl", "src/font_hershey.*"), ("forgectrl", "src/lens.*"), + ("forgectrl", "src/ui/wizard.*"), ("forgectrl", "src/wizrun.h"), + ("grblhal-glowforge", "src/glowforge_laser.c")): + self.assertIn(want, covers, "%s lacks %s" % (tid, want)) + + 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")): + text = " ".join(self.reg[tid].steps) + self.assertIn(size, text, tid) + self.assertIn("back-left corner", text, tid) + self.assertIn("gantry stops", text, tid) + self.assertIn("press the machine's button once", text, tid) + + +class ScriptedDaemon: + """The wizard runner as the suite sees it: /wiz//start opens the + first prompt, each answer opens the next (a thickness out of range in + the machine's units opens it again), the press prompt 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 carries the + versions, and the preview and the program are served per burn.""" + + def __init__(self, fake): + self.fake = fake + fake.state["settings"].update(FOUND) + self.dark = {"id": "", "running": False, "phase": "", "progress": 0, "elapsed_s": 0, + "log": [], "prompt": None, "result": None, "error": "", + "shots": {"lid": False, "head": False}} + self.versions = {} + self.answers = {} # wid -> [values] + self.order = [] # the burns as started + self.aborted = [] + self.settings_at_start = {} + self.previews = [] + self.queue = [] + self.seq = 0 + self.press_seen = False + self.thickness = None + self.lens_at_start = None + self.units = None # the units the number prompts read; None = ui_units + self.start_writes = {} # wid -> settings the burn writes as it starts + self.patch = {} # wid -> result fields that differ from the script + self.stray_z = None # a Z the served program commands beyond the reach, when set + fake.on_get = self.on_get + fake.on_post = self.on_post + + def open_next(self): + if not self.queue: + return False + pid, kind, options = self.queue[0] + self.seq += 1 + self.dark["prompt"] = {"seq": self.seq, "id": pid, "kind": kind, "text": pid, "options": options} + return True + + def finish(self, wid): + s = self.fake.state["settings"] + r = dict(SCRIPT[wid][1], thickness_mm=round(self.thickness, 3)) + if wid == "laser.focus": + lens = self.lens_at_start + r["window"] = {k: lens[k] for k in ("below", "above", "reach_min", "reach_max")} + r["ladder_z"] = ladder_z(lens["edge_z"], lens["below"], lens["above"]) + r.update(self.patch.get(wid, {})) + s.update(WRITES[wid]) + self.dark.update({"running": False, "prompt": None, "progress": 100, "result": r}) + self.versions[wid] = 1 + + def on_get(self, path, q): + s = self.fake.state["settings"] + if path == "/status": + # the lens block follows the settings, as the daemon's does + return 200, dict(self.fake.state["status"], lens=lens_block(s)) + if path == "/wiz/dark": + p = self.dark.get("prompt") + if p and p["id"] == "press": + if self.press_seen: + self.press_seen = False + self.queue.pop(0) + if not self.open_next(): + self.finish(self.dark["id"]) + else: + self.press_seen = True + return 200, dict(self.dark) + if path == "/wiz": + return 200, {"versions": dict(self.versions), "recommended": [], "completed": True} + if path == "/wiz/sheet.svg": + self.previews.append(q.get("card")) + if q.get("card") not in LIVE: + return 404, {"error": "no such card"} + return 200, b"", "image/svg+xml" + if path == "/wiz/sheet.gcode": + # the text is M4 at the mark dose, as the daemon serves it; the + # head's Z is the edge, the tail goes back to it, and the focus + # ladder spans the window the settings hold now + lens = lens_block(s) + lines = ["; mock", "G0 Z%.2f" % lens["edge_z"], "M4 S400", "G0 X12 Y12", "G1 X40 Y12 F3000"] + if q.get("card") == "laser.focus": + lines += ["G0 Z%.2f" % z for z in ladder_z(lens["edge_z"], lens["below"], lens["above"])] + if self.stray_z is not None: + lines.append("G0 Z%.2f" % self.stray_z) + lines += ["M5", "G0 Z%.2f" % lens["edge_z"], "G0 X0 Y0", "M2"] + return 200, ("\n".join(lines) + "\n").encode(), "text/plain" + return None + + def on_post(self, path, form): + s = self.fake.state["settings"] + if path.startswith("/wiz/") and path.endswith("/start"): + wid = path[len("/wiz/"):-len("/start")] + if wid not in SCRIPT: + return 404, {"error": "no such wizard"} + self.order.append(wid) + self.settings_at_start[wid] = dict(s) + s.update(self.start_writes.get(wid, {})) + self.lens_at_start = lens_block(s) + self.queue = list(SCRIPT[wid][0]) + self.press_seen = False + self.dark.update({"id": wid, "running": True, "phase": "starting", "result": None, "error": ""}) + self.open_next() + return 200, {"started": True, "id": wid} + if path.startswith("/wiz/") and path.endswith("/answer"): + wid = path[len("/wiz/"):-len("/answer")] + p = self.dark.get("prompt") + if not p or form.get("seq") != str(p["seq"]): + return 409, {"error": "no such prompt is open"} + value = form.get("value") + self.answers.setdefault(wid, []).append(value) + if p["id"] == "thickness": + imperial = (self.units or s.get("ui_units")) == "imperial" + mm = float(value) * (25.4 if imperial else 1.0) + if not 0 < mm <= 30: + self.open_next() # the same prompt again, a new sequence number + return 200, {"ok": True} + self.thickness = mm + self.queue.pop(0) + if not self.open_next(): + self.finish(wid) + return 200, {"ok": True} + if path.startswith("/wiz/") and path.endswith("/abort"): + self.aborted.append(path[len("/wiz/"):-len("/abort")]) + self.dark.update({"running": False, "prompt": None, "error": "aborted"}) + return 200, {"ok": True} + if path == "/settings": + for k, v in form.items(): + if v == "": + s.pop(k, None) + else: + s[k] = v + return 200, dict(s) + return None + + +class BurnRunTests(unittest.TestCase): + def setUp(self): + self.fake = helpers.FakeForgectrl().start() + self.daemon = ScriptedDaemon(self.fake) + self.poll = setup_dark.POLL_S + setup_dark.POLL_S = 0.01 + + def tearDown(self): + setup_dark.POLL_S = self.poll + self.fake.stop() + + def run_test(self, tid): + t = catalog.load_suite()[tid] + run = Run("test", t.id, t.title) + run.unattended = True # the ready gate passes; the arm cue is a notice + t.fn(Context(run, None, t)) + return run + + def assert_found(self, keys=None): + s = self.fake.state["settings"] + for k, v in FOUND.items(): + if keys is None or k in keys: + self.assertEqual(s.get(k), v, k) + + def test_the_focus_burn_in_the_lens_travel_window(self): + run = self.run_test("setup.focus") + d = self.daemon + self.assertEqual(d.order, ["laser.focus"]) + self.assertEqual(d.answers["laser.focus"], ["Yes", HOME, "Continue", "3.2", "Continue", "11"]) + self.assertEqual(d.previews, ["laser.focus", "laser.focus"]) # before the burn, and after it + # a fresh head's focus: the edge cleared, the window the Lens travel check's + start = d.settings_at_start["laser.focus"] + self.assertNotIn("lens_hall_edge_z_mm", start) + self.assertEqual((start["lens_stop_below_steps"], start["lens_stop_above_steps"]), ("14", "20")) + self.assertEqual(run.evidence["result"]["emission"]["hv_max"], 610) + fw = run.evidence["focus_window"] + self.assertEqual(fw["window"]["below"], 14) + self.assertEqual(fw["window"]["above"], 20) + self.assertEqual(fw["edge"], "6.96") + self.assertEqual(fw["program_z_before"]["reach"], [-1.44, 10.19]) # the default edge + self.assertEqual(fw["program_z_after"]["reach"], [2.17, 13.8]) # the edge the burn wrote + self.assertEqual(len(fw["program_z_after"]["z"]), 14) # head, twelve lines, tail + self.assertEqual(run.evidence["settings_before"], + {k: FOUND[k] for k in ("lens_hall_edge_z_mm", "lens_stop_below_steps", + "lens_stop_above_steps")}) + self.assert_found() # restored + + def test_the_focus_burn_in_the_fallback_window(self): + # A head the Lens travel check has not measured: the settings hold + # no window, so the ladder runs in the fallback and the burn leaves + # the keys unset. + s = self.fake.state["settings"] + s.pop("lens_stop_below_steps") + s.pop("lens_stop_above_steps") + run = self.run_test("setup.focus") + fw = run.evidence["focus_window"] + self.assertEqual((fw["window"]["below"], fw["window"]["above"]), setup_dark.FALLBACK_WINDOW) + self.assertNotIn("lens_stop_below_steps", s) + self.assertNotIn("lens_stop_above_steps", s) + self.assertEqual(s["lens_hall_edge_z_mm"], "4.10") + + 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.assert_found() + + def test_imperial_units_answer_the_thickness_in_inches(self): + # The number prompts read ui_units: on an imperial machine the + # thickness is 0.125 in and the record carries 3.175 mm. + self.fake.state["settings"]["ui_units"] = "imperial" + run = self.run_test("setup.floor") + self.assertEqual(self.daemon.answers["laser.floor"][3], "0.125") + self.assertEqual(run.evidence["thickness_answer"], "0.125") + self.assertEqual(run.evidence["result"]["thickness_mm"], 3.175) + + def test_a_thickness_the_burn_refuses_ends_it_before_anything_fires(self): + # 3.2 read as inches is over the 30 mm cap, and the burn asks again: + # the test ends the burn rather than answer the same thing forever. + self.daemon.units = "imperial" + with self.assertRaises(Failed) as cm: + self.run_test("setup.floor") + self.assertIn("asked thickness again", str(cm.exception)) + self.assertEqual(self.daemon.answers["laser.floor"], ["Yes", HOME, "Continue", "3.2"]) + self.assertEqual(self.daemon.aborted, ["laser.floor"]) + self.assert_found() + + 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 + # by hand, which the bench never needs. + SCRIPT["laser.floor"][0].insert(1, ("home-push", "continue", ["Continue"])) + try: + with self.assertRaises(Failed) as cm: + self.run_test("setup.floor") + finally: + SCRIPT["laser.floor"][0].pop(1) + self.assertIn("asked home-push, which the test does not answer", str(cm.exception)) + self.assertEqual(self.daemon.aborted, ["laser.floor"]) + self.assert_found() + + 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}} + with self.assertRaises(Failed) as cm: + self.run_test("setup.floor") + self.assertIn("the tube current never rose", str(cm.exception)) + self.assert_found() + + def test_a_focus_burn_that_writes_the_window_fails(self): + # The focus burn writes the edge alone: the window is the Lens + # travel check's. A burn that moves it fails, and it goes back. + self.daemon.start_writes["laser.focus"] = {"lens_stop_below_steps": "16", "lens_stop_above_steps": "22"} + with self.assertRaises(Failed) as cm: + self.run_test("setup.focus") + self.assertIn("the focus burn wrote the lens window", str(cm.exception)) + self.assert_found() + + def test_a_focus_ladder_outside_the_window_held_fails(self): + self.daemon.patch["laser.focus"] = {"window": {"below": 10, "above": 12, "reach_min": -0.07, + "reach_max": 7.46}} + with self.assertRaises(Failed) as cm: + self.run_test("setup.focus") + self.assertIn("the ladder ran in 10/12", str(cm.exception)) + self.assert_found() + + def test_a_focus_burn_that_reports_a_stop_search_fails(self): + self.daemon.patch["laser.focus"] = {"stops": {"found": True, "below": 14, "above": 20}} + with self.assertRaises(Failed) as cm: + self.run_test("setup.focus") + self.assertIn("searched for the stops", str(cm.exception)) + self.assert_found() + + def test_a_program_z_beyond_the_reach_fails_before_the_burn(self): + # A served program that commands a Z the lens does not reach is + # refused before the burn starts: nothing burns. + self.daemon.stray_z = 12.6 + with self.assertRaises(Failed) as cm: + self.run_test("setup.floor") + self.assertIn("outside the lens window", str(cm.exception)) + self.assertIn("12.6", str(cm.exception)) + self.assertEqual(self.daemon.order, []) + self.assert_found() + + def test_the_result_keys_the_tests_require(self): + # every key either 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) + <= set(SCRIPT["laser.focus"][1]) | {"thickness_mm", "window", "ladder_z"}) + + +if __name__ == "__main__": + unittest.main() diff --git a/forgetest/tests/test_setup_dark.py b/forgetest/tests/test_setup_dark.py index 4bdc502..5e49e76 100644 --- a/forgetest/tests/test_setup_dark.py +++ b/forgetest/tests/test_setup_dark.py @@ -3,11 +3,13 @@ # https://community.openglow.org # SPDX-License-Identifier: MIT -"""The setup's checks in the catalog: registered as automatic tests with -no precheck, the takeover ones marked so, every covers map naming the -check runner; and the check driver itself against the fake daemon: a -start, the run polled to its end with no question asked, the result -taken, the record read at version 1, and the settings read as before.""" +"""The setup's checks in the catalog: registered as automatic tests, the +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 +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 +record read at version 1, and the settings read as before or put back +as found.""" import os import sys import unittest @@ -17,8 +19,11 @@ import helpers # noqa: E402 from forgetest import catalog # noqa: E402 from forgetest.runner import Context, Run # noqa: E402 -IDS = ["setup.check-switches", "setup.check-sensors", "setup.check-airflow", - "setup.check-cameras", "setup.check-motion", "setup.check-flow-verify"] +IDS = ["setup.check-switches", "setup.check-sensors", "setup.check-airflow", "setup.check-intakes", + "setup.check-cameras", "setup.check-motion", "setup.check-lens", "setup.check-flow-verify"] +# 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 them. +COUNTED = ("setup.check-airflow", "setup.check-intakes") class RegistrationTests(unittest.TestCase): @@ -33,18 +38,42 @@ class RegistrationTests(unittest.TestCase): for tid in ("setup.check-switches", "setup.check-cameras"): self.assertEqual(self.reg[tid].kind, "operator", tid) self.assertTrue(self.reg[tid].fixture_runnable(("button", "lid", "interlock")), tid) - for tid in ("setup.check-sensors", "setup.check-airflow", "setup.check-motion", - "setup.check-flow-verify"): + 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) def test_no_check_asks_the_operator_to_prepare_anything(self): for tid in IDS: - self.assertIsNone(getattr(self.reg[tid], "precheck", None), tid) + if tid not in COUNTED: + self.assertIsNone(getattr(self.reg[tid], "precheck", None), tid) + + def test_the_intake_count_must_be_set_before_a_check_puts_it_back(self): + fake = helpers.FakeForgectrl().start() + try: + for tid in COUNTED: + fake.state["settings"].pop("cool_intake_fans", None) + why = self.reg[tid].cannot_start() + self.assertIn("cool_intake_fans is unset", why, tid) + self.assertIn("Intake fans step", why, tid) + fake.state["settings"]["cool_intake_fans"] = "2" + self.assertIsNone(self.reg[tid].cannot_start(), tid) + finally: + fake.stop() def test_the_checks_that_take_the_machine_say_so(self): - for tid in ("setup.check-airflow", "setup.check-motion", "setup.check-flow-verify"): + for tid in ("setup.check-airflow", "setup.check-intakes", "setup.check-motion", "setup.check-lens", + "setup.check-flow-verify"): self.assertEqual(self.reg[tid].hardware, "takeover", tid) + def test_the_lens_check_covers_the_finder_and_the_frame(self): + covers = set(self.reg["setup.check-lens"].covers) + for want in (("forgectrl", "src/lensfind.*"), ("forgectrl", "src/lens.*")): + self.assertIn(want, covers) + 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): for tid in IDS: covers = set(self.reg[tid].covers) @@ -256,5 +285,184 @@ class AirflowCheckTests(unittest.TestCase): self.assert_put_back() + +class ScriptedCheck(unittest.TestCase): + """A dark check against the fake daemon: the start, one poll that sees + the run, the next that sees its end (`finish` writes the settings and + returns the result), the record at version 1, and /status's lens block + read from the settings as the daemon reads them.""" + + WID = "" + + def setUp(self): + self.fake = helpers.FakeForgectrl().start() + 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.polls = 0 + self.prompt = None # a prompt the run opens on its first poll, when set + self.answers = [] + + def on_get(path, q): + s = self.fake.state["settings"] + if path == "/wiz/dark": + if self.dark["running"]: + self.polls += 1 + if self.polls == 1 and self.prompt: + self.dark["prompt"] = dict(self.prompt, seq=1) + elif self.polls >= 2 and not self.dark["prompt"]: + self.dark.update({"running": False, "progress": 100, "result": self.finish(s)}) + self.versions[self.WID] = 1 + return 200, dict(self.dark) + if path == "/wiz": + return 200, {"versions": dict(self.versions), "recommended": [], "completed": True} + if path == "/status": + def count(key, fallback): + v = s.get(key) + return int(v) if v and v.isdigit() and 1 <= int(v) <= 40 else fallback + lens = {"below": count("lens_stop_below_steps", 10), "above": count("lens_stop_above_steps", 12)} + return 200, dict(self.fake.state["status"], lens=lens) + return None + + def on_post(path, form): + s = self.fake.state["settings"] + if path == "/wiz/%s/start" % self.WID: + self.dark.update({"id": self.WID, "running": True, "phase": "running", "prompt": None}) + return 200, {"started": True, "id": self.WID} + if path == "/wiz/%s/answer" % self.WID: + self.answers.append(dict(form)) + self.dark["prompt"] = None + return 200, {"ok": True} + if path == "/wiz/%s/abort" % self.WID: + self.dark.update({"running": False, "prompt": None, "error": "aborted"}) + return 200, {"ok": True} + if path == "/settings": + for k, v in form.items(): + if v == "": + s.pop(k, None) + else: + s[k] = v + return 200, dict(s) + return None + self.fake.on_get = on_get + self.fake.on_post = on_post + + def tearDown(self): + self.fake.stop() + + def run_check(self, tid): + t = catalog.load_suite()[tid] + run = Run("test", t.id, t.title) + t.fn(Context(run, None, t)) + return run + + +class LensCheckTests(ScriptedCheck): + """setup.check-lens's judge: the window is the stops found or the + fallback, the settings and /status hold it, and both keys go back as + found.""" + + WID = "lens.travel" + FOUND = {"lens_stop_below_steps": "12", "lens_stop_above_steps": "18"} + + def setUp(self): + super().setUp() + self.fake.state["settings"].update(self.FOUND) + self.stops = {"found": True, "below": 14, "above": 20, "contact_below": 16, "contact_above": 22, "why": ""} + self.writes = None # what the check writes; None = the window of its result + + def finish(self, s): + st = self.stops + below, above = (st["below"], st["above"]) if st["found"] else (10, 12) + w = self.writes or (below, above) + s.update({"lens_stop_below_steps": str(w[0]), "lens_stop_above_steps": str(w[1])}) + return {"z_passes": [100, 101], "stops": dict(st), + "window": {"below": below, "above": above, "found": st["found"]}, "summary": "x"} + + def assert_put_back(self): + s = self.fake.state["settings"] + for k, v in self.FOUND.items(): + self.assertEqual(s.get(k), v, k) + + def test_the_stops_found_are_the_window(self): + run = self.run_check("setup.check-lens") + self.assertEqual(run.evidence["window"]["settings"], ("14", "20")) + self.assertEqual(run.evidence["window"]["status"]["below"], 14) + self.assertEqual(run.evidence["result"]["stops"]["contact_above"], 22) + self.assert_put_back() + + def test_stops_not_found_write_the_fallback(self): + self.stops = {"found": False, "below": 0, "above": 0, "contact_below": 0, "contact_above": 0, + "why": "the lens steps do not ring clearly enough on this head"} + run = self.run_check("setup.check-lens") + self.assertEqual(run.evidence["window"]["settings"], ("10", "12")) + self.assert_put_back() + + def test_a_window_the_settings_do_not_hold_fails_and_still_restores(self): + self.writes = (12, 18) + with self.assertRaises(Exception) as cm: + self.run_check("setup.check-lens") + self.assertIn("the settings hold", str(cm.exception)) + self.assert_put_back() + + def test_a_question_fails_the_check(self): + self.prompt = {"id": "lens-stop", "kind": "continue", "text": "?", "options": ["Continue"]} + with self.assertRaises(Exception) as cm: + self.run_check("setup.check-lens") + self.assertIn("asked a question", str(cm.exception)) + self.assertEqual(self.answers, []) + self.assertIn(("/wiz/lens.travel/abort", {}), self.fake.posts) + self.assert_put_back() + + +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.""" + + WID = "airflow.intakes" + + def setUp(self): + super().setUp() + self.fake.state["settings"]["cool_intake_fans"] = "2" + self.count = 2 + + def finish(self, s): + s["cool_intake_fans"] = str(self.count) + return {"intake_fans": self.count, "intake_1_rpm": 4150.0, + "intake_2_rpm": 0.0 if self.count == 1 else 4100.0, "intake_2_silent": self.count == 1, + "counted_by": "tachs", "summary": "x"} + + 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(self.fake.state["settings"]["cool_intake_fans"], "2") + self.assertIn(("/settings", {"cool_intake_fans": "2"}), self.fake.posts) # the put-back + + def test_one_intake_fan_on_a_machine_built_with_one(self): + self.fake.state["settings"]["cool_intake_fans"] = "1" + self.count = 1 + run = self.run_check("setup.check-intakes") + self.assertEqual(run.evidence["count"]["counted"], 1) + self.assertEqual(self.fake.state["settings"]["cool_intake_fans"], "1") + + def test_a_count_other_than_the_machine_held_fails_and_puts_it_back(self): + self.count = 1 + with self.assertRaises(Exception) as cm: + self.run_check("setup.check-intakes") + self.assertIn("the tachs count 1 intake fans, the machine held 2", str(cm.exception)) + self.assertEqual(self.fake.state["settings"]["cool_intake_fans"], "2") + + def test_the_operator_question_fails_the_check(self): + # intake 1 never came up: the step asks a person to count, which + # is not a count by the tachs + self.prompt = {"id": "intake-count", "kind": "choice", "text": "how many?", "options": ["One", "Two"]} + with self.assertRaises(Exception) as cm: + self.run_check("setup.check-intakes") + 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") + + if __name__ == "__main__": unittest.main() diff --git a/forgetest/tests/test_setup_sheet.py b/forgetest/tests/test_setup_sheet.py deleted file mode 100644 index 45d0f34..0000000 --- a/forgetest/tests/test_setup_sheet.py +++ /dev/null @@ -1,385 +0,0 @@ -# Copyright 2026 514 LLC d/b/a OpenGlow -# Written by Scott Wiederhold -# https://community.openglow.org -# SPDX-License-Identifier: MIT - -"""The setup's sheet in the catalog: one live test on the button with a -piece of wood at hand, its covers map naming the live runner, the -renderer, the sender, the page, and the driver's laser module; and the -whole run against a scripted daemon: the placement's jogs and origin, -then each card's preview and program read, its start, the arm prompt and -the pick answered, the witnesses judged, the record read at version 1, -every card's evidence filed under its own key, and the settings the cards -wrote put back as found at the end.""" -import os -import sys -import unittest - -sys.path.insert(0, os.path.dirname(__file__)) -import helpers # noqa: E402 -from forgetest import catalog # noqa: E402 -from forgetest.runner import Context, Run # noqa: E402 -from forgetest.suite import setup_sheet # noqa: E402 - -LIVE = ["sheet.frame", "laser.focus", "laser.floor", "laser.dose-curve", "laser.corner", "cooling.flow-load"] -# The focus card's program is read again after it ran: the served ladder -# must then lie in the window the card wrote. -PREVIEWS = ["sheet.frame", "laser.focus", "laser.focus", "laser.floor", "laser.dose-curve", "laser.corner", - "cooling.flow-load"] -EMISSION = {"hv_max": 610, "laser_on_samples": 900, "thermopile_delta": 600, "lit_s": 30.0} -SPM = 2.922 # the lens screw, half-steps per mm ($102) -EDGE_DEFAULT = 3.35 - - -def lens_block(s): - """The /status lens block as the daemon builds it from the settings: a - stop count from 1 to 40 stands, anything else is the fallback.""" - def count(key, fallback): - try: - n = float(s.get(key, "")) - except ValueError: - return fallback, False - return (int(n), True) if 1 <= n <= 40 else (fallback, False) - try: - edge = float(s.get("lens_hall_edge_z_mm", "")) - except ValueError: - edge = EDGE_DEFAULT - below, a = count("lens_stop_below_steps", 10) - above, b = count("lens_stop_above_steps", 12) - return {"edge_z": round(edge, 2), "below": below, "above": above, "stops_found": a and b, - "reach_min": round(edge - below / SPM, 2), "reach_max": round(edge + above / SPM, 2)} - - -def ladder_z(edge, below, above): - """The focus ladder the daemon builds: twelve whole half-steps from - the top of the window to the bottom.""" - return [round(edge + round(above - (above + below) * i / 11) / SPM, 2) for i in range(12)] - - -# The window the focus card finds and writes before its controller starts. -STOPS = {"lens_stop_below_steps": "14", "lens_stop_above_steps": "20"} -FOCUS_WINDOW = {"below": 14, "above": 20, "reach_min": round(EDGE_DEFAULT - 14 / SPM, 2), - "reach_max": round(EDGE_DEFAULT + 20 / SPM, 2)} - -# Each wizard as the daemon runs it: its prompts in order (id, kind, -# options) and the result it ends with. A "jog" prompt repeats until the -# origin is set. -SCRIPT = { - "sheet.place": ([("place", "jog", ["X-10", "X+10", "Set origin"]), - ("sheet-kind", "choice", ["Full sheet", "One card"]), - ("thickness", "number", ["Set", "Skip"])], - {"origin_x": 100, "origin_y": 200, "origin_z": 0, "steps_per_mm": {"x": 40.0, "y": 40.0}, - "z_referenced": True, "thickness_mm": 3.2, "alone": False}), - "sheet.frame": ([("frame-arm", "continue", ["Continue"]), ("frame-ok", "choice", ["Yes", "No", "Again"])], - {"mark_s": 400, "mark_feed": 3000, "emission": EMISSION}), - "laser.focus": ([("focus-arm", "continue", ["Continue"]), - ("focus-pick", "multichoice", [str(i) for i in range(1, 13)]), - ("thickness", "number", ["Keep", "Set"])], - {"pick": 11, "thickness_mm": 3.2, "pick_half_steps": -11, "edge_z_mm": 6.96, - "steps_per_mm": 2.922, "max_height_mm": 13.81, - "focus_range_mm": {"min": 2.17, "max": 13.81}, - "stops": {"found": True, "below": 14, "above": 20, "contact_below": 16, - "contact_above": 22, "why": ""}, - "window": dict(FOCUS_WINDOW), "ladder_z": ladder_z(EDGE_DEFAULT, 14, 20), - "emission": EMISSION}), - "laser.floor": ([("floor-arm", "continue", ["Continue"]), - ("floor-pick", "choice", ["2", "4", "6", "8", "None"])], - {"faintest_density": 8.0, "floor_density": 10.0, "emission": EMISSION}), - "laser.dose-curve": ([("dose-arm", "continue", ["Continue"])], - {"points": [[10, 0.44], [100, 100]], "curve": "10:0.44,100:100", "emission": EMISSION}), - "laser.corner": ([("corner-arm", "continue", ["Continue"]), - ("corner-pick", "choice", ["1.00", "1.25", "1.50", "1.75", "2.00"])], - {"gamma": 1.5, "emission": EMISSION}), - "cooling.flow-load": ([("load-arm", "continue", ["Continue"])], - {"lit_s": 30.0, "dose_raw_s": 12000, "peak_c": 0.72, "k_density": 2.14e-5, - "k_cw": 2.78e-5, "emission": EMISSION}), -} -# What a card writes when it starts (the focus card's window, before its -# controller starts) and when it ends. -START_WRITES = {"laser.focus": dict(STOPS)} -WRITES = { - "laser.focus": {"lens_hall_edge_z_mm": "6.96"}, - "laser.floor": {"laser_floor_density": "10"}, - "laser.dose-curve": {"laser_dose_curve": "10:0.44,100:100"}, - "laser.corner": {"laser_corner_gamma": "1.5"}, - "cooling.flow-load": {"cool_laser_heat_density": "2.14e-5", "cool_laser_heat_cw": "2.78e-5"}, -} -FOUND = {"lens_hall_edge_z_mm": "3.35", "lens_stop_below_steps": "10", "lens_stop_above_steps": "12", - "laser_floor_density": "12", - "laser_dose_curve": "off", "laser_corner_gamma": "2", "cool_laser_heat_density": "1e-5", - "cool_laser_heat_cw": "1e-5"} - - -class RegistrationTests(unittest.TestCase): - def setUp(self): - self.t = catalog.load_suite()["setup.sheet"] - - def test_one_live_test_on_the_button_with_the_sheet_at_hand(self): - t = self.t - self.assertEqual(t.kind, "live") - self.assertEqual(t.mode, "grbl") - self.assertEqual(t.hardware, "takeover") - self.assertIn("button", t.actions) - self.assertTrue(t.hands) - self.assertFalse(t.fixture_runnable(("button", "lid", "interlock"))) - self.assertIsNone(getattr(t, "precheck", None)) - for want in ("setup.check-motion", "laser.emission-witness", "cooling.flow-verify"): - self.assertIn(want, t.requires) - - def test_no_other_sheet_test_remains(self): - ids = [i for i in catalog.load_suite() if i.startswith("setup.sheet")] - self.assertEqual(ids, ["setup.sheet"]) - - def test_covers_the_live_runner_the_renderer_the_sender_the_page_and_the_driver(self): - covers = set(self.t.covers) - for want in (("forgectrl", "src/wizlive.*"), ("forgectrl", "src/sheet.*"), ("forgectrl", "src/jobstream.*"), - ("forgectrl", "src/curverec.*"), ("forgectrl", "src/font_hershey.*"), - ("forgectrl", "src/ui/wizard.*"), ("forgectrl", "src/wizrun.h"), - ("grblhal-glowforge", "src/glowforge_laser.c")): - self.assertIn(want, covers, "lacks %s" % (want,)) - - def test_the_steps_ask_for_one_press_and_no_page(self): - text = " ".join(self.t.steps).lower() - self.assertIn("press the machine's button once", text) - self.assertNotIn("when the page", text) - self.assertEqual(len(setup_sheet.CARDS), 6) - - -class ScriptedDaemon: - """The wizard runner as the suite sees it: /wiz//start opens the - first prompt, each answer opens the next or ends the wizard with its - result and its writes, /wiz/dark reports it, GET /wiz carries the - versions, the sheet preview and program are served per card.""" - - def __init__(self, fake): - self.fake = fake - fake.state["settings"].update(FOUND) - self.dark = {"id": "", "running": False, "phase": "", "progress": 0, "elapsed_s": 0, - "log": [], "prompt": None, "result": None, "error": "", - "shots": {"lid": False, "head": False}} - self.versions = {} - self.answers = {} # wid -> [values] - self.order = [] # the wizards as started - self.settings_at_start = {} # wid -> the settings as the wizard found them - self.previews = [] - self.queue = [] - self.seq = 0 - self.stray_z = None # a Z the served program commands beyond the reach, when set - fake.on_get = self.on_get - fake.on_post = self.on_post - - def open_next(self): - if not self.queue: - return False - pid, kind, options = self.queue[0] - self.seq += 1 - self.dark["prompt"] = {"seq": self.seq, "id": pid, "kind": kind, "text": pid, "options": options} - return True - - def on_get(self, path, q): - s = self.fake.state["settings"] - if path == "/status": - # the lens block follows the settings, as the daemon's does - return 200, dict(self.fake.state["status"], lens=lens_block(s)) - if path == "/wiz/dark": - return 200, dict(self.dark) - if path == "/wiz": - return 200, {"versions": dict(self.versions), "required": [], "completed": True} - if path == "/wiz/sheet.svg": - self.previews.append(q.get("card")) - if q.get("card") not in LIVE: - return 404, {"error": "no such card"} - return 200, b"", "image/svg+xml" - if path == "/wiz/sheet.gcode": - # the frame and the text are M4 at the mark dose, as the daemon - # serves them; the head's Z is the edge, the tail goes back to - # it, and the focus card's ladder spans the window the settings - # hold now - lens = lens_block(s) - lines = ["; mock", "G0 Z%.2f" % lens["edge_z"], "M4 S400", "G0 X60 Y42", "G1 X120 Y42 F3000"] - if q.get("card") == "laser.focus": - lines += ["G0 Z%.2f" % z for z in ladder_z(lens["edge_z"], lens["below"], lens["above"])] - if self.stray_z is not None: - lines.append("G0 Z%.2f" % self.stray_z) - lines += ["M5", "G0 Z%.2f" % lens["edge_z"], "G0 X0 Y0", "M2"] - return 200, ("\n".join(lines) + "\n").encode(), "text/plain" - return None - - def on_post(self, path, form): - s = self.fake.state["settings"] - if path.startswith("/wiz/") and path.endswith("/start"): - wid = path[len("/wiz/"):-len("/start")] - if wid not in SCRIPT: - return 404, {"error": "no such wizard"} - self.order.append(wid) - self.settings_at_start[wid] = dict(s) - s.update(START_WRITES.get(wid, {})) - self.queue = list(SCRIPT[wid][0]) - self.dark.update({"id": wid, "running": True, "phase": "starting", "result": None, "error": ""}) - self.open_next() - return 200, {"started": True, "id": wid} - if path.startswith("/wiz/") and path.endswith("/answer"): - wid = path[len("/wiz/"):-len("/answer")] - p = self.dark.get("prompt") - if not p or form.get("seq") != str(p["seq"]): - return 409, {"error": "no such prompt is open"} - self.answers.setdefault(wid, []).append(form.get("value")) - if not (p["kind"] == "jog" and form.get("value") != "Set origin"): - self.queue.pop(0) - if not self.open_next(): - s.update(WRITES.get(wid, {})) - self.dark.update({"running": False, "prompt": None, "progress": 100, - "result": dict(SCRIPT[wid][1])}) - self.versions[wid] = 1 - return 200, {"ok": True} - if path == "/settings": - for k, v in form.items(): - if v == "": - s.pop(k, None) - else: - s[k] = v - return 200, dict(s) - return None - - -class SheetRunTests(unittest.TestCase): - def setUp(self): - self.fake = helpers.FakeForgectrl().start() - self.daemon = ScriptedDaemon(self.fake) - - def tearDown(self): - self.fake.stop() - - def test_the_whole_sheet_on_one_piece(self): - t = catalog.load_suite()["setup.sheet"] - run = Run("test", t.id, t.title) - run.unattended = True # the ready gate passes; the arm cue is a notice - t.fn(Context(run, None, t)) - d = self.daemon - self.assertEqual(d.order, ["sheet.place"] + LIVE) - self.assertEqual(d.answers["sheet.place"], ["X+10", "X-10", "Set origin", "Full sheet", "3.2"]) - self.assertEqual(d.answers["sheet.frame"], ["Continue", "Yes"]) - self.assertEqual(d.answers["laser.focus"], ["Continue", "11", "Keep"]) - self.assertEqual(d.answers["laser.floor"], ["Continue", "8"]) - self.assertEqual(d.answers["laser.corner"], ["Continue", "1.50"]) - self.assertEqual(d.previews, PREVIEWS) - cards = run.evidence["cards"] - self.assertEqual(sorted(cards), sorted(["sheet.place"] + LIVE)) - for wid in LIVE: - self.assertEqual(cards[wid]["emission"]["hv_max"], 610, wid) - self.assertEqual(cards[wid]["result"], SCRIPT[wid][1], wid) - self.assertNotIn("emission", run.evidence) - self.assertEqual(cards["sheet.place"]["result"]["origin_x"], 100) - self.assertEqual(run.evidence["settings_before"], FOUND) - # A fresh machine's run: the lens settings were gone before the - # frame, the frame's program ran in the fallback window, the focus - # card's window was in the settings when the floor card started, - # and the window check filed what it compared. - for k in setup_sheet.LENS_SETTINGS: - self.assertNotIn(k, d.settings_at_start["sheet.frame"], k) - self.assertEqual(run.evidence["lens_fresh"]["stops_found"], False) - self.assertEqual(run.evidence["program_z"]["sheet.frame"]["reach"], [-0.07, 7.46]) - floor_start = d.settings_at_start["laser.floor"] - self.assertEqual({k: floor_start[k] for k in setup_sheet.LENS_SETTINGS}, - {"lens_hall_edge_z_mm": "6.96", "lens_stop_below_steps": "14", "lens_stop_above_steps": "20"}) - fw = run.evidence["focus_window"] - self.assertEqual(fw["window"], FOCUS_WINDOW) - self.assertEqual(fw["settings"]["lens_stop_below_steps"], "14") - self.assertEqual(len(fw["ladder_z"]), 12) - self.assertEqual(fw["program_z_before"]["reach"], [-0.07, 7.46]) # the fallback ladder - self.assertEqual(fw["program_z_after"]["reach"], [2.17, 13.8]) # the written window at the new edge - self.assertEqual(len(fw["program_z_after"]["z"]), 14) # head, twelve lines, tail - for k, v in FOUND.items(): - self.assertEqual(self.fake.state["settings"][k], v, k) # restored - - def test_imperial_units_answer_the_thickness_in_inches(self): - # The number prompts read ui_units: on an imperial machine the - # placement's thickness is 0.125 in and the record carries 3.175 mm. - self.fake.state["settings"]["ui_units"] = "imperial" - place_prompts, place_result = SCRIPT["sheet.place"] - SCRIPT["sheet.place"] = (place_prompts, dict(place_result, thickness_mm=3.175)) - try: - t = catalog.load_suite()["setup.sheet"] - run = Run("test", t.id, t.title) - run.unattended = True - t.fn(Context(run, None, t)) - self.assertEqual(self.daemon.answers["sheet.place"][-1], "0.125") - self.assertEqual(run.evidence["thickness_answer"], "0.125") - finally: - SCRIPT["sheet.place"] = (place_prompts, place_result) - - def test_a_metric_answer_on_an_imperial_machine_fails_the_placement(self): - # The defect the bench found: 3.2 read as inches is over the 30 mm - # cap and the wizard records 0; the test must not read that as a pass. - from forgetest.runner import Failed - place_prompts, place_result = SCRIPT["sheet.place"] - SCRIPT["sheet.place"] = (place_prompts, dict(place_result, thickness_mm=0.0)) - try: - t = catalog.load_suite()["setup.sheet"] - run = Run("test", t.id, t.title) - run.unattended = True - with self.assertRaises(Failed) as cm: - t.fn(Context(run, None, t)) - self.assertIn("thickness 0.0 mm", str(cm.exception)) - self.assertEqual(self.daemon.order, ["sheet.place"]) # nothing burned - finally: - SCRIPT["sheet.place"] = (place_prompts, place_result) - - def test_a_dark_burn_fails_the_run_and_still_restores(self): - from forgetest.runner import Failed - SCRIPT_DARK = dict(SCRIPT["laser.floor"][1]) - SCRIPT_DARK["emission"] = {"hv_max": 0, "laser_on_samples": 0, "thermopile_delta": 0} - SCRIPT["laser.floor"] = (SCRIPT["laser.floor"][0], SCRIPT_DARK) - try: - t = catalog.load_suite()["setup.sheet"] - run = Run("test", t.id, t.title) - run.unattended = True - with self.assertRaises(Failed): - t.fn(Context(run, None, t)) - self.assertEqual(self.daemon.order, ["sheet.place", "sheet.frame", "laser.focus", "laser.floor"]) - for k, v in FOUND.items(): - self.assertEqual(self.fake.state["settings"][k], v, k) - finally: - SCRIPT["laser.floor"] = (SCRIPT["laser.floor"][0], dict(SCRIPT_DARK, emission=EMISSION)) - - def test_a_focus_window_the_settings_do_not_hold_fails_the_run(self): - # The defect the bench found: the focus card burned its ladder over - # the stops it found while the controller's Z limit stood on the - # settings, which did not hold them. A daemon whose focus result - # names a window the settings do not hold fails the run before the - # floor card, and the settings still go back. - from forgetest.runner import Failed - saved = dict(START_WRITES["laser.focus"]) - START_WRITES["laser.focus"] = {} - try: - t = catalog.load_suite()["setup.sheet"] - run = Run("test", t.id, t.title) - run.unattended = True - with self.assertRaises(Failed) as cm: - t.fn(Context(run, None, t)) - self.assertIn("the settings hold", str(cm.exception)) - self.assertEqual(self.daemon.order, ["sheet.place", "sheet.frame", "laser.focus"]) - for k, v in FOUND.items(): - self.assertEqual(self.fake.state["settings"][k], v, k) - finally: - START_WRITES["laser.focus"] = saved - - def test_a_program_z_beyond_the_reach_fails_before_the_burn(self): - # A served program that commands a Z the lens does not reach is - # refused before the card starts: nothing burns. - from forgetest.runner import Failed - self.daemon.stray_z = 10.6 - t = catalog.load_suite()["setup.sheet"] - run = Run("test", t.id, t.title) - run.unattended = True - with self.assertRaises(Failed) as cm: - t.fn(Context(run, None, t)) - self.assertIn("outside the lens window", str(cm.exception)) - self.assertIn("10.6", str(cm.exception)) - self.assertEqual(self.daemon.order, ["sheet.place"]) - for k, v in FOUND.items(): - self.assertEqual(self.fake.state["settings"][k], v, k) - - -if __name__ == "__main__": - unittest.main() diff --git a/forgetest/tests/test_setup_suite.py b/forgetest/tests/test_setup_suite.py index 8e5acb1..cd98eee 100644 --- a/forgetest/tests/test_setup_suite.py +++ b/forgetest/tests/test_setup_suite.py @@ -19,7 +19,7 @@ from forgetest import baseline, catalog from forgetest.runner import Context, Failed, Run from forgetest.suite import setup -IDS = ("setup.gate-blocks-controllers", "setup.override-until-reboot", +IDS = ("setup.gate-blocks-controllers", "setup.advisories-rehash", "setup.account-login", "setup.https-only-writes", "setup.ssh-until-reboot", "setup.cloud-disabled-surface", "setup.factory-return", "setup.machine-name", "setup.first-run-flow", @@ -42,6 +42,8 @@ class RegistrationTests(unittest.TestCase): self.assertEqual(kinds[tid], "auto", tid) self.assertNotIn("setup.root-ssh-refused", self.reg) # folded into ssh-until-reboot self.assertNotIn("setup.wizard-first-run", self.reg) # split into the flow and the page + self.assertNotIn("setup.override-until-reboot", self.reg) # the gate has nothing to lift + self.assertNotIn("setup.sheet", self.reg) # the two burns, each on its own scrap def test_the_record_swapping_tests_are_takeovers(self): for tid in ("setup.gate-blocks-controllers", "setup.advisories-rehash", @@ -61,13 +63,36 @@ class RegistrationTests(unittest.TestCase): self.assertIn("setup.check-sensors", self.reg["setup.mirror"].requires) def test_flag_of_reads_the_status_document(self): - w = {"required": ["airflow", {"id": "laser.floor", "reason": "the tube was replaced"}], - "recommended": [{"id": "laser.focus", "reason": "the tray was replaced"}]} - self.assertEqual(setup.flag_of(w, "required", "laser.floor"), "the tube was replaced") - self.assertEqual(setup.flag_of(w, "recommended", "laser.focus"), "the tray was replaced") - self.assertIsNone(setup.flag_of(w, "required", "airflow")) # a plain id is the table, not a flag - self.assertIsNone(setup.flag_of(w, "required", "laser.focus")) - self.assertIsNone(setup.flag_of({}, "recommended", "x")) + w = {"recommended": [{"id": "laser.floor", "reason": "the tube was replaced"}, + {"id": "laser.focus", "reason": "the tray was replaced"}, + {"id": "airflow", "reason": "not run yet"}, "stray"]} + self.assertEqual(setup.flag_of(w, "laser.floor"), "the tube was replaced") + self.assertEqual(setup.flag_of(w, "laser.focus"), "the tray was replaced") + self.assertEqual(setup.flag_of(w, "airflow"), "not run yet") + self.assertIsNone(setup.flag_of(w, "stray")) # only an entry with an id counts + self.assertIsNone(setup.flag_of(w, "cameras")) + self.assertIsNone(setup.flag_of({}, "x")) + self.assertEqual(setup.recommended(w), {"laser.floor": "the tube was replaced", + "laser.focus": "the tray was replaced", + "airflow": "not run yet"}) + + def test_the_gate_parts_and_the_optional_steps_are_never_recommended(self): + self.assertFalse(set(setup.RECOMMENDED_STEPS) & set(setup.NEVER_RECOMMENDED)) + self.assertFalse(set(setup.RECOMMENDED_STEPS) & set(setup.REQUIRED_STEPS)) + for wid in ("advisories", "account", "airflow.intakes", "laser.floor", "motion.envelope"): + self.assertIn(wid, setup.NEVER_RECOMMENDED) + for wid in ("lens.travel", "laser.focus", "airflow", "switches"): + self.assertIn(wid, setup.RECOMMENDED_STEPS) + self.assertEqual(setup.FORM_WIZARDS, ("advisories", "account", "preferences", "machine", "cloud")) + # in the order the gate's reason names them + self.assertEqual(setup.REQUIRED_STEPS, ("preferences", "machine", "cloud")) + + def test_the_gate_test_starts_from_an_open_gate_and_the_intake_tests_from_a_count(self): + reg = self.reg + self.assertIs(reg["setup.gate-blocks-controllers"].precheck, setup.gate_is_open) + from forgetest.suite import setup_dark + for tid in ("setup.check-intakes", "setup.check-airflow"): + self.assertIs(reg[tid].precheck, setup_dark.intakes_counted, tid) def test_the_first_run_flow_runs_unattended_with_the_fixture_and_the_page_walk_does_not(self): flow = self.reg["setup.first-run-flow"] @@ -89,14 +114,14 @@ class RegistrationTests(unittest.TestCase): for want in (("forgectrl", "src/ui/wizard.*"), ("forgectrl", "src/ui/md.js"), ("forgectrl", "src/tls.*")): self.assertIn(want, page, want) - def test_the_only_attended_tests_need_a_workstation_or_the_sheet(self): - # With the bench actuator up, three setup tests keep a - # person: the page walk, the print from the Glowforge app, and - # the sheet. Everything else runs from the queue. + def test_the_only_attended_tests_need_a_workstation_or_a_scrap(self): + # With the bench actuator up, four setup tests keep a person: the + # page walk, the print from the Glowforge app, and the two burns, + # each on a fresh scrap. Everything else runs from the queue. attended = sorted(tid for tid, t in self.reg.items() if tid.startswith("setup.") and t.kind != "auto" and not t.fixture_runnable(("button", "lid", "interlock"))) self.assertEqual(attended, ["setup.cloud-header-capture", "setup.first-run-page", - "setup.sheet"]) + "setup.floor", "setup.focus"]) def test_the_factory_return_never_runs_the_return(self): import inspect @@ -152,13 +177,43 @@ class RecordTests(unittest.TestCase): def test_without_wizards_keeps_consent_and_account(self): rec = setup.complete_record(self.STATUS) - rec["flags"] = {"machine": {"level": "required", "reason": "x"}} + rec["flags"] = {"machine": {"reason": "x"}} out = setup.record_without_wizards(rec) self.assertEqual(out["wizards"], {}) self.assertEqual(out["flags"], {}) self.assertEqual(out["advisories"], rec["advisories"]) self.assertEqual(out["account"], rec["account"]) - self.assertEqual(rec["flags"], {"machine": {"level": "required", "reason": "x"}}) # a copy + self.assertEqual(rec["flags"], {"machine": {"reason": "x"}}) # a copy + + def test_with_required_steps_keeps_them_and_nothing_else(self): + rec = setup.complete_record(self.STATUS) + rec["wizards"]["machine"]["result"] = {"model": "basic"} + rec["wizards"]["switches"] = {"version": 1} + rec["flags"] = {"machine": {"reason": "the head was replaced"}} + out = setup.record_with_required_steps(rec) + self.assertEqual(sorted(out["wizards"]), ["cloud", "machine", "preferences"]) + self.assertEqual(out["wizards"]["machine"]["result"], {"model": "basic"}) # kept from the record + self.assertEqual(out["wizards"]["preferences"]["version"], 1) # written at version 1 + self.assertEqual(out["flags"], {}) + self.assertEqual(out["account"], rec["account"]) + self.assertIn("switches", rec["wizards"]) # a copy + + def test_without_consent_keeps_the_account_and_the_wizards(self): + rec = setup.complete_record(self.STATUS) + out = setup.record_without_consent(rec) + self.assertEqual(out["advisories"], {}) + self.assertNotIn("acceptance", out) + self.assertEqual(out["account"], rec["account"]) + self.assertEqual(out["wizards"], rec["wizards"]) + self.assertIn("pressed_at", rec["acceptance"]) # a copy + + def test_without_account_keeps_the_consent(self): + rec = setup.complete_record(self.STATUS) + out = setup.record_without_account(rec) + self.assertNotIn("account", out) + self.assertEqual(out["advisories"], rec["advisories"]) + self.assertEqual(out["acceptance"], rec["acceptance"]) + self.assertEqual(rec["account"]["name"], "bench") # a copy def test_write_file_and_remove(self): tmp = tempfile.mkdtemp(prefix="forgetest-comm-") @@ -180,7 +235,6 @@ class RecordTests(unittest.TestCase): try: self.assertEqual(setup.record_path(), "/tmp/ffdata/setup.json") self.assertEqual(setup.users_path(), "/tmp/ffdata/users") - self.assertEqual(setup.override_path(), "/tmp/ffrun/setup-override") self.assertEqual(setup.ssh_flag_path(), "/tmp/ffrun/ssh-enabled") finally: os.environ.pop("FORGECTRL_DATA_DIR", None) @@ -266,7 +320,7 @@ class GatedSettleTests(unittest.TestCase): def test_gated_returns_at_once(self): self.fake.state["mode"] = {"mode": "grbl", "controller": "gated", "pid": 0, "motion": "unverified", - "gated": True, "why": "setup required: machine"} + "gated": True, "why": "the intake fans are not counted"} lines = [] body = baseline.Baseline(lines.append).wait_settled(timeout=8) self.assertEqual(body["controller"], "gated") @@ -275,14 +329,14 @@ class GatedSettleTests(unittest.TestCase): class FirstRunSeedTests(unittest.TestCase): """The record the first-run tests install: a first run behind the - checks and the sheet.""" + checks and the burns.""" STATUS = {"wizards": [{"id": "advisories", "version": 1, "class": "form"}, {"id": "account", "version": 1, "class": "form"}, {"id": "cloud", "version": 1, "class": "form"}, {"id": "switches", "version": 1, "class": "dark"}, {"id": "motion", "version": 2, "class": "dark"}, - {"id": "sheet.frame", "version": 1, "class": "live"}]} + {"id": "laser.focus", "version": 1, "class": "live"}]} def test_no_consent_no_account_not_complete_and_every_check_done(self): base = {"created": "2026-09-01T00:00:00Z", "completed": "2026-09-02T00:00:00Z", @@ -296,14 +350,14 @@ class FirstRunSeedTests(unittest.TestCase): self.assertEqual(rec["advisories"], {}) self.assertEqual(rec["created"], "2026-09-01T00:00:00Z") self.assertEqual(rec["flags"], {"flow_thin": True}) - self.assertEqual(sorted(rec["wizards"]), ["motion", "sheet.frame", "switches"]) + self.assertEqual(sorted(rec["wizards"]), ["laser.focus", "motion", "switches"]) self.assertEqual(rec["wizards"]["switches"]["result"], {"lid": True}) # kept from the base self.assertEqual(rec["wizards"]["motion"]["version"], 2) # written at the catalog version def test_without_a_base_record(self): rec = setup.first_run_record(self.STATUS, None) - self.assertEqual(sorted(rec["wizards"]), ["motion", "sheet.frame", "switches"]) - self.assertEqual(rec["wizards"]["sheet.frame"]["version"], 1) + self.assertEqual(sorted(rec["wizards"]), ["laser.focus", "motion", "switches"]) + self.assertEqual(rec["wizards"]["laser.focus"]["version"], 1) class SshdPolicyTests(unittest.TestCase):