vh/forgefirm
1
0
Fork 0
mirror of https://github.com/openglow-org/forgefirm.git synced 2026-10-16 01:29:55 -07:00
Code Issues Packages Projects Releases Wiki Activity

forgetest tests: the confirmed trims of the tool's own unit tests

Browse Source 
The cases that tested a mock, a timer, a wording or a table, or that
replayed a judgment a host test already holds, leave forgetest/tests.
What catches a false PASS, keeps the bench safe, or guards the gate
stays.

- test_baseline: the fake sysfs tree is a mixin with no test_ methods,
  so ExtensionFreeTests and BaselineModeTests no longer re-run
  BaselineTests' 25 cases (133 runs, now 45). CoolPublishTests, the
  lamp-fade cases and TransientNotLeftoverTests go; PositionDeadbandTests
  keeps a reading past the band and an unreadable reading, both never
  excused. Both latch-relock cases stay.
- test_fixture: MdnsTests, the three timing cases, the probe's log
  wording, and nine prompt-routing variants go. A live test never runs
  unattended, the arm-press opt-in, the short-press edge, the presence
  press handing over the arm press, and the arm notice dropping at the
  press stay (35 cases, now 21).
- test_cloud_suite: three false-PASS guards stay: the latch open at a
  dark print's button, a lid stop that is not edge-driven, the kernel
  refusing the resume. The guards for the latch unlocking in the hold
  and a paused print resuming left with their catalog tests in 6a434bb.
  The rest go, with the harness pieces and the hunt-lid fixture only
  they read (41 cases, now 3).
- test_cooling_suite: GateOffTests, CriticalTierTests, FanGateTests and
  FansQuietTests go with their harness; the flow verdict's log reads
  stay: only the text after the offset, never an older verdict before it
  (the flow-under-load false-PASS guard), a rotated log read whole.
- test_queue: RunClockTests and three ordering cases go, and refusals
  are judged by a reason given, not its words; both live
  acknowledgment cases stay.
- test_takeover_position: the six cases that judge the head and move
  nothing stay.
- test_camera_home_return: the five cases where the head must not move
  stay, and so do the three that check the hand-back jog's direction and
  distance (homed, unhomed with the counters not counted twice, after
  cloud mode): a wrong sign or a double count drives the head into the
  gantry stops (15 cases, now 8).
- test_server: the keep-alive and ETag cases, the takeover and scope
  half of the bench case, and the page's self-contained asserts
  (test_page.py's) go.
- test_operator: the wording cases, the fixture case (test_fixture.py's)
  and JournalTests go.
- test_mode: the supervisor timeout table goes.
- test_takeover_client: the listener, the offline service back, the
  online client without the hunt, and GRBL under no marker stay.
- test_responsiveness: PageTests and ImplementationCostTests go; the
  log cache keeps appends, a partial line, a replaced file and a fresh
  list, and the fingerprint cache a changed manifest and one with no
  content hash.
- test_bench_registry: GfbenchTests and two description and argument
  pins go.
- test_installer: the log-format and TLS-wording cases and
  InstallerShapeTests go; the die case checks the reason is in the log,
  not its format.
- test_image: KernelIdentTests and the registration pin go.

Each file's __main__ guard now ends the file, and the docstrings say what
each still proves.

Proof: each of the fifteen files passes under Linux as forgetest-ci
discovers it, and pyflakes reports nothing new in them.

Acceptance: none; test-only.
This commit is contained in:
ScottW514
2026-10-13 20:29:25 -04:00
parent 8536ae7076
commit 520d62ac05
16 changed files with 78 additions and 2848 deletions
Download Patch File Download Diff File Expand all files Collapse all files
forgetest/tests/test_fixture.py
+8 -230
View File
@@ -7,19 +7,16 @@
The fixture is a box on the network that opens the lid loop, pulls the
interlock loop and presses the button on request. What has to hold: the
tool finds it by name without a resolver on the image, speaks its API
under the key, falls back to the operator when it cannot act, moves an
operator test it can perform alone into the unattended queue (never a
live one, never one that needs a person for anything else), refuses a
prompt during such a run rather than hanging on it, releases what the
box still holds after every run, and presses the arm button only where
the bench opted in.
tool speaks its API under the key, falls back to the operator when it
cannot act, moves an operator test it can perform alone into the
unattended queue (never a live one, never one that needs a person for
anything else), refuses a prompt during such a run rather than hanging on
it, releases what the box still holds after every run, and presses the
arm button only where the bench opted in.
"""
import json
import os
import shutil
import socket
import struct
import tempfile
import threading
import time
@@ -30,7 +27,7 @@ import helpers
from forgetest import fixture as fx
from forgetest import runner as runner_mod
from forgetest.log import Log
from forgetest.runner import Context, Failed, Run, Runner
from forgetest.runner import Context, Run, Runner
KEY = "0123456789abcdef0123456789abcdef"
@@ -113,38 +110,6 @@ class FakeFixture:
self.httpd.server_close()
class MdnsTests(unittest.TestCase):
def test_query_packet(self):
q = fx.mdns_query("forgefixture.local")
self.assertEqual(q[:12], struct.pack("!HHHHHH", 0, 0, 1, 0, 0, 0))
self.assertEqual(q[12:], b"\x0cforgefixture\x05local\x00" + struct.pack("!HH", 1, 0x8001))
def _response(self, name, ip, extra_name=None):
labels = b"".join(struct.pack("B", len(p)) + p.encode() for p in name.split(".")) + b"\x00"
hdr = struct.pack("!HHHHHH", 0, 0x8400, 0, 2 if extra_name else 1, 0, 0)
rr = labels + struct.pack("!HHIH", 1, 0x8001, 120, 4) + socket.inet_aton(ip)
if extra_name:
# a second answer naming the first through a compression pointer
# to offset 12 (the first name), plus its own first label
other = struct.pack("B", len(extra_name)) + extra_name.encode() + struct.pack("!H", 0xC000 | 12)
rr += other + struct.pack("!HHIH", 1, 0x8001, 120, 4) + socket.inet_aton("10.0.0.9")
return hdr + rr
def test_answers_are_the_named_a_records(self):
data = self._response("forgefixture.local", "192.0.2.50")
self.assertEqual(fx.mdns_answers(data, "forgefixture.local"), ["192.0.2.50"])
self.assertEqual(fx.mdns_answers(data, "FORGEFIXTURE.local."), ["192.0.2.50"])
self.assertEqual(fx.mdns_answers(data, "other.local"), [])
# a query (QR clear) is never an answer
self.assertEqual(fx.mdns_answers(fx.mdns_query("forgefixture.local"), "forgefixture.local"), [])
# compression pointers are followed; the other record is not ours
data2 = self._response("forgefixture.local", "192.0.2.50", extra_name="printer")
self.assertEqual(fx.mdns_answers(data2, "forgefixture.local"), ["192.0.2.50"])
self.assertEqual(fx.mdns_answers(data2, "printer.forgefixture.local"), ["10.0.0.9"])
# garbage does not raise
self.assertEqual(fx.mdns_answers(b"\x00\x01", "forgefixture.local"), [])
class ConfigTests(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.mkdtemp(prefix="forgetest-fixture-")
@@ -231,37 +196,6 @@ class ClientTests(unittest.TestCase):
f.act("button", "press")
self.assertIn("jumper", str(cm.exception))
def test_two_presses_are_spaced_so_the_controller_sees_the_release(self):
f = self.client()
f.status()
t0 = time.time()
f.act("button", "press")
f.act("button", "press")
took = time.time() - t0
posts = [c for c in self.dev.calls if c[:2] == ("POST", "/button")]
self.assertEqual(len(posts), 2) # no 409 round trip was needed
self.assertEqual(self.dev.presses, 2)
# the second waited for the first pulse (200 ms as reported) and the gap
self.assertGreaterEqual(took, 0.2 + fx.BUTTON_GAP_S - 0.05)
def test_a_pulse_in_progress_is_waited_out_then_retried(self):
f = self.client()
f.status()
self.dev.state["button"] = "pressed" # a press this client did not time
threading.Timer(0.3, lambda: self.dev.state.__setitem__("button", "idle")).start()
f.act("button", "press")
posts = [c for c in self.dev.calls if c[:2] == ("POST", "/button")]
self.assertEqual(len(posts), 2) # the 409, then the press
self.assertEqual(self.dev.presses, 1)
def test_a_pulse_that_never_ends_is_the_fixtures_error(self):
f = self.client()
f.status()
self.dev.state["button"] = "pressed"
with self.assertRaises(fx.FixtureError) as cm:
f.act("button", "press")
self.assertIn("in progress", str(cm.exception))
def test_a_wrong_key_is_refused(self):
f = self.client(key="wrong")
with self.assertRaises(fx.FixtureError) as cm:
@@ -279,10 +213,9 @@ class ClientTests(unittest.TestCase):
self.assertTrue(f.covers("lid"))
self.assertFalse(f.covers("interlock"))
def test_probe_without_a_config_is_none_and_a_silent_box_is_logged(self):
def test_probe_without_a_config_or_with_a_silent_box_is_none(self):
lines = []
self.assertIsNone(fx.probe(lines.append, path=os.path.join(tempfile.gettempdir(), "no-such-fixture.json")))
self.assertEqual(lines, [])
tmp = tempfile.mkdtemp(prefix="forgetest-fixture-")
try:
p = os.path.join(tmp, "fixture.json")
@@ -291,7 +224,6 @@ class ClientTests(unittest.TestCase):
# port 1 on localhost: nothing listens there
f2 = fx.probe(lines.append, path=p, resolver=lambda h: None)
self.assertIsNone(f2)
self.assertTrue(lines and "running without it" in lines[-1], lines)
finally:
shutil.rmtree(tmp, ignore_errors=True)
@@ -430,39 +362,6 @@ class RoutingTests(unittest.TestCase):
self.assertEqual(av["attended"], ["r.app", "r.live"]) # needs the app; live never moves
self.assertEqual(summary["hostname"], "forgefixture")
def test_without_the_jumper_the_button_tests_stay_attended(self):
self.stub.button_enabled = False
av, _ = self.selection()
self.assertIn("r.btn", av["attended"])
self.assertIn("r.lid", av["unattended"])
def test_no_fixture_is_the_old_routing(self):
fx.probe = lambda log, path=None, resolver=None: None
self.runner._fixture_probed = 0.0
av, summary = self.selection()
self.assertEqual(av["unattended"], ["r.auto"])
self.assertIsNone(summary)
def test_a_queue_started_during_a_probe_waits_for_the_fixture(self):
# The daemon just came up: nothing probed yet, a page poll starts
# the first probe (slow: an mDNS answer), and the queue start
# lands while it is in flight. The start must see the fixture.
stub = self.stub
def slow_probe(log, path=None, resolver=None):
time.sleep(0.5)
return stub
fx.probe = slow_probe
self.runner._fixture_probed = 0.0
self.runner.fixture = None
poll = threading.Thread(target=self.runner.probe_fixture)
poll.start()
time.sleep(0.1)
order = self.runner.batch_selection("unattended")
poll.join()
self.assertIn("r.lid", order)
self.assertIn("r.btn", order)
def run_queue(self, group, ack_live=False):
ok, msg, order = self.runner.start_batch(group, ack_live=ack_live)
self.assertTrue(ok, msg)
@@ -495,20 +394,6 @@ class RoutingTests(unittest.TestCase):
asks = self.last_result("r.asks")
self.assertIn("declare the step in hands=", asks["message"])
def test_a_failing_box_hands_the_action_to_the_operator(self):
self.stub.fail = True
run = Run("test", "r.lid", "r.lid")
ctx = Context(run, self.runner, self.reg["r.lid"])
self.runner.probe_fixture(force=True)
seen = []
ctx.wait_for = lambda cond, timeout, poll=0.25: 1.0 # the machine shows the state
run.set_notice = lambda text: seen.append(text)
ctx.act("lid", "open")
rec = run.evidence["actions"][0]
self.assertEqual(rec["by"], "operator")
self.assertIn("the box is off", rec["fixture_error"])
self.assertTrue(seen and "lid" in seen[0].lower())
def test_a_failing_box_in_an_unattended_run_ends_the_test_as_an_error(self):
# nobody is in the room: the run must not wait ACT_TIMEOUT_S for a
# hand that is not there; it ends at once, as the harness's error
@@ -563,36 +448,6 @@ class RoutingTests(unittest.TestCase):
self.assertEqual(rec[0]["by"], "operator")
self.assertTrue(any("press the button on the machine" in ln.lower() for ln in run.lines))
def test_presence_is_proved_once_per_test_not_once_per_gate(self):
"""A test with two armed halves reaches the ready gate twice. The
second one must not ask for another press: the operator proved
presence a moment earlier and the actuator has the presses. The
setup line still goes up, because the second half may want the
scrap moved."""
run = Run("test", "r.live", "r.live")
ctx = Context(run, self.runner, self.reg["r.live"])
self.runner.probe_fixture(force=True)
asked = []
run.ask = lambda q, o: asked.append(q)
self.press_button(False)
threading.Timer(0.3, self.press_button, args=(True,)).start()
threading.Timer(0.8, self.press_button, args=(False,)).start()
ctx.ready("First half.")
self.assertTrue(run.fixture_takeover)
presses = [r for r in run.evidence["actions"] if r["state"] == "presence"]
self.assertEqual(len(presses), 1)
# the second gate: returns at once, no new press, notice still shown
seen = []
run.set_notice = lambda text: seen.append(text)
t0 = time.time()
ctx.ready("Second half. Move the scrap.")
self.assertLess(time.time() - t0, 1.0)
self.assertEqual(asked, [])
self.assertEqual(len([r for r in run.evidence["actions"] if r["state"] == "presence"]), 1)
self.assertIn("Second half. Move the scrap.", seen)
self.assertTrue(any("presence already proved" in ln for ln in run.lines))
def test_the_presence_press_hands_the_arm_press_to_the_actuator(self):
"""The bench's standing opt-in is not needed once the operator has
proved presence: that press is what the opt-in existed to
@@ -613,68 +468,6 @@ class RoutingTests(unittest.TestCase):
finally:
runner_mod.hw.button_lit = saved
def test_the_press_prompt_is_what_the_actuator_presses_on(self):
"""The machine's own press prompt, not a button LED read from
here: the button is lit through parts of a live check that are
not the arm, and a press then lands before the job waits for it
and is lost. press_now presses at the moment the check asks, with
no LED reading and no waiting thread."""
run = Run("test", "r.live", "r.live")
ctx = Context(run, self.runner, self.reg["r.live"])
self.runner.probe_fixture(force=True)
self.stub.arm_press = False
run.fixture_takeover = True
saved = runner_mod.hw.button_lit
runner_mod.hw.button_lit = lambda: False # never consulted
try:
self.assertTrue(ctx.press_now())
self.assertIn(("button", "press"), self.stub.acts)
finally:
runner_mod.hw.button_lit = saved
rec = [r for r in run.evidence["actions"] if r["state"] == "arm"][-1]
self.assertEqual(rec["by"], "fixture")
def test_the_press_prompt_goes_to_the_operator_without_an_actuator(self):
"""No actuator, or one that never took this test's presses: the
notice goes up for a person, who is looking at the same prompt."""
run = Run("test", "r.live", "r.live")
ctx = Context(run, self.runner, self.reg["r.live"])
self.runner.probe_fixture(force=True)
self.stub.arm_press = False
run.fixture_takeover = False
seen = []
run.set_notice = lambda text: seen.append(text)
self.assertFalse(ctx.press_now("Press it now."))
self.assertNotIn(("button", "press"), self.stub.acts)
self.assertIn("Press it now.", seen)
def test_the_arm_press_waits_as_long_as_the_card_says_for_the_light(self):
"""A card that settles the coolant before it lights the button
takes minutes; the caller names the wait, and the actuator presses
when the light comes instead of handing the press to a person at
the default 60 s (found on the bench by the flow-load card)."""
run = Run("test", "r.live", "r.live")
ctx = Context(run, self.runner, self.reg["r.live"])
self.runner.probe_fixture(force=True)
run.fixture_takeover = True
lit_at = time.time() + 0.6
saved = runner_mod.hw.button_lit
runner_mod.hw.button_lit = lambda: time.time() >= lit_at
seen = []
run.set_notice = lambda text: seen.append(text)
try:
self.assertTrue(ctx.arm_press(lit_timeout=5))
deadline = time.time() + 5
while ("button", "press") not in self.stub.acts and time.time() < deadline:
time.sleep(0.05)
self.assertIn(("button", "press"), self.stub.acts)
self.assertEqual([n for n in seen if n], []) # nobody was asked
rec = run.evidence["actions"][-1]
self.assertEqual(rec["by"], "fixture")
self.assertGreaterEqual(rec["took_s"], 0.5)
finally:
runner_mod.hw.button_lit = saved
def test_an_actuator_lost_after_the_takeover_is_said_out_loud(self):
"""Falling back to the operator without a word is how one dropped
actuator becomes a press nobody can account for afterwards."""
@@ -689,21 +482,6 @@ class RoutingTests(unittest.TestCase):
rec = run.evidence["actions"][-1]
self.assertTrue(rec.get("fixture_lost"))
self.assertEqual(rec["by"], "operator")
self.assertTrue(any("WARNING" in ln and "now gone" in ln for ln in run.lines))
def test_ready_still_asks_the_page_with_no_actuator(self):
run = Run("test", "r.live", "r.live")
ctx = Context(run, self.runner, self.reg["r.live"])
self.runner.fixture = None
asked = []
def ask(q, o):
asked.append((q, list(o)))
return "Ready"
run.ask = ask
ctx.ready("LIVE FIRE. Scrap under the head.")
self.assertEqual(asked, [("LIVE FIRE. Scrap under the head.", ["Ready", "Cannot"])])
self.assertFalse(run.fixture_takeover)
def test_the_arm_press_is_the_operators_unless_the_bench_opted_in(self):
run = Run("test", "r.live", "r.live")