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

bench: every board-runnable tool ported to the bench page

Browse Source 
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.
This commit is contained in:
ScottW514
2026-08-16 16:28:39 -04:00
parent 03947cd94c
commit cd8a01a3d9
21 changed files with 691 additions and 316 deletions
Download Patch File Download Diff File Expand all files Collapse all files
forgetest/tests/test_bench_registry.py
+114
View File
@@ -0,0 +1,114 @@
"""The bench registry against scripts/bench: every python tool is
registered (or named as not-a-tool), every registered script exists,
every argument spec builds a command line, every script compiles, and
the shared helper resolves host and local mode as documented."""
import glob
import os
import py_compile
import subprocess
import sys
import unittest
import helpers # noqa: F401 (sys.path)
from forgetest import bench as bench_mod
REPO = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
BENCH = os.path.join(REPO, "scripts", "bench")
class RegistryTests(unittest.TestCase):
def setUp(self):
self.bench = bench_mod.Bench(tool_dir=BENCH, index_path=os.devnull)
self.scripts = sorted(os.path.basename(p) for p in glob.glob(os.path.join(BENCH, "*.py")))
def test_every_python_tool_is_registered(self):
registered = {t["script"] for t in bench_mod.TOOLS}
missing = [s for s in self.scripts if s not in registered and s not in bench_mod.NOT_TOOLS]
self.assertEqual(missing, [], "scripts/bench python files missing from the registry")
def test_every_registered_script_exists(self):
for t in bench_mod.TOOLS:
self.assertTrue(os.path.exists(os.path.join(BENCH, t["script"])), t["script"])
def test_ids_unique_and_fields_valid(self):
ids = [t["id"] for t in bench_mod.TOOLS]
self.assertEqual(len(ids), len(set(ids)))
for t in bench_mod.TOOLS:
self.assertIn(t["safety"], ("dry", "takeover", "scope", "live"), t["id"])
self.assertIn(t["where"], ("board", "host"), t["id"])
self.assertIsInstance(t["ported"], bool, t["id"])
self.assertTrue(t["desc"], t["id"])
for a in t.get("args", []):
self.assertIn(a["type"], ("str", "int", "float", "choice"), (t["id"], a["name"]))
if a["type"] == "choice":
self.assertIn(a["default"], a["choices"], (t["id"], a["name"]))
def test_ported_tools_build_a_command_with_defaults(self):
for t in bench_mod.TOOLS:
if not t["ported"]:
continue
ok, argv, err = self.bench.command(t, {})
self.assertTrue(ok, "%s: %s" % (t["id"], err))
self.assertEqual(argv[1], os.path.join(BENCH, t["script"]))
def test_unported_tools_are_host_only(self):
# what stays unported is what cannot run on the machine at all
for t in bench_mod.TOOLS:
if not t["ported"]:
self.assertEqual(t["where"], "host", t["id"])
self.assertIn("not a bench-page tool", t["desc"], t["id"])
def test_scripts_compile(self):
for s in self.scripts:
py_compile.compile(os.path.join(BENCH, s), doraise=True)
class GfbenchTests(unittest.TestCase):
"""gfbench.py in a subprocess (module-level host resolution)."""
def run_snippet(self, code, env_extra):
env = dict(os.environ)
env.pop("GF_HOST", None)
env.pop("FORGETEST_BENCH_DATA", None)
env.update(env_extra)
r = subprocess.run([sys.executable, "-c", code], cwd=BENCH, env=env,
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, timeout=30)
return r.returncode, r.stdout.strip()
def test_host_mode(self):
rc, out = self.run_snippet("import gfbench; print(gfbench.HOST, gfbench.LOCAL, gfbench.SSH[0])",
{"GF_HOST": "192.0.2.7", "GF_SSH": "fakessh -x"})
self.assertEqual(rc, 0, out)
self.assertEqual(out, "192.0.2.7 False fakessh")
def test_local_when_named_local(self):
rc, out = self.run_snippet("import gfbench; print(gfbench.HOST, gfbench.LOCAL); "
"print(gfbench.board('echo hi').strip())",
{"GF_HOST": "127.0.0.1"})
self.assertEqual(rc, 0, out)
self.assertEqual(out.splitlines(), ["127.0.0.1 True", "hi"])
def test_refuses_without_a_machine(self):
if os.path.isdir("/sys/glowforge"):
self.skipTest("running on the machine")
rc, out = self.run_snippet("import gfbench", {})
self.assertNotEqual(rc, 0)
self.assertIn("GF_HOST", out)
def test_degc_and_data_dir(self):
import tempfile
d = tempfile.mkdtemp(prefix="forgetest-bench-")
rc, out = self.run_snippet(
"import gfbench, os; print(round(gfbench.degc(700), 2), gfbench.degc('x') != gfbench.degc('x'), "
"os.path.dirname(gfbench.data_path('f.json')) == os.environ['FORGETEST_BENCH_DATA'])",
{"GF_HOST": "127.0.0.1", "FORGETEST_BENCH_DATA": d})
self.assertEqual(rc, 0, out)
val, nan, indir = out.split()
# a plausible coolant reading (700 counts is room temperature territory)
self.assertTrue(15.0 < float(val) < 35.0, val)
self.assertEqual(nan, "True")
self.assertEqual(indir, "True")
if __name__ == "__main__":
unittest.main()
forgetest/tests/test_server.py
+16 -2
View File
@@ -62,7 +62,9 @@ class ServerTests(unittest.TestCase):
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 sys, time\nprint('args', sys.argv[1:])\nsys.stdout.flush()\n"
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",
@@ -73,6 +75,8 @@ class ServerTests(unittest.TestCase):
{"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"))
@@ -261,7 +265,7 @@ class ServerTests(unittest.TestCase):
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"])
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"})
@@ -274,6 +278,9 @@ class ServerTests(unittest.TestCase):
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
@@ -295,6 +302,13 @@ class ServerTests(unittest.TestCase):
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))