forgetest: the burns, the lens and the intake checks follow forgectrl's review fixes

- baseline: the lid lamp fades to a new level (the kernel's smooth
  trigger), so the hand-back reads it until it holds the wanted level
  twice in a row, for up to 3 s, before it is a leftover.
  motion.deadman failed its hand-back reading 232 against 236 0.05 s
  after the supervisor re-applied 236 to a respawned controller.
- setup.check-lens and setup.focus: grblHAL reads the lens window and
  the edge only at its start, and their restore came after the wizard
  had restarted the controller on the test's values. Each test now waits
  for the step to end, puts the settings back, and restarts the
  controller (stop, start, running and verified with a new pid) on every
  exit path; cloud mode restarts nothing.
- setup.focus and setup.floor: the served program ends with G0 Z<edge>,
  G0 X0 Y0, G92.1, M2; setup.floor requires setup.focus and checks the
  scrap's thickness against the lens reach; the burns cover lenshome.
- setup.check-intakes: the two assertions that could not fail are gone;
  an intake-count prompt fails the test unanswered (the bench
  reference's intakes both turn). The precheck's refusal names no
  setting.
- setup.check-airflow: the machine's count rules (intake 2 turning
  writes 2; silent on a one-fan machine, the count is not written).

Shared text: setup_dark.py's precheck string (all setup_dark tests,
setup_burns, setup.check-envelope), setup_burns.py's imports, covers,
docstring and preview check, cooling_intake.py's docstring; every test
they move covers paths forgectrl changed in the same work.

Proof: the forgetest host suite passes (548), with new host tests for
the lamp's fade, the restarts on pass, fail and abort, the floor's Focus
refusal and reach, the tail without G92.1, the intakes prompt, and the
airflow count rules; pyflakes is clean on the changed modules.

Acceptance: these are the tests.
This commit is contained in:
ScottW514
2026-10-12 19:53:20 -04:00
parent 005ff8008b
commit 4c5dde2549
7 changed files with 666 additions and 146 deletions
+54
View File
@@ -686,11 +686,65 @@ class BaselineModeTests(BaselineTests):
import helpers
self.fc = helpers.FakeForgectrl().start()
baseline.Baseline._unreachable_until = 0.0
# the lamp's settle, short: five polls, so a lamp that never holds
# costs the suite no real seconds
self._lamp_timing = (baseline.LID_LAMP_SETTLE_S, baseline.LID_LAMP_POLL_S)
baseline.LID_LAMP_SETTLE_S, baseline.LID_LAMP_POLL_S = 0.05, 0.01
def tearDown(self):
baseline.LID_LAMP_SETTLE_S, baseline.LID_LAMP_POLL_S = self._lamp_timing
self.fc.stop()
super().tearDown()
def lamp_reads(self, *reads):
"""The lid lamp reads `reads` in turn, then what its file holds:
the smooth trigger's fade as the baseline polls it, one read per
poll, so the script does not depend on the clock."""
seq = list(reads)
real = baseline.hw.sysfs_read
def read(attr, default=None):
if attr == baseline.LID_LAMP_ATTR and seq:
return seq.pop(0)
return real(attr, default)
baseline.hw.sysfs_read = read
self.addCleanup(setattr, baseline.hw, "sysfs_read", real)
def lamp_polls(self):
return max(2, int(round(baseline.LID_LAMP_SETTLE_S / baseline.LID_LAMP_POLL_S)))
def test_a_lamp_still_fading_to_its_idle_level_is_not_a_leftover(self):
# The supervisor re-applied 236 a moment before the check: the lamp
# reads 232 on its way there and then holds 236. Nothing to restore,
# nothing to record.
self._attr("pic/lid_led", "236")
self.lamp_reads("232", "234", "236", "236")
writes = []
real_write = baseline.hw.sysfs_write
baseline.hw.sysfs_write = lambda attr, value: (writes.append(attr), real_write(attr, value))
self.addCleanup(setattr, baseline.hw, "sysfs_write", real_write)
left = self.bl().enforce("post", captured=None)
self.assertEqual(left, [])
self.assertNotIn(baseline.LID_LAMP_ATTR, writes)
self.assertTrue(any("the fade toward a level just written" in ln for ln in self.lines), self.lines)
def test_a_fade_that_only_passes_the_idle_level_is_a_leftover(self):
# A lamp left fading to 77 passes 236 on its way down: one read at
# the level is not the level held. It is restored and recorded at
# the level it settled on.
self.lamp_reads("240", "236", *(["77"] * (self.lamp_polls() - 1)))
left = self.bl().enforce("post", captured=None)
self.assertEqual([(x.item, x.found, x.expected, x.action) for x in left],
[("pic/lid_led", "77", "236", "restored")])
self.assertEqual(self._read("pic/lid_led"), "236")
def test_a_lamp_that_never_reaches_the_level_after_the_restore_says_so(self):
# the write does not take: the restore is recorded as failed, with
# the level the lamp holds
self.lamp_reads(*(["77"] * (2 * self.lamp_polls() + 1)))
left = self.bl().enforce("post", captured=None)
self.assertEqual([(x.item, x.action) for x in left], [("pic/lid_led", "failed: reads 77")])
def run(self, result=None):
# only this class's own tests, not the base class's
if self._testMethodName not in BaselineModeTests.__dict__:
+139 -12
View File
@@ -10,7 +10,9 @@ scripted daemon: the preview and the program read, the start, the
prompts answered in order (the gantry stops, the head at home, the scrap,
its thickness in the machine's units, the arm, the machine's own press
prompt, the pick), the witnesses judged, the record read at version 1,
and the setting the burn wrote put back as found."""
the setting the burn wrote put back as found, and for the focus burn the
controller started again on the edge and the window put back. The floor
starts only with Focus on the record and a scrap the lens reaches."""
import os
import sys
import unittest
@@ -18,7 +20,7 @@ import unittest
sys.path.insert(0, os.path.dirname(__file__))
import helpers # noqa: E402
from forgetest import catalog # noqa: E402
from forgetest.runner import Context, Failed, Run # noqa: E402
from forgetest.runner import Aborted, Context, Failed, Run # noqa: E402
from forgetest.suite import setup_burns, setup_dark # noqa: E402
LIVE = ["laser.focus", "laser.floor"]
@@ -78,6 +80,8 @@ WRITES = {"laser.focus": {"lens_hall_edge_z_mm": "6.96"}, "laser.floor": {"laser
# travel check wrote, its floor.
FOUND = {"lens_hall_edge_z_mm": "4.10", "lens_stop_below_steps": "14", "lens_stop_above_steps": "20",
"laser_floor_density": "12", "ui_units": "metric"}
# What grblHAL reads at its start: the edge and the window.
LENS_KEYS = ("lens_hall_edge_z_mm", "lens_stop_below_steps", "lens_stop_above_steps")
HOME = "The head is home"
@@ -97,8 +101,9 @@ class RegistrationTests(unittest.TestCase):
self.assertIsNone(getattr(t, "precheck", None), tid)
self.assertEqual(list(self.reg["setup.focus"].requires),
["forgectrl.auth", "setup.check-motion", "laser.emission-witness", "setup.check-lens"])
# the floor burns at the focus Focus measured, and refuses without it
self.assertEqual(list(self.reg["setup.floor"].requires),
["forgectrl.auth", "setup.check-motion", "laser.emission-witness"])
["forgectrl.auth", "setup.check-motion", "laser.emission-witness", "setup.focus"])
def test_the_sheet_is_gone(self):
self.assertFalse([i for i in self.reg if i.startswith("setup.sheet")])
@@ -109,6 +114,7 @@ class RegistrationTests(unittest.TestCase):
for want in (("forgectrl", "src/wizlive.*"), ("forgectrl", "src/sheet.*"),
("forgectrl", "src/jobstream.*"), ("forgectrl", "src/curverec.*"),
("forgectrl", "src/font_hershey.*"), ("forgectrl", "src/lens.*"),
("forgectrl", "src/lenshome.*"), # the lens back on its reference at the end
("forgectrl", "src/ui/wizard.*"), ("forgectrl", "src/wizrun.h"),
("grblhal-glowforge", "src/glowforge_laser.c")):
self.assertIn(want, covers, "%s lacks %s" % (tid, want))
@@ -124,11 +130,18 @@ class RegistrationTests(unittest.TestCase):
class ScriptedDaemon:
"""The wizard runner as the suite sees it: /wiz/<id>/start opens the
first prompt, each answer opens the next (a thickness out of range in
the machine's units opens it again), the press prompt closes on the
poll after the one that served it, the last answer ends the burn with
its result and its writes, /wiz/dark reports it, GET /wiz carries the
versions, and the preview and the program are served per burn."""
first prompt (the floor ends at once, with the wizard's error, when the
record carries no Focus), each answer opens the next (a thickness that
is not a plain number, out of range in the machine's units, or, for
the floor, outside the lens reach opens it again), the press prompt
closes on the poll after the one that served it, the last answer ends
the burn with its result and its writes, /wiz/dark reports it, GET /wiz
carries the versions (the record carries Focus already, as the bench's
does), the preview and the program are served per burn, the program
ending in the wizard's tail, and the supervisor's stop and start levers
work (a start records the settings the new controller reads)."""
TAIL = ["G0 X0 Y0", "G92.1", "M2"] # after the lens's return to the edge
def __init__(self, fake):
self.fake = fake
@@ -136,7 +149,9 @@ class ScriptedDaemon:
self.dark = {"id": "", "running": False, "phase": "", "progress": 0, "elapsed_s": 0,
"log": [], "prompt": None, "result": None, "error": "",
"shots": {"lid": False, "head": False}}
self.versions = {}
self.versions = {"laser.focus": 1}
self.starts = [] # the lens settings each controller start read
self.tail = list(self.TAIL)
self.answers = {} # wid -> [values]
self.order = [] # the burns as started
self.aborted = []
@@ -207,7 +222,7 @@ class ScriptedDaemon:
lines += ["G0 Z%.2f" % z for z in ladder_z(lens["edge_z"], lens["below"], lens["above"])]
if self.stray_z is not None:
lines.append("G0 Z%.2f" % self.stray_z)
lines += ["M5", "G0 Z%.2f" % lens["edge_z"], "G0 X0 Y0", "M2"]
lines += ["M5", "G0 Z%.2f" % lens["edge_z"]] + self.tail
return 200, ("\n".join(lines) + "\n").encode(), "text/plain"
return None
@@ -219,6 +234,11 @@ class ScriptedDaemon:
return 404, {"error": "no such wizard"}
self.order.append(wid)
self.settings_at_start[wid] = dict(s)
if wid == "laser.floor" and not self.versions.get("laser.focus"):
# the floor burns at the focus Focus measured: no prompt, no motion
self.dark.update({"id": wid, "running": False, "phase": "", "prompt": None, "result": None,
"error": "run Focus first: the floor burns at the focus it measures"})
return 200, {"started": True, "id": wid}
s.update(self.start_writes.get(wid, {}))
self.lens_at_start = lens_block(s)
self.queue = list(SCRIPT[wid][0])
@@ -235,8 +255,13 @@ class ScriptedDaemon:
self.answers.setdefault(wid, []).append(value)
if p["id"] == "thickness":
imperial = (self.units or s.get("ui_units")) == "imperial"
mm = float(value) * (25.4 if imperial else 1.0)
if not 0 < mm <= 30:
try:
mm = float(value.strip()) * (25.4 if imperial else 1.0)
except ValueError:
mm = -1 # "1/8" or "3 mm": not a plain number
lens = self.lens_at_start
if not 0 < mm <= 30 or (wid == "laser.floor"
and not lens["reach_min"] <= mm <= lens["reach_max"]):
self.open_next() # the same prompt again, a new sequence number
return 200, {"ok": True}
self.thickness = mm
@@ -255,8 +280,19 @@ class ScriptedDaemon:
else:
s[k] = v
return 200, dict(s)
if path == "/controller/stop":
self.fake.state["mode"] = dict(self.fake.state["mode"], controller="standby")
return 200, {"stopped": True}
if path == "/controller/start":
m = self.fake.state["mode"]
self.fake.state["mode"] = dict(m, controller="running", motion="verified", pid=m["pid"] + 1)
self.starts.append({k: s.get(k) for k in LENS_KEYS})
return 200, {"started": True}
return None
def lever_posts(self):
return [p for p, _ in self.fake.posts if p.startswith("/controller/")]
class BurnRunTests(unittest.TestCase):
def setUp(self):
@@ -300,10 +336,23 @@ class BurnRunTests(unittest.TestCase):
self.assertEqual(fw["program_z_before"]["reach"], [-1.44, 10.19]) # the default edge
self.assertEqual(fw["program_z_after"]["reach"], [2.17, 13.8]) # the edge the burn wrote
self.assertEqual(len(fw["program_z_after"]["z"]), 14) # head, twelve lines, tail
# the program served after the burn ends in the tail at the edge the burn wrote
self.assertEqual(run.evidence["program_tail"]["laser.focus"], ["G0 Z6.96", "G0 X0 Y0", "G92.1", "M2"])
self.assertEqual(run.evidence["settings_before"],
{k: FOUND[k] for k in ("lens_hall_edge_z_mm", "lens_stop_below_steps",
"lens_stop_above_steps")})
self.assert_found() # restored
self.assert_restarted_on_found()
def assert_restarted_on_found(self, found=None):
"""grblHAL reads the edge and the window only at its start: the
hand-back is one stop and one start after the restore, the start
reading what was found."""
d = self.daemon
self.assertEqual(d.lever_posts(), ["/controller/stop", "/controller/start"])
self.assertEqual(d.starts, [found or {k: FOUND[k] for k in LENS_KEYS}])
self.assertEqual(self.fake.state["mode"]["controller"], "running")
self.assertEqual(self.fake.state["mode"]["pid"], 101)
def test_the_focus_burn_in_the_fallback_window(self):
# A head the Lens travel check has not measured: the settings hold
@@ -318,6 +367,8 @@ class BurnRunTests(unittest.TestCase):
self.assertNotIn("lens_stop_below_steps", s)
self.assertNotIn("lens_stop_above_steps", s)
self.assertEqual(s["lens_hall_edge_z_mm"], "4.10")
self.assert_restarted_on_found({"lens_hall_edge_z_mm": "4.10", "lens_stop_below_steps": None,
"lens_stop_above_steps": None})
def test_the_floor_burn(self):
run = self.run_test("setup.floor")
@@ -327,6 +378,61 @@ class BurnRunTests(unittest.TestCase):
self.assertEqual(d.previews, ["laser.floor"])
self.assertEqual(run.evidence["result"]["floor_density"], 10.0)
self.assertEqual(run.evidence["settings_before"], {"laser_floor_density": "12"})
self.assertEqual(run.evidence["focus_on_record"], 1)
self.assertEqual(run.evidence["program_tail"]["laser.floor"], ["G0 Z4.10", "G0 X0 Y0", "G92.1", "M2"])
self.assert_found()
# the controller loads the floor at every arm: nothing to restart
self.assertEqual(d.lever_posts(), [])
def test_the_floor_needs_focus_on_the_record(self):
# The floor burns at the focus Focus measured, and the wizard
# refuses without it: the test says so before the scrap is asked
# for, and nothing starts.
self.daemon.versions = {}
with self.assertRaises(Failed) as cm:
self.run_test("setup.floor")
self.assertIn("the record carries no Focus", str(cm.exception))
self.assertEqual(self.daemon.order, [])
self.assertEqual(self.daemon.previews, [])
self.assert_found()
def test_the_floors_own_refusal_without_focus_ends_the_run(self):
# Past the test's own reading (a record that lost Focus in
# between): the wizard's refusal ends the run with its word, before
# any prompt.
real = self.daemon.on_get
def on_get(path, q):
if path == "/wiz" and not self.daemon.order:
return 200, {"versions": {"laser.focus": 1}, "recommended": [], "completed": True}
return real(path, q)
self.fake.on_get = on_get
self.daemon.versions = {}
with self.assertRaises(Failed) as cm:
self.run_test("setup.floor")
self.assertIn("run Focus first", str(cm.exception))
self.assertEqual(self.daemon.order, ["laser.floor"])
self.assertEqual(self.daemon.answers, {})
self.assert_found()
def test_a_scrap_the_lens_does_not_reach_fails_before_the_floor(self):
# An edge high enough that the lens cannot focus 3.2 mm above the
# tray: the floor would refuse the thickness, so nothing starts.
self.fake.state["settings"]["lens_hall_edge_z_mm"] = "9.50"
with self.assertRaises(Failed) as cm:
self.run_test("setup.floor")
self.assertIn("leaves out the 3.2 mm scrap", str(cm.exception))
self.assertEqual(self.daemon.order, [])
def test_a_program_that_keeps_its_g92_origin_fails_before_the_burn(self):
# The tail clears the G92 origin, which the controller would
# otherwise carry to its next start under a sender's own offsets.
self.daemon.tail = ["G0 X0 Y0", "M2"]
with self.assertRaises(Failed) as cm:
self.run_test("setup.floor")
self.assertIn("not the tail", str(cm.exception))
self.assertIn("G92.1", str(cm.exception))
self.assertEqual(self.daemon.order, [])
self.assert_found()
def test_imperial_units_answer_the_thickness_in_inches(self):
@@ -392,6 +498,27 @@ class BurnRunTests(unittest.TestCase):
self.run_test("setup.focus")
self.assertIn("the focus burn wrote the lens window", str(cm.exception))
self.assert_found()
self.assert_restarted_on_found() # a failed run hands back the controller too
def test_an_aborted_focus_burn_still_restarts_the_controller_on_the_settings_as_found(self):
# The run aborted at the arm press: the burn is ended, the settings
# go back, and the controller starts again on them.
t = catalog.load_suite()["setup.focus"]
run = Run("test", t.id, t.title)
run.unattended = True
real = self.daemon.on_get
def on_get(path, q):
p = self.daemon.dark.get("prompt")
if path == "/wiz/dark" and p and p["id"] == "press":
run.aborted.set()
return real(path, q)
self.fake.on_get = on_get
with self.assertRaises(Aborted):
t.fn(Context(run, None, t))
self.assertEqual(self.daemon.aborted, ["laser.focus"])
self.assert_found()
self.assert_restarted_on_found()
def test_a_focus_ladder_outside_the_window_held_fails(self):
self.daemon.patch["laser.focus"] = {"window": {"below": 10, "above": 12, "reach_min": -0.07,
+189 -45
View File
@@ -8,8 +8,8 @@ takeover ones marked so, every covers map naming the check runner, and
the two that put the intake fan count back as found refused unless it is
set; and the check driver itself against the fake daemon: a start, the
run polled to its end with no question asked, the result taken, the
record read at version 1, and the settings read as before or put back
as found."""
record read at version 1, the settings read as before or put back as
found, and the controller started again on a lens window put back."""
import os
import sys
import unittest
@@ -48,13 +48,17 @@ class RegistrationTests(unittest.TestCase):
self.assertIsNone(getattr(self.reg[tid], "precheck", None), tid)
def test_the_intake_count_must_be_set_before_a_check_puts_it_back(self):
# The refusal says why the test cannot start; it never names a
# setting for the operator to change.
fake = helpers.FakeForgectrl().start()
try:
for tid in COUNTED:
fake.state["settings"].pop("cool_intake_fans", None)
why = self.reg[tid].cannot_start()
self.assertIn("cool_intake_fans is unset", why, tid)
self.assertIn("Intake fans step", why, tid)
self.assertIn("the intake fans are not counted", why, tid)
self.assertIn("the setup gate holds every controller", why, tid)
self.assertNotIn("cool_intake_fans", why, tid)
self.assertNotIn("Intake fans step", why, tid)
fake.state["settings"]["cool_intake_fans"] = "2"
self.assertIsNone(self.reg[tid].cannot_start(), tid)
finally:
@@ -168,22 +172,25 @@ class SensorsCheckTests(unittest.TestCase):
class AirflowCheckTests(unittest.TestCase):
"""setup.check-airflow's judge against a scripted daemon: the check
writes its floors and the intake fans fitted, and the test reads the
fans the count says are there. A machine built with one intake fan
reports intake 2 not fitted; the bench reference has two."""
writes its floors, and the intake fan count is the machine's. Intake
2 turning proves two fans and the check writes the count as 2; intake
2 silent on a machine counted with one is reported not fitted and not
turning, and the count is not written; intake 2 silent on a machine
counted with two fails the check. The bench reference has two, both
turning."""
FLOORS = {"cool_tach_exhaust_min_rpm": "6400", "cool_tach_intake_min_rpm": "2290",
"cool_tach_air_assist_min_rpm": "6050", "cool_purge_min_current": "345",
"cool_fan_grace_s": "12"}
FAN = {"steady_rpm": 4150, "spinup_s": 6.8, "ok": True}
def result(self, intakes):
def result(self, silent2):
fans = {"exhaust": dict(self.FAN), "intake 1": dict(self.FAN), "intake 2": dict(self.FAN),
"air assist": dict(self.FAN)}
if intakes == 1:
fans["intake 2"] = {"fitted": False}
return {"fans": fans, "intake_fans": intakes, "purge_current_on": 628, "purge_current_off": 74,
"floors": dict(self.FLOORS)}
if silent2:
fans["intake 2"] = {"fitted": False, "turning": False}
return {"fans": fans, "intake_fans": 1 if silent2 else 2, "purge_current_on": 628,
"purge_current_off": 74, "floors": dict(self.FLOORS)}
def setUp(self):
import tempfile
@@ -194,25 +201,36 @@ class AirflowCheckTests(unittest.TestCase):
self.saved_root = os.environ.get("GF_SYSFS_ROOT")
os.environ["GF_SYSFS_ROOT"] = self.sysfs + "/"
self.fake = helpers.FakeForgectrl().start()
self.fake.state["settings"].update({"cool_tach_intake_min_rpm": "2300"})
# the bench reference: two intake fans, counted
self.fake.state["settings"].update({"cool_tach_intake_min_rpm": "2300", "cool_intake_fans": "2"})
self.fake.state["cool"] = {"phase": "idle", "fan_gates": {"purge": {"reading": 628, "floor": 345}}}
self.dark = {"id": "", "running": False, "phase": "", "progress": 0, "elapsed_s": 0,
"log": [], "prompt": None, "result": None, "error": "",
"shots": {"lid": False, "head": False}}
self.versions = {}
self.polls = 0
self.wizard_result = self.result(2)
self.wizard_writes = None # what the check writes; None = what its result says
self.silent2 = False # intake 2 gives no tach pulse
self.wizard_result = None # the result the check ends with; None = the daemon's own
self.wizard_writes = None # the count the check writes; None = the daemon's own rule
def on_get(path, q):
if path == "/wiz/dark":
if self.dark["running"]:
self.polls += 1
if self.polls >= 2:
r = self.wizard_result
s = self.fake.state["settings"]
if self.silent2 and s.get("cool_intake_fans") == "2":
# a silent intake 2 on a machine counted with two
self.dark.update({"running": False, "progress": 100, "result": None,
"error": "a fan did not come up to speed at the run "
"profile: see the log"})
return 200, dict(self.dark)
r = self.wizard_result or self.result(self.silent2)
s.update(r["floors"])
s["cool_intake_fans"] = self.wizard_writes or str(r["intake_fans"])
if self.wizard_writes:
s["cool_intake_fans"] = self.wizard_writes
elif not self.silent2:
s["cool_intake_fans"] = "2" # intake 2 turned: two fans
self.dark.update({"running": False, "progress": 100, "result": r})
self.versions["airflow"] = 1
return 200, dict(self.dark)
@@ -251,34 +269,77 @@ class AirflowCheckTests(unittest.TestCase):
t.fn(Context(run, None, t))
return run
def assert_put_back(self):
def count(self, n):
self.fake.state["settings"]["cool_intake_fans"] = n
def assert_put_back(self, count="2"):
s = self.fake.state["settings"]
self.assertEqual(s.get("cool_tach_intake_min_rpm"), "2300")
self.assertNotIn("cool_intake_fans", s)
self.assertEqual(s.get("cool_intake_fans"), count)
self.assertNotIn("cool_tach_exhaust_min_rpm", s)
def test_two_intake_fans_pass_and_are_put_back(self):
self.run_check()
# the bench path: both intakes turn, the count is written as 2
run = self.run_check()
self.assertEqual(run.evidence["intakes"], {"counted": "2", "found": 2, "intake_2_silent": False})
self.assert_put_back()
def test_one_intake_fan_passes_with_intake_2_not_fitted(self):
self.wizard_result = self.result(1)
def test_a_machine_counted_with_one_whose_intake_2_turns_is_counted_two_and_put_back(self):
# intake 2 turning proves two fans: the check writes 2, and the
# count goes back as found
self.count("1")
self.run_check()
self.assert_put_back("1")
self.assertIn(("/settings", {"cool_intake_fans": "1"}), self.fake.posts)
def test_one_intake_fan_counted_one_passes_and_the_count_is_not_written(self):
self.count("1")
self.silent2 = True
run = self.run_check()
self.assertEqual(run.evidence["intakes"], {"counted": "1", "found": 1, "intake_2_silent": True})
self.assert_put_back("1")
def test_a_silent_intake_2_on_a_machine_counted_with_two_fails_the_check(self):
self.silent2 = True
with self.assertRaises(Exception) as cm:
self.run_check()
self.assertIn("a fan did not come up to speed at the run profile", str(cm.exception))
self.assert_put_back()
def test_one_intake_fan_with_the_setting_left_at_two_fails(self):
self.wizard_result = self.result(1)
def test_a_silent_intake_2_whose_count_was_written_fails(self):
# a silent intake 2 never writes the count
self.count("1")
self.silent2 = True
self.wizard_writes = "2"
with self.assertRaises(Exception) as cm:
self.run_check()
self.assertIn("cool_intake_fans reads '2', the check found 1", str(cm.exception))
self.assertIn("cool_intake_fans reads '2', expected '1' (intake 2 silent", str(cm.exception))
self.assert_put_back("1")
def test_a_turning_intake_2_counted_one_fails(self):
r = self.result(False)
r["intake_fans"] = 1
self.wizard_result = r
with self.assertRaises(Exception) as cm:
self.run_check()
self.assertIn("intake 2 turned, and the check found 1 intake fans", str(cm.exception))
self.assert_put_back()
def test_a_silent_intake_2_reported_fitted_fails(self):
self.count("1")
self.silent2 = True
r = self.result(True)
r["fans"]["intake 2"] = {"fitted": True, "turning": False}
self.wizard_result = r
with self.assertRaises(Exception) as cm:
self.run_check()
self.assertIn("intake 2 with one intake fan", str(cm.exception))
self.assert_put_back("1")
def test_a_result_without_the_count_fails(self):
r = self.result(2)
r = self.result(False)
del r["intake_fans"]
self.wizard_result = r
self.wizard_writes = "2"
with self.assertRaises(Exception) as cm:
self.run_check()
self.assertIn("no intake fan count", str(cm.exception))
@@ -289,10 +350,13 @@ class AirflowCheckTests(unittest.TestCase):
class ScriptedCheck(unittest.TestCase):
"""A dark check against the fake daemon: the start, one poll that sees
the run, the next that sees its end (`finish` writes the settings and
returns the result), the record at version 1, and /status's lens block
read from the settings as the daemon reads them."""
returns the result), the record at version 1, /status's lens block
read from the settings as the daemon reads them, and the supervisor's
stop and start levers (a start records the settings the new controller
reads)."""
WID = ""
LENS_KEYS = ("lens_stop_below_steps", "lens_stop_above_steps", "lens_hall_edge_z_mm")
def setUp(self):
self.fake = helpers.FakeForgectrl().start()
@@ -303,6 +367,7 @@ class ScriptedCheck(unittest.TestCase):
self.polls = 0
self.prompt = None # a prompt the run opens on its first poll, when set
self.answers = []
self.starts = [] # the lens settings each controller start read
def on_get(path, q):
s = self.fake.state["settings"]
@@ -344,6 +409,14 @@ class ScriptedCheck(unittest.TestCase):
else:
s[k] = v
return 200, dict(s)
if path == "/controller/stop":
self.fake.state["mode"] = dict(self.fake.state["mode"], controller="standby")
return 200, {"stopped": True}
if path == "/controller/start":
m = self.fake.state["mode"]
self.fake.state["mode"] = dict(m, controller="running", motion="verified", pid=m["pid"] + 1)
self.starts.append({k: s.get(k) for k in self.LENS_KEYS})
return 200, {"started": True}
return None
self.fake.on_get = on_get
self.fake.on_post = on_post
@@ -351,6 +424,9 @@ class ScriptedCheck(unittest.TestCase):
def tearDown(self):
self.fake.stop()
def lever_posts(self):
return [p for p, _ in self.fake.posts if p.startswith("/controller/")]
def run_check(self, tid):
t = catalog.load_suite()[tid]
run = Run("test", t.id, t.title)
@@ -381,15 +457,24 @@ class LensCheckTests(ScriptedCheck):
"window": {"below": below, "above": above, "found": st["found"]}, "summary": "x"}
def assert_put_back(self):
"""The window as found, in the settings and in the controller:
grblHAL reads it only at its start, so the hand-back is one stop
and one start after the put-back, the start reading the window
as found."""
s = self.fake.state["settings"]
for k, v in self.FOUND.items():
self.assertEqual(s.get(k), v, k)
self.assertEqual(self.lever_posts(), ["/controller/stop", "/controller/start"])
self.assertEqual([{k: st[k] for k in self.FOUND} for st in self.starts], [self.FOUND])
self.assertEqual(self.fake.state["mode"]["controller"], "running")
self.assertEqual(self.fake.state["mode"]["pid"], 101)
def test_the_stops_found_are_the_window(self):
run = self.run_check("setup.check-lens")
self.assertEqual(run.evidence["window"]["settings"], ("14", "20"))
self.assertEqual(run.evidence["window"]["status"]["below"], 14)
self.assertEqual(run.evidence["result"]["stops"]["contact_above"], 22)
self.assertEqual(run.evidence["restart"], {"mode": "grbl", "pid": 100, "stop": 200, "start": 200})
self.assert_put_back()
def test_stops_not_found_write_the_fallback(self):
@@ -415,54 +500,113 @@ class LensCheckTests(ScriptedCheck):
self.assertIn(("/wiz/lens.travel/abort", {}), self.fake.posts)
self.assert_put_back()
def test_a_controller_that_does_not_restart_fails_the_hand_back(self):
# The start refused: the controller stays stopped on the stale
# window's heels, and the run says so rather than pass.
real = self.fake.on_post
def on_post(path, form):
if path == "/controller/start":
return 500, {"error": "no"}
return real(path, form)
self.fake.on_post = on_post
with self.assertRaises(Exception) as cm:
self.run_check("setup.check-lens")
self.assertIn("the controller was not restarted on the window as found", str(cm.exception))
def test_cloud_mode_restarts_nothing(self):
# The cloud client reads no lens window: nothing to restart.
self.fake.state["mode"] = dict(self.fake.state["mode"], mode="cloud")
run = self.run_check("setup.check-lens")
self.assertEqual(self.lever_posts(), [])
self.assertEqual(run.evidence["restart"], {"mode": "cloud", "pid": 100})
class IntakesCheckTests(ScriptedCheck):
"""setup.check-intakes's judge: the tachs count the intake fans the
machine held, the step writes the count, and it goes back as found."""
machine held, the step writes the count, and it goes back as found.
The tachs count only two turning fans: intake 2 giving no reading
reads the same as a machine built with one, so the step asks the
operator, and so it does when intake 1 does not come up. Both intakes
turn on the bench reference, so either question fails the test
unanswered."""
WID = "airflow.intakes"
# The step's two questions, as the daemon asks them.
SILENT_2 = ("Intake 1 turns; intake 2 gives no reading. Look at the back of the machine: how many "
"intake fans does it have? If it has two, the second one needs attention before the "
"machine cuts.")
SLOW_1 = ("The intake fans could not be counted: intake 1 did not come up to speed. Look at the back "
"of the machine: how many intake fans does it have? A fan that does not turn needs "
"attention before the machine cuts.")
def setUp(self):
super().setUp()
self.fake.state["settings"]["cool_intake_fans"] = "2"
self.count = 2
def finish(self, s):
s["cool_intake_fans"] = str(self.count)
return {"intake_fans": self.count, "intake_1_rpm": 4150.0,
"intake_2_rpm": 0.0 if self.count == 1 else 4100.0, "intake_2_silent": self.count == 1,
# both intakes turned: two, by the tachs
s["cool_intake_fans"] = "2"
return {"intake_fans": 2, "intake_1_rpm": 4150.0, "intake_2_rpm": 4100.0, "intake_2_silent": False,
"counted_by": "tachs", "summary": "x"}
def ask(self, text):
self.prompt = {"id": "intake-count", "kind": "choice", "text": text, "options": ["One", "Two"]}
def test_two_intake_fans_counted_and_put_back(self):
run = self.run_check("setup.check-intakes")
self.assertEqual(run.evidence["count"]["counted"], 2)
self.assertEqual(run.evidence["count"]["by"], "tachs")
self.assertEqual(self.fake.state["settings"]["cool_intake_fans"], "2")
self.assertIn(("/settings", {"cool_intake_fans": "2"}), self.fake.posts) # the put-back
def test_one_intake_fan_on_a_machine_built_with_one(self):
def test_a_count_other_than_the_machine_held_fails_and_puts_it_back(self):
# a machine that held one intake fan, whose tachs saw two turn
self.fake.state["settings"]["cool_intake_fans"] = "1"
self.count = 1
run = self.run_check("setup.check-intakes")
self.assertEqual(run.evidence["count"]["counted"], 1)
with self.assertRaises(Exception) as cm:
self.run_check("setup.check-intakes")
self.assertIn("the tachs count 2 intake fans, the machine held 1", str(cm.exception))
self.assertEqual(self.fake.state["settings"]["cool_intake_fans"], "1")
def test_a_count_other_than_the_machine_held_fails_and_puts_it_back(self):
self.count = 1
def test_a_silent_intake_2_is_a_fault_the_test_does_not_answer(self):
# intake 1 turns, intake 2 gives no reading: the step asks the
# operator, and on the bench reference that is a fan needing attention
self.ask(self.SILENT_2)
with self.assertRaises(Exception) as cm:
self.run_check("setup.check-intakes")
self.assertIn("the tachs count 1 intake fans, the machine held 2", str(cm.exception))
self.assertIn("asked for the count, which the tachs could not make", str(cm.exception))
self.assertIn("intake 2 gives no reading", str(cm.exception))
self.assertEqual(self.answers, [])
self.assertIn(("/wiz/airflow.intakes/abort", {}), self.fake.posts)
self.assertEqual(self.fake.state["settings"]["cool_intake_fans"], "2")
def test_the_operator_question_fails_the_check(self):
# intake 1 never came up: the step asks a person to count, which
# is not a count by the tachs
self.prompt = {"id": "intake-count", "kind": "choice", "text": "how many?", "options": ["One", "Two"]}
def test_a_silent_intake_2_on_a_machine_that_held_one_still_fails(self):
# the test never answers the operator's question, whatever was held
self.fake.state["settings"]["cool_intake_fans"] = "1"
self.ask(self.SILENT_2)
with self.assertRaises(Exception) as cm:
self.run_check("setup.check-intakes")
self.assertIn("asked for the count", str(cm.exception))
self.assertEqual(self.answers, [])
self.assertEqual(self.fake.state["settings"]["cool_intake_fans"], "1")
def test_intake_1_not_up_to_speed_fails_the_check(self):
self.ask(self.SLOW_1)
with self.assertRaises(Exception) as cm:
self.run_check("setup.check-intakes")
self.assertIn("asked for the count", str(cm.exception))
self.assertIn("intake 1 did not come up to speed", str(cm.exception))
self.assertEqual(self.answers, [])
self.assertEqual(self.fake.state["settings"]["cool_intake_fans"], "2")
def test_the_tachs_count_is_judged_by_the_count_alone(self):
# counted by the tachs, the count is the whole judgment: the speeds
# are recorded, not judged again
real = self.finish
self.finish = lambda s: dict(real(s), intake_1_rpm=None)
run = self.run_check("setup.check-intakes")
self.assertIsNone(run.evidence["count"]["intake_1_rpm"])
if __name__ == "__main__":
unittest.main()