diff --git a/forgetest/forgetest/suite/__init__.py b/forgetest/forgetest/suite/__init__.py index ff2245c..45e0a06 100644 --- a/forgetest/forgetest/suite/__init__.py +++ b/forgetest/forgetest/suite/__init__.py @@ -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 diff --git a/forgetest/forgetest/suite/evmore.py b/forgetest/forgetest/suite/evmore.py new file mode 100644 index 0000000..6a3b8f2 --- /dev/null +++ b/forgetest/forgetest/suite/evmore.py @@ -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) diff --git a/forgetest/forgetest/suite/extlife.py b/forgetest/forgetest/suite/extlife.py new file mode 100644 index 0000000..e3fbe7d --- /dev/null +++ b/forgetest/forgetest/suite/extlife.py @@ -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)