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
05d68ba0a8bd0c80e9d6ade6f0afefebc062928b
forgefirm/forgetest/tests/test_server.py
T
ScottW514 86ce0419e5 forgetest: cloud tests stay in cloud mode; the page can ignore prerequisites 
The cloud job tests (lid-abort, lid-during-button-wait, hunt-lid-open,
pause-resume) run in cloud mode and leave the machine there: enter_cloud
reuses a live session (pid-scoped from the client's own websocket state
lines) and switches once from GRBL mode, declaring the change to the
baseline; nothing switches back. Each test judges the log from its own
window, the print by its own "print [id]: finished" line, and waits the
service's follow-up moves out (wait_quiet) before it ends. hunt-lid-open
restarts the cloud client through the supervisor's stop/start lever for
a fresh connect. The former switch-back is what failed the last bench run
of hunt-lid-open (409 machine is not idle: the service was still
re-finding the head after the lid closed).

The baseline is mode-aware: in cloud mode the client owns the GRBL
controller's init values, the lid lamp, and the position counters; the
mode itself is preserved unless the run declared the change
(Context.mode_changed); controller_mode is never handed back as a bare
setting (a bare write left the persisted mode out of step with the live
one). The undeclared-change restore now uses the captured state, which
the post pass never saw before.

The acceptance page gets an "Ignore prerequisites" switch (remembered by
the browser): POST /start {ignore_requires} starts a test whose requires
are unmet, and the run records the unmet prerequisites in its evidence
and log; they stay required for the release. The cloud tests' requires
no longer chain through cloud.mode-switch.

Proof: tests/test_cloud_suite.py replays the four tests on the bench's
own gfcloud excerpts (fixtures/) and the run loop's emitted pause/resume
lines under the real runner Context against a fake forgectrl; baseline
mode tests and the server override test; the whole suite (98) and the
coverage lint pass. Catalog consequence: cloud.* fingerprints move with
the module; the catalog hash moves with the requires.
2026-08-17 06:42:48 -04:00

339 lines
16 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"])
# the operator's override: the test runs alone and the record says so
st, d = self.call("POST", "/start", {"test": "fake.needs", "ignore_requires": True})
self.assertEqual(st, 200)
state = self.wait_idle()
self.assertEqual(state["tests"]["fake.needs"]["status"], "pass")
st, rec = self.call("GET", "/result?test=fake.needs")
self.assertEqual(rec["evidence"]["prerequisites"]["missing"], ["fake.prompt"])
self.assertTrue(rec["evidence"]["prerequisites"]["overridden"])
self.assertTrue(any("prerequisites overridden" in l for l in rec["log"]))
# its prerequisite is still required for the release
self.assertFalse(state["tests"]["fake.prompt"]["satisfied"])
self.assertFalse(state["authorized"])
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