mirror of
https://github.com/openglow-org/forgefirm.git
synced 2026-10-15 17:19:55 -07:00
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).
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user