vh/forgefirm
1
0
Fork 0
mirror of https://github.com/openglow-org/forgefirm.git synced 2026-09-28 01:01:12 -07:00
Code Issues Packages Projects Releases Wiki Activity
Files
d218b3de25bc4c96fb18e78fb840c09ecfd00a93
forgefirm/forgetest/tests/test_server.py
T
ScottW514 cd8a01a3d9 bench: every board-runnable tool ported to the bench page 
The remaining bench diagnostics run from forgetest's #bench tab. The
tools that also run from a LAN host share scripts/bench/gfbench.py:
GF_HOST names a remote machine (host mode, sysfs through ssh, Grbl and
forgectrl over the LAN); unset, the tool runs on the board itself
(local mode, sysfs directly, everything on 127.0.0.1), which is how the
page runs them - with GF_HOST=127.0.0.1, the panel token in GF_TOKEN
and their data files under <data>/bench/ (FORGETEST_BENCH_DATA). The
helper also reads a machine setting from forgectrl, or from the settings
file on the board while forgectrl is stopped.

Ported: pwm_sweep / pwm_hold (scope = a takeover; the latch relocked,
the write refused if FIRE or LASER_ON reads active), pwm_stream_test
(PASS/FAIL exit), flow_characterize, flow_recheck_char,
flow_warm_validate and flow_matrix (takeovers: forgectrl owns the
thermal hardware, so the page's takeover replaces the tools' own
controller stop/restart, whose command line predated the supervisor;
results and logs in the bench data directory), flow_sustained,
fan_test, temp_calibrate (dry; watch bounded in seconds; the threshold
and the coolant conversion from the shared code), flow_escalate_drill
(cool_confirm_max_s shortened through forgectrl's settings for the
drill and restored; the setting's minimum is the default budget), and
live_fire_drills (<drill> [S] [F], all six drills, host from GF_HOST,
token from the board). flow_matrix joins the registry. What stays
unported cannot run against the machine at all: the two null-sink CI
harnesses and the .puls decoder.

Runner: a scope tool runs inside the takeover wrapper; the bench
environment above is passed to every tool. Tests: test_bench_registry
(registry <-> scripts/bench consistency, every ported tool builds its
command line, every script compiles, gfbench host/local modes) and the
server test (scope tool takeover, the environment reaching the tool).
Local mode smoke-run on the bench (temp_calibrate watch, setting, token)
from /tmp, removed after.

No catalog consequence: bench tools are not image components (dev-only
forgetest); the acceptance catalog is unchanged.
2026-08-16 16:28:39 -04:00

327 lines
15 KiB
Python
Raw Blame History

"""Runner + HTTP API end to end on localhost with a fake catalog and a
fake bench tool. No hardware, no forgectrl."""
import json
import os
import shutil
import sys
import tempfile
import threading
import time
import unittest
import urllib.error
import urllib.request
import helpers
from forgetest import bench as bench_mod
from forgetest import catalog, server
from forgetest.log import Log
from forgetest.runner import Failed, Runner
def t_pass(ctx):
ctx.log("hello")
ctx.evidence["k"] = 1
def t_prompt(ctx):
ans = ctx.prompt("Did the light blink?", ("Yes", "No"))
if ans != "Yes":
raise Failed("operator said no")
def t_fail(ctx):
ctx.check(False, "deliberate")
def t_slow(ctx):
ctx.sleep(30)
def t_error(ctx):
raise RuntimeError("boom")
class ServerTests(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.tmp = tempfile.mkdtemp(prefix="forgetest-")
os.environ["FORGETEST_DATA"] = cls.tmp
os.environ["FORGETEST_MARKER"] = os.path.join(cls.tmp, "marker")
cls.man = helpers.make_manifest()
cls.reg = helpers.registry(
helpers.make_test("fake.pass", [("forgectrl", "src/ui.c")], always=True, fn=t_pass),
helpers.make_test("fake.prompt", [("forgectrl", "src/auth.c")], fn=t_prompt, kind="operator"),
helpers.make_test("fake.fail", [("forgectrl", "src/cool.c")], fn=t_fail),
helpers.make_test("fake.slow", [("forgectrl", "src/main.c")], fn=t_slow),
helpers.make_test("fake.error", [("forgectrl", "src/main.c")], fn=t_error),
helpers.make_test("fake.live", [("grblhal-glowforge", "src/**")], fn=t_pass, kind="live"),
helpers.make_test("fake.needs", [("kernel-module-glowforge", "**")], fn=t_pass,
requires=("fake.prompt",)),
)
# a fake bench tool
cls.tooldir = os.path.join(cls.tmp, "bench")
os.makedirs(cls.tooldir)
with open(os.path.join(cls.tooldir, "echo_tool.py"), "w") as f:
f.write("import os, sys, time\nprint('args', sys.argv[1:])\n"
"print('env', os.environ.get('GF_HOST'), os.environ.get('FORGETEST_BENCH_DATA'))\n"
"sys.stdout.flush()\n"
"if 'slow' in sys.argv: time.sleep(30)\nsys.exit(0 if 'fail' not in sys.argv else 3)\n")
tools = [{"id": "echo", "title": "Echo", "script": "echo_tool.py", "safety": "dry", "where": "board",
"ported": True, "desc": "echo",
"args": [{"name": "word", "type": "str", "default": "hi", "help": ""},
{"name": "n", "type": "int", "default": 2, "help": ""}]},
{"id": "unported", "title": "U", "script": "nope.py", "safety": "dry", "where": "host",
"ported": False, "desc": "", "args": []},
{"id": "hot", "title": "H", "script": "echo_tool.py", "safety": "live", "where": "board",
"ported": True, "desc": "", "args": []},
{"id": "tk", "title": "T", "script": "echo_tool.py", "safety": "takeover", "where": "board",
"ported": True, "desc": "", "args": []},
{"id": "sc", "title": "S", "script": "echo_tool.py", "safety": "scope", "where": "board",
"ported": True, "desc": "", "args": []}]
cls.bench = bench_mod.Bench(tools, tool_dir=cls.tooldir,
index_path=os.path.join(cls.tmp, "bench.jsonl"))
cls.log = Log(os.path.join(cls.tmp, "results.jsonl"))
cls.runner = Runner(cls.log, cls.man, cls.reg, cls.bench)
cls.token = server.load_token(os.path.join(cls.tmp, "token"))
cls.app = server.App(cls.runner, cls.token, export_dir=os.path.join(cls.tmp, "export"))
cls.srv = server.make_server(cls.app, "127.0.0.1", 0)
cls.port = cls.srv.server_address[1]
cls.th = threading.Thread(target=cls.srv.serve_forever, daemon=True)
cls.th.start()
@classmethod
def tearDownClass(cls):
cls.srv.shutdown()
cls.srv.server_close()
shutil.rmtree(cls.tmp, ignore_errors=True)
# -- helpers ----------------------------------------------------------
def call(self, method, path, body=None, token=True, headers=None):
url = "http://127.0.0.1:%d%s" % (self.port, path)
hdrs = {"Host": "127.0.0.1:%d" % self.port}
if token:
hdrs["X-ForgeFIRM-Token"] = self.token
data = None
if body is not None:
data = json.dumps(body).encode()
hdrs["Content-Type"] = "application/json"
if headers:
hdrs.update(headers)
req = urllib.request.Request(url, data=data, method=method, headers=hdrs)
try:
with urllib.request.urlopen(req, timeout=10) as r:
raw = r.read()
st = r.status
ct = r.headers.get("Content-Type", "")
except urllib.error.HTTPError as e:
raw = e.read()
st = e.code
ct = e.headers.get("Content-Type", "")
if "json" in ct:
return st, json.loads(raw.decode())
return st, raw
def wait_idle(self, timeout=10):
deadline = time.time() + timeout
while time.time() < deadline:
st, d = self.call("GET", "/state")
if not d["running"]:
return d
time.sleep(0.1)
self.fail("run did not finish")
def wait_prompt(self, timeout=10):
deadline = time.time() + timeout
while time.time() < deadline:
st, d = self.call("GET", "/state")
if d["running"] and d["running"]["prompt"]:
return d["running"]["prompt"]
time.sleep(0.05)
self.fail("no prompt appeared")
# -- tests -----------------------------------------------------------------
def test_01_auth_and_page(self):
st, d = self.call("GET", "/state", token=False)
self.assertEqual(st, 200)
st, d = self.call("GET", "/", token=False)
self.assertEqual(st, 200)
self.assertIn(self.token.encode(), d)
st, d = self.call("GET", "/state", headers={"Host": "evil.example.net"})
self.assertEqual(st, 403)
st, d = self.call("GET", "/state", headers={"Origin": "http://evil.example.net"})
self.assertEqual(st, 403)
st, d = self.call("GET", "/state", headers={"Sec-Fetch-Site": "cross-site"})
self.assertEqual(st, 403)
st, d = self.call("POST", "/start", {"test": "fake.pass"}, token=False)
self.assertEqual(st, 403)
self.assertEqual(d["error"], "authentication required")
st, d = self.call("GET", "/catalog")
self.assertEqual(st, 200)
self.assertEqual(len(d["tests"]), 7)
st, d = self.call("GET", "/nope")
self.assertEqual(st, 404)
def test_02_run_pass_opens_campaign(self):
st, d = self.call("GET", "/state")
self.assertIsNone(d["campaign"])
st, d = self.call("POST", "/start", {"test": "fake.pass"})
self.assertEqual(st, 200, d)
state = self.wait_idle()
self.assertIsNotNone(state["campaign"])
self.assertEqual(state["tests"]["fake.pass"]["status"], "pass")
self.assertEqual(state["last_run"]["finished"]["result"], "PASS")
st, rec = self.call("GET", "/result?test=fake.pass")
self.assertEqual(st, 200)
self.assertEqual(rec["evidence"]["k"], 1)
# the baseline passes ran (no machine on the host: nothing to restore)
self.assertEqual(rec["evidence"]["baseline"], {"pre": [], "post": []})
self.assertTrue(any("hello" in l for l in rec["log"]))
def test_03_requires_and_live_gate(self):
st, d = self.call("POST", "/start", {"test": "fake.needs"})
self.assertEqual(st, 409)
self.assertIn("prerequisites", d["message"])
st, d = self.call("POST", "/start", {"test": "fake.live"})
self.assertEqual(st, 409)
self.assertIn("live", d["message"])
st, d = self.call("POST", "/start", {"test": "fake.live", "ack_live": True})
self.assertEqual(st, 200)
state = self.wait_idle()
self.assertEqual(state["tests"]["fake.live"]["status"], "pass")
st, rec = self.call("GET", "/result?test=fake.live")
self.assertTrue(rec["evidence"]["operator"]["ack_live"])
def test_04_prompt_flow(self):
st, d = self.call("POST", "/start", {"test": "fake.prompt"})
self.assertEqual(st, 200)
p = self.wait_prompt()
self.assertEqual(p["options"], ["Yes", "No"])
st, d = self.call("POST", "/answer", {"prompt_id": p["id"], "value": "Maybe"})
self.assertEqual(st, 409)
st, d = self.call("POST", "/answer", {"prompt_id": p["id"], "value": "Yes"})
self.assertEqual(st, 200)
state = self.wait_idle()
self.assertEqual(state["tests"]["fake.prompt"]["status"], "pass")
st, rec = self.call("GET", "/result?test=fake.prompt")
self.assertEqual(rec["answers"][0]["answer"], "Yes")
# now the dependent test may start
st, d = self.call("POST", "/start", {"test": "fake.needs"})
self.assertEqual(st, 200)
self.wait_idle()
def test_05_busy_abort(self):
st, d = self.call("POST", "/start", {"test": "fake.slow"})
self.assertEqual(st, 200)
st, d = self.call("POST", "/start", {"test": "fake.pass"})
self.assertEqual(st, 409)
st, d = self.call("POST", "/abort")
self.assertEqual(st, 200)
state = self.wait_idle()
self.assertEqual(state["tests"]["fake.slow"]["status"], "aborted")
self.assertIsNotNone(state["campaign"], "an abort does not close the campaign")
def test_06_error_and_fail_close_campaign(self):
st, d = self.call("GET", "/state")
cid = d["campaign"]["id"]
st, d = self.call("POST", "/start", {"test": "fake.error"})
state = self.wait_idle()
self.assertEqual(state["tests"]["fake.error"]["status"], "error")
self.assertIsNone(state["campaign"])
self.assertEqual(state["closed_by"], "fail")
# inherited passes survive; the core is required again
self.assertEqual(state["tests"]["fake.prompt"]["status"], "inherited")
self.assertTrue(state["tests"]["fake.pass"]["required"])
# a new start opens a new campaign
st, d = self.call("POST", "/start", {"test": "fake.pass"})
state = self.wait_idle()
self.assertIsNotNone(state["campaign"])
self.assertNotEqual(state["campaign"]["id"], cid)
def test_07_export_and_invalidate(self):
st, d = self.call("POST", "/export")
self.assertEqual(st, 200)
self.assertFalse(d["authorized"])
st, raw = self.call("GET", "/export/acceptance.json")
self.assertEqual(st, 200)
art = raw if isinstance(raw, dict) else json.loads(raw)
self.assertEqual(art["manifest_sha"], self.man.content_sha)
st, raw = self.call("GET", "/export/acceptance.md")
self.assertEqual(st, 200)
self.assertIn(b"Release authorized: NO", raw)
st, d = self.call("POST", "/invalidate", {"reason": ""})
self.assertEqual(st, 400)
st, d = self.call("POST", "/invalidate", {"reason": "new tube"})
self.assertEqual(st, 200)
st, d = self.call("GET", "/state")
self.assertEqual(d["invalidate"]["reason"], "new tube")
self.assertIsNone(d["campaign"])
for tid, ts in d["tests"].items():
self.assertNotEqual(ts["status"], "inherited", tid)
st, raw = self.call("GET", "/log", token=False)
self.assertEqual(st, 200)
self.assertIn(b'"t": "invalidate"'.replace(b" ", b""), raw.replace(b" ", b""))
def test_08_bench(self):
st, d = self.call("GET", "/bench")
self.assertEqual(st, 200)
ids = [t["id"] for t in d["tools"]]
self.assertEqual(ids, ["echo", "unported", "hot", "tk", "sc"])
st, d = self.call("POST", "/bench/start", {"tool": "unported"})
self.assertEqual(st, 409)
st, d = self.call("POST", "/bench/start", {"tool": "hot"})
self.assertEqual(st, 409)
st, d = self.call("POST", "/bench/start", {"tool": "echo", "args": {"word": "yo", "n": "x"}})
self.assertEqual(st, 409)
st, d = self.call("POST", "/bench/start", {"tool": "echo", "args": {"word": "yo", "n": 5}})
self.assertEqual(st, 200, d)
state = self.wait_idle()
self.assertEqual(state["last_run"]["kind"], "bench")
self.assertEqual(state["last_run"]["finished"]["result"], "OK")
self.assertTrue(any("['yo', '5']" in l for l in state["last_run"]["log"]))
# the bench environment: the machine is local, data under <data>/bench
self.assertTrue(any(" env 127.0.0.1 " in l and l.endswith("bench")
for l in state["last_run"]["log"]), state["last_run"]["log"])
st, d = self.call("GET", "/bench")
self.assertEqual(d["tools"][0]["last"]["result"]["result"], "OK")
# a failing tool and an aborted one
st, d = self.call("POST", "/bench/start", {"tool": "echo", "args": {"word": "fail", "n": 1}})
state = self.wait_idle()
self.assertEqual(state["last_run"]["finished"]["result"], "EXIT 3")
st, d = self.call("POST", "/bench/start", {"tool": "echo", "args": {"word": "slow", "n": 1}})
self.assertEqual(st, 200)
time.sleep(0.5)
st, d = self.call("POST", "/abort")
state = self.wait_idle()
self.assertEqual(state["last_run"]["finished"]["result"], "ABORTED")
# a takeover tool runs inside the takeover wrapper (init.d is absent on the host: rc 127)
st, d = self.call("POST", "/bench/start", {"tool": "tk"})
self.assertEqual(st, 200, d)
state = self.wait_idle(timeout=40) # two unreachable-forgectrl settle waits
log = "\n".join(state["last_run"]["log"])
self.assertIn("takeover: pulse device free", log)
self.assertIn("takeover: forgectrl start", log)
self.assertIn("baseline: forgectrl unreachable for 10 s", log)
self.assertFalse(os.path.exists(os.environ["FORGETEST_MARKER"]))
# a scope tool is a takeover too
st, d = self.call("POST", "/bench/start", {"tool": "sc"})
self.assertEqual(st, 200, d)
state = self.wait_idle(timeout=40)
log = "\n".join(state["last_run"]["log"])
self.assertIn("takeover: pulse device free", log)
self.assertIn("takeover: forgectrl start", log)
# bench runs never touched the acceptance log
recs = self.log.read()
self.assertFalse(any(r.get("t") == "result" and r.get("test") == "echo" for r in recs))
def test_09_recovery_marker(self):
marker = os.environ["FORGETEST_MARKER"]
with open(marker, "w") as f:
f.write("x fake.slow\n")
r = Runner(self.log, self.man, self.reg, self.bench)
self.assertTrue(any("recovered" in m for m in r.messages))
self.assertFalse(os.path.exists(marker))
if __name__ == "__main__":
unittest.main()
Reference in New Issue View Git Blame Copy Permalink