From 67ac8f4064e229102559667060a5b3e91c4362dc Mon Sep 17 00:00:00 2001 From: ScottW514 Date: Tue, 13 Oct 2026 18:34:04 -0400 Subject: [PATCH] forgetest: events.button-telemetry leaves the catalog events.button-telemetry held on the machine that one physical press reaches the event stream as pressed then released through its 25 Hz switch poll, and the telemetry event's fields. The host proves the rules now: - the press between two 5 Hz gathers, idle snapshot, scripted switch word: forgectrl events_test test_short_press (d8d07fc); - the telemetry field string, and that a press the machine is waiting for (arming, armed, leased, a button wait) is no event: events_test test_button. The physical button reaching the switch word stays in the catalog: setup.check-switches. - evmore.py, the test's own module, goes with its import in suite/__init__.py; nothing else imported it. - Its covers (forgectrl src/events.*, main.c, status.*) stay covered: events.stream holds all three, and other catalog tests hold main.c and status.*. No HOST_TESTED row is needed. Proof: the coverage lint reports 0 uncovered paths and 0 host-tested row problems over 107 tests (its one failure is setup.check-lens's src/lensfind.* entry, which the forgectrl pin bump clears); the gate self-check refuses an empty artifact cleanly. The unit modules that load the whole catalog pass under Linux: test_artifact, test_campaign, test_cooling_order, test_fixture, test_manifest, test_operator. Acceptance: the catalog hash moves; no remaining test's fingerprint moves (the module goes whole). --- forgetest/forgetest/suite/__init__.py | 1 - forgetest/forgetest/suite/evmore.py | 103 -------------------------- 2 files changed, 104 deletions(-) delete mode 100644 forgetest/forgetest/suite/evmore.py diff --git a/forgetest/forgetest/suite/__init__.py b/forgetest/forgetest/suite/__init__.py index 9ed31f6..cb13fa3 100644 --- a/forgetest/forgetest/suite/__init__.py +++ b/forgetest/forgetest/suite/__init__.py @@ -22,7 +22,6 @@ 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 extcall # noqa: F401,E402 -from . import evmore # noqa: F401,E402 from . import extcat # noqa: F401,E402 from . import extwizard # noqa: F401,E402 from . import updlock # noqa: F401,E402 diff --git a/forgetest/forgetest/suite/evmore.py b/forgetest/forgetest/suite/evmore.py deleted file mode 100644 index 3046d3f..0000000 --- a/forgetest/forgetest/suite/evmore.py +++ /dev/null @@ -1,103 +0,0 @@ -# Copyright 2026 514 LLC d/b/a OpenGlow -# Written by Scott Wiederhold -# https://community.openglow.org -# SPDX-License-Identifier: MIT - -"""The event stream's button and its telemetry, on the machine. - -Its own module, so that no other test's fingerprint moves. The stream is -opened straight at forgectrl's read-only listener, as a client off the -machine would, from a loopback source address of its own so that it takes -a place of its own among the three. -""" - -import json -import threading -import time - -from ..catalog import test -from .exthost import _stream - -SOURCE = "127.0.0.9" - - -def _reader(sock, seen, stop): - """Every event on the stream into seen, as (name, data), until stop.""" - buf = b"" - sock.settimeout(1.0) - while not stop.is_set(): - try: - k = sock.recv(4096) - except OSError: - continue - if not k: - break - buf += k - while b"\n\n" in buf: - block, buf = buf.split(b"\n\n", 1) - name, data = None, None - for line in block.decode("utf-8", "replace").split("\n"): - if line.startswith("event: "): - name = line[7:] - elif line.startswith("data: "): - data = line[6:] - if name: - try: - seen.append((name, json.loads(data or "null"), time.time())) - except ValueError: - seen.append((name, data, time.time())) - - -@test("events.button-telemetry", title="The event stream tells a press of the button, and the telemetry", - subsystem="forgectrl", kind="operator", mode="grbl", est_min=2, - covers=[("forgectrl", "src/events.*"), ("forgectrl", "src/main.c"), ("forgectrl", "src/status.*")], - actions=["button"], - steps=["The machine idle; you will be told to press the button once.", - "Nothing moves and nothing fires: nothing is waiting for the press."], - description="A stream opened at forgectrl's read-only listener gets a telemetry.tick within 12 s, " - "and it is what the daemon already holds: the cooling phase, the verdict, fire_ok, the " - "two coolant temperatures, the controller's state, and the lid, with no sensor read of " - "its own. The machine idle and nothing waiting for the button, one press (the bench " - "fixture's, or the operator's) arrives as two button events, pressed true and then false, " - "in that order: the button is looked at 25 times a second, so a press shorter than the " - "rest of the stream's 5 Hz look is not lost. That a press the machine is waiting for - a " - "job arming or armed, a wizard under the lease, the daemon's own wait - is no event is " - "forgectrl's events_test's, as are update.available and setup.flag, which a machine in " - "use cannot be made to show on demand.") -def button_telemetry(ctx): - fc = ctx.forgectrl - ev = ctx.evidence - ctx.check(fc.wait_idle(timeout=30, abort=ctx.aborted), "machine not idle") - code, why, sock = _stream("127.0.0.1", keep=True, source=SOURCE) - ctx.check(code == 200 and sock, "the event stream: %s %s", code, why) - seen, stop = [], threading.Event() - th = threading.Thread(target=_reader, args=(sock, seen, stop), daemon=True) - th.start() - try: - ctx.wait_for(lambda: any(n == "telemetry.tick" for n, _d, _t in seen), 14) - tick = next((d for n, d, _t in seen if n == "telemetry.tick"), None) - ev["telemetry"] = tick - ctx.log("telemetry.tick: %s", tick) - ctx.check(isinstance(tick, dict) and set(tick) == {"phase", "verdict", "fire_ok", "down_c", "up_c", "state", "lid"}, - "the telemetry's fields: %s", tick) - ctx.check(tick.get("lid") in ("closed", "open") and isinstance(tick.get("down_c"), (int, float)), - "its values: %s", tick) - mark = len(seen) - - def both(): - got = [d for n, d, _t in seen[mark:] if n == "button"] - return got if len(got) >= 2 else None - ctx.act("button", "press", until=both, text="Press the button once: nothing is waiting for it, and " - "the press is only reported.") - presses = [d for n, d, _t in seen[mark:] if n == "button"] - ev["button"] = presses - ctx.log("button events: %s", presses) - ctx.check(presses[:2] == [{"pressed": True}, {"pressed": False}], "a press is pressed, then released: %s", - presses) - finally: - stop.set() - try: - sock.close() - except OSError: - pass - th.join(3)