mirror of
https://github.com/openglow-org/forgefirm.git
synced 2026-09-27 08:41:13 -07:00
forgetest: events.button-telemetry and exthost.lifecycle
Two new tests, each in its own module so that no other test's fingerprint moves. events.button-telemetry (suite/evmore.py): a stream opened at forgectrl's read-only listener, from a loopback source of its own, gets a telemetry.tick within 14 s with exactly its fields (phase, verdict, fire_ok, down_c, up_c, state, lid). With the machine idle and nothing waiting for the button, one press (the fixture's, or the operator's) arrives as pressed true and then false, in that order. That a press the machine is waiting for is no event, and the update and setup edges, are events_test's. exthost.lifecycle (suite/extlife.py): a package that follows its feed is running when extensions are turned off. It reads ext.shutdown with the host's reason and writes it to its data directory before its process ends, the host stops it a second later, and nothing is frozen in that second. Its covers name forgeext's run, machine (the armed question asked during the grace), evfeed, and api. Proof: on the bench reference, with forgectrl a9c45fc and forgeext dbb00e5 bind-mounted, both PASS. exthost.lifecycle FAILed on the host before its grace fix (the service frozen for want of an armed reading) and PASSed after it. forgetest's unit tests pass (451), and the coverage lint passes with --enforce.
This commit is contained in:
ScottW514
@@ -24,3 +24,5 @@ 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
|
||||
|
||||
@@ -0,0 +1,100 @@
|
||||
# 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="auto", mode="grbl", est_min=2,
|
||||
covers=[("forgectrl", "src/events.*"), ("forgectrl", "src/main.c"), ("forgectrl", "src/status.*")],
|
||||
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)
|
||||
@@ -0,0 +1,154 @@
|
||||
# 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)
|
||||
Reference in New Issue
Block a user