mirror of
https://github.com/openglow-org/forgefirm.git
synced 2026-09-28 01:01:12 -07:00
ScottW514
4aaedc8080
A baseline pass brackets every test and bench tool: before the run the machine is verified against the fresh-boot idle state and anything off it is restored; after the run - pass, fail, or abort - it is restored again. Fixed items are the resting values the boot establishes (module defaults, forgectrl's start-up writes, the GRBL controller's init writes) and forgectrl's idle picture (controller running with motion verified, no diagnostic, camera and cooling engines idle); preserved items (lid lamp level, position counters, settings map, controller mode) are captured before and handed back after. Deviations are leftovers: in the run pane, in the result's evidence, and on the page - attributed to the previous run when found before, to the run itself when found after. forgetest takes a fresh-boot reference once per boot (within ten minutes of boot, after the supervisor settles) as the session's resting lid-lamp level and the check on the fixed values; the values were confirmed against a fresh boot of the dev image on the bench (step_freq rests at 28160, the controller's default tick, not the probe's 10000). Takeover runs capture the controller-owned kernel attributes on entry and write them back before forgectrl restarts: the bench found the kernel tests leaving motor_lock=15 behind, which masked the supervisor's liveness probe - no motion by construction, a false driver-wedge verdict, the rail-off ladder, and finally motion-fault. The takeover wrapper also waits for the supervisor to settle on both sides (moved into baseline). Catalog consequence: none beyond the runner; the tests' own drills are unchanged.
313 lines
14 KiB
Python
313 lines
14 KiB
Python
"""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 sys, time\nprint('args', sys.argv[1:])\nsys.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": []}]
|
|
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"])
|
|
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"]))
|
|
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"]))
|
|
# 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()
|