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
1c8197faa3420d3ea759343dad8ff054db4e2ee6
forgefirm/forgetest/tests/test_server.py
T
ScottW514 de324cc32f forgetest: the operator's part asked for by name, and fewer hands in a campaign 
A campaign asked a person for about eighty things: lid, button and
interlock actions, app jobs, and sixteen confirmations by eye, most of
them as popups to read and answer while the head was already moving.
This is the forgetest-only step of cutting that down.

The operator channel. A test asks for its operator's part in four ways:
ctx.ready() pre-announces a timed step and waits for the click that
starts it; ctx.notice() is a standing instruction with no button, the
test watching the machine for the result; ctx.act(channel, state) is a
machine action by name (lid, interlock, button) - a notice for the
operator today, proven done by the switch reading or an `until`
condition, recorded in evidence.actions with who performed it, and the
seam a bench actuator plugs into through runner.fixture; ctx.confirm()
stays for the yes/no the evidence cannot answer. Tests declare
`actions`; a `precheck` refuses a start the machine cannot honor (a
reason, no result, a queue skips it and carries on).

The page shows what you will do before it is asked: the running test's
steps, a queue's attended tests still waiting, or the test whose title
you clicked while idle; notices and prompts sit under it. The campaign
card no longer carries baseline, queue and leftover notes: those go to
the runner journal (daemon.log, syslog as `forgetest`, and the run in
progress), with a Runner journal button in the footer.

The catalog, 43 tests (27 auto, 8 operator, 8 live; was 45: 25/12/8):
cloud.mode-switch absorbs cloud.hunt-lid-open and cloud.gfhome-homing
(the connect made with the lid open, the hunt judged lid-open with its
Z cycle, the re-hunt waited out after the close, the switch back, then
$H judged by gfhome's own "homing complete" line with its motion
windows; precheck homing_mode = gfcloud). kernel.fire-line is auto with
the HV-not-good precheck, camera.snapshot is auto (a second frame with
the lid lamp off differs and is smaller), motion.jog-roundtrip is auto
(the head accelerometer per leg, the supervisor's own witness). The
remaining attended tests use Ready gates and act(); the head's beam
detector and the button LEDs replace the eye, leaving two confirms: the
emission witness's mark and the app's display in cloud.pause-resume.

Proof: tests/test_operator.py (the channel, the precheck, the journal),
the cloud replays re-targeted to the merged round trip over a fake grbl
and the bench excerpts, 196 host tests green, coverage lint 0 uncovered.
Every re-ported attended test is owed one bench run on the next dev
image (BRINGUP). Catalog consequence: the merged and reclassified
tests' implementation hashes move; nothing else is invalidated.
2026-08-22 16:36:03 -04:00

425 lines
20 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):
st, payload, _ = self.call_h(method, path, body, token, headers)
return st, payload
def call_h(self, method, path, body=None, token=True, headers=None):
"""As call, plus the response headers."""
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
rh = r.headers
except urllib.error.HTTPError as e:
raw = e.read()
st = e.code
rh = e.headers
ct = rh.get("Content-Type", "")
if "json" in ct:
return st, json.loads(raw.decode()), rh
return st, raw, rh
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_01c_connection_is_kept_alive(self):
"""A poll per second over a fresh TCP connection each time is
waste the board does not need to pay."""
import http.client
c = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
try:
for _ in range(3):
c.request("GET", "/state", headers={"Host": "127.0.0.1:%d" % self.port})
r = c.getresponse()
r.read()
self.assertEqual(r.status, 200)
self.assertEqual(r.version, 11)
self.assertNotEqual((r.getheader("Connection") or "").lower(), "close")
self.assertIsNotNone(r.getheader("Content-Length"))
finally:
c.close()
def test_01d_refused_post_does_not_poison_the_connection(self):
"""A POST refused before its body is read leaves that body in the
socket. On a kept-alive connection the next read would take it for
a request line, so a refusal must end the connection instead."""
import http.client
c = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
try:
body = json.dumps({"test": "fake.pass"}).encode()
c.request("POST", "/start", body=body,
headers={"Host": "127.0.0.1:%d" % self.port,
"Content-Type": "application/json"}) # no token
r = c.getresponse()
payload = json.loads(r.read().decode())
self.assertEqual(r.status, 403)
self.assertEqual(payload["error"], "authentication required")
self.assertEqual((r.getheader("Connection") or "").lower(), "close",
"an unread body was left on a connection kept alive")
finally:
c.close()
# the server is still healthy, and the body was never taken for a request
st, d = self.call("GET", "/state")
self.assertEqual(st, 200)
self.assertIsNone(d["running"])
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")
import logging
from forgetest.runner import journal
seen = []
class Catch(logging.Handler):
def emit(self, rec):
seen.append(rec.getMessage())
h = Catch()
journal.addHandler(h)
try:
r = Runner(self.log, self.man, self.reg, self.bench)
finally:
journal.removeHandler(h)
# the recovery is a journal line, not a page message
self.assertTrue(any("recovered" in m for m in seen), seen)
self.assertNotIn("messages", r.state()[0])
self.assertFalse(os.path.exists(marker))
def test_10_state_is_conditional(self):
"""The page polls; an unchanged state must cost a 304, and the
ETag must move as soon as the state does.
Runs last because it invalidates: timestamps are whole seconds, and
a PASS stamped in the same second as an invalidate is deliberately
not inheritable, so an invalidate early in this class would decide
the inheritance the earlier tests are checking.
"""
st, d, hdrs = self.call_h("GET", "/state")
self.assertEqual(st, 200)
etag = hdrs.get("ETag")
self.assertTrue(etag and etag.startswith('"'), "no ETag on /state")
st, _, hdrs2 = self.call_h("GET", "/state", headers={"If-None-Match": etag})
self.assertEqual(st, 304)
self.assertEqual(hdrs2.get("ETag"), etag)
# a stale validator must not be honored
st, _, _ = self.call_h("GET", "/state", headers={"If-None-Match": '"stale"'})
self.assertEqual(st, 200)
# and the validator moves with the state
self.call("POST", "/invalidate", {"reason": "etag check"})
st, _, hdrs3 = self.call_h("GET", "/state", headers={"If-None-Match": etag})
self.assertEqual(st, 200)
self.assertNotEqual(hdrs3.get("ETag"), etag)
if __name__ == "__main__":
unittest.main()
Reference in New Issue View Git Blame Copy Permalink