From 6a434bbf772f8a2b5d81415723ed73e9fe23d9e0 Mon Sep 17 00:00:00 2001 From: ScottW514 Date: Tue, 13 Oct 2026 16:05:14 -0400 Subject: [PATCH] forgetest: nineteen tests a host test already proves leave the catalog The catalog is for what needs the machine or an operator. Each test below asserted software rules that a host test in CI already proves, read assertion by assertion next to that host test before it left; the record of that reading is kept with the audit. Safety rules (each approved for deletion by the owner on that reading): - laser.disarm-in-hold: laser_lifecycle_test test_resume_after_grace. - laser.arm-wait-lid: laser_arm_test E-G, laser_lifecycle_test test_lid_open_in_wait. - laser.port-dark: ctlport_test test_port_dark (port jogs, a long jog and its cancel, four injected words, the panel ops: no FIRE tick under an open window with M3 modal). - motion.respawn-gate: super_test A and D. - motion.interlock-cancel-home: the lifecycle loop cases, laser_arm_test G. - motion.lid-policy-hold: laser_lifecycle_test test_lid_policy_hold. - cloud.lid-during-button-wait, cloud.dark-print, cloud.verdict-refuse, cloud.paused-lid-cancel: test_machine_lid_button.py. - cooling.fan-duty-readback: cool_fanwrite_test. - cooling.intake-one-fan: cool_intake_test A-E. - exthost.hold-pause-tier: holdkeep_test, run_test.py, cool_flow_test P, holds_test. Other rules: - homing.cloud-offsets: manual_home_test [own-offsets], test_gfhome_motion; the real session stays in cloud.mode-switch. - exthost.core-range: version_test, install_test. - exthost.operator-destinations: run_test.py, netrules_test, install_test, extpkg_test. - exthost.lifecycle: run_test.py's ext.shutdown read before the stop. - exthost.mcode: the null-sink mcode_test, forgectrl mcode_test, sdk_test, install_test. - motion.soft-limits: z_envelope_test's homed cases, which now assert no step before the alarm, and envelope_test. - cooling_intake.py, extcore.py, extdest.py, extlife.py and extmcode.py go whole; homeoff.py keeps only the camera-home helpers that cloud.mode-switch, setup.check-envelope and their meta-tests use. - The helpers only the removed tests used go with them. - exthost.armed-freeze still runs the cloud suite's dark-print helper, whose latch checks still run; its description says so. It now requires cooling.floor-and-warm-up, the dark print's own prerequisite, in place of the removed cloud.dark-print. - HOST_TESTED: forgectrl src/holds.* (holds_test) and src/mcode.* (mcode_test), whose only catalog home left with these tests. - The meta-tests that replayed the removed tests go: test_cooling_intake_suite.py, the cloud replay cases and their fixture. The one that catches a false PASS stays and runs the dark print as armed-freeze does: a latch open at the button fails. - Text that named a removed test (exthost.motion-job's description and comment, test_baseline's docstring) no longer does. Proof: the coverage lint on a manifest from the recipe pins reports 0 uncovered paths and 0 host-tested row problems over 111 tests (its one failure is setup.check-lens's src/lensfind.* entry, which the forgectrl pin bump clears); pyflakes adds no finding; the gate self-check refuses an empty artifact cleanly; the forgetest host suite passes under Linux (547 tests); the catalog loads with 111 tests after armed-freeze's description edit. Acceptance: the catalog hash moves, so the next image opens a new campaign. Every remaining test in laser.py, motion.py, cloud.py, cooling.py, exthost.py, homeoff.py and bedsize.py moves its fingerprint (their modules' shared text changed); exthost.armed-freeze moves by its requires. --- forgetest/forgetest/manifest.py | 6 + forgetest/forgetest/suite/__init__.py | 5 - forgetest/forgetest/suite/bedsize.py | 2 +- forgetest/forgetest/suite/cloud.py | 268 +-------------- forgetest/forgetest/suite/cooling.py | 67 ---- forgetest/forgetest/suite/cooling_intake.py | 114 ------ forgetest/forgetest/suite/extcore.py | 117 ------- forgetest/forgetest/suite/extdest.py | 237 ------------- forgetest/forgetest/suite/exthost.py | 199 +---------- forgetest/forgetest/suite/extlife.py | 154 --------- forgetest/forgetest/suite/extmcode.py | 238 ------------- forgetest/forgetest/suite/homeoff.py | 87 +---- forgetest/forgetest/suite/laser.py | 249 +------------- forgetest/forgetest/suite/motion.py | 324 ------------------ .../tests/fixtures/gfcloud-buttonwait.log | 41 --- forgetest/tests/test_baseline.py | 6 +- forgetest/tests/test_cloud_suite.py | 197 +---------- forgetest/tests/test_cooling_intake_suite.py | 176 ---------- 18 files changed, 41 insertions(+), 2446 deletions(-) delete mode 100644 forgetest/forgetest/suite/cooling_intake.py delete mode 100644 forgetest/forgetest/suite/extcore.py delete mode 100644 forgetest/forgetest/suite/extdest.py delete mode 100644 forgetest/forgetest/suite/extlife.py delete mode 100644 forgetest/forgetest/suite/extmcode.py delete mode 100644 forgetest/tests/fixtures/gfcloud-buttonwait.log delete mode 100644 forgetest/tests/test_cooling_intake_suite.py diff --git a/forgetest/forgetest/manifest.py b/forgetest/forgetest/manifest.py index b9ff82d..87c9cf4 100644 --- a/forgetest/forgetest/manifest.py +++ b/forgetest/forgetest/manifest.py @@ -125,6 +125,12 @@ BEHAVIORAL = [ # catalog test exercises on the hardware belongs in that test's covers, not # here. HOST_TESTED = [ + # An extension package's holds as the engine reads them (forgeext's + # holdkeep_test and run_test prove the writer's side). + ("forgectrl", "src/holds.*", "forgectrl:tests/holds_test.c", "forgectrl:.github/workflows/build.yml"), + # The M-code relay to the package that answers (the controller's side is + # the null-sink mcode_test harness, under grblhal-glowforge's src/**). + ("forgectrl", "src/mcode.*", "forgectrl:tests/mcode_test.c", "forgectrl:.github/workflows/build.yml"), ] # This repository's root, when the package runs from a checkout of it; a diff --git a/forgetest/forgetest/suite/__init__.py b/forgetest/forgetest/suite/__init__.py index 01441e3..9ed31f6 100644 --- a/forgetest/forgetest/suite/__init__.py +++ b/forgetest/forgetest/suite/__init__.py @@ -15,20 +15,15 @@ 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 -from . import cooling_intake # noqa: F401,E402 from . import laser # noqa: F401,E402 from . import camera # noqa: F401,E402 from . import update # noqa: F401,E402 from . import cloud # noqa: F401,E402 # Last: it stands on the setup record's consent and on the cloud suite's dark print. from . import exthost # noqa: F401,E402 -from . import extcore # noqa: F401,E402 from . import extcall # noqa: F401,E402 -from . import extdest # noqa: F401,E402 -from . import extlife # noqa: F401,E402 from . import evmore # noqa: F401,E402 from . import extcat # noqa: F401,E402 -from . import extmcode # noqa: F401,E402 from . import extwizard # noqa: F401,E402 from . import updlock # noqa: F401,E402 from . import homeoff # noqa: F401,E402 diff --git a/forgetest/forgetest/suite/bedsize.py b/forgetest/forgetest/suite/bedsize.py index 00a6d36..7f2cebb 100644 --- a/forgetest/forgetest/suite/bedsize.py +++ b/forgetest/forgetest/suite/bedsize.py @@ -9,7 +9,7 @@ Its own module, so that no other test's fingerprint moves. The check is driven from here inside the travel the machine already knows: a few jogs out from the camera home, never toward an end, so it needs no operator. Measuring the real ends is the operator's own run of the check. The -homing is the web-service session homing.cloud-offsets runs. +homing is the web-service session cloud.mode-switch runs. """ import json diff --git a/forgetest/forgetest/suite/cloud.py b/forgetest/forgetest/suite/cloud.py index 06150cd..490baa7 100644 --- a/forgetest/forgetest/suite/cloud.py +++ b/forgetest/forgetest/suite/cloud.py @@ -1262,123 +1262,7 @@ def lid_interlock_abort_body(ctx, ev, off, job, offset): ev.get("edge_to_stop_ms")) -@test("cloud.lid-during-button-wait", title="Lid open at the cloud button prompt cancels the print", - subsystem="cloud", kind="operator", est_min=8, - covers=_MACHINE_RUN, requires=[], actions=["lid"], - steps=[OFFLINE_STEP, LID_STEP, - "The job is longer than the ring holds, so the ring is filled before the button " - "lights; that takes a minute. When the button lights white, do NOT press it - open " - "the lid, and close it when told. Nothing moves and nothing fires."], - description="A cloud print waiting for the button is canceled by the lid: the wait ends " - "with the lid named as the reason, the laser latch relocks, the armed window " - "closes, no run starts, and the job ends ':cancelled'. The job is longer than " - "the ring, so its feeder is alive through the wait with the rest of the print " - "in hand: the cancel has to stop that feeder before the park clears the ring, " - "and the ring has to stay empty afterward. A feeder left alive would refill " - "it and the park, which ignores the lid, would play the print's path.") -def lid_during_button_wait(ctx): - ev = ctx.evidence - offset = enter_offline(ctx) - job = offline_job(ctx, "wait.puls", seconds=3500) - off = Offline().__enter__() - try: - lid_during_button_wait_body(ctx, ev, off, job, offset) - finally: - off.__exit__(None, None, None) - offline_cleanup(ctx) - - -def lid_during_button_wait_body(ctx, ev, off, job, offset): - off.print_ready(9003, job) - got = wait_button_wait(ctx, offset, 180, also=["job is longer than the ring"]) - ctx.check(got["job is longer than the ring"], - "the job fit the ring: this test needs a job the ring cannot hold") - ctx.act("lid", "open", text="The button is lit: do NOT press it.", timeout=120) - relock = "button wait lid opened - relocking the laser" - got = wait_log(ctx, offset, [relock], 60) - fin = wait_action_finished(ctx, offset, "print", 60) - ev["log"] = {relock: bool(got[relock]), "print finished": message(fin)} - ctx.check(got[relock], "the lid did not end the button wait") - ctx.check(fin and CANCELLED in fin, "the print did not end ':cancelled': %s", - message(fin) or "no finish line") - # No run may start between the button wait and the cancel: the print - # itself, or a park (the head never moved, there is nothing to park). - # The service's moves BEFORE the print are legitimate runs and are - # outside this window. - lines = log_lines_since(GFCLOUD_LOG, offset) - wait_i = next((i for i, ln in enumerate(lines) if "waiting for button" in ln), None) - end_i = action_finish_index(lines, "print") - if end_i is None or wait_i is None or end_i < wait_i: - end_i = len(lines) - ran = ([ln for ln in lines[wait_i:end_i] if "starting run" in ln] - if wait_i is not None else []) - ev["runs_started_after_wait"] = len(ran) - ctx.check(not ran, "a run started after the lid-open cancel (%d)", len(ran)) - st, cs = ctx.forgectrl.get("/cool/status") - ev["armed_after"] = cs.get("armed") if isinstance(cs, dict) else None - ev["latch_locked"] = latch_locked() - ctx.check(not ev["armed_after"], "armed window still open after the cancel") - ctx.check(ev["latch_locked"], "kernel latch not locked after the cancel") - ev["button_dark"] = hw.button_lit() - ctx.check(ev["button_dark"] is False, "the button is still lit after the cancel (%s)", ev["button_dark"]) - # The park cleared the ring and nothing refilled it: the feeder is gone, - # the device is out of live-feed mode, and the program total stands at - # zero and stays there. A feeder left alive shows here as a total that - # comes back within its retry period. - total = read_program_total() - ctx.sleep(2) - later = read_program_total() - streaming = hw.sysfs_int("cnc/streaming", 0) - ev["program_total_after"] = {"first": total, "later": later} - ev["streaming_after"] = streaming - ctx.check(total == 0 and later == 0, - "the ring is not empty after the cancel (program total %s, then %s): the feeder " - "outlived the job", total, later) - ctx.check(streaming == 0, "cnc/streaming is %s after the cancel", streaming) - ctx.act("lid", "close") - settle_cloud(ctx, offset) - ev["events"] = offline_events(off) - ctx.log("PASS: lid open at the button prompt canceled the print; latch locked, armed=false, " - "button dark, ring empty with the feed stopped") - - -# ---------------------------------------------------------------- the cooling verdict - -HOLD_KEYS = ("cool_temp_min", "cool_temp_start") -REFUSE_KEYS = HOLD_KEYS + ("cloud_hold_max_s",) -HOLD_REFUSE_S = 60 # the shortest bound the client takes -HOLD_REFUSE_LINE = "cooling hold held 60 s - relocking the laser" -WARMUP_WAIT_LINE = "waiting on the cooling engine: WARMUP" - - -@test("cloud.dark-print", title="A dark cloud print: the latch locked until the run, unlocked " - "for it, locked after it", - subsystem="cloud", kind="operator", est_min=3, - covers=_MACHINE_RUN + [("python3-gfhardware", "gfhardware/coolsvc.py"), - ("forgectrl", "src/cool.*"), ("forgectrl", "src/main.c")], - requires=["cloud.lid-during-button-wait", "cooling.floor-and-warm-up"], actions=["button"], - steps=[OFFLINE_STEP, - "Bed clear (the job is dark: a 30 s square at S0, nothing fires). Press the button " - "when it lights: the print waits for the engine, runs, and finishes."], - description="The cooling-engine contract in cloud mode at its ordinary end, on the engine " - "itself: the print arms on the press, waits for the engine's acknowledgment (the " - "fans at their floors), and runs. The laser latch is locked at the button and " - "through the wait, unlocks only for the run (immediately before it starts), and " - "is locked again when the job ends; the print completes. A hold the engine never " - "clears is cloud.verdict-refuse; the engine's own warm-up release is " - "cooling.floor-and-warm-up.") -def dark_print(ctx): - ev = ctx.evidence - fc = ctx.forgectrl - offset = enter_offline(ctx) - job = offline_job(ctx, "dark.puls", seconds=30) - off = Offline().__enter__() - try: - dark_print_body(ctx, ev, off, job, offset, fc) - finally: - off.__exit__(None, None, None) - offline_cleanup(ctx) - +# ---------------------------------------------------------------- the dark print (exthost.armed-freeze) def dark_print_body(ctx, ev, off, job, offset, fc): off.print_ready(9005, job) @@ -1404,91 +1288,6 @@ def dark_print_body(ctx, ev, off, job, offset, fc): ctx.log("PASS: locked at the button, unlocked for the run, locked after it; the print completed") -@test("cloud.verdict-refuse", title="A cloud print held by the engine past the bound is canceled with " - "the latch never unlocked", - subsystem="cloud", kind="operator", est_min=4, - covers=_MACHINE_RUN + [("python3-gfhardware", "gfhardware/coolsvc.py"), - ("forgectrl", "src/cool.*")], - requires=["cloud.dark-print"], actions=["button"], - steps=[OFFLINE_STEP, - "Coolant at room temperature (10 to 30 C). The test sets the warm-up gate well above " - "the coolant and the hold bound to its minimum, and restores both. Press the button " - "when it lights: nothing runs; a minute later the print cancels by itself."], - description="The other end of the cooling-engine contract before a run: a verdict that " - "refuses fire and never clears. The start gate far above the coolant keeps the " - "armed session under the warm-up for longer than the hold bound (cloud_hold_max_s " - "at its minimum, 60 s). The client waits with the laser latch locked the whole " - "time - it is never unlocked before the verdict passes, so there is nothing to " - "relock - then cancels the print at the bound: no run, the armed window closed, " - "the print ':cancelled'. Dark by construction.") -def verdict_refuse(ctx): - ev = ctx.evidence - fc = ctx.forgectrl - before = fc.settings() - orig = {k: before.get(k, "") for k in REFUSE_KEYS} - ev["orig"] = orig - up = (fc.status().get("coolant") or {}).get("up_c") - ctx.check(up is not None and 10.0 <= up <= 30.0, "coolant outside the test's range (up_c %s)", up) - offset = enter_offline(ctx) - job = offline_job(ctx, "refuse.puls", seconds=30) - off = Offline().__enter__() - try: - verdict_refuse_body(ctx, ev, off, job, offset, fc, up, orig) - finally: - st, body = fc.post("/settings", params=orig) - ctx.log("restore: POST /settings %s -> %s", orig, st) - off.__exit__(None, None, None) - offline_cleanup(ctx) - - -def verdict_refuse_body(ctx, ev, off, job, offset, fc, up, orig): - gate = round(up + 8.0, 1) # eight degrees: the heater cannot get there in a minute - st, body = fc.post("/settings", params={"cool_temp_start": str(gate), - "cool_temp_min": orig["cool_temp_min"], - "cloud_hold_max_s": str(HOLD_REFUSE_S)}) - ctx.check(st == 200, "POST /settings cool_temp_start=%s cloud_hold_max_s=%s -> %s %s", - gate, HOLD_REFUSE_S, st, body) - ev["gate"] = gate - off.print_ready(9006, job) - wait_button_wait(ctx, offset, 120) - ev["latch_locked_at_button"] = latch_locked() - ctx.check(ev["latch_locked_at_button"], "the latch is unlocked at the button wait") - ctx.act("button", "press", text="The button is lit: press it. Nothing runs; the print cancels " - "itself a minute later.", - until=lambda: log_has(offset, WARMUP_WAIT_LINE), timeout=PRESS_TIMEOUT_S) - got = wait_log(ctx, offset, [WARMUP_WAIT_LINE], 30) - ctx.check(got[WARMUP_WAIT_LINE], "the armed print did not wait on the warm-up") - t0 = time.time() - samples = [] - ctx.notice("The engine holds the print; the client gives up at the bound. Do nothing.") - try: - while time.time() - t0 < HOLD_REFUSE_S + 30: - ctx.checkpoint() - samples.append(latch_locked()) - if log_has(offset, HOLD_REFUSE_LINE): - break - ctx.sleep(2) - finally: - ctx.clear_notice() - ev["hold_s"] = round(time.time() - t0) - ev["latch_samples"] = {"n": len(samples), "unlocked": samples.count(False)} - ctx.check(log_has(offset, HOLD_REFUSE_LINE), "the client did not cancel at the hold bound (%d s)", - ev["hold_s"]) - ctx.check(all(samples), "the latch unlocked during the hold (%d of %d samples)", - samples.count(False), len(samples)) - ctx.check(wait_print_running(ctx, offset, 1) is None, "the run started under the refused verdict") - fin = wait_action_finished(ctx, offset, "print", 120) - ev["print finished"] = message(fin) - ctx.check(fin and CANCELLED in fin, "the held print did not end ':cancelled': %s", - message(fin) or "no finish line") - st, cs = fc.get("/cool/status") - ev["armed_after"] = cs.get("armed") if isinstance(cs, dict) else None - ctx.check(not ev["armed_after"], "armed window still open after the cancel") - ctx.check(latch_locked(), "the latch is unlocked after the cancel") - ev["events"] = offline_events(off) - ctx.log("PASS: held %s s with the latch locked throughout, then canceled at the bound", ev["hold_s"]) - - @test("cloud.pause-resume", title="Button pauses and resumes a cloud print (factory backtrack + lead)", subsystem="cloud", kind="live", est_min=8, covers=_CLOUD_ALL, @@ -1722,68 +1521,3 @@ def oversize_stream_body(ctx, ev, off, job, offset, before): "stopped it, parked, relocked and reported ':cancelled'") -@test("cloud.paused-lid-cancel", title="A paused cloud print is canceled by the lid", - subsystem="cloud", kind="live", est_min=6, - covers=_MACHINE_RUN, requires=["cloud.pause-resume", "cloud.lid-interlock-abort"], - actions=["button", "lid"], - steps=[OFFLINE_STEP, ARM_STEP, LID_STEP, - "The head needs 40 mm of free +X and +Y travel.", - "Press the button to start; when asked, press it again (pause), then open the lid and " - "leave it open until the head is back; close it when told."], - description="A job paused on the button is canceled by the lid, from the state the factory " - "ends it in - there is no resume past a lid open. The same tail as every abort: " - "the motion stops, the head parks back at the job start, the latch relocks and " - "the armed window closes, the button goes dark, and the print ends ':cancelled'. " - "(The app's own cancel of a running print is exercised where a print has to be " - "ended that way: cloud.oversize-stream.)") -def paused_lid_cancel(ctx): - ev = ctx.evidence - offset = enter_offline(ctx) - job = offline_job(ctx, "paused.puls", seconds=60) - off = Offline().__enter__() - try: - paused_lid_cancel_body(ctx, ev, off, job, offset) - finally: - off.__exit__(None, None, None) - offline_cleanup(ctx) - - -def paused_lid_cancel_body(ctx, ev, off, job, offset): - # -- paused on the button, then the lid --------------------------------- - offline_start_print(ctx, off, 9005, job, offset) - ctx.act("button", "press", text="The head is moving: the press pauses the print.", - until=lambda: log_has(offset, PAUSE_LINES[0]), timeout=PRESS_TIMEOUT_S, fail=False) - got = wait_log(ctx, offset, list(PAUSE_LINES), 10) - ev["paused"] = {k: bool(v) for k, v in got.items()} - ctx.check(got["button pressed mid-run; pausing"], "the press did not pause the print") - ctx.check(got["paused at"], "the pause did not settle (no 'paused at')") - ctx.act("lid", "open", text="The print is paused: leave the lid open until the head has returned " - "to the corner.", timeout=60) - lid_stop = "lid opened mid-run; stopping motion" - got = wait_log(ctx, offset, [lid_stop, "start return home", "return home complete"], 120) - fin1 = wait_action_finished(ctx, offset, "print", 60) - ev["lid_from_pause"] = {k: message(v) for k, v in got.items()} - ev["lid_from_pause"]["print finished"] = message(fin1) - for k, v in ev["lid_from_pause"].items(): - ctx.log(" [print] %s: %s", k, "seen" if v else "MISSING") - ctx.check(got[lid_stop], "the lid did not end the paused print") - ctx.check(got["return home complete"], "the paused print did not park to completion") - ctx.check(fin1 and CANCELLED in fin1, "the print did not end ':cancelled': %s", - message(fin1) or "no finish line") - ctx.check(ctx.forgectrl.wait_idle(15, abort=ctx.aborted), "machine not idle after the park") - kpos = read_position() - ev["counters_after"] = kpos - ctx.check(kpos is not None and abs(kpos[0]) <= 3 and abs(kpos[1]) <= 3, - "the head did not come back to the job start (kernel counters %s)", kpos) - st, cs = ctx.forgectrl.get("/cool/status") - ev["armed_after"] = cs.get("armed") if isinstance(cs, dict) else None - ev["latch_locked_after"] = latch_locked() - ctx.check(not ev["armed_after"], "armed window still open after the paused print was canceled") - ctx.check(ev["latch_locked_after"], - "kernel latch not locked after the paused print was canceled") - ev["button_dark"] = hw.button_lit() - ctx.check(ev["button_dark"] is False, "the button is still lit after the cancel (%s)", ev["button_dark"]) - ctx.act("lid", "close") - settle_cloud(ctx, offset) - ev["events"] = offline_events(off) - ctx.log("PASS: a paused print canceled by the lid stopped, parked, relocked and reported ':cancelled'") diff --git a/forgetest/forgetest/suite/cooling.py b/forgetest/forgetest/suite/cooling.py index 455f0ce..acff552 100644 --- a/forgetest/forgetest/suite/cooling.py +++ b/forgetest/forgetest/suite/cooling.py @@ -1404,73 +1404,6 @@ def flow_under_load(ctx): "(dT %.1f), window closed %.1f s after M2", raw, share, rise, float(limit), float(dtc), dt) -@test("cooling.fan-duty-readback", title="A fan duty a device lost is read back and put back inside a tick", - subsystem="cooling", kind="auto", mode="grbl", est_min=2, - covers=_COOL_COVERS + [("kernel-module-glowforge", "src/head*")], - requires=["kernel.latch-locked-idle"], - description="M8 opens a run session and every fan reads its run duty back from its device. " - "The head's air-assist register is then written back to the idle duty behind the " - "engine's back (what a head reset or a dropped transaction leaves) and the exhaust " - "PWM to zero: within a few ticks each reads its run duty again, the log names each " - "loss and its put-back, and the session stays OK with no airflow fault (a fan put " - "back gets its spin-up grace). Then M9 and the idle duties.") -def fan_duty_readback(ctx): - from .motion import FORGECTRL_LOG, _log_lines, _log_offset - fc = ctx.forgectrl - ev = ctx.evidence - run = {"head/air_assist_pwm": 1023, "thermal/exhaust_pwm": 65535, "thermal/intake_pwm": 43278} - with ctx.grbl() as g: - c = _hold_session(ctx, g, fc, lambda c: c.get("phase") == "run", "readback", 20) - try: - ctx.check(c.get("phase") == "run", "no run session opened: %s", c) - ctx.sleep(2) - before = {k: hw.sysfs_int(k) for k in run} - ctx.log("duties at run: %s", before) - ev["run_duties"] = before - ctx.check(all(before.get(k) == v for k, v in run.items()), - "not every fan reads its run duty: %s", before) - at_loss = _log_offset(FORGECTRL_LOG) # the put-back lines are ones written after this - hw.sysfs_write("head/air_assist_pwm", 204) - hw.sysfs_write("thermal/exhaust_pwm", 0) - lost = {k: hw.sysfs_int(k) for k in ("head/air_assist_pwm", "thermal/exhaust_pwm")} - ctx.log("written behind the engine's back: %s", lost) - back = {} - t0 = time.time() - while time.time() - t0 < 6: - ctx.sleep(0.5) - back = {k: hw.sysfs_int(k) for k in ("head/air_assist_pwm", "thermal/exhaust_pwm")} - if back.get("head/air_assist_pwm") == 1023 and back.get("thermal/exhaust_pwm") == 65535: - break - took = round(time.time() - t0, 1) - ev["put_back"] = {"duties": back, "after_s": took} - ctx.log("put back after %s s: %s", took, back) - ctx.check(back.get("head/air_assist_pwm") == 1023, "the air-assist duty was not put back: %s", back) - ctx.check(back.get("thermal/exhaust_pwm") == 65535, "the exhaust duty was not put back: %s", back) - ctx.sleep(2) - line = "head/air_assist_pwm read 204 with 1023 commanded: put back" - ctx.wait_for(lambda: bool(_log_lines(FORGECTRL_LOG, at_loss, line)), 5, poll=0.25) - ctx.check(_log_lines(FORGECTRL_LOG, at_loss, line), - "the air-assist loss is not named in the log") - line = "thermal/exhaust_pwm read 0 with 65535 commanded: put back" - ctx.wait_for(lambda: bool(_log_lines(FORGECTRL_LOG, at_loss, line)), 5, poll=0.25) - ctx.check(_log_lines(FORGECTRL_LOG, at_loss, line), - "the exhaust loss is not named in the log") - for _ in range(8): - ctx.sleep(1) - c = _cool(fc) - if c.get("verdict") != "OK": - break - ev["verdict_after"] = {"verdict": c.get("verdict"), "hold": c.get("hold"), "reason": c.get("reason")} - ctx.check(c.get("verdict") == "OK" and not c.get("hold"), - "the session did not stay OK after the put-back: %s %s", c.get("verdict"), c.get("reason")) - finally: - g.command("M9") - _session_ended(ctx, fc, "readback") - idle = {k: hw.sysfs_int(k) for k in run} - ev["idle_duties"] = idle - ctx.log("duties at idle: %s", idle) - - @test("cooling.fail-tier-stop", title="A fail-tier verdict ends the controller: CRASH stops motion, " "locks the laser, and the supervisor restarts the controller", subsystem="cooling", kind="operator", mode="grbl", est_min=4, diff --git a/forgetest/forgetest/suite/cooling_intake.py b/forgetest/forgetest/suite/cooling_intake.py deleted file mode 100644 index 9f75ae6..0000000 --- a/forgetest/forgetest/suite/cooling_intake.py +++ /dev/null @@ -1,114 +0,0 @@ -# Copyright 2026 514 LLC d/b/a OpenGlow -# Written by Scott Wiederhold -# https://community.openglow.org -# SPDX-License-Identifier: MIT - -"""cooling.intake-one-fan - a machine built with one intake fan -(cool_intake_fans 1, which the setup's Intake fans step writes when the -operator counts one): the second -intake tach is published as not fitted and never judged, and intake 1 -still holds the intake floor. On a machine with two intake fans the -setting alone proves both halves: intake 2 turns and is not judged, and -an intake floor no fan can meet trips with intake 2 still out of it.""" -from ..catalog import test -from .cooling import _COOL_COVERS, _after_session, _cool, _fan_gate, _run_session, _set_gates - -INTAKE_KEYS = ("cool_intake_fans", "cool_tach_intake_min_rpm", "cool_fan_grace_s") -INTAKE_GRACE_S = "8" # past the intakes' 7 s to 90 percent, so only the leg's floor trips -INTAKE_TRIP_WAIT_S = 20 # grace + three ticks, with slack for the 1 Hz pipeline - - -@test("cooling.intake-one-fan", title="With one intake fan set, intake 2 is not judged and intake 1 still is", - subsystem="cooling", kind="auto", mode="grbl", est_min=3, - covers=_COOL_COVERS + [("forgectrl", "src/main.c")], requires=["cooling.fan-gate-trips"], - steps=["Machine idle, fans quiet. The test sets one intake fan, the intake floor and the spin-up " - "grace and restores all three; three short M8/M9 cycles spin the fans."], - description="cool_intake_fans 1 must take intake 2 out of the airflow gates and leave intake 1 in " - "them. With it set, a session at the cut fan profile must publish intake_2 as not fitted " - "with no floor while it reads its real speed, run OK, and judge intake_1 ok at the intake " - "floor; an intake floor at its legal maximum must then trip AIRFLOW (hold, fire blocked) " - "with intake_1 TRIPPED and intake_2 still not fitted, and the fault must end with the " - "session; restored, the next session judges intake_2 again and every fan reads at or " - "above its floor.") -def intake_one_fan(ctx): - fc = ctx.forgectrl - ev = ctx.evidence - before = fc.settings() - orig = {k: before.get(k, "") for k in INTAKE_KEYS} - ev["orig"] = orig - ctx.log("original: %s", orig) - inl = (before.get("gates") or {}).get("cool_tach_intake_min_rpm") or {} - ctx.check(inl.get("gate") == "intake" and inl.get("off") == "low", - "settings reply does not describe the intake floor as the intake gate, off at zero: %s", inl) - ctx.check(orig["cool_intake_fans"] != "1", - "cool_intake_fans is already 1: the test starts from a machine with both intakes judged") - c0 = _cool(fc) - ctx.check(c0.get("verdict") == "OK" and c0.get("gates_off") == [], - "engine is not at OK with every gate on before the test: %s", c0) - - restored = False - with ctx.grbl() as grbl: - st = grbl.status_report() - ctx.check(st["state"].startswith("Idle"), "controller is %s", st["state"]) - try: - # Leg 1: one intake fan. Intake 2 still turns on this machine; it - # is measured and published not fitted, and nothing judges it. - _set_gates(ctx, fc, {"cool_intake_fans": "1", "cool_fan_grace_s": INTAKE_GRACE_S}) - c = _run_session(ctx, grbl, fc, - lambda c: c.get("verdict") == "OK" and _fan_gate(c, "intake_1").get("state") == "ok", - "one intake fan leg", wait=INTAKE_TRIP_WAIT_S + 10) - g1, g2 = _fan_gate(c, "intake_1"), _fan_gate(c, "intake_2") - ev["one_fan"] = {"cool": c, "intake_1": g1, "intake_2": g2} - ctx.check(g2.get("state") == "not fitted" and (g2.get("floor") or 0) == 0, - "intake_2 is not published not fitted with no floor: %s", g2) - ctx.check((g2.get("reading") or 0) > 0, - "intake_2 reads %s: the test needs a machine whose second intake turns", g2.get("reading")) - ctx.check(c.get("verdict") == "OK", "the one intake fan session is not OK: %s", c) - ctx.check(g1.get("state") == "ok" and (g1.get("floor") or 0) > 0, - "intake_1 is not judged ok at the intake floor: %s", g1) - - # Leg 2: the intake floor at its legal maximum. Intake 1 is the - # one intake left in the gates; intake 2 stays out of them. - _set_gates(ctx, fc, {"cool_tach_intake_min_rpm": str(inl.get("hi"))}) - c = _run_session(ctx, grbl, fc, lambda c: c.get("verdict") == "AIRFLOW", "intake 1 trip leg", - wait=INTAKE_TRIP_WAIT_S) - g1, g2 = _fan_gate(c, "intake_1"), _fan_gate(c, "intake_2") - ev["intake_1_trip"] = {"cool": c, "intake_1": g1, "intake_2": g2} - ctx.check(c.get("verdict") == "AIRFLOW", "an intake floor of %s rpm did not trip: %s", inl.get("hi"), c) - ctx.check(c.get("fire_ok") is False and c.get("hold") is True, - "AIRFLOW without fire blocked and a hold: %s", c) - ctx.check(g1.get("state") == "TRIPPED", "intake_1 does not read TRIPPED: %s", g1) - ctx.check(g2.get("state") == "not fitted", "intake_2 was judged with one intake fan set: %s", g2) - ctx.check("intake_1" in (c.get("reason") or ""), "the reason does not name intake_1: %r", c.get("reason")) - c = _after_session(ctx, fc) - ev["intake_1_trip_after"] = c - ctx.check(c.get("verdict") == "OK" and c.get("hold") is False, - "the fan fault outlived its run session: %s", c) - - # Restore, and prove it: intake 2 judged again, every fan at or - # above its floor. The restored grace is the shipped one. - _set_gates(ctx, fc, orig) - restored = True - c = _run_session(ctx, grbl, fc, - lambda c: c.get("verdict") == "OK" and not c.get("gates_off") - and all(g.get("state") == "ok" for g in (c.get("fan_gates") or {}).values()), - "restored", wait=INTAKE_TRIP_WAIT_S + 20) - ev["restored"] = c.get("fan_gates") - ctx.check(c.get("verdict") == "OK" and c.get("gates_off") == [], - "engine did not return to OK with every gate on after the restore: %s", c) - bad = {k: g for k, g in (c.get("fan_gates") or {}).items() if g.get("state") != "ok"} - ctx.check(not bad, "fans not at or above their floors after the restore: %s", bad) - finally: - if not restored: - st, body = fc.post("/settings", params=orig) - ctx.log("restore on failure: POST /settings %s -> %s", orig, st) - try: - c = _run_session(ctx, grbl, fc, lambda c: c.get("verdict") == "OK" and not c.get("gates_off"), - "restore on failure") - ctx.log("restore on failure: engine %s gates_off %s", c.get("verdict"), c.get("gates_off")) - except Exception as e: # the original failure is the one to report - ctx.log("restore on failure: run session did not complete (%s)", e) - after = fc.settings() - ctx.check(all(after.get(k, "") == orig[k] for k in INTAKE_KEYS), - "settings not restored: %s", {k: after.get(k) for k in INTAKE_KEYS}) - ctx.check(fc.wait_idle(60, abort=ctx.aborted), "machine did not return to idle") diff --git a/forgetest/forgetest/suite/extcore.py b/forgetest/forgetest/suite/extcore.py deleted file mode 100644 index 31bac57..0000000 --- a/forgetest/forgetest/suite/extcore.py +++ /dev/null @@ -1,117 +0,0 @@ -# Copyright 2026 514 LLC d/b/a OpenGlow -# Written by Scott Wiederhold -# https://community.openglow.org -# SPDX-License-Identifier: MIT - -"""What the extension host makes of the firmware's own version file. - -A campaign only ever runs on a dev image - the release image does not -ship this suite - and a dev image stamps ${DATETIME} (dev), which is no -version at all. So the branch where a core range is actually judged -cannot be reached from here by any machine this suite runs on, and it is -not this module's to prove: that is forgeext's version_test, over the -file as each image build writes it. What is left here is the half a -machine can show, and it is the half that ties the daemon to the image: -that the host reads the image's own file, finds a build stamp, and -judges nothing rather than guessing. - -Its own module rather than a test in exthost.py: a test's fingerprint -holds the shared text of the module it lives in, and the blank lines -between tests are shared text, so a test added to exthost.py moves the -fingerprint of every other test there and costs a campaign a re-run of -each - exthost.panel-install among them, which is an operator drill with -the button held. The helpers come from exthost, and importing them puts -that module's shared text in this test's fingerprint and not the other -way round. -""" - -import json -import os -import subprocess - -from ..catalog import test -from .exthost import FWUP, _forgeext, _read, _write - - -@test("exthost.core-range", title="The host reads the image's own version file and judges by it", - subsystem="exthost", kind="auto", est_min=2, - covers=[("forgeext", "src/install.*"), ("forgeext", "src/manifest.*"), ("forgeext", "src/main.c")], - description="The firmware's version reaches the host from /etc/forgefirm-version, which the image " - "build writes. This test asks the host what it made of that file rather than reading the " - "file and believing it. On the dev image a campaign runs on, the first word is a build " - "stamp and no version: inspect answers core_checked false, and a package asking for a " - "minimum far above or a maximum far below is accepted all the same, because a range " - "cannot be judged against a stamp and the host does not guess one. A dev image that " - "began stamping something version-shaped would fail here, which is the regression this " - "test is for; the judging branch belongs to forgeext's version_test, since no image " - "carrying this suite can reach it. Nothing is installed, nothing runs, and nothing " - "moves: inspect changes nothing and the work directory goes at the end.") -def core_range(ctx): - import io - import re - import shutil - import tarfile - import tempfile - ev = ctx.evidence - - stamp = (_read("/etc/forgefirm-version", "").split() or [""])[0] - ctx.check(stamp, "/etc/forgefirm-version says nothing: the image did not stamp itself") - version = stamp[1:] if stamp[:1] == "v" and stamp[1:2].isdigit() else stamp - looks_like_a_version = bool(re.match(r"^\d+\.\d+\.\d+(-[0-9A-Za-z.-]+)?$", version)) - ctx.log("/etc/forgefirm-version: %r -> %r", stamp, version) - ev["firmware_version"] = {"file": stamp, "version": version, "looks_like_a_version": looks_like_a_version} - - # The suite ships on the dev image alone, so this is the only shape a - # campaign can meet. If it ever is not, the image changed and the - # judging path needs a proof that is not here. - ctx.check(not looks_like_a_version, - "this image stamped %r, which is a version: a campaign cannot run on a release image, so " - "either the dev image's stamp changed or this is not the image this suite belongs on", stamp) - - work = tempfile.mkdtemp(prefix="forgetest-core-") - try: - def packed(core): - """A data package that asks for nothing, with the core range given.""" - m = {"manifest": 1, "id": "org.forgetest.corerange", "name": "forgetest core range", - "version": "1.0.0", "author": "forgetest", "license": "MIT", "api": "0.1", - "runtime": "data", "capabilities": []} - if core: - m["core"] = core - payload = os.path.join(work, "payload.tar.gz") - with tarfile.open(payload, "w:gz") as t: - data = json.dumps(m).encode() - info = tarfile.TarInfo("manifest.json") - info.size, info.mode = len(data), 0o644 - t.addfile(info, io.BytesIO(data)) - conf = os.path.join(work, "fwup.conf") - _write(conf, 'meta-product = "ForgeFIRM extension"\nmeta-description = "core range"\n' - 'meta-version = "1.0.0"\nmeta-platform = "forgefirm-ext"\n' - 'file-resource payload.tar.gz {\n host-path = "%s"\n}\n' % payload) - out = os.path.join(work, "corerange.ffx") - if os.path.exists(out): - os.remove(out) - subprocess.run([FWUP, "-c", "-f", conf, "-o", out], check=True, capture_output=True, - timeout=60, cwd=work) - return out - - r = _forgeext("inspect", packed(None)) - ctx.check(r.get("ok") is True, "a package with no core range was refused: %s", r.get("error")) - ctx.check(r.get("core_checked") is False, - "the host judged a range against the stamp %r: core_checked %s", stamp, r.get("core_checked")) - ev["no_range"] = r - - # A stamp is no version, so neither end of a range is judged and - # neither refuses the package. The host says so rather than - # refusing on a comparison it cannot make. - ev["ranges"] = [] - for core in ({"min": "99.0.0"}, {"max": "0.0.1"}, {"min": "0.0.1", "max": "0.0.2"}): - r = _forgeext("inspect", packed(core)) - ctx.check(r.get("ok") is True and r.get("core_checked") is False, - "the host judged %s against the stamp anyway: %s", core, r) - ev["ranges"].append({"core": core, "inspect": r}) - - ctx.check("org.forgetest.corerange" not in - [x.get("id") for x in (_forgeext("list").get("packages") or [])], - "inspect installed the package: it must change nothing") - finally: - shutil.rmtree(work, ignore_errors=True) diff --git a/forgetest/forgetest/suite/extdest.py b/forgetest/forgetest/suite/extdest.py deleted file mode 100644 index 67c65c0..0000000 --- a/forgetest/forgetest/suite/extdest.py +++ /dev/null @@ -1,237 +0,0 @@ -# Copyright 2026 514 LLC d/b/a OpenGlow -# Written by Scott Wiederhold -# https://community.openglow.org -# SPDX-License-Identifier: MIT - -"""The destinations the operator names for a package that asks for them. - -Its own module rather than a test in exthost.py, for the reason extcore.py -gives. The package here is the reference package's id and key with a -service of its own, so exthost's put-back takes it away like the others. - -What it dials is the machine's own DNS resolver on TCP port 53: an address -every machine on a network has, that is not the machine, and that answers. -Named by address, it gives the service no resolver of its own, so the -only way through to it is the one the operator named. -""" - -import json -import os -import socket -import subprocess - -from ..catalog import test -from .exthost import (EXT_ROOT, FWUP, NFT, REF_ID, REF_KEY, _as_found, _forgeext, _put_back, _read, _svc, _tree, - _until, _write) -from .forgectrl import lan_ip -from .setup import SAFETY_PHRASE, read_file, record_path, request - -# The service: at every start, where it may connect and what one dial does. -DEST_SERVICE = r''' -import errno, json, os, socket, sys, time -data, target, port = os.environ["FFX_DATA"], sys.argv[1], int(sys.argv[2]) - - -def api(method, path): - s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - s.settimeout(5) - try: - s.connect(os.environ["FFX_API"]) - s.sendall(("%s %s HTTP/1.1\r\nHost: forgeext\r\n\r\n" % (method, path)).encode()) - buf = b"" - while True: - c = s.recv(65536) - if not c: - break - buf += c - return json.loads(buf.partition(b"\r\n\r\n")[2] or b"null") - finally: - s.close() - - -def dial(p): - s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - s.settimeout(4) - try: - s.connect((target, p)) - return "connected" - except socket.timeout: - return "timeout" - except OSError as e: - return errno.errorcode.get(e.errno, str(e.errno)) - finally: - s.close() - - -report = {"pid": os.getpid(), "destinations": (api("GET", "/v0/self") or {}).get("destinations"), - "dial": dial(port), "dial_other_port": dial(port + 1)} -with open(os.path.join(data, "report.json.new"), "w") as f: - json.dump(report, f) -os.rename(os.path.join(data, "report.json.new"), os.path.join(data, "report.json")) -print("destinations service up: %s" % report, flush=True) -while True: - time.sleep(60) -''' - - -def _resolver(): - """The first IPv4 nameserver of /etc/resolv.conf, or None.""" - for line in _read("/etc/resolv.conf").splitlines(): - parts = line.split() - if len(parts) >= 2 and parts[0] == "nameserver" and parts[1].count(".") == 3: - return parts[1] - return None - - -def _pack_dest(work, target, port): - """The reference package's id and key, asking for the operator's destinations: (archive, public key).""" - import io - import tarfile - manifest = {"manifest": 1, "id": REF_ID, "name": "forgetest destinations", "version": "1.0.0", - "author": "forgetest", "license": "MIT", "api": "0.1", "runtime": "python", - "service": {"exec": "bin/dest.py", "args": [target, str(port)]}, - "capabilities": ["net.outbound.operator"]} - payload = os.path.join(work, "payload.tar.gz") - with tarfile.open(payload, "w:gz") as t: - for name, text, mode in (("manifest.json", json.dumps(manifest), 0o644), ("bin/dest.py", DEST_SERVICE, 0o755)): - info = tarfile.TarInfo(name) - data = text.encode() - info.size, info.mode = len(data), mode - t.addfile(info, io.BytesIO(data)) - conf = os.path.join(work, "fwup.conf") - _write(conf, 'meta-product = "ForgeFIRM extension"\nmeta-description = "%s"\nmeta-version = "1.0.0"\n' - 'meta-platform = "forgefirm-ext"\nfile-resource payload.tar.gz {\n host-path = "%s"\n}\n' - % (REF_ID, payload)) - key = os.path.join(work, REF_KEY) - raw, signed = os.path.join(work, "raw.ffx"), os.path.join(work, "dest.ffx") - for cmd in ([FWUP, "-g", "-o", key], [FWUP, "-c", "-f", conf, "-o", raw], - [FWUP, "-S", "-s", key + ".priv", "-i", raw, "-o", signed]): - subprocess.run(cmd, check=True, capture_output=True, timeout=60, cwd=work) - return signed, key + ".pub" - - -@test("exthost.operator-destinations", title="The operator names where a package may connect, and takes it away", - subsystem="exthost", kind="auto", hardware="api", est_min=4, - covers=[("forgeext", "src/install.*"), ("forgeext", "src/state.*"), ("forgeext", "src/run.*"), - ("forgeext", "src/super.*"), ("forgeext", "src/api.*"), ("forgeext", "src/caps.*"), - ("forgeext", "src/manifest.*"), ("forgeext", "src/netrules.*"), ("forgeext", "src/main.c"), - ("forgectrl", "src/extpkg.*"), ("forgectrl", "src/main.c")], - requires=["exthost.service"], - description="A package that asks for net.outbound.operator may reach the places the operator names for it " - "and nothing else. Installed, it has none: its own GET /v0/self names no destination and its " - "dial to the machine's DNS resolver on TCP port 53 is refused inside the sandbox (EACCES). " - "The operator names that resolver through POST /ext/dest; the host starts the service again " - "(said in its log, and no crash), its answer names the destination, its dial connects, the " - "next port up is still refused, and the account's chain in the rule table carries the " - "address. The machine's own LAN address and loopback are refused in words, a destination " - "out of form is 400, and taking away one that was never named is refused. Taken away, the " - "service starts again with no destination, the dial is refused again, and the chain no longer " - "carries the address. The package, the key, the setting, and the setup record are put back as " - "found.") -def operator_destinations(ctx): - import shutil - import tempfile - fc = ctx.forgectrl - ev = ctx.evidence - target = _resolver() - ctx.check(target, "/etc/resolv.conf names no IPv4 resolver: there is nothing to dial") - s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - s.settimeout(4) - try: - s.connect((target, 53)) - reachable = True - except OSError as e: - reachable = str(e) - finally: - s.close() - ev["resolver"] = {"address": target, "tcp_53_from_root": reachable} - ctx.check(reachable is True, "the resolver %s does not take a TCP connection on port 53 even from outside the " - "sandbox (%s): this network cannot show a way out opening", target, reachable) - ctx.check(fc.wait_idle(timeout=30, abort=ctx.aborted), "machine not idle: settings are locked") - prior = fc.settings().get("ext_enabled") or "" - raw = read_file(record_path()) - found_tree = _tree(EXT_ROOT) - dir_mode = os.stat(os.path.dirname(EXT_ROOT)).st_mode & 0o7777 - st, body, hdrs = request(fc.base, "GET", "/advisories/extensions", headers={"Host": fc.host_header()}) - etag = hdrs.get("etag") - work = tempfile.mkdtemp(prefix="forgetest-dest.") - owner_key = os.path.join(EXT_ROOT, "keys", REF_KEY + ".pub") - report = os.path.join(EXT_ROOT, "data", REF_ID, "report.json") - where = "%s:53" % target - - def seen(): - try: - with open(report) as f: - return json.load(f) - except (OSError, ValueError): - return None - - def fresh(than): - r = seen() - return r if r and r.get("pid") != than and _svc(REF_ID).get("state") == "running" else None - - def chain_has(addr): - """Does the package's own chain let it through to addr on TCP 53? No chain is no.""" - acct = _svc(REF_ID).get("account") or "" - if not acct.startswith("ffx"): - return False - out = subprocess.run([NFT, "list", "chain", "inet", "ffx", "u%d" % (800 + int(acct[3:]))], capture_output=True, - text=True, timeout=30).stdout - return ("daddr %s tcp dport 53 accept" % addr) in out - - try: - archive, pub = _pack_dest(work, target, 53) - shutil.copy(pub, owner_key) - os.chmod(owner_key, 0o644) - r = _forgeext("install", archive, "--consent-community") - ctx.check(r.get("ok") is True, "the install of a package asking for the operator's destinations -> %s", - r.get("error")) - st, reply = fc.post("/settings", data={"ext_enabled": "1", "advisory": etag, "phrase": SAFETY_PHRASE}) - ctx.check(st == 200, "ext_enabled=1 -> %s %r", st, reply) - first = _until(ctx, lambda: fresh(None), 60) - ctx.check(first, "the service did not come up: %s", _svc(REF_ID)) - ev["none_named"] = first - ctx.log("none named: %s", first) - ctx.check(first.get("destinations") == [] and first.get("dial") == "EACCES", - "with nothing named it may reach nothing, and the sandbox refuses the dial: %s", first) - ctx.check(not chain_has(target), "the rule table carries %s before the operator named it", target) - - # Refusals, in the host's words or the relay's. - refusals = [] - for dest, want, words in (("%s:80" % lan_ip(), 409, "is this machine"), ("127.0.0.1:80", 409, "is this machine"), - ("--root", 400, "host:port"), ("%s" % target, 409, "host:port")): - st_, why = fc.post("/ext/dest", data={"id": REF_ID, "action": "add", "dest": dest}) - refusals.append([dest, st_, why]) - ctx.check(st_ == want and words in json.dumps(why), "add %s -> %s %s, wanted %s %s", dest, st_, why, want, words) - st_, why = fc.post("/ext/dest", data={"id": REF_ID, "action": "remove", "dest": where}) - refusals.append(["remove " + where, st_, why]) - ctx.check(st_ == 409 and "is not one of" in json.dumps(why), "taking away one never named -> %s %s", st_, why) - ev["refusals"] = refusals - - # Named: started again with its way out. - st_, doc = fc.post("/ext/dest", data={"id": REF_ID, "action": "add", "dest": where}) - mine = [p for p in (doc or {}).get("packages", []) if p.get("id") == REF_ID] if isinstance(doc, dict) else [] - ctx.check(st_ == 200 and mine and mine[0].get("destinations") == [where], "POST /ext/dest add %s -> %s %s", - where, st_, doc if st_ != 200 else mine) - named = _until(ctx, lambda: fresh(first.get("pid")), 60) - ev["named"] = named - ctx.log("named: %s", named) - ctx.check(named and named.get("destinations") == [where] and named.get("dial") == "connected" - and named.get("dial_other_port") == "EACCES", - "named, it is started again, names it, reaches it, and still not the next port: %s", named) - ctx.check(chain_has(target), "the rule table does not carry %s once named", target) - svc = _svc(REF_ID) - ctx.check(svc.get("state") == "running" and "quarantined" not in (svc.get("reason") or ""), - "the restart was no crash: %s", svc) - - # Taken away: started again with none. - st_, doc = fc.post("/ext/dest", data={"id": REF_ID, "action": "remove", "dest": where}) - ctx.check(st_ == 200, "POST /ext/dest remove %s -> %s %s", where, st_, doc) - gone = _until(ctx, lambda: fresh(named.get("pid") if named else None), 60) - ev["taken_away"] = gone - ctx.check(gone and gone.get("destinations") == [] and gone.get("dial") == "EACCES", - "taken away, it starts again and reaches nothing: %s", gone) - ctx.check(not chain_has(target), "the rule table still carries %s after it was taken away", target) - finally: - _put_back(ctx, fc, work, prior, etag, raw, dir_mode) - _as_found(ctx, fc, prior, raw, dir_mode, found_tree) diff --git a/forgetest/forgetest/suite/exthost.py b/forgetest/forgetest/suite/exthost.py index 4c322c9..426af2d 100644 --- a/forgetest/forgetest/suite/exthost.py +++ b/forgetest/forgetest/suite/exthost.py @@ -1214,13 +1214,14 @@ def _frozen(id_): covers=[("forgeext", "src/super.*"), ("forgeext", "src/run.*"), ("forgeext", "src/machine.*"), ("forgeext", "src/cgroup.*"), ("forgectrl", "src/cool.*"), ("forgectrl", "src/cam.*"), ("forgectrl", "src/main.c")], - requires=["exthost.service", "cloud.dark-print"], actions=["button"], + requires=["exthost.service", "cooling.floor-and-warm-up"], actions=["button"], steps=[OFFLINE_STEP, "Bed clear (the job is dark: a 30 s square at S0, nothing fires). Press the button " "when it lights."], description="The reference package runs, its heartbeat advancing twice a second, and a dark " - "cloud print (cloud.dark-print's own job and checks) opens a real armed window over " - "it. Sampled five times a second from before the button to after the end: the " + "cloud print opens a real armed window over it: the cloud suite's dark-print " + "helper, whose latch checks still run (locked at the button, unlocked for the run, " + "locked after it), and the print completes. Sampled five times a second from before the button to after the end: the " "engine's armed flag, the group's frozen state as the kernel reports it, and the " "heartbeat. The window is open for at least 10 s. From 2 s after it opens to its " "close the group reads frozen in every sample and the heartbeat does not move, the " @@ -1350,195 +1351,9 @@ def armed_freeze(ctx): # ------------------------------------------------------------- the hold -HOLDS_DIR = "/run/forgefirm/holds" REQUIRED_HOLDS = EXT_ROOT + "/required-holds" -@test("exthost.hold-pause-tier", title="A package's hold withholds fire, and the operator's exits end it", - subsystem="exthost", kind="auto", hardware="takeover", est_min=7, - covers=[("forgeext", "src/holdkeep.*"), ("forgeext", "src/run.*"), ("forgeext", "src/install.*"), - ("forgeext", "src/state.*"), ("forgeext", "src/main.c"), ("forgectrl", "src/holds.*"), - ("forgectrl", "src/cool.*")], - requires=["exthost.service"], - description="The reference package is installed with the hold grant and its hold is marked required " - "(the package is then named under required-holds for forgectrl). Until its service has " - "run healthy (60 s) the required hold stands in the host's words; from then on the host " - "keeps its hold file fresh and clear and the engine's verdict is OK. The package then raises " - "its hold itself over its API socket (POST /v0/hold): EXT with the package's own words, " - "words the form does not take refused with the hold unchanged, and cleared again. The test then " - "makes the service end at every start: the host raises the hold in its own words and GET " - "/cool/status reads verdict EXT, fire_ok false, hold true, with the reason naming the " - "package. Safe mode ends the hold within two ticks and leaving it brings the hold back; " - "marked advisory the hold is dropped and the verdict is OK; marked required again, with " - "the host suspended the file goes stale and the reason becomes that the host is not " - "answering, and resumed it is the package's again; ext_enabled=0 ends it. Nothing moves " - "and nothing fires: the verdict is read at idle. Everything is put back as " - "exthost.service puts it back, and the required holds are empty at the end.") -def hold_pause_tier(ctx): - import tempfile - fc = ctx.forgectrl - ev = ctx.evidence - ctx.check(len(_host_pids()) == 1, "the extension host is not one running process: %s", _host_pids()) - ctx.check(fc.wait_idle(timeout=30, abort=ctx.aborted), "machine not idle: settings are locked") - prior = fc.settings().get("ext_enabled") or "" - raw = read_file(record_path()) - ctx.check(raw, "no setup record at %s", record_path()) - ctx.check(not os.path.exists(SAFE_FILE), "%s exists: the machine is in safe mode", SAFE_FILE) - found_tree = _tree(EXT_ROOT) - dir_mode = os.stat(os.path.dirname(EXT_ROOT)).st_mode & 0o7777 - st, body, hdrs = request(fc.base, "GET", "/advisories/extensions", headers={"Host": fc.host_header()}) - etag = hdrs.get("etag") - ctx.check(st == 200 and etag, "GET /advisories/extensions -> %s", st) - - def cool(): - st_, c = fc.get("/cool/status") - return c if st_ == 200 and isinstance(c, dict) else {} - - def verdict_is(name, seconds, reason=None): - def ok(): - c = cool() - return c if c.get("verdict") == name and (reason is None or reason in (c.get("reason") or "")) else None - return _until(ctx, ok, seconds, poll=0.25) - - def hold_file(): - try: - h = json.loads(_read("%s/%s.json" % (HOLDS_DIR, REF_ID))) - except ValueError: - return None - h["age"] = time.monotonic() - h["ts_mono"] - return h - - start = cool() - ev["verdict_at_start"] = {k: start.get(k) for k in ("verdict", "fire_ok", "hold", "reason")} - ctx.check(start.get("verdict") == "OK", "the engine's own verdict %r stands above a hold: the test needs OK", - start.get("verdict")) - work = tempfile.mkdtemp(prefix="forgetest-ffx.") - stop_path = os.path.join(EXT_ROOT, "data", REF_ID, "stop") - why_not_running = REF_ID + ": the extension is not running" - try: - archive, pub = _pack_reference(work, lan_ip(), more_caps=["hold"]) - with open(pub, "rb") as f: - write_file(os.path.join(EXT_ROOT, "keys", REF_KEY + ".pub"), f.read()) - os.chmod(os.path.join(EXT_ROOT, "keys", REF_KEY + ".pub"), 0o644) - r = _forgeext("install", archive, "--consent-community") - ev["install_without_the_grant"] = r.get("error") - ctx.check(r.get("ok") is False and "hold" in (r.get("error") or ""), "a hold was installed without the operator's grant: %s", r) - r = _forgeext("install", archive, "--consent-community", "--grant", "hold") - ctx.check(r.get("ok") is True, "the install -> %s", r.get("error")) - ctx.check(not os.listdir(REQUIRED_HOLDS), "a granted hold is required before the operator marks it: %s", - os.listdir(REQUIRED_HOLDS)) - r = _forgeext("hold", REF_ID, "required") - ctx.check(r.get("ok") is True and os.listdir(REQUIRED_HOLDS) == [REF_ID], "marked required -> %s, named %s", - r.get("error"), os.listdir(REQUIRED_HOLDS)) - - st, reply = fc.post("/settings", data={"ext_enabled": "1", "advisory": etag, "phrase": SAFETY_PHRASE}) - ctx.check(st == 200, "ext_enabled=1 over the advisory -> %s %r", st, reply) - beat_path = os.path.join(EXT_ROOT, "data", REF_ID, "beat") - seen = set() - ctx.check(_until(ctx, lambda: seen.add(_read(beat_path)) or len(seen) >= 3, 120, poll=0.2), - "the reference service's heartbeat does not advance: %s", _svc(REF_ID) or _host_status()) - h = hold_file() - ev["running"] = {"hold_file": h, "verdict": cool().get("verdict"), "status_hold": _svc(REF_ID).get("hold")} - ctx.check(h and h["required"] is True and h["age"] < 1.5, "a running package's required hold is not fresh: %s", h) - # a required hold stands until its package has run healthy (60 s): one that ends at every - # start reads as running for a moment each time, and must not flicker clear - ctx.check(cool().get("verdict") == "EXT" and "only just started" in (cool().get("reason") or ""), - "a required hold is clear before its package has run healthy: %s", cool()) - ctx.check(verdict_is("OK", 100), "with the package running healthy the verdict is %s (%s)", cool().get("verdict"), - cool().get("reason")) - h = hold_file() - ctx.check(h and h["raised"] is False and h["age"] < 1.5, "healthy: the hold file is not fresh and clear: %s", h) - ctx.check(_svc(REF_ID).get("hold") == "required", "the status file does not say the hold is required: %s", _svc(REF_ID)) - - # the package's own word, over its API socket: raised in its words, and cleared - def tell(words): - said = os.path.join(EXT_ROOT, "data", REF_ID, "said") - if os.path.exists(said): - os.remove(said) - _write(os.path.join(EXT_ROOT, "data", REF_ID, "say"), json.dumps(words)) - ctx.check(_until(ctx, lambda: os.path.exists(said), 10), "the service did not pass %s on", words) - return json.loads(_read(said)) - - r = tell({"raised": True, "reason": "no badge presented"}) - ctx.check(r[0] == 200 and r[1] == {"raised": True, "reason": "no badge presented"}, "POST /v0/hold raise -> %s", r) - t0 = time.time() - c = verdict_is("EXT", 6, REF_ID + ": no badge presented") - ev["raised_by_the_package"] = {"after_s": round(time.time() - t0, 1), "cool": {k: (c or cool()).get(k) for k in ("verdict", "fire_ok", "hold", "reason")}} - ctx.log("the package raised its hold: %s", ev["raised_by_the_package"]) - ctx.check(c and c.get("fire_ok") is False and c.get("hold") is True, "the package's own hold does not withhold fire: %s", c or cool()) - r = tell({"raised": True, "reason": "a \"quote\""}) - ctx.check(r[0] == 400 and verdict_is("EXT", 3, REF_ID + ": no badge presented"), "words the form does not take -> %s, and the " - "hold now reads %s", r, cool().get("reason")) - r = tell({"raised": False}) - ctx.check(r[0] == 200 and r[1].get("raised") is False and verdict_is("OK", 6), "POST /v0/hold clear -> %s, verdict %s", r, - cool().get("verdict")) - - # the package can no longer speak for itself - _write(stop_path, "") - os.kill(_svc(REF_ID)["pid"], signal.SIGKILL) - t0 = time.time() - c = verdict_is("EXT", 40, why_not_running) - ev["held"] = {"after_s": round(time.time() - t0, 1), "cool": {k: (c or {}).get(k) for k in ("verdict", "fire_ok", "hold", "reason")}, - "hold_file": hold_file()} - ctx.log("the service ends at every start: %s", ev["held"]) - ctx.check(c and c.get("fire_ok") is False and c.get("hold") is True, - "the hold does not withhold fire: %s", c or cool()) - # it reads as running for a moment at every start on its way to quarantine: never clear - looks = [] - for _ in range(24): - ctx.sleep(0.5) - c = cool() - looks.append((c.get("verdict"), c.get("fire_ok"))) - ev["looks_over_the_crash_loop"] = sorted(set(looks), key=str) - ctx.check(all(v == "EXT" and ok is False for v, ok in looks), - "the required hold flickered clear while its package kept ending: %s", ev["looks_over_the_crash_loop"]) - - # the operator's exits - _write(SAFE_FILE, "") - t0 = time.time() - ctx.check(verdict_is("OK", 4), "safe mode did not end the hold within four seconds: %s", cool()) - ev["safe_mode_released_s"] = round(time.time() - t0, 1) - os.remove(SAFE_FILE) - ctx.check(verdict_is("EXT", 30, why_not_running), "out of safe mode the hold did not come back: %s", cool()) - - r = _forgeext("hold", REF_ID, "advisory") - ctx.check(r.get("ok") is True and not os.listdir(REQUIRED_HOLDS), "marked advisory -> %s, still named %s", - r.get("error"), os.listdir(REQUIRED_HOLDS)) - ctx.check(verdict_is("OK", 6), "an advisory hold of a package that is not running was not dropped: %s", cool()) - r = _forgeext("hold", REF_ID, "required") - ctx.check(r.get("ok") is True and verdict_is("EXT", 6, why_not_running), "marked required again: %s", cool()) - - # the host itself gone quiet: the file goes stale, and that is what stands - host = _host_pids()[0] - os.kill(host, signal.SIGSTOP) - try: - c = verdict_is("EXT", 8, REF_ID + ": the extension host is not answering") - ev["host_suspended"] = {k: (c or cool()).get(k) for k in ("verdict", "fire_ok", "reason")} - ctx.check(c and c.get("fire_ok") is False, "with the host suspended the required hold does not stand on its " - "stale file: %s", c or cool()) - finally: - os.kill(host, signal.SIGCONT) - ctx.check(verdict_is("EXT", 8, why_not_running), "the host resumed: the hold is not the package's again: %s", cool()) - - st, reply = fc.post("/settings", data={"ext_enabled": "0"}) - ctx.check(st == 200, "ext_enabled=0 -> %s %r", st, reply) - t0 = time.time() - ctx.check(verdict_is("OK", 4), "ext_enabled=0 did not end the hold within four seconds: %s", cool()) - ev["ext_off_released_s"] = round(time.time() - t0, 1) - ctx.log("PASS: held in %s s; safe mode released it in %s s, extensions off in %s s", ev["held"]["after_s"], - ev["safe_mode_released_s"], ev["ext_off_released_s"]) - finally: - for pid in _host_pids(): - try: - os.kill(pid, signal.SIGCONT) - except OSError: - pass - _put_back(ctx, fc, work, prior, etag, raw, dir_mode) - _as_found(ctx, fc, prior, raw, dir_mode, found_tree) - ctx.check(not os.listdir(REQUIRED_HOLDS), "a required hold is still named at the end: %s", os.listdir(REQUIRED_HOLDS)) - ctx.check(_until(ctx, lambda: cool().get("verdict") == "OK", 6), "the verdict at the end is %s", cool().get("verdict")) - - # ------------------------------------------------- the operator's door @test("exthost.events", title="The machine's events reach a package, over one subscription", @@ -1932,8 +1747,8 @@ def ui_delivery(ctx): "package, under the machine lease, and the machine's own job record names it as the " "owner and counts the program's lines. A dark program commands no laser, so no armed " "window opens and there is nothing for a press to arm: the arm gates are in force and " - "are never reached, and a job that fires is cloud.dark-print's and " - "laser.emission-witness's. Every line is sent and acknowledged, and the runner ends the " + "are never reached, and a job that fires is laser.emission-witness's. " + "Every line is sent and acknowledged, and the runner ends the " "job as one that ended without a discharge - which is what a program that commands no " "laser is, and is the outcome this test wants. The job's own emission witnesses are all " "zero. The head accelerometer sees the motion and the " @@ -2003,7 +1818,7 @@ def motion_job(ctx): # it as this package. A dark program commands no laser, so no # armed window opens and there is nothing for a press to arm - # the arm gates are in force and simply never reached. A job - # that fires is cloud.dark-print's and laser.emission-witness's. + # that fires is laser.emission-witness's. def job_rec(): st_, j_ = fc.get("/job") return j_ if isinstance(j_, dict) else {} diff --git a/forgetest/forgetest/suite/extlife.py b/forgetest/forgetest/suite/extlife.py deleted file mode 100644 index e3fbe7d..0000000 --- a/forgetest/forgetest/suite/extlife.py +++ /dev/null @@ -1,154 +0,0 @@ -# Copyright 2026 514 LLC d/b/a OpenGlow -# Written by Scott Wiederhold -# https://community.openglow.org -# SPDX-License-Identifier: MIT - -"""The host's own events: a service is told before it is stopped. - -Its own module, for the reason extcore.py gives. The package here is the -reference package's id and key with a service of its own, so exthost's -put-back takes it away like the others. -""" - -import json -import os -import subprocess -import time - -from ..catalog import test -from .exthost import (EXT_ROOT, FWUP, REF_ID, REF_KEY, _as_found, _forgeext, _put_back, _svc, _tree, _until, _write) -from .setup import SAFETY_PHRASE, read_file, record_path, request - -# The service: it follows the host's event feed and keeps every host event it reads, with the -# monotonic time it read it, in its data directory. -LIFE_SERVICE = r''' -import json, os, socket, time -data = os.environ["FFX_DATA"] - - -def api(body): - s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - s.settimeout(40) - try: - s.connect(os.environ["FFX_API"]) - payload = json.dumps(body).encode() - s.sendall(b"POST /v0/events HTTP/1.1\r\nHost: forgeext\r\nContent-Type: application/json\r\n" - b"Content-Length: %d\r\n\r\n" % len(payload) + payload) - buf = b"" - while True: - c = s.recv(65536) - if not c: - break - buf += c - return json.loads(buf.partition(b"\r\n\r\n")[2] or b"null") - finally: - s.close() - - -def keep(name, doc): - with open(os.path.join(data, name + ".new"), "w") as f: - json.dump(doc, f) - os.rename(os.path.join(data, name + ".new"), os.path.join(data, name)) - - -place = api({})["next"] -keep("ready.json", {"place": place, "t": time.monotonic()}) -got = [] -while True: - ans = api({"since": place, "wait": 20}) - for e in ans.get("events") or []: - if e.get("event", "").startswith("ext."): - got.append({"event": e["event"], "data": e.get("data"), "t": time.monotonic()}) - keep("host-events.json", got) - place = ans.get("next", place) -''' - - -def _pack_life(work): - import io - import tarfile - manifest = {"manifest": 1, "id": REF_ID, "name": "forgetest lifecycle", "version": "1.0.0", - "author": "forgetest", "license": "MIT", "api": "0.1", "runtime": "python", - "service": {"exec": "bin/life.py"}, "capabilities": ["events"]} - payload = os.path.join(work, "payload.tar.gz") - with tarfile.open(payload, "w:gz") as t: - for name, text, mode in (("manifest.json", json.dumps(manifest), 0o644), ("bin/life.py", LIFE_SERVICE, 0o755)): - info = tarfile.TarInfo(name) - data = text.encode() - info.size, info.mode = len(data), mode - t.addfile(info, io.BytesIO(data)) - conf = os.path.join(work, "fwup.conf") - _write(conf, 'meta-product = "ForgeFIRM extension"\nmeta-description = "%s"\nmeta-version = "1.0.0"\n' - 'meta-platform = "forgefirm-ext"\nfile-resource payload.tar.gz {\n host-path = "%s"\n}\n' - % (REF_ID, payload)) - key = os.path.join(work, REF_KEY) - raw, signed = os.path.join(work, "raw.ffx"), os.path.join(work, "life.ffx") - for cmd in ([FWUP, "-g", "-o", key], [FWUP, "-c", "-f", conf, "-o", raw], - [FWUP, "-S", "-s", key + ".priv", "-i", raw, "-o", signed]): - subprocess.run(cmd, check=True, capture_output=True, timeout=60, cwd=work) - return signed, key + ".pub" - - -@test("exthost.lifecycle", title="A package is told, in the feed it reads, before the host stops it", - subsystem="exthost", kind="auto", hardware="api", est_min=3, - covers=[("forgeext", "src/run.*"), ("forgeext", "src/machine.*"), ("forgeext", "src/evfeed.*"), - ("forgeext", "src/api.*")], - requires=["exthost.service"], - description="A package that holds events reads the host's own events beside the machine's. With the " - "package running and following its feed, extensions are turned off: the host puts " - "ext.shutdown into the feed with the reason (\"extensions are off (ext_enabled)\"), and " - "stops the service a second later, so that the package has read it - the event is in its " - "data directory, read before its process ended. Holds do not wait for that second: they " - "follow extensions off at once, as exthost.hold-pause-tier holds. ext.will_freeze and " - "ext.thawed, which an armed window brings, are the automation package's host test's " - "(automation_test.py, where a service with job_time.run reads both). The package, the " - "key, the setting, and the setup record are put back as found.") -def lifecycle(ctx): - import shutil - import tempfile - fc = ctx.forgectrl - ev = ctx.evidence - ctx.check(fc.wait_idle(timeout=30, abort=ctx.aborted), "machine not idle: settings are locked") - prior = fc.settings().get("ext_enabled") or "" - raw = read_file(record_path()) - found_tree = _tree(EXT_ROOT) - dir_mode = os.stat(os.path.dirname(EXT_ROOT)).st_mode & 0o7777 - st, body, hdrs = request(fc.base, "GET", "/advisories/extensions", headers={"Host": fc.host_header()}) - etag = hdrs.get("etag") - work = tempfile.mkdtemp(prefix="forgetest-life.") - owner_key = os.path.join(EXT_ROOT, "keys", REF_KEY + ".pub") - data = os.path.join(EXT_ROOT, "data", REF_ID) - - def doc(name): - try: - with open(os.path.join(data, name)) as f: - return json.load(f) - except (OSError, ValueError): - return None - - try: - archive, pub = _pack_life(work) - shutil.copy(pub, owner_key) - os.chmod(owner_key, 0o644) - r = _forgeext("install", archive, "--consent-community") - ctx.check(r.get("ok") is True, "the install -> %s", r.get("error")) - st, reply = fc.post("/settings", data={"ext_enabled": "1", "advisory": etag, "phrase": SAFETY_PHRASE}) - ctx.check(st == 200, "ext_enabled=1 -> %s %r", st, reply) - ready = _until(ctx, lambda: _svc(REF_ID).get("state") == "running" and doc("ready.json"), 60) - ctx.check(ready, "the service did not come up and follow its feed: %s", _svc(REF_ID)) - time.sleep(1.0) - pid = _svc(REF_ID).get("pid") - st, reply = fc.post("/settings", data={"ext_enabled": "0"}) - ctx.check(st == 200, "ext_enabled=0 -> %s %r", st, reply) - stopped = _until(ctx, lambda: _svc(REF_ID).get("state") != "running" and time.monotonic(), 20) - ctx.check(stopped, "the service was not stopped: %s", _svc(REF_ID)) - ctx.check(not os.path.exists("/proc/%s" % pid), "its process %s outlived the stop", pid) - got = doc("host-events.json") or [] - ev["host_events"] = got - ctx.log("what it read: %s", got) - ctx.check(len(got) >= 1 and got[0].get("event") == "ext.shutdown" - and (got[0].get("data") or {}).get("reason") == "extensions are off (ext_enabled)", - "it read ext.shutdown, with the reason, before it was stopped: %s", got) - finally: - _put_back(ctx, fc, work, prior, etag, raw, dir_mode) - _as_found(ctx, fc, prior, raw, dir_mode, found_tree) diff --git a/forgetest/forgetest/suite/extmcode.py b/forgetest/forgetest/suite/extmcode.py deleted file mode 100644 index 4423319..0000000 --- a/forgetest/forgetest/suite/extmcode.py +++ /dev/null @@ -1,238 +0,0 @@ -# Copyright 2026 514 LLC d/b/a OpenGlow -# Written by Scott Wiederhold -# https://community.openglow.org -# SPDX-License-Identifier: MIT - -"""An M-code a package answers, on the machine. - -Its own module, for the reason extcore.py gives. The package is the -reference package's id and key with a service of its own that answers -M160 on its call socket, so exthost's put-back takes it away like the -others. The jobs are dark: they move and never command the laser. -""" - -import json -import os -import subprocess - -from ..catalog import test -from .exthost import (EXT_ROOT, FWUP, REF_ID, REF_KEY, _as_found, _forgeext, _put_back, _svc, _tree, _until, _write) -from .motion import _job, _job_post, _job_wait, _words, kernel_start, kernel_xy_mm, machine_idle -from .setup import SAFETY_PHRASE, read_file, record_path, request - -# The service: it answers M160 on the call socket the host hands it (fd 4), and keeps every call it was -# asked, with the monotonic time it was asked, in its data directory. P1 takes 2 s and is done; any -# other P is refused. -MC_SERVICE = r''' -import json, os, socket, time -data = os.environ["FFX_DATA"] -lst = socket.socket(fileno=int(os.environ["FFX_CALL_FD"])) -calls = [] -while True: - c, _ = lst.accept() - try: - buf = b"" - while b"\r\n\r\n" not in buf: - k = c.recv(4096) - if not k: - break - buf += k - head, _, body = buf.partition(b"\r\n\r\n") - n = 0 - for line in head.split(b"\r\n")[1:]: - k, _, v = line.partition(b":") - if k.strip().lower() == b"content-length": - n = int(v) - while len(body) < n: - k = c.recv(4096) - if not k: - break - body += k - req = json.loads(body[:n] or b"{}") - path = head.split(b" ")[1].decode() - t0 = time.monotonic() - words = req.get("words", {}) - if path == "/mcode" and words.get("P") == 1: - time.sleep(2.0) - status, ans = 200, {"message": "forgetest P1"} - elif path == "/mcode": - status, ans = 409, {"error": "the stand-in refused"} - else: - status, ans = 404, {"error": "no such call"} - calls.append({"path": path, "req": req, "t": t0, "status": status}) - with open(os.path.join(data, "calls.json.new"), "w") as f: - json.dump(calls, f) - os.rename(os.path.join(data, "calls.json.new"), os.path.join(data, "calls.json")) - out = json.dumps(ans).encode() - c.sendall(b"HTTP/1.1 %d X\r\nContent-Type: application/json\r\nContent-Length: %d\r\nConnection: close\r\n\r\n" - % (status, len(out)) + out) - except Exception: - pass - finally: - c.close() -''' - - -def _pack_mcode(work): - import io - import tarfile - manifest = {"manifest": 1, "id": REF_ID, "name": "forgetest M-code", "version": "1.0.0", - "author": "forgetest", "license": "MIT", "api": "0.1", "runtime": "python", - "service": {"exec": "bin/mcode.py"}, "capabilities": ["mcode:160", "job_time.run"]} - payload = os.path.join(work, "payload.tar.gz") - with tarfile.open(payload, "w:gz") as t: - for name, text, mode in (("manifest.json", json.dumps(manifest), 0o644), ("bin/mcode.py", MC_SERVICE, 0o755)): - info = tarfile.TarInfo(name) - data = text.encode() - info.size, info.mode = len(data), mode - t.addfile(info, io.BytesIO(data)) - conf = os.path.join(work, "fwup.conf") - _write(conf, 'meta-product = "ForgeFIRM extension"\nmeta-description = "%s"\nmeta-version = "1.0.0"\n' - 'meta-platform = "forgefirm-ext"\nfile-resource payload.tar.gz {\n host-path = "%s"\n}\n' - % (REF_ID, payload)) - key = os.path.join(work, REF_KEY) - raw, signed = os.path.join(work, "raw.ffx"), os.path.join(work, "mcode.ffx") - for cmd in ([FWUP, "-g", "-o", key], [FWUP, "-c", "-f", conf, "-o", raw], - [FWUP, "-S", "-s", key + ".priv", "-i", raw, "-o", signed]): - subprocess.run(cmd, check=True, capture_output=True, timeout=60, cwd=work) - return signed, key + ".pub" - - -def _port_state(fc): - st, body = fc.get("/motion/state") - return body if st == 200 and isinstance(body, dict) else {} - - -@test("exthost.mcode", title="A job waits at an M-code a package answers, dark and still", - subsystem="exthost", kind="auto", mode="grbl", est_min=4, - covers=[("grblhal-glowforge", "src/glowforge_mcode.*"), ("grblhal-glowforge", "src/ctlport.*"), - ("forgectrl", "src/mcode.*"), ("forgectrl", "src/extpkg.*"), ("forgectrl", "src/grblport.*"), - ("forgeext", "src/main.c"), ("forgeext", "src/run.*"), ("forgeext", "src/caps.*"), - ("forgeext", "src/manifest.*"), ("forgeext", "src/call.*")], - requires=["exthost.service", "motion.job"], - steps=["Bed clear; the head needs 10 mm of free travel toward +X. Nobody touches the gantry."], - description="A package that asks for mcode:160 and job_time.run, the one granted, is installed and " - "turned on; its service answers POST /mcode on its call socket. The host's status names it " - "as the one that answers M160. A dark job (out 5 mm, M160 P1, back) plays through POST " - "/job: at the M-code the controller's port state names M160 with P 1, the head stands " - "still across the wait (the kernel's counters and the port's position unchanged over " - "1 s), and the service answers after 2 s; the job then goes on and ends done, every line " - "acknowledged, the head back where it began, the service asked once with the code and " - "its words, no discharge. A job that names M161, which nothing answers, fails at that " - "line with nothing moved. A job at M160 P2, which the service refuses, is held: the " - "port's state goes to Hold with the head still, and the job is aborted from there. " - "Everything is put back as exthost.service puts it back; the refusals the controller " - "makes of a table or an answer out of form, the timeout, the reset, and the barrier's " - "darkness under an open armed window are the driver's mcode_test harness's.") -def mcode(ctx): - import tempfile - fc = ctx.forgectrl - ev = ctx.evidence - machine_idle(ctx) - prior = fc.settings().get("ext_enabled") or "" - raw = read_file(record_path()) - found_tree = _tree(EXT_ROOT) - dir_mode = os.stat(os.path.dirname(EXT_ROOT)).st_mode & 0o7777 - st, body, hdrs = request(fc.base, "GET", "/advisories/extensions", headers={"Host": fc.host_header()}) - etag = hdrs.get("etag") - work = tempfile.mkdtemp(prefix="forgetest-ffx.") - owner_key = os.path.join(EXT_ROOT, "keys", REF_KEY + ".pub") - calls_file = os.path.join(EXT_ROOT, "data", REF_ID, "calls.json") - - def calls(): - try: - return json.loads(read_file(calls_file) or "[]") - except ValueError: - return [] - - try: - archive, pub = _pack_mcode(work) - import shutil - shutil.copy(pub, owner_key) - os.chmod(owner_key, 0o644) - r = _forgeext("install", archive, "--consent-community", "--grant", "job_time.run") - ctx.check(r.get("ok") is True, "the install with job_time.run granted -> %s", r.get("error")) - st, reply = fc.post("/settings", data={"ext_enabled": "1", "advisory": etag, "phrase": SAFETY_PHRASE}) - ctx.check(st == 200, "ext_enabled=1 -> %s %r", st, reply) - x = _until(ctx, lambda: _svc(REF_ID) if _svc(REF_ID).get("state") == "running" else None, 90, poll=0.5) - ctx.check(x, "the service is not running: %s", _svc(REF_ID)) - - def listed(): - try: - doc = json.loads(read_file("/run/forgefirm/ext/status.json") or "{}") - except ValueError: - return None - return doc.get("mcodes") if doc.get("mcodes") == [{"code": 160, "id": REF_ID}] else None - ev["status_mcodes"] = _until(ctx, listed, 30, poll=0.5) - ctx.check(ev["status_mcodes"], "the host's status does not name the package as the one that answers M160") - ctx.sleep(3.0) # the relay tells the controller within its 2 s - - # A dark job that waits at M160 P1 for 2 s. - x0, _y0 = kernel_start(ctx) - st, body = _job_post(fc, "(forgetest exthost.mcode)\nG21\nG91\nG1 X5 F1200\nM160 P1\nG1 X-5 F1200\nG90\n") - ctx.check(st == 200 and isinstance(body, dict) and body.get("state") == "running", - "POST /job -> %s %s", st, _words(body)[:200]) - waiting = _until(ctx, lambda: _port_state(fc) if isinstance(_port_state(fc).get("mcode"), dict) else None, - 20, poll=0.1) - ctx.check(waiting and waiting["mcode"].get("code") == 160 and waiting["mcode"].get("words") == {"P": 1}, - "the port's state does not name M160 P1 while the job waits: %s", waiting) - k1 = kernel_xy_mm(ctx) - ctx.sleep(1.0) - k2 = kernel_xy_mm(ctx) - still = _port_state(fc) - ev["wait"] = {"port": waiting, "kernel": [k1, k2], "port_after_1s": still} - ctx.log("the job waits at M160: port %s, kernel %s -> %s", waiting, k1, k2) - ctx.check(abs(k2[0] - k1[0]) < 0.001 and abs(k2[1] - k1[1]) < 0.001, - "the head moved while the job waited: %s -> %s", k1, k2) - ctx.check(still.get("mpos") == waiting.get("mpos") and still.get("state") == "Idle", - "the port's position or state moved while the job waited: %s -> %s", waiting, still) - ctx.check(abs(k1[0] - x0 - 5.0) < 0.05, "the job did not wait at the M-code 5 mm out: %.3f", k1[0] - x0) - rec, far = _job_wait(ctx, fc, 40, x0) - x1 = kernel_xy_mm(ctx)[0] - got = calls() - ev["job"] = {"record": rec, "farthest_mm": round(far, 3), "end_mm": round(x1 - x0, 3), "calls": got} - ctx.log("the job at M160 P1: %s", ev["job"]) - ctx.check(rec["state"] == "done" and rec["reason"] == "", "the job did not end well: %s", rec) - ctx.check(rec["sent"] == rec["acked"] == rec["lines"] + 1, "lines %s, sent %s, acked %s", - rec["lines"], rec["sent"], rec["acked"]) - ctx.check(rec["lit"] is False and rec["emission"]["laser_on_samples"] == 0, "a dark job's witnesses: %s", - rec["emission"]) - ctx.check(abs(x1 - x0) < 0.1, "the job did not end where it began: %.3f mm off", x1 - x0) - ctx.check(len(got) == 1 and got[0]["path"] == "/mcode" and got[0]["req"] == {"code": 160, "words": {"P": 1}} - and got[0]["status"] == 200, "the service was not asked M160 P1 once: %s", got) - - # M161: nothing answers it, and the job stops where it is parsed. - x0 = kernel_xy_mm(ctx)[0] - st, body = _job_post(fc, "G21\nG91\nM161\nG1 X5 F1200\nG1 X-5 F1200\nG90\n") - rec, far = _job_wait(ctx, fc, 20, x0) - ev["unanswered"] = {"record": rec, "farthest_mm": round(far, 3)} - ctx.log("the job at M161: %s", ev["unanswered"]) - ctx.check(rec["state"] == "failed" and "answered error:20" in (rec.get("reason") or ""), - "a job at M161 did not fail with error:20: %s", rec) - ctx.check(far < 0.01, "a job at an M-code nothing answers moved the head %.3f mm", far) - - # M160 P2: the service refuses it, and the job is held there. - x0 = kernel_xy_mm(ctx)[0] - st, body = _job_post(fc, "G21\nG91\nM160 P2\nG1 X5 F1200\nG1 X-5 F1200\nG90\n") - held = _until(ctx, lambda: _port_state(fc) if _port_state(fc).get("state") == "Hold" else None, 30, poll=0.2) - ev["refused"] = {"port": held, "calls": calls()} - ctx.log("the job at M160 P2: %s", ev["refused"]) - ctx.check(held and held.get("mcode") is None, "a job at a refused M160 is not held: %s", _port_state(fc)) - ctx.sleep(1.0) - ctx.check(abs(kernel_xy_mm(ctx)[0] - x0) < 0.01, "the held job moved the head") - st, body = fc.post("/job/abort") - ctx.check(st == 200, "POST /job/abort -> %s %s", st, _words(body)[:160]) - rec, _far = _job_wait(ctx, fc, 20, x0) - ctx.check(rec["state"] == "failed", "the held job did not end aborted: %s", rec) - if _port_state(fc).get("state") == "Alarm": - st, body = _job_post(fc, "G21\n", unlock="1") - _job_wait(ctx, fc, 20, x0) - ctx.check(_port_state(fc).get("state") == "Idle", "the controller is not idle after the abort: %s", - _port_state(fc)) - finally: - # A check that failed mid-job leaves the job running; it holds the settings until it ends. - if _job(fc).get("state") == "running": - fc.post("/job/abort") - _job_wait(ctx, fc, 20) - _put_back(ctx, fc, work, prior, etag, raw, dir_mode) - _as_found(ctx, fc, prior, raw, dir_mode, found_tree) diff --git a/forgetest/forgetest/suite/homeoff.py b/forgetest/forgetest/suite/homeoff.py index a700cb2..021aa0b 100644 --- a/forgetest/forgetest/suite/homeoff.py +++ b/forgetest/forgetest/suite/homeoff.py @@ -3,13 +3,12 @@ # https://community.openglow.org # SPDX-License-Identifier: MIT -"""A camera home's offset pair, on the machine. +"""The helpers for a test that lets the web service move the head. -Its own module, so that no other test's fingerprint moves. The homing is -the web-service session cloud.mode-switch runs. The helpers for a test -that lets the service move the head live here too: the motions ran whole, -and the head goes back to where the test found it, by the travel the -service's motions logged. No counter reading can say where that was: every +The homing is the web-service session cloud.mode-switch runs. These judge +that its motions ran whole, and take the head back to where the test found +it by the travel the service's motions logged. No counter reading can say +where that was: every service motion zeroes the step counters at its start, and a home and every controller start zero them again. """ @@ -18,13 +17,11 @@ import re import sys from ..baseline import XY_STEPS_PER_MM, counter_frame, counter_steps_per_mm, read_position -from ..catalog import test from ..runner import Failed from . import cloud as logs # the log paths, read at each call -from .cloud import _HOMING_PATH, gfhome_homing, log_lines_since, log_size -from .motion import _drop_reference, clean_slate, machine_idle, wait_idle, wait_state +from .cloud import log_lines_since, log_size +from .motion import _drop_reference, clean_slate, machine_idle, wait_idle -HOME_X, HOME_Y = 4.5, -3.25 RETURN_MAX_MM = 600.0 # no hand-back jogs farther on an axis: the bed is smaller # The client's own record of a motion's end: the kernel counters, which # the motion zeroed at its start (the "(actual/expected)" line of the same @@ -169,73 +166,3 @@ def _jog_back(ctx, rec, travel, since): _drop_reference(ctx, ctx.forgectrl) ctx.counters_rezeroed() ctx.log("head returned to where the test found it (%.3f/%.3f mm)", -dx, -dy) - - -@test("homing.cloud-offsets", title="A camera home declares its offset pair, and the envelope reaches back to a " - "negative one", - subsystem="homing", kind="auto", mode="grbl", est_min=7, - covers=_HOMING_PATH + [("grblhal-glowforge", "src/glowforge_homing.*"), - ("grblhal-glowforge", "src/glowforge_cooling.*")], - requires=["forgectrl.auth", "motion.pacing"], - steps=["Bed clear, lid closed; cloud credentials configured; the machine on the network. The head " - "travels to the camera home and back to where it was."], - description="With cloud mode on, homing_mode = gfcloud and the camera-home offsets set to (4.5, -3.25), " - "$H runs the web-service homing session. Every service motion of the session runs whole: " - "the runner reports to the cooling engine in the controller's place, and one stopped short " - "(the engine's stop, a fault) leaves the head short of the home and the position declared " - "there false. When the session ends, the controller declares the position (4.5, -3.25): a " - "camera-home offset may be negative, and the declared position is snapped to the step " - "grid. The work envelope reaches back to the home on the negative axis, since the head " - "stands there: a jog 2 mm out in +Y and 2 mm back ends at the home, accepted, and the same " - "in X. The homing mode and the offsets are put back as found, the camera home is dropped, " - "and the head is jogged back to where the test found it by the travel the session's " - "motions logged; a travel that cannot be known fails the test and moves nothing. That an " - "offset past the axis travel is refused, and the envelope's edge on the grid, are the " - "driver's unit and harness cases.") -def cloud_offsets(ctx): - fc = ctx.forgectrl - ev = ctx.evidence - s0 = fc.settings() or {} - ctx.check(s0.get("cloud_enabled") == "1", "cloud mode is off: a camera home needs it on") - found = {k: s0.get(k) or "" for k in ("homing_mode", "gfcloud_home_x", "gfcloud_home_y")} - ev["found"] = found - session_at = None - try: - for k, v in (("homing_mode", "gfcloud"), ("gfcloud_home_x", str(HOME_X)), ("gfcloud_home_y", str(HOME_Y))): - st, body = fc.post("/settings", data={k: v}) - ctx.check(st == 200, "%s=%s -> %s %s", k, v, st, body) - with ctx.grbl() as g: - session_at = session_mark() - gfhome_homing(ctx, ev, g) - judge_whole_motions(ctx, ev, session_at) - mpos = g.status_report().get("MPos") or (None, None, None) - ev["declared"] = mpos - ctx.log("the camera home declared %s", mpos) - ctx.check(mpos[0] is not None and abs(mpos[0] - HOME_X) < 0.02 and abs(mpos[1] - HOME_Y) < 0.02, - "the camera home did not declare (%.2f, %.2f): %s", HOME_X, HOME_Y, mpos) - jogs = {} - for axis, home in (("Y", HOME_Y), ("X", HOME_X)): - out = g.command("$J=G91 G21 %s2 F1000" % axis, timeout=10) - wait_state(ctx, g, "Idle", 20) - back = g.command("$J=G91 G21 %s-2 F1000" % axis, timeout=10) - wait_state(ctx, g, "Idle", 20) - at = g.status_report().get("MPos") - jogs[axis] = {"out": out[-1:], "back": back[-1:], "at": at} - ctx.check(out[-1:] == ["ok"] and back[-1:] == ["ok"], - "a jog out and back to the home in %s was refused: %s", axis, jogs[axis]) - i = "XY".index(axis) - ctx.check(at and abs(at[i] - home) < 0.02, "the jog back in %s did not end at the home: %s", axis, at) - ev["jogs"] = jogs - ctx.log("out and back to the home: %s", jogs) - finally: - # The settings are refused for a moment after the suite's Grbl client closes; each is put back when taken. - for k, v in found.items(): - def put(): - st_, body_ = (fc.post("/settings", params={k: ""}) if v == "" else fc.post("/settings", data={k: v})) - if st_ != 200: - ctx.log("restore %s=%r -> %s %s", k, v, st_, body_) - return st_ == 200 or None - took = ctx.wait_for(put, 20, poll=0.5) - ctx.log("restore %s=%r -> %s", k, v, "taken" if took is not None else "not taken") - if session_at is not None: - camera_home_return(ctx, ev, session_at) diff --git a/forgetest/forgetest/suite/laser.py b/forgetest/forgetest/suite/laser.py index 5114b7f..ac9077e 100644 --- a/forgetest/forgetest/suite/laser.py +++ b/forgetest/forgetest/suite/laser.py @@ -30,7 +30,7 @@ from ..catalog import test from .. import hw from ..runner import Failed from .motion import (kernel_xy_mm, kernel_start, check_kernel_returned, wait_state, - wait_state_text, wait_left_state, wait_idle, drain_text, Watch, machine_idle, + wait_state_text, wait_left_state, drain_text, Watch, machine_idle, _job, _job_post, _job_wait, _lease_holder, _port_state, _words) _LASER_COVERS = [("grblhal-glowforge", "src/**"), ("kernel-module-glowforge", "**"), @@ -652,86 +652,6 @@ def m5_rapid_dark(ctx): peak, hv_tail) -@test("laser.disarm-in-hold", title="Disarm grace counts down in Hold", subsystem="laser", - kind="live", mode="grbl", est_min=4, - covers=_LASER_COVERS, - requires=["laser.emission-witness"], actions=["button"], - steps=["Scrap under the head with 40 mm of free +X travel; lid closed; exhaust on.", - "Press the physical button when it lights white. Nothing else: the test holds the " - "job itself and waits about a minute for the disarm."], - description="Arm and start a +X move at S400/F300, feed-hold it after ~2 s of motion, and " - "hold: the disarm grace must count down while held and close the armed window " - "(armed -> false) without the job resuming, and the button goes dark with it.") -def disarm_in_hold(ctx): - ev = ctx.evidence - with ctx.grbl() as g, LiveJob(ctx, g): - prepare(ctx, g) - k0 = kernel_start(ctx) - ctx.ready(ARM_CUE % "40 mm +X") - stream(g, ["G91", "G21", "M4", "S400", "G1 X40 F300"]) - ctx.log("armed; waiting for motion to start (arm + your button press)...") - w = Watch(g) - # The press waits for the button to light: pressed before the job - # reaches its arm wait, it is lost, and the move never starts. - ctx.arm_press() - ctx.wait_for(w.in_state("Run"), 180) - ctx.clear_notice() - st = w.last["state"] if w.last else None - ctx.check(st and st.startswith("Run"), "motion never started (state=%s) - arm refused or no press", st) - ctx.log("moving under laser: %s; feed-hold in 2 s", st) - ctx.sleep(2) - g.realtime(ord("!")) - t1 = time.time() - while time.time() - t1 < 5: - st = g.status_report()["state"] - if st.startswith("Hold"): - break - time.sleep(0.1) - ev["held_state"] = st - ctx.log("feed-held mid-move: %s; watching the disarm grace count down IN HOLD", st) - ctx.check(st.startswith("Hold"), "feed hold did not park (state %s)", st) - t0 = time.time() - disarmed_at = None - left_hold = None - while time.time() - t0 < 120: - ctx.checkpoint() - s = sample(ctx) - held = g.status_report()["state"].startswith("Hold") - if s and not s["armed"]: - disarmed_at = time.time() - t0 - break - if not held and left_hold is None: - left_hold = g.status_report()["state"] - ctx.log("note: left Hold (state=%s) before disarm", left_hold) - time.sleep(1) - ev["disarmed_after_s"] = round(disarmed_at, 1) if disarmed_at is not None else None - ev["left_hold"] = left_hold - # recover: laser off, abort out of hold - g.command("M5", timeout=1) - g.realtime(0x18) - ctx.sleep(1) - if "Alarm" in g.status_report()["state"]: - g.command("$X") - # The hold parks the head part-way through the move and the reset - # abandons the rest of it, so the job ends wherever the feed hold - # caught it - about 11 mm in, on the bench reference. That is the - # test's own doing and it goes back: the kernel counters say how - # far the head actually went, which is what a jog has to undo. - wait_idle(ctx, g, 15) - dx, dy = kernel_xy_mm(ctx) - back_x, back_y = k0[0] - dx, k0[1] - dy - ev["left_at_mm"] = [round(dx - k0[0], 3), round(dy - k0[1], 3)] - if abs(back_x) >= 0.02 or abs(back_y) >= 0.02: - ctx.log("returning the head: %+.3f/%+.3f mm", back_x, back_y) - g.command("$J=G91X%.3fY%.3fF1200" % (back_x, back_y)) - wait_idle(ctx, g, 30) - ev["returned_mm"] = list(kernel_xy_mm(ctx)) - ctx.check(disarmed_at is not None, "still armed after 120 s in Hold") - ctx.check(left_hold is None, "the job left Hold (%s) before the disarm", left_hold) - check_button_dark(ctx, ev) - ctx.log("PASS: disarmed in Hold after %.1f s, button dark", disarmed_at) - - @test("laser.armed-kill", title="Armed kill mid-fire: the expected stop, then a SIGKILL", subsystem="laser", kind="live", mode="grbl", est_min=6, covers=_LASER_COVERS + [("forgectrl", "src/main.c")], @@ -1090,78 +1010,6 @@ def verdict_cut(ctx): DISARMED_MSG = "laser disarmed - latch locked" -@test("laser.arm-wait-lid", title="Lid open during the arm wait cancels the job", - subsystem="laser", kind="operator", mode="grbl", est_min=3, - covers=_LASER_COVERS + [("grblhal-glowforge", "src/glowforge_laser.c"), - ("grblhal-glowforge", "src/glowforge_switches.c"), - ("grblhal-glowforge", "src/glowforge_switch_map.h")], - requires=["kernel.latch-locked-idle", "motion.jog-roundtrip"], actions=["lid"], - steps=["Lid closed; nothing under the head needs to be in place - the machine will not fire.", - "When the white button lights, do NOT press it: open the lid instead, then close it " - "when told."], - description="Start a laser job so the controller unlocks the latch and lights the button, " - "then open the lid while it waits. The wait must abort with the lid named as the " - "reason, canceled with a soft reset (no alarm - nothing to unlock), the armed " - "window closed (armed -> false), the kernel latch back to locked, and no emission; " - "the controller ends Idle. No press is given, so nothing can fire.") -def arm_wait_lid(ctx): - ev = ctx.evidence - with ctx.grbl() as g, LiveJob(ctx, g): - prepare(ctx, g) - base = sample(ctx) - ctx.check(base, "forgectrl /status or /cool/status unavailable") - ctx.check(not base["armed"], "armed window already open before the job") - stream(g, ["G91", "G21", "M4", "S400", "G1 X5 F600"]) - # The prompt: the latch is unlocked and the button lit from here on. - t0 = time.time() - text = "" - while time.time() - t0 < 30 and "press the button" not in text: - ctx.checkpoint() - text += g.drain() - time.sleep(0.1) - ctx.check("press the button" in text, "no arm prompt within 30 s (job did not reach the arm)") - s = sample(ctx) - ev["armed_during_wait"] = s["armed"] if s else None - ctx.log("arm prompt seen; armed=%s; asking the operator to open the lid", ev["armed_during_wait"]) - ctx.act("lid", "open", text="The button is lit white: do NOT press it.", timeout=60) - t1 = time.time() - while time.time() - t1 < 15: - ctx.checkpoint() - text += g.drain() - if "job canceled" in text and "help]" in text: - break - time.sleep(0.1) - ev["messages"] = [ln for ln in text.splitlines() if ln.startswith("[MSG:") or ln.startswith("ALARM")] - ctx.log("controller: %s", ev["messages"]) - ctx.check("lid opened during arm - job canceled" in text, - "the lid open was not reported as canceling the arm") - ctx.check("help]" in text, "no reset banner after the lid-open cancel (it must be a clean cancel)") - ctx.check("ALARM" not in text, "an alarm was raised on the lid-open cancel") - # Armed window closed and the kernel latch locked. - t2 = time.time() - s = None - while time.time() - t2 < 10: - s = sample(ctx) - if s and not s["armed"]: - break - time.sleep(0.25) - ev["armed_after"] = s["armed"] if s else None - ilk = hw.sysfs_int("cnc/interlock_circuit") - locked = ilk is not None and bool(ilk & (1 << 3)) - ev["latch_locked"] = locked - ev["emission"] = s["emission"] if s else None - ctx.check(s and not s["armed"], "the armed window stayed open after the lid-open cancel") - ctx.check(locked, "kernel latch not locked after the lid-open cancel (interlock_circuit=%s)", ilk) - ctx.check(not ev["emission"], "emission_samples nonzero (%s) - nothing may have fired", ev["emission"]) - check_button_dark(ctx, ev) - ctx.act("lid", "close") - ctx.sleep(1) - st = g.status_report()["state"] - ev["state_after"] = st - ctx.check(st.startswith("Idle"), "controller is %s after the cancel, expected Idle (no unlock needed)", st) - ctx.log("PASS: lid open during the arm wait canceled the job (clean reset, no alarm), armed=false, latch locked") - - @test("laser.pause-resume-lid-cancel", title="One live cut: the button pauses and resumes it, the lid " "cancels it and sends the head home", subsystem="laser", kind="live", mode="grbl", est_min=7, @@ -1327,101 +1175,6 @@ def pause_resume_lid_cancel(ctx): "%.3f mm), button latch SET", zero_at, drift) -def _port_jog_sampled(ctx, fc, params, samples, timeout=30.0): - """One jog through the controller port, sampled at 8 Hz until the port - says it is over. Returns the HTTP status and body of the jog request.""" - st, body = fc.post("/motion/jog", params=params) - t0 = time.time() - while time.time() - t0 < timeout: - ctx.checkpoint() - smp = sample(ctx) - if smp: - samples.append(smp) - try: - ps = fc.get("/motion/state")[1] - except hw.HwError: - ps = None - if st != 200 or (isinstance(ps, dict) and ps.get("state") == "Idle" and not ps.get("port_jog")): - break - time.sleep(0.125) - return st, body - - -@test("laser.port-dark", title="A panel jog under an open armed window ships dark", - subsystem="laser", kind="live", mode="grbl", est_min=3, - covers=_LASER_COVERS + [("forgectrl", "src/grblport.*")], - requires=["laser.emission-witness", "motion.port-jog"], actions=["button"], - steps=["Scrap under the head with 20 mm of free +X travel; lid closed; exhaust on.", - "Press the physical button when it lights white (the arm)."], - description="A 20 mm line at M3 S400/F600 with no M5 and no program end after it, so the " - "armed window stays open with M3 modal and S above zero: the state in which an " - "injected G1 would fire. The line itself must be witnessed lit, or the case " - "proves nothing. Then POST /motion/jog takes the head back over the line and out " - "again, through the controller port. The window must read armed before and " - "after the jogs, and through both the kernel's LASER_ON sample count stays 0, " - "the HV current stays at its idle reading, and the head's beam detector does " - "not rise over its level before the jogs. The client then ends the job (M5, " - "M2) and the window closes.") -def port_dark(ctx): - ev = ctx.evidence - fc = ctx.forgectrl - with ctx.grbl() as g, LiveJob(ctx, g): - prepare(ctx, g) - base = sample(ctx) - ctx.check(base, "forgectrl /status or /cool/status unavailable") - ctx.check(not base["emission"], "emission_samples nonzero before the job (%s)", base["emission"]) - ctx.ready(ARM_CUE % "20 mm +X") - ctx.arm_press() - jog_samples = [] - try: - # No M5 and no M2: the window stays open, M3 S400 stays modal. - lit = run_and_sample(ctx, g, ["G91", "G21", "M3", "S400", "G1 X20 F600"]) - peak = max((s["emission"] or 0 for s in lit), default=0) - ctx.check(peak > 0, "no emission witnessed on the G1: the dark jogs would prove nothing") - - # The cut is over and dark again; the window is still open. - ok = ctx.wait_for(lambda: (sample(ctx) or {}).get("emission") == 0, 10, poll=0.125) - ctx.check(ok is not None, "emission_samples did not return to 0 after the G1") - before = sample(ctx) - ctx.check(before and before["armed"], "the armed window closed before the jogs: %s", - before and before["phase"]) - ctx.sleep(0.6) # past the port's sender-quiet interval - - for params in ({"x": "-20", "feed": "600"}, {"x": "20", "feed": "2400"}, - {"x": "-20", "feed": "2400"}): - st, body = _port_jog_sampled(ctx, fc, params, jog_samples) - ctx.check(st == 200, "POST /motion/jog %s under the open window -> %s %s", params, st, body) - ctx.sleep(0.6) - after = sample(ctx) - ctx.check(after and after["armed"], - "the armed window closed during the jogs: the dark samples were not under it") - finally: - ctx.clear_notice() - r = g.command("M5") - r += g.command("M2") - ctx.check(not any(x.startswith("error") for x in r), "the client could not end the job: %s", r) - ev["disarm_s"] = wait_disarm(ctx, 30) - ctx.check(ev["disarm_s"] is not None, "the armed window did not close after the program end") - - lit_in_jog = [s for s in jog_samples if s["emission"]] - hv_max = max((s["hv"] for s in jog_samples if s["hv"] is not None), default=0) - beams = [s["beam"] for s in jog_samples if s.get("beam") is not None] - beam_rise = (max(beams) - before["beam"]) if beams and before.get("beam") is not None else None - ev.update({"emission_peak_g1": peak, "jog_samples": len(jog_samples), - "jog_emission_samples": len(lit_in_jog), "jog_hv_max": hv_max, - "jog_beam_rise": beam_rise, "trail": trail(jog_samples)}) - ctx.log("G1 emission peak %s; %d samples across three port jogs: %d with emission, HV max %s, " - "beam detector rise %s", peak, len(jog_samples), len(lit_in_jog), hv_max, beam_rise) - ctx.check(len(jog_samples) >= 16, "only %d samples across the jogs", len(jog_samples)) - ctx.check(not lit_in_jog, "a port jog fired under the open window: %d samples with emission, " - "first emission_samples %s", len(lit_in_jog), lit_in_jog[0]["emission"] if lit_in_jog else None) - ctx.check(hv_max <= HV_DARK_MAX, "HV current %s during a port jog (idle reads ~0)", hv_max) - ctx.check(beam_rise is not None and beam_rise < BEAM_DELTA_MIN, - "the head's beam detector rose %s during a port jog", beam_rise) - ctx.log("PASS: lit on the G1 (peak %s), and three port jogs under the same open window with M3 S400 " - "modal shipped dark by all three witnesses", peak) - - # ---------------------------------------------------------------- the recorder, as a job of the runner's def _laser_keys(fc): diff --git a/forgetest/forgetest/suite/motion.py b/forgetest/forgetest/suite/motion.py index ea3c2b5..a07f3ad 100644 --- a/forgetest/forgetest/suite/motion.py +++ b/forgetest/forgetest/suite/motion.py @@ -1146,192 +1146,6 @@ def homing_kill(ctx, wait_running): ctx.log("restore homing_mode=%r -> %s", hm, st) -@test("motion.respawn-gate", title="A respawn waits for the lid the way a first spawn does", - subsystem="motion", kind="operator", mode="grbl", est_min=2, - covers=_MOTION_COVERS, requires=["motion.deadman", "motion.gate-waits-for-lid"], actions=["lid"], - steps=["Bed clear. Open the lid when told and close it when told; nothing moves."], - description="The enclosure check runs before every controller spawn, respawns included. " - "With the lid open, the controller is killed: the supervisor safes the machine " - "(latch locked), reports waiting with why naming the lid, and starts no controller " - "while the lid stays open. When the lid closes the controller comes back verified, " - "without a second motion probe (the probe is once per broker hold).") -def respawn_gate(ctx): - import os as _os - import signal as _signal - fc = ctx.forgectrl - ev = ctx.evidence - ctx.check(fc.wait_idle(15, abort=ctx.aborted), "machine not idle at the start") - st, m0 = fc.get("/mode") - ctx.check(st == 200 and isinstance(m0, dict) and m0.get("controller") == "running", "controller not running: %s", m0) - pid0 = m0.get("pid") - off = _log_offset(FORGECTRL_LOG) - ctx.act("lid", "open") - try: - ctx.sleep(1.0) - _os.kill(pid0, _signal.SIGKILL) - t_kill = time.time() - ctx.log("SIGKILL sent to controller pid %d with the lid open", pid0) - waited = ctx.wait_for(lambda: (fc.get("/mode")[1] or {}).get("controller") == "waiting", 15) - st, m1 = fc.get("/mode") - ev["after_kill"] = {"waiting_s": waited, "mode": m1} - ctx.log("mode %.1f s after the kill: %s", time.time() - t_kill, m1) - ctx.check(waited is not None, "the supervisor did not report waiting for the lid: %s", m1) - ctx.check("lid" in (m1.get("why") or ""), "why does not name the lid: %r", m1.get("why")) - ilk = hw.sysfs_int("cnc/interlock_circuit") - ctx.check(ilk is not None and ilk & (1 << 3), "latch not locked after the kill") - ctx.sleep(4.0) - st, m2 = fc.get("/mode") - ctx.check(isinstance(m2, dict) and m2.get("controller") == "waiting" and not m2.get("pid"), - "the wait did not hold with the lid open: %s", m2) - finally: - ctx.act("lid", "close") - t_close = time.time() - came = ctx.wait_for(lambda: ((fc.get("/mode")[1] or {}).get("controller") == "running" - and (fc.get("/mode")[1] or {}).get("pid") != pid0), 60) - st, m3 = fc.get("/mode") - ev["after_close"] = {"running_s": came, "mode": m3} - ctx.log("mode %.1f s after the lid closed: %s", time.time() - t_close, m3) - ctx.check(came is not None and m3.get("motion") == "verified", - "the controller did not come back verified after the lid closed: %s", m3) - lines = _probe_lines(FORGECTRL_LOG, off) - ev["probe_lines"] = lines[-2:] - ctx.check(not lines, "the motion probe ran again for the respawn: %s", lines[:1]) - ctx.check(fc.wait_idle(15, abort=ctx.aborted), "machine not idle at the end") - ctx.log("PASS: the respawn waited %.1f s for the lid, then came back verified in %.1f s with no probe", - t_close - t_kill, came) - - -@test("motion.soft-limits", title="After a home the bed is the X/Y envelope", - subsystem="motion", kind="auto", mode="grbl", est_min=5, - covers=_MOTION_COVERS, requires=["motion.jog-roundtrip", "cloud.mode-switch"], - steps=["Bed clear, lid closed; cloud credentials configured; the machine on the network. The test " - "homes the machine through the web service (about a minute), sends moves the controller " - "must refuse, and hands the head back where it found it."], - description="The machine has no limit switches, so the core's $20 cannot be turned on and " - "the X/Y soft limits are the driver's: off while the position is not trusted, on " - "after a home, when the envelope is the bed from the home corner: the far edges the " - "bed check measured (envelope_x_mm, envelope_y_mm, held to 50 mm up to the travel " - "plus 30 mm), or $130 by $131 on a machine that was never measured. " - "With homing_mode = gfcloud and the camera-home offsets unset for the run, $H runs " - "the web-service homing session, every motion of it run whole. Homed, a program " - "move 5 mm past the far edge in X or in Y, or past the near edge, raises ALARM:2 " - "before any motion (the kernel counters do not move), a jog past the far edge is " - "refused with error 15, and a move inside the bed runs. A move or jog that is not " - "refused is stopped before the test fails. The Z envelope is the lens window, as " - "before. The settings are put back as found, the camera home is dropped, and the " - "head is jogged back to where the test found it by the travel the session's " - "motions logged.") -def soft_limits(ctx): - # The camera-home helpers, imported here: homeoff imports this module. - from .cloud import gfhome_homing - from .homeoff import camera_home_return, judge_whole_motions, session_mark - fc = ctx.forgectrl - ev = ctx.evidence - s0 = fc.settings() or {} - ctx.check(s0.get("cloud_enabled") == "1", "cloud mode is off: a camera home needs it on") - found = {k: s0.get(k) or "" for k in ("homing_mode", "gfcloud_home_x", "gfcloud_home_y")} - ev["found"] = found - session_at = None - - def put(k, v): - # The controller refuses a setting for a moment after a Grbl client - # closes: each is written when it is taken. An empty value unsets it. - def write(): - st_, body_ = fc.post("/settings", params={k: ""}) if v == "" else fc.post("/settings", data={k: v}) - if st_ != 200: - ctx.log("%s=%r -> %s %s", k, v, st_, body_) - return st_ == 200 or None - return ctx.wait_for(write, 20, poll=0.5) is not None - - try: - # A camera home at the origin: the envelope is the bed from the home corner. - for k, v in (("homing_mode", "gfcloud"), ("gfcloud_home_x", ""), ("gfcloud_home_y", "")): - ctx.check(put(k, v), "%s=%r was not taken", k, v) - with ctx.grbl() as g: - clean_slate(ctx, g) - session_at = session_mark() - gfhome_homing(ctx, ev, g) - judge_whole_motions(ctx, ev, session_at) - ctx.check(fc.status().get("homed"), "the machine is not homed") - x_travel = float(grbl_setting(g, "$130")) - y_travel = float(grbl_setting(g, "$131")) - ev["travel"] = {"x": x_travel, "y": y_travel} - - # The far edges the home armed, as the driver takes them - # (gfhome_clamp_envelope_mm): the measured key held to 50 mm up - # to the travel plus 30 mm, or the travel when it is unset. - def far_edge(key, travel): - try: - mm = float(s0.get(key) or 0) - except ValueError: - mm = 0.0 - if not mm > 0: - return travel - return min(max(mm, 50.0), travel + 30.0) - - x_edge = far_edge("envelope_x_mm", x_travel) - y_edge = far_edge("envelope_y_mm", y_travel) - ev["envelope"] = {"x": x_edge, "y": y_edge} - k0 = kernel_xy_mm(ctx) - - def refused(cmd): - lines = g.command(cmd, timeout=2) - ctx.sleep(0.5) - text = "\n".join(lines) + g.drain() - k1 = kernel_xy_mm(ctx) - moved = max(abs(k1[0] - k0[0]), abs(k1[1] - k0[1])) - st = g.status_report()["state"] - rec = {"cmd": cmd, "reply": lines, "alarm": "ALARM:2" in text, "moved_mm": round(moved, 3), - "state": st} - ctx.log("%s", rec) - # Stopped first: a move that was not refused must not run on - # to its target while the test fails. - g.realtime(0x18) - ctx.sleep(1.5) - g.drain() - ctx.check(rec["alarm"], "%s was not refused with ALARM:2: %s", cmd, lines) - ctx.check(moved < 0.05, "%s moved the kernel %.3f mm before the alarm", cmd, moved) - unlock = g.command("$X") - ctx.check(unlock and unlock[-1] == "ok", "$X after the soft-limit alarm: %s", unlock) - st = g.status_report()["state"] - ctx.check(st.startswith("Idle"), "controller is %s after the recovery", st) - return rec - - ev["refused"] = [refused("G90 G1 X%.1f F600" % (x_edge + 5)), - refused("G90 G1 Y%.1f F600" % (y_edge + 5)), - refused("G90 G1 X-1 F600")] - # Homed still: the soft-limit alarm and its reset keep the reference. - ctx.check(fc.status().get("homed"), "the soft-limit alarm and the reset un-homed the machine") - jog = g.command("$J=G91X%.1fF1200" % (x_edge + 5)) - ev["jog"] = jog - if not any(l.startswith("error:15") for l in jog): - g.realtime(0x85) # the jog cancel, before the test fails - ctx.sleep(1.5) - g.drain() - ctx.check(any(l.startswith("error:15") for l in jog), "a jog past the bed was not refused with error 15: %s", - jog) - # The core answers the line after an error with that error again - # until an empty line clears it (the sender's acknowledgment). - g.command("") - g.command("G90 G1 X10 Y10 F600", timeout=0.5) - peak, states, st = wait_idle(ctx, g, 20) - ev["inside"] = states - ctx.check("TIMEOUT" not in states, "a move inside the bed did not run") - g.command("G90 G1 X0 Y0 F600", timeout=0.5) - wait_idle(ctx, g, 20) - k1 = kernel_start(ctx) # at rest: grbl's Idle leads the kernel's tail - ev["kernel_back_mm"] = [round(k1[0] - k0[0], 3), round(k1[1] - k0[1], 3)] - ctx.check(max(abs(k1[0] - k0[0]), abs(k1[1] - k0[1])) < 0.1, "the head did not come back to the corner: %s", - ev["kernel_back_mm"]) - finally: - for k, v in found.items(): - ctx.log("restore %s=%r -> %s", k, v, "taken" if put(k, v) else "not taken") - if session_at is not None: - camera_home_return(ctx, ev, session_at) - ctx.log("PASS: X max, Y max and X min refused with ALARM:2 and no motion, the jog refused with " - "error 15, a move inside the bed ran") - - # ------------------------------------------------- lid / button (the factory's) _LID_COVERS = _MOTION_COVERS + [("grblhal-glowforge", "src/glowforge_switches.c"), @@ -1597,144 +1411,6 @@ def lid_cancel_home(ctx): "reset without alarm, head returned to the start both times", drift, hold_drift) -@test("motion.interlock-cancel-home", title="The interlock loop cancels a job like the lid and returns to " - "the job start", - subsystem="motion", kind="operator", mode="grbl", est_min=4, - covers=_LID_COVERS, requires=["motion.lid-cancel-home"], actions=["interlock"], - steps=["Bed clear; the head needs 60 mm of free +X travel. No laser is involved.", - "Be able to open the remote-interlock loop: unplug the Pro's interlock plug, or pull the " - "jumper at J8 on a Basic/Plus.", - "On Ready the head starts a 12 s move: open the interlock while it moves and leave it " - "open until the head has come back, then restore it."], - description="The remote-interlock loop is the lid's equal in the cancel policy: opening it mid-job " - "cancels the job with 'interlock open' named as the reason - the lid's own message would " - "be wrong here - and sends the head back to the job start with the loop still open. " - "(An open lid not stopping the park is the lid test's park, which runs with the lid open " - "throughout; in GRBL mode the park is a rapid too short to open a lid during, so the " - "mid-park lid edge is the cloud test's.)") -def interlock_cancel_home(ctx): - ev = ctx.evidence - policy = (ctx.forgectrl.settings() or {}).get("lid_policy") or "cancel" - ev["lid_policy"] = policy - ctx.check(policy == "cancel", "lid_policy is %r; this test needs cancel", policy) - sw = (ctx.forgectrl.status().get("switches") or {}) - ctx.check(sw.get("interlock_ok"), "the interlock loop already reads open - close it before this test") - with ctx.grbl() as g: - clean_slate(ctx, g) - k0 = kernel_start(ctx) - start = g.status_report()["MPos"] - ev["start"] = start - ev["kernel_start"] = k0 - ctx.ready("On Ready the head starts a 12 s move along +X. Open the interlock loop while it " - "moves and leave it open until the head has come back on its own.") - g.command("M5") - g.command("G91") - start_move(ctx, g, "G1X60F300") # a 12 s move - g.drain() - ctx.act("interlock", "open", text="Leave it open until the head has come back.", timeout=16) - sw = (ctx.forgectrl.status().get("switches") or {}) - ev["interlock_ok_after_pull"] = sw.get("interlock_ok") - ctx.check(sw.get("interlock_ok") is False, - "the interlock still reads closed - the loop was not opened (switches: %s)", sw) - drift = expect_cancel_and_return(ctx, g, ev, start, k0, "interlock open", "interlock") - sw = (ctx.forgectrl.status().get("switches") or {}) - ev["switches_at_return"] = {"lid": sw.get("lid"), "interlock_ok": sw.get("interlock_ok")} - ctx.log("at the end of the park: %s", ev["switches_at_return"]) - ctx.check(sw.get("interlock_ok") is False, - "the interlock was closed again before the park finished - the park ran with the loop " - "restored, not open") - ctx.act("interlock", "close") - ctx.sleep(1) - sw = (ctx.forgectrl.status().get("switches") or {}) - ev["restored"] = {"lid": sw.get("lid"), "interlock_ok": sw.get("interlock_ok")} - ctx.check(sw.get("interlock_ok"), "the interlock loop is still open - restore it before continuing") - r = g.command("$J=G91X5F1200") - ctx.check(not any(x.startswith("error") for x in r), "jog refused after the cancel: %s", r) - wait_idle(ctx, g, 15) - g.command("$J=G91X-5F1200") - wait_idle(ctx, g, 15) - g.command("G90") - machine_idle(ctx) - ctx.log("PASS: interlock open canceled the job with its own reason and the head returned to the " - "start (drift %.3f mm) with the loop still open", drift) - - -@test("motion.lid-policy-hold", title="lid_policy=hold parks the job in Door and a cycle start resumes it", - subsystem="motion", kind="operator", mode="grbl", est_min=4, - covers=_LID_COVERS + [("forgectrl", "src/settings.*")], requires=["motion.lid-cancel-home"], - actions=["lid"], - steps=["Bed clear; the head needs 40 mm of free +X travel. No laser is involved.", - "On Ready the head starts an 8 s move: open the lid while it moves (the job parks in " - "Door), then close it when told; the job finishes after that."], - description="The other lid policy, kept for senders that expect stock grblHAL: with " - "lid_policy=hold a lid open parks the job in the door state and holds it there - " - "no cancel, no return home - and once the lid is closed a cycle start finishes the " - "move with its position intact. The setting is restored to cancel at the end.") -def lid_policy_hold(ctx): - ev = ctx.evidence - fc = ctx.forgectrl - # What the machine had, exactly: an unset lid_policy reads as the empty - # string and behaves as cancel, and writing the word back where the - # machine had nothing is a leftover the hand-back reports. The default - # is for reading the value, never for restoring it. - was = (fc.settings() or {}).get("lid_policy", "") - ev["lid_policy_before"] = was - ev["lid_policy_in_force"] = was or "cancel" - st, _b = fc.post("/settings", data={"lid_policy": "hold"}) - ctx.check(st == 200, "could not set lid_policy=hold (%s)", st) - ctx.check(((fc.settings() or {}).get("lid_policy")) == "hold", "lid_policy did not take") - try: - with ctx.grbl() as g: - clean_slate(ctx, g) - start = g.status_report()["MPos"] - ctx.ready("On Ready the head starts an 8 s move along +X. Open the lid while it moves; " - "the job parks in the door state and waits.") - g.command("M5") - g.command("G91") - start_move(ctx, g, "G1X40F300") # an 8 s move - g.drain() - ctx.act("lid", "open", timeout=12) - st, text = wait_state_text(ctx, g, "Door", 8) - ev["door_state"] = st["state"] if st else g.status_report()["state"] - ctx.check(st is not None, "the lid did not park the job in Door (state %s)", ev["door_state"]) - ev["messages"] = [ln for ln in text.splitlines() if ln.startswith("[MSG:")] - ctx.check("job canceled" not in text, "the job was canceled under lid_policy=hold: %s", ev["messages"]) - ctx.check("returned to the job start" not in text, - "the head returned to the job start under lid_policy=hold") - held = g.status_report()["MPos"] - ev["parked_at"] = held - ctx.log("parked in %s at %s", ev["door_state"], held) - ctx.act("lid", "close") - ctx.sleep(1) - ev["state_after_close"] = g.status_report()["state"] - ctx.log("after the lid closed: %s (a cycle start is needed)", ev["state_after_close"]) - g.realtime(0x7E) # ~ cycle start - peak, states, st = wait_idle(ctx, g, 30) - ev["states_after_resume"] = states - ctx.check("TIMEOUT" not in states, "the job did not finish after the resume: %s", states) - final = st["MPos"] - moved = final[0] - start[0] - ev["moved_mm"] = round(moved, 3) - ctx.log("finished: moved %.3f mm of 40 (states %s)", moved, states) - ctx.check(abs(moved - 40.0) <= 0.05, - "the resumed job did not finish its move (%.3f mm of 40)", moved) - g.command("$J=G91X-40F1200") - wait_idle(ctx, g, 30) - g.command("G90") - machine_idle(ctx) - finally: - # An empty value clears the key, and only the query-string form - # carries one: an empty form field never reaches the request, and - # the write is refused with "no known setting in request". - st, _b = (fc.post("/settings", params={"lid_policy": ""}) if not was - else fc.post("/settings", data={"lid_policy": was})) - ev["lid_policy_restored"] = (fc.settings() or {}).get("lid_policy", "") - ctx.log("lid_policy restored to %s", ev["lid_policy_restored"]) - ctx.check(ev["lid_policy_restored"] == was, "lid_policy was not restored to %r", was) - ctx.log("PASS: lid_policy=hold parked the job in Door and the cycle start finished it (%.3f mm)", - ev["moved_mm"]) - - GRBLHAL_LOG = "/data/log/forgefirm/grblhal/grblhal.log" SCHED_FIFO = 1 diff --git a/forgetest/tests/fixtures/gfcloud-buttonwait.log b/forgetest/tests/fixtures/gfcloud-buttonwait.log deleted file mode 100644 index 6e5456b..0000000 --- a/forgetest/tests/fixtures/gfcloud-buttonwait.log +++ /dev/null @@ -1,41 +0,0 @@ -2026-08-17T09:44:00.417337+00:00 gfcloud[1927] INFO cnc:set_step_freq 28160 -2026-08-17T09:44:00.418627+00:00 gfcloud[1927] INFO cnc:set_x_decay 1 -2026-08-17T09:44:00.419858+00:00 gfcloud[1927] INFO cnc:set_x_mode 8 -2026-08-17T09:44:00.421083+00:00 gfcloud[1927] INFO cnc:set_x_current 135 -2026-08-17T09:44:00.422535+00:00 gfcloud[1927] INFO cnc:set_y_decay 1 -2026-08-17T09:44:00.423794+00:00 gfcloud[1927] INFO cnc:set_y_mode 8 -2026-08-17T09:44:00.425014+00:00 gfcloud[1927] INFO cnc:set_y_current 22 -2026-08-17T09:44:00.432591+00:00 gfcloud[1927] INFO machine:_run_loop starting run -2026-08-17T09:44:00.434005+00:00 gfcloud[1927] INFO machine:_run_loop current state: MachineState.IDLE -2026-08-17T09:44:00.438277+00:00 gfcloud[1927] INFO machine:_run_loop current state: MachineState.RUNNING -2026-08-17T09:44:01.785230+00:00 gfcloud[1927] INFO machine:_run_loop current state: MachineState.IDLE -2026-08-17T09:44:01.797728+00:00 gfcloud[1927] INFO machine:_run_loop finished run -2026-08-17T09:44:01.800661+00:00 gfcloud[1927] INFO cnc:laser_latch 1 -2026-08-17T09:44:01.808504+00:00 gfcloud[1927] INFO machine:_motion_locked end positions (actual/expected): X (-10313/-10313), Y (-649/-649), Z (0/0) -2026-08-17T09:44:01.809326+00:00 gfcloud[1927] INFO machine:_motion_locked motion bytes actual:34952, expected: 34952 -2026-08-17T09:44:01.810066+00:00 gfcloud[1927] INFO machine:_motion_locked start idle -2026-08-17T09:44:01.811352+00:00 gfcloud[1927] INFO cnc:set_x_current 33 -2026-08-17T09:44:01.812874+00:00 gfcloud[1927] INFO cnc:set_y_current 5 -2026-08-17T09:44:01.820631+00:00 gfcloud[1927] INFO machine:_motion_locked end positions (-10313, -649, 0) -2026-08-17T09:44:01.822062+00:00 gfcloud[1927] INFO machine:_motion end motion -2026-08-17T09:44:01.822829+00:00 gfcloud[1927] INFO basemachine:_finish_action motion [1576550614]: finished with event ":completed" -2026-08-17T09:44:01.824200+00:00 gfcloud[1927] INFO cnc:laser_latch 1 -2026-08-17T09:44:07.993647+00:00 gfcloud[1927] INFO gfuiservice:run service action request: print (ready) -2026-08-17T09:44:07.998655+00:00 gfcloud[1927] INFO machine:_motion start motion -2026-08-17T09:44:07.999643+00:00 gfcloud[1927] INFO gfuiservice:run print (ready) service action print -2026-08-17T09:44:14.199666+00:00 gfcloud[1927] INFO cnc:laser_latch 0 -2026-08-17T09:44:14.208291+00:00 gfcloud[1927] INFO machine:_button_wait waiting for button -2026-08-17T09:44:21.270846+00:00 gfcloud[1927] INFO machine:_switch_event lid opened -2026-08-17T09:44:21.275703+00:00 gfcloud[1927] WARNING machine:_button_wait button wait lid opened - relocking the laser -2026-08-17T09:44:21.278615+00:00 gfcloud[1927] INFO cnc:laser_latch 1 -2026-08-17T09:44:21.288305+00:00 gfcloud[1927] INFO machine:_return_home start return home -2026-08-17T09:44:21.291373+00:00 gfcloud[1927] INFO machine:_return_home return home: already at the job start -2026-08-17T09:44:21.292501+00:00 gfcloud[1927] INFO machine:_motion_locked start cool down -2026-08-17T09:44:21.305578+00:00 gfcloud[1927] INFO machine:_motion_locked start idle -2026-08-17T09:44:21.307322+00:00 gfcloud[1927] INFO cnc:set_x_current 33 -2026-08-17T09:44:21.309185+00:00 gfcloud[1927] INFO cnc:set_y_current 5 -2026-08-17T09:44:21.316340+00:00 gfcloud[1927] INFO machine:_motion_locked end positions (0, 0, 0) -2026-08-17T09:44:21.318139+00:00 gfcloud[1927] INFO machine:_motion end motion -2026-08-17T09:44:21.318860+00:00 gfcloud[1927] INFO basemachine:_finish_action print [1576550621]: finished with event ":cancelled" -2026-08-17T09:44:21.320214+00:00 gfcloud[1927] INFO cnc:laser_latch 1 -2026-08-17T09:44:34.058320+00:00 gfcloud[1927] INFO machine:_switch_event lid closed diff --git a/forgetest/tests/test_baseline.py b/forgetest/tests/test_baseline.py index 3a0ef5a..28a43c0 100644 --- a/forgetest/tests/test_baseline.py +++ b/forgetest/tests/test_baseline.py @@ -531,9 +531,9 @@ class ExtensionFreeTests(BaselineTests): class TransientNotLeftoverTests(BaselineTests): """A leftover is what a run left behind, not the machine part-way - through its own work. Found on the bench reference, where - laser.arm-wait-lid passed every check it makes and failed its - hand-back on the post-job smoke clear.""" + through its own work. Found on the bench reference, where a laser + test passed every check it makes and failed its hand-back on the + post-job smoke clear.""" def test_a_fading_button_led_is_not_a_leftover(self): # the trigger fades brightness toward target: target 0 with diff --git a/forgetest/tests/test_cloud_suite.py b/forgetest/tests/test_cloud_suite.py index ea7e9ef..67e111b 100644 --- a/forgetest/tests/test_cloud_suite.py +++ b/forgetest/tests/test_cloud_suite.py @@ -69,6 +69,19 @@ def cut(lines, marker, count=1): raise KeyError(marker) +def dark_print(ctx): + """The dark print as exthost.armed-freeze runs it: the cloud suite's + dark_print_body between the offline service's setup and cleanup.""" + offset = cloud.enter_offline(ctx) + job = cloud.offline_job(ctx, "dark.puls", seconds=30) + off = cloud.Offline().__enter__() + try: + cloud.dark_print_body(ctx, ctx.evidence, off, job, offset, ctx.forgectrl) + finally: + off.__exit__(None, None, None) + cloud.offline_cleanup(ctx) + + class Script: """Answers every prompt with its first option; a hook per prompt or notice substring drives the fake machine and the log replay (a @@ -1040,30 +1053,7 @@ class CloudSuiteTests(unittest.TestCase): self.assertFails(cloud.lid_interlock_abort, "not edge-driven", hooks=self.abort_hooks(pre, lid_tail, ilk_stop, ilk_park, [])) - def test_lid_during_button_wait_on_the_bench_excerpt(self): - self.in_offline() - lines = fixture("buttonwait") - pre, rest = cut(lines, "waiting for button") - # The bench excerpt is a print that fit the ring; the test now wants - # one that did not, so the feeder is alive through the wait. - long_job = ("2026-08-17T09:44:14.100000+00:00 gfcloud[1927] INFO machine:_feed_and_run " - "job is longer than the ring: 33554432 bytes enqueued, feeding the rest as it plays") - pre, rest = pre + [long_job, rest[0]], rest[1:] - self.offline_print_hooks(pre) - hooks = {"do NOT press it": lambda: (self.lid(False), self.append(rest, delay=0.05)), - "Close the lid.": lambda: self.lid(True)} - run = self.run_test(cloud.lid_during_button_wait, hooks=hooks, test_id="cloud.lid-during-button-wait") - ev = run.evidence - self.assertEqual([m["status"] for m in FakeOffline.sent], ["ready"]) - self.assertEqual(ev["runs_started_after_wait"], 0) - self.assertIn(":cancelled", ev["log"]["print finished"]) - self.assertTrue(ev["latch_locked"]) - self.assertEqual(ev["program_total_after"], {"first": 0, "later": 0}) - self.assertEqual(ev["streaming_after"], 0) - self.assertEqual(self.fc.posts, []) - self.assertTrue(any("PASS: lid open at the button prompt" in l for l in run.lines)) - - # -- the cooling verdict: locked until the run, a hold that never clears -- + # -- the dark print exthost.armed-freeze runs: locked at the button -------- def dark_parts(self): """(the print prologue through the button prompt, the run and the print's end with no pause in it).""" @@ -1074,170 +1064,13 @@ class CloudSuiteTests(unittest.TestCase): run = [l for l in run if "paus" not in l and "resum" not in l] return pre, run - def test_dark_print_on_the_bench_excerpt(self): - self.in_offline() - self.fc.state["cool"].update({"verdict": "OK", "hold": False, "fire_ok": True, "armed": False}) - pre, run = self.dark_parts() - self.offline_print_hooks(pre) - - def press(): - # the run: the client unlocks the latch right before it, and - # relocks when the job ends - self._attr("cnc/interlock_circuit", "37") - self.append(run[:1], delay=0.05) - - def ended(): - self._attr("cnc/interlock_circuit", "45") - self.append(run[1:], delay=0.0) - threading.Timer(2.0, ended).start() - hooks = {"press it. The print runs dark": press} - run_ = self.run_test(cloud.dark_print, hooks=hooks, test_id="cloud.dark-print") - ev = run_.evidence - self.assertTrue(ev["latch_locked_at_button"]) - self.assertFalse(ev["latch_locked_in_run"]) - self.assertTrue(ev["latch_locked_after"]) - self.assertIn(":completed", ev["print finished"]) - self.assertTrue(any("PASS: locked at the button" in l for l in run_.lines)) - def test_dark_print_fails_when_the_latch_is_open_at_the_button(self): self.in_offline() pre, run = self.dark_parts() self._attr("cnc/interlock_circuit", "37") # unlocked before the press self.offline_print_hooks(pre) hooks = {"press it. The print runs dark": lambda: self.append(run, delay=0.05)} - self.assertFails(cloud.dark_print, "the latch is unlocked at the button wait", hooks=hooks) - - def test_a_print_turned_away_before_the_button_fails_at_once_with_the_reason(self): - """The bench reference's own lines, 2026-09-20: the coolant read 27.4 - against the client's start ceiling of 27, and the test then sat out - its two minutes and said only that the button wait never came.""" - self.in_offline() - stamp = "2026-09-20T21:28:16.83+00:00 gfcloud[782] INFO " - refused = [stamp + "gfuiservice:run service action request: print (ready)", - stamp + "machine:_motion start motion", - stamp + "machine:_safe_to_move machine temp is too high, temp: 27.4", - stamp + "machine:_motion end motion", - stamp + 'basemachine:_finish_action print [9005]: finished with event ":cancelled"'] - self.offline_print_hooks(refused) - t0 = time.time() - self.assertFails(cloud.dark_print, "turned the print away before the button wait: " - "INFO machine:_safe_to_move machine temp is too high, temp: 27.4") - self.assertLess(time.time() - t0, 30, "the refusal was not seen until the wait timed out") - - def test_verdict_refuse_on_the_bench_excerpt(self): - self.in_offline() - self.fc.state["status"]["coolant"] = {"up_c": 21.5, "down_c": 21.4} - self.fc.state["settings"].update({"cool_temp_min": "5", "cool_temp_start": "16", - "cloud_hold_max_s": ""}) - self.fc.state["cool"].update({"verdict": "OK", "hold": False, "fire_ok": True, "armed": False}) - pre, run = self.dark_parts() - stamp = "2026-08-17T09:44:14.%06d+00:00 gfcloud[1927] %s machine:_verdict_wait " - wait = [stamp % (100000, "INFO") + "waiting on the cooling engine: WARMUP (WARM-UP: coolant " - "21.5 C under the 29.5 C start gate - heater on, hold)"] - refuse = [stamp % (200000, "WARNING") + "cooling hold held 60 s - relocking the laser", - "2026-08-17T09:44:15.000000+00:00 gfcloud[1927] INFO basemachine:_finish_action print " - "[9006]: finished with event \":cancelled\""] - - def on_post(path, form): - if path == "/settings" and form.get("cool_temp_start") not in (None, "0", "16"): - self.fc.state["cool"].update({"verdict": "WARMUP", "hold": True, "fire_ok": False, - "armed": True, "phase": "warm-up"}) - return None - self.fc.on_post = on_post - self.offline_print_hooks(pre) - - def press(): - self.append(wait, delay=0.05) - - def refused(): - self.fc.state["cool"].update({"verdict": "OK", "hold": False, "fire_ok": True, - "armed": False, "phase": "idle"}) - self.append(refuse, delay=0.0) - threading.Timer(2.5, refused).start() - hooks = {"press it. Nothing runs": press} - run_ = self.run_test(cloud.verdict_refuse, hooks=hooks, test_id="cloud.verdict-refuse") - ev = run_.evidence - self.assertEqual(ev["gate"], 29.5) - self.assertTrue(ev["latch_locked_at_button"]) - self.assertEqual(ev["latch_samples"]["unlocked"], 0) - self.assertGreaterEqual(ev["latch_samples"]["n"], 1) - self.assertIn(":cancelled", ev["print finished"]) - posted = [f for p_, f in self.fc.posts if p_ == "/settings"] - self.assertEqual(posted[0]["cool_temp_start"], "29.5") - self.assertEqual(posted[0]["cloud_hold_max_s"], "60") - self.assertEqual(posted[-1], {"cool_temp_min": "5", "cool_temp_start": "16", "cloud_hold_max_s": ""}) - self.assertTrue(any("PASS: held" in l for l in run_.lines)) - - def test_verdict_refuse_fails_when_the_latch_unlocks_during_the_hold(self): - self.in_offline() - self.fc.state["status"]["coolant"] = {"up_c": 21.5, "down_c": 21.4} - self.fc.state["settings"].update({"cool_temp_min": "5", "cool_temp_start": "16"}) - self.fc.state["cool"].update({"verdict": "WARMUP", "hold": True, "fire_ok": False, "armed": True}) - pre, run = self.dark_parts() - stamp = "2026-08-17T09:44:14.%06d+00:00 gfcloud[1927] %s machine:_verdict_wait " - wait = [stamp % (100000, "INFO") + "waiting on the cooling engine: WARMUP (no reason given)"] - refuse = [stamp % (200000, "WARNING") + "cooling hold held 60 s - relocking the laser", - "2026-08-17T09:44:15.000000+00:00 gfcloud[1927] INFO basemachine:_finish_action print " - "[9006]: finished with event \":cancelled\""] - self.offline_print_hooks(pre) - - def press(): - self.append(wait, delay=0.05) - self._attr("cnc/interlock_circuit", "37") # unlocked under the hold - threading.Timer(2.5, lambda: self.append(refuse, delay=0.0)).start() - hooks = {"press it. Nothing runs": press} - self.assertFails(cloud.verdict_refuse, "the latch unlocked during the hold", hooks=hooks) - - # -- a paused print canceled by the lid, a running one by the app -------- - def cancel_parts(self): - """(print prologue, the pause lines, the lid stop + park + cancel, - the same tail with the app's cancel as the trigger - what - cloud.oversize-stream's ending looks like).""" - lines = fixture("lidabort") - pre, rest = cut(lines, "waiting for button") - run_pre, rest = cut(rest, "machine:_run_loop starting run") - pre, rest = pre + run_pre + [rest[0]], rest[1:] - paused = ["2026-08-17T09:42:25.500000+00:00 gfcloud[1522] INFO machine:_run_loop " - "button pressed mid-run; pausing", - "2026-08-17T09:42:26.100000+00:00 gfcloud[1522] INFO machine:_run_loop " - "paused at Position(x=41.2, y=17.0, z=0.0)"] - app_cancel = [l.replace("lid opened mid-run; stopping motion", - "action canceled mid-run; stopping motion") for l in rest] - return pre, paused, rest, app_cancel - - def test_paused_lid_cancel_on_the_bench_excerpt(self): - self.in_offline() - pre, paused, lid_tail, _app = self.cancel_parts() - self.offline_print_hooks(pre) - run = self.run_test(cloud.paused_lid_cancel, - hooks={"the press pauses the print": lambda: self.append(paused, delay=0.05), - "The print is paused: leave the lid open": lambda: ( - self.lid(False), self.append(lid_tail, delay=0.05)), - "Close the lid.": lambda: self.lid(True)}, - test_id="cloud.paused-lid-cancel") - ev = run.evidence - self.assertEqual(len(FakeOffline.sent), 1) # one print, handed over the socket - self.assertEqual(ev["paused"], {"button pressed mid-run; pausing": True, "paused at": True}) - self.assertIn(":cancelled", ev["lid_from_pause"]["print finished"]) - self.assertTrue(ev["lid_from_pause"]["return home complete"]) - self.assertEqual(ev["counters_after"], [0, 0, 3]) - self.assertTrue(ev["latch_locked_after"]) - self.assertFalse(ev["armed_after"]) - self.assertFalse(ev["button_dark"]) - self.assertEqual(self.fc.posts, []) - self.assertTrue(any("PASS: a paused print canceled by the lid" in l for l in run.lines), run.lines) - - def test_paused_lid_cancel_fails_when_the_paused_print_resumes_instead(self): - # a lid that resumed (or was ignored) leaves the print ':completed' - self.in_offline() - pre, paused, lid_tail, _app = self.cancel_parts() - lid_tail = [l.replace(':cancelled"', ':completed"') for l in lid_tail] - self.offline_print_hooks(pre) - self.assertFails( - cloud.paused_lid_cancel, "the print did not end ':cancelled'", - hooks={"the press pauses the print": lambda: self.append(paused, delay=0.05), - "The print is paused: leave the lid open": lambda: ( - self.lid(False), self.append(lid_tail, delay=0.05))}) + self.assertFails(dark_print, "the latch is unlocked at the button wait", hooks=hooks) def test_print_finish_is_the_prints_not_another_actions(self): # a motion that completes before the print must not satisfy the print's finish diff --git a/forgetest/tests/test_cooling_intake_suite.py b/forgetest/tests/test_cooling_intake_suite.py deleted file mode 100644 index 59e7b53..0000000 --- a/forgetest/tests/test_cooling_intake_suite.py +++ /dev/null @@ -1,176 +0,0 @@ -# Copyright 2026 514 LLC d/b/a OpenGlow -# Written by Scott Wiederhold -# https://community.openglow.org -# SPDX-License-Identifier: MIT - -"""cooling.intake-one-fan replayed host-side under the real runner -Context against a scripted engine: the fake reads cool_intake_fans and -the floors at every M8 (the engine reloads at run start), publishes -intake_2 not fitted with no floor when one intake fan is set, holds every -fan in grace for the configured seconds, then judges the bench readings -against the floors. The negative engines are the two ways the change -could go wrong: the setting ignored (intake 2 still judged) and the -setting taking intake 1 out with it.""" -import threading -import time -import unittest - -import helpers -from forgetest.runner import Context, Failed, Run -from forgetest.suite import cooling, cooling_intake -from test_cooling_suite import FakeGrbl - - -class IntakeOneFanTests(unittest.TestCase): - READINGS = {"exhaust": 11640, "intake_1": 4157, "intake_2": 4158, "air_assist": 11050, "purge": 625} - FLOORS = {"exhaust": ("cool_tach_exhaust_min_rpm", 6400.0, 20000.0), - "intake_1": ("cool_tach_intake_min_rpm", 2290.0, 20000.0), - "intake_2": ("cool_tach_intake_min_rpm", 2290.0, 20000.0), - "air_assist": ("cool_tach_air_assist_min_rpm", 6000.0, 30000.0), - "purge": ("cool_purge_min_current", 300.0, 1023.0)} - GATE_OF = {"exhaust": "exhaust", "intake_1": "intake", "intake_2": "intake", - "air_assist": "air_assist", "purge": "purge"} - - def setUp(self): - self.fc = helpers.FakeForgectrl().start() - self.grbl = FakeGrbl() - self.saved = (cooling.VERDICT_WAIT_S, cooling.SESSION_END_WAIT_S, cooling_intake.INTAKE_TRIP_WAIT_S) - cooling.VERDICT_WAIT_S = 4 - cooling.SESSION_END_WAIT_S = 3 - cooling_intake.INTAKE_TRIP_WAIT_S = 4 - self.fc.state["status"] = dict(self.fc.state["status"], gates_off=[]) - self.fc.state["cool"] = {"phase": "idle", "verdict": "OK", "fire_ok": False, "hold": False, - "gates_off": [], "fan_gates": {}} - for key, _d, _h in self.FLOORS.values(): - self.fc.state["settings"].setdefault(key, "") - self.fc.state["settings"].setdefault("cool_fan_grace_s", "") - self.fc.state["settings"].setdefault("cool_intake_fans", "") - self.honors_setting = True # the engine takes intake 2 out with one intake fan - self.drops_both = False # the engine takes intake 1 out with it - self.grace_scale = 0.1 # seconds of fake grace per configured second - self._describe() - self.grbl.on_command = self._engine - self.fc.on_post = self._on_post - - def tearDown(self): - cooling.VERDICT_WAIT_S, cooling.SESSION_END_WAIT_S, cooling_intake.INTAKE_TRIP_WAIT_S = self.saved - self.grbl.close() - self.fc.stop() - - def setting(self, key, default): - v = self.fc.state["settings"].get(key) or "" - return float(v) if v else default - - def _describe(self): - gates = {} - for fan, (key, default, hi) in self.FLOORS.items(): - v = self.setting(key, default) - gates[key] = {"gate": self.GATE_OF[fan], "def": default, "lo": 0.0, "hi": hi, - "band": [default * 0.9, default * 1.1], "off": "low", "value": v, - "state": "off" if v <= 0 else "ok"} - self.fc.state["settings"]["gates"] = gates - - def _on_post(self, path, form): - if path != "/settings": - return None - self.fc.state["settings"].update(form) - self._describe() - return (200, self.fc.state["settings"]) - - def _engine(self, line): - self._describe() - cool = self.fc.state["cool"] - if line == "M9": - def end(): - time.sleep(0.3) - cool["phase"] = "idle" - cool.update(verdict="OK", fire_ok=True, hold=False, reason="") - for g in cool.get("fan_gates", {}).values(): - if g["state"] != "not fitted": - g["state"] = "idle" - threading.Thread(target=end, daemon=True).start() - return - if line != "M8": - return - cool["phase"] = "run" - one = self.fc.state["settings"].get("cool_intake_fans") == "1" and self.honors_setting - out = {"intake_2"} | ({"intake_1"} if self.drops_both else set()) if one else set() - floors = {fan: 0.0 if fan in out else self.setting(key, default) - for fan, (key, default, _h) in self.FLOORS.items()} - cool.update(verdict="OK", fire_ok=True, hold=False, resume_ok=True, reason="", gates_off=[], - fan_gates={f: {"reading": self.READINGS[f], "floor": floors[f], - "state": "not fitted" if f in out else "grace"} for f in floors}) - grace = self.setting("cool_fan_grace_s", 15.0) * self.grace_scale - - def judge(): - time.sleep(grace) - if cool["phase"] != "run": - return - for f in floors: - if f not in out: - cool["fan_gates"][f]["state"] = "ok" if self.READINGS[f] >= floors[f] else "under" - time.sleep(0.3) - if cool["phase"] != "run": - return - for f in floors: - if f not in out and floors[f] > self.READINGS[f]: - cool["fan_gates"][f]["state"] = "TRIPPED" - if cool["verdict"] != "AIRFLOW": - cool.update(verdict="AIRFLOW", fire_ok=False, hold=True, resume_ok=False, - reason="AIRFLOW: %s %d under the %d floor for 3 s - hold, no resume this job" - % (f, self.READINGS[f], floors[f])) - threading.Thread(target=judge, daemon=True).start() - - def run_test(self): - run = Run("test", "cooling.intake-one-fan", "t") - ctx = Context(run, None, helpers.make_test("cooling.intake-one-fan", [])) - cooling_intake.intake_one_fan(ctx) - return run - - def settings_posts(self): - return [f for p, f in self.fc.posts if p == "/settings"] - - def restored(self): - return {"cool_intake_fans": "", "cool_tach_intake_min_rpm": "", "cool_fan_grace_s": ""} - - def test_one_fan_then_intake_1_trip_then_restore_pass(self): - run = self.run_test() - ev = run.evidence - self.assertEqual(ev["one_fan"]["intake_2"]["state"], "not fitted") - self.assertEqual(ev["one_fan"]["intake_1"]["state"], "ok") - self.assertEqual(ev["intake_1_trip"]["intake_1"]["state"], "TRIPPED") - self.assertEqual(ev["intake_1_trip"]["intake_2"]["state"], "not fitted") - self.assertTrue(all(g["state"] == "ok" for g in ev["restored"].values())) - posts = self.settings_posts() - self.assertEqual(posts[0], {"cool_intake_fans": "1", "cool_fan_grace_s": "8"}) - self.assertEqual(posts[1], {"cool_tach_intake_min_rpm": "20000.0"}) - self.assertEqual(posts[-1], self.restored()) - self.assertEqual(self.grbl.commands.count("M8"), 3) - self.assertEqual(self.grbl.commands.count("M9"), 3) - - def test_an_engine_that_ignores_the_setting_fails_and_restores(self): - self.honors_setting = False - with self.assertRaises(Failed) as cm: - self.run_test() - self.assertIn("intake_2 is not published not fitted", str(cm.exception)) - self.assertEqual(self.settings_posts()[-1], self.restored()) - self.assertEqual(self.fc.state["cool"]["verdict"], "OK") - - def test_an_engine_that_drops_intake_1_too_fails_and_restores(self): - self.drops_both = True - with self.assertRaises(Failed) as cm: - self.run_test() - self.assertIn("intake_1 is not judged ok at the intake floor", str(cm.exception)) - self.assertEqual(self.settings_posts()[-1], self.restored()) - - def test_a_machine_already_at_one_intake_fan_is_refused_untouched(self): - self.fc.state["settings"]["cool_intake_fans"] = "1" - with self.assertRaises(Failed) as cm: - self.run_test() - self.assertIn("already 1", str(cm.exception)) - self.assertEqual(self.settings_posts(), []) - self.assertEqual(self.grbl.commands.count("M8"), 0) - - -if __name__ == "__main__": - unittest.main()