mirror of
https://github.com/openglow-org/forgefirm.git
synced 2026-10-16 01:29:55 -07:00
ScottW514
520d62ac05
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.
729 lines
35 KiB
Python
729 lines
35 KiB
Python
# Copyright 2026 514 LLC d/b/a OpenGlow
|
|
# Written by Scott Wiederhold
|
|
# https://community.openglow.org
|
|
# SPDX-License-Identifier: MIT
|
|
|
|
"""The baseline: fixed resting values are restored, preserved values are
|
|
handed back, every deviation is a recorded leftover. Runs against a fake
|
|
sysfs tree; forgectrl is unreachable (service-side checks skip)."""
|
|
import json
|
|
import os
|
|
import shutil
|
|
import struct
|
|
import tempfile
|
|
import unittest
|
|
|
|
from forgetest import baseline
|
|
|
|
# The fixed machine tick at the mode an unset xy_microsteps reads as.
|
|
# Taken from baseline rather than typed, so a change to the default mode
|
|
# moves the expectations with it.
|
|
DEFAULT_TICK = dict(baseline.fixed_sysfs())["cnc/step_freq"]
|
|
|
|
|
|
class BaselineFixture:
|
|
"""The fake sysfs tree and LED class a baseline runs against. It has
|
|
no tests of its own, so a class that mixes it in runs only its own."""
|
|
|
|
def setUp(self):
|
|
self.tmp = tempfile.mkdtemp(prefix="forgetest-bl-")
|
|
self.sysfs = os.path.join(self.tmp, "sysfs") + os.sep
|
|
self.leds = os.path.join(self.tmp, "leds") + os.sep
|
|
for group in ("cnc", "pic", "head", "thermal"):
|
|
os.makedirs(self.sysfs + group)
|
|
for name in baseline.BUTTON_LEDS + ("lid_led",):
|
|
os.makedirs(self.leds + name)
|
|
self._led(name, "0")
|
|
# A clean machine at the mode an unset xy_microsteps reads as:
|
|
# x/y_mode, step_freq and ramp_rate are the mode's, not the x8
|
|
# literals in FIXED_SYSFS, and that is what enforce() compares
|
|
# against.
|
|
for attr, val in baseline.fixed_sysfs() + baseline.IDLE_READBACKS:
|
|
self._attr(attr, val)
|
|
self._attr("cnc/interlock_circuit", "45")
|
|
self._attr("pic/lid_led", "0")
|
|
self._pos(0, 0, 0)
|
|
os.environ["GF_SYSFS_ROOT"] = self.sysfs
|
|
os.environ["GF_LEDS_ROOT"] = self.leds
|
|
os.environ["FORGECTRL_URL"] = "http://127.0.0.1:1" # nothing listens
|
|
baseline.Baseline._unreachable_until = 0.0
|
|
self.lines = []
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.tmp, ignore_errors=True)
|
|
for k in ("GF_SYSFS_ROOT", "GF_LEDS_ROOT", "FORGECTRL_URL"):
|
|
os.environ.pop(k, None)
|
|
|
|
def _attr(self, attr, val):
|
|
with open(self.sysfs + attr, "w") as f:
|
|
f.write(str(val))
|
|
|
|
def _read(self, attr):
|
|
with open(self.sysfs + attr) as f:
|
|
return f.read().strip()
|
|
|
|
def _led(self, name, val):
|
|
with open(self.leds + name + "/brightness", "w") as f:
|
|
f.write(val)
|
|
# the class interface writes 'target'; the fake mirrors it into brightness
|
|
# only when the test asks (see _sync_leds)
|
|
|
|
def _sync_leds(self):
|
|
for name in baseline.BUTTON_LEDS:
|
|
p = self.leds + name + "/target"
|
|
if os.path.exists(p):
|
|
with open(p) as f:
|
|
v = f.read().strip()
|
|
with open(self.leds + name + "/brightness", "w") as f:
|
|
f.write(v)
|
|
|
|
def _pos(self, x, y, z):
|
|
with open(self.sysfs + "cnc/position", "wb") as f:
|
|
f.write(struct.pack("<5i", x, y, z, 0, 0))
|
|
|
|
def bl(self):
|
|
return baseline.Baseline(self.lines.append)
|
|
|
|
def _pos_bytes(self, x, y, z, processed, total):
|
|
with open(self.sysfs + "cnc/position", "wb") as f:
|
|
f.write(struct.pack("<3i2I", x, y, z, processed, total))
|
|
|
|
def _probe_state(self):
|
|
# what the kernel shows between the supervisor's motion probe and
|
|
# the GRBL controller's init writes
|
|
self._attr("cnc/motor_lock", "0")
|
|
self._attr("cnc/step_freq", "10000")
|
|
self._attr("cnc/y_mode", "1")
|
|
|
|
|
|
class BaselineTests(BaselineFixture, unittest.TestCase):
|
|
def test_clean_machine_has_no_leftovers(self):
|
|
left = self.bl().enforce("pre", captured=None)
|
|
self.assertEqual(left, [])
|
|
self.assertTrue(any("pre: clean" in l for l in self.lines))
|
|
|
|
def test_fixed_values_are_restored_and_recorded(self):
|
|
self._attr("cnc/motor_lock", "15")
|
|
self._attr("cnc/step_freq", "10000")
|
|
self._attr("cnc/streaming", "1")
|
|
left = self.bl().enforce("post", captured=None)
|
|
items = {x.item: x for x in left}
|
|
self.assertEqual(set(items), {"cnc/motor_lock", "cnc/step_freq", "cnc/streaming"})
|
|
for x in left:
|
|
self.assertEqual(x.action, "restored", str(x))
|
|
self.assertEqual(self._read("cnc/motor_lock"), "0")
|
|
self.assertEqual(self._read("cnc/step_freq"), DEFAULT_TICK)
|
|
self.assertEqual(self._read("cnc/streaming"), "0")
|
|
self.assertEqual(items["cnc/motor_lock"].found, "15")
|
|
self.assertEqual(items["cnc/motor_lock"].expected, "0")
|
|
|
|
def test_unlocked_latch_is_relocked(self):
|
|
self._attr("cnc/interlock_circuit", "5") # bit 3 clear = unlocked
|
|
left = self.bl().enforce("post", captured=None)
|
|
self.assertEqual([x.item for x in left], ["laser_latch"])
|
|
self.assertEqual(self._read("cnc/laser_latch"), "1")
|
|
|
|
def test_readonly_deviation_is_unrestorable(self):
|
|
self._attr("cnc/state", "disabled")
|
|
left = self.bl().enforce("post", captured=None)
|
|
self.assertEqual([(x.item, x.action) for x in left], [("cnc/state", "unrestorable")])
|
|
|
|
def test_button_leds_are_turned_off(self):
|
|
self._led("button_led_2", "255")
|
|
left = self.bl().enforce("post", captured=None)
|
|
self.assertEqual([x.item for x in left], ["leds/button_led_2"])
|
|
self._sync_leds()
|
|
self.assertEqual(baseline.read_led("button_led_2"), "0")
|
|
|
|
def test_preserved_position(self):
|
|
b = self.bl()
|
|
cap = b.capture()
|
|
self.assertEqual(cap["position"], [0, 0, 0])
|
|
# the run shifted the counters
|
|
self._pos(1000, 0, 0)
|
|
left = b.enforce("post", captured=cap)
|
|
items = {x.item: x for x in left}
|
|
self.assertEqual(set(items), {"position"})
|
|
# no GRBL controller on the host: the head cannot be jogged back
|
|
self.assertTrue(items["position"].action.startswith("unrestorable"), items["position"].action)
|
|
self.assertEqual(items["position"].found, [1000, 0, 0])
|
|
|
|
def test_a_lost_counter_frame_never_moves_the_head(self):
|
|
# The controller zeroes the step counters at every start and at
|
|
# every home, and rewrites its anchor then. A run that began at
|
|
# -6400 steps and restarted the controller ends at 0 with the head
|
|
# where it was: the difference is not a distance, and the hand-back
|
|
# must neither jog the head nor call it a leftover.
|
|
anchor = os.path.join(self.tmp, "grblhal.homed")
|
|
old_path, baseline.ANCHOR_PATH = baseline.ANCHOR_PATH, anchor
|
|
try:
|
|
with open(anchor, "w") as f:
|
|
f.write("0 0 0 4 startup")
|
|
self._pos(-6400, 0, 0)
|
|
b = self.bl()
|
|
cap = b.capture()
|
|
self.assertIsNotNone(cap["frame"])
|
|
os.unlink(anchor) # a controller start: a new anchor file
|
|
with open(anchor + ".new", "w") as f:
|
|
f.write("0 0 0 4 startup")
|
|
os.replace(anchor + ".new", anchor)
|
|
os.utime(anchor, ns=(5, 5)) # a freed inode can come straight back, inside one clock tick
|
|
self._pos(0, 0, 0)
|
|
left = b.enforce("post", captured=cap)
|
|
self.assertEqual([x.item for x in left], [])
|
|
self.assertTrue(any("re-zeroed its counters during the run" in l for l in self.lines), self.lines)
|
|
# the same counters with the frame intact are a displaced head, as before
|
|
self.lines.clear()
|
|
self._pos(-6400, 0, 0)
|
|
cap = b.capture()
|
|
self._pos(0, 0, 0)
|
|
left = b.enforce("post", captured=cap)
|
|
self.assertEqual([x.item for x in left], ["position"])
|
|
# and a test that vouches for the new frame is held to it
|
|
self._pos(-6400, 0, 0)
|
|
cap = b.capture()
|
|
os.utime(anchor, ns=(1, 1)) # re-zeroed, and declared (ctx.counters_rezeroed)
|
|
cap["position"], cap["rezero_declared"] = [0, 0, 0], True
|
|
self._pos(0, 0, 0)
|
|
self.assertEqual(b.enforce("post", captured=cap), [])
|
|
self._pos(1000, 0, 0)
|
|
self.assertEqual([x.item for x in b.enforce("post", captured=cap)], ["position"])
|
|
finally:
|
|
baseline.ANCHOR_PATH = old_path
|
|
|
|
def test_the_counters_are_read_at_the_modes_scale(self):
|
|
# 6400 steps is 30 mm at x32 (a return within the bound; on the
|
|
# host it stops at the missing controller) and 120 mm at x8
|
|
# (beyond the bound). Read at the x8 scale, an x32 machine's
|
|
# displaced head is never jogged back.
|
|
b = self.bl()
|
|
cap = b.capture()
|
|
self._attr("cnc/x_mode", "32")
|
|
self._pos(6400, 0, 0)
|
|
items = {x.item: x for x in b.enforce("post", captured=cap)}
|
|
self.assertEqual(items["position"].action, "unrestorable: no running GRBL controller")
|
|
self._attr("cnc/x_mode", "8")
|
|
items = {x.item: x for x in b.enforce("post", captured=cap)}
|
|
self.assertEqual(items["position"].action, "unrestorable: 120.0/0.0 mm exceeds 100 mm")
|
|
|
|
def test_a_held_controller_is_reset_before_the_return_jog(self):
|
|
# a pause test that failed while held leaves the controller in
|
|
# Hold, which refuses a jog: the baseline resets out of it first
|
|
import helpers
|
|
fc = helpers.FakeForgectrl().start()
|
|
dev = helpers.FakeGrbl().start()
|
|
try:
|
|
dev.state = "Hold:0"
|
|
dev.reset_to = "Idle"
|
|
dev.on_command = lambda line: self._pos(0, 0, 0) if line.startswith("$J=") else None
|
|
b = self.bl()
|
|
cap = b.capture()
|
|
# 4.144 mm into the held move, in the steps of the mode in force
|
|
held_mm = 4.144
|
|
self._pos(round(held_mm * baseline.xy_steps_per_mm(baseline.XY_MODE_DEFAULT)), 0, 0)
|
|
left = b.enforce("post", captured=cap)
|
|
items = {x.item: x for x in left}
|
|
self.assertTrue(items["position"].action.startswith("restored"), items["position"].action)
|
|
self.assertEqual(dev.sent[0], "^X")
|
|
jogs = [l for l in dev.sent if l.startswith("$J=")]
|
|
self.assertEqual(len(jogs), 1)
|
|
self.assertIn("X-%.3f" % held_mm, jogs[0])
|
|
self.assertTrue(any("reset out of Hold:0" in l for l in self.lines), self.lines)
|
|
finally:
|
|
dev.stop()
|
|
fc.stop()
|
|
for k in ("GRBL_HOST", "GRBL_PORT"):
|
|
os.environ.pop(k, None)
|
|
|
|
def test_ring_residue_is_a_leftover_and_blocks_the_return_jog(self):
|
|
b = self.bl()
|
|
cap = b.capture()
|
|
# the run left 40 unplayed bytes queued in the kernel ring and the
|
|
# head 1000 counts out: the hand-back stands the machine down to
|
|
# empty the ring, and only refuses the return jog when the bytes
|
|
# are still there afterwards (they are here: no daemon to restart
|
|
# a controller)
|
|
self._pos_bytes(1000, 0, 0, 100, 140)
|
|
left = b.enforce("post", captured=cap)
|
|
items = {x.item: x for x in left}
|
|
self.assertIn("pulse ring", items)
|
|
self.assertEqual(items["pulse ring"].found, "40 unplayed bytes")
|
|
self.assertIn("still queued", items["position"].action)
|
|
self.assertIn("after a controller restart", items["position"].action)
|
|
self.assertEqual(baseline.read_ring_residue(), 40)
|
|
|
|
def test_the_hand_back_stands_the_machine_down_for_residue(self):
|
|
"""It acts, it does not report: a run that left bytes in the ring
|
|
gets a controller restart out of the hand-back, because that is
|
|
what empties the ring."""
|
|
b = self.bl()
|
|
cap = b.capture()
|
|
called = []
|
|
b.stand_down = lambda why: called.append(why)
|
|
self._pos_bytes(1000, 0, 0, 100, 140)
|
|
b.enforce("post", captured=cap)
|
|
self.assertEqual(len(called), 1)
|
|
self.assertIn("unplayed bytes", called[0])
|
|
|
|
def test_clean_ring_reads_zero_residue(self):
|
|
self.assertEqual(baseline.read_ring_residue(), 0)
|
|
|
|
def test_lamp_needs_forgectrl(self):
|
|
# the lamp's idle level comes from forgectrl's settings: without the
|
|
# daemon there is nothing to compare against
|
|
self._attr("pic/lid_led", "77")
|
|
left = self.bl().enforce("pre", captured=None)
|
|
self.assertEqual(left, [])
|
|
self.assertEqual(self._read("pic/lid_led"), "77")
|
|
|
|
def test_no_sysfs_means_skip(self):
|
|
os.environ["GF_SYSFS_ROOT"] = os.path.join(self.tmp, "nope") + os.sep
|
|
left = self.bl().enforce("pre", captured=None)
|
|
self.assertEqual(left, [])
|
|
self.assertTrue(any("kernel sysfs not present" in l for l in self.lines))
|
|
|
|
def test_boot_reference_needs_a_recent_boot(self):
|
|
os.environ["FORGETEST_BOOT_ID"] = "test-boot"
|
|
try:
|
|
# no reference file, uptime unknown on a host without /proc/uptime,
|
|
# or too old: None, with the reason logged
|
|
ref = baseline.boot_reference(self.lines.append, self.tmp)
|
|
up = baseline.uptime_s()
|
|
if up is None or up > baseline.BOOT_MAX_AGE_S:
|
|
self.assertIsNone(ref)
|
|
self.assertTrue(any("no fresh-boot reference" in l for l in self.lines))
|
|
else:
|
|
# a young host: the reference is taken from the fake tree
|
|
self.assertIsNotNone(ref)
|
|
self.assertEqual(ref["sysfs"]["cnc/motor_lock"], "0")
|
|
self.assertTrue(os.path.exists(os.path.join(self.tmp, "boot-test-boot.json")))
|
|
# and loaded back the second time
|
|
self.lines[:] = []
|
|
ref2 = baseline.boot_reference(self.lines.append, self.tmp)
|
|
self.assertEqual(ref2["ts"], ref["ts"])
|
|
self.assertTrue(any("reference loaded" in l for l in self.lines))
|
|
finally:
|
|
os.environ.pop("FORGETEST_BOOT_ID", None)
|
|
|
|
def test_fixed_constants_checked_against_a_dump(self):
|
|
ref = {"sysfs": {"cnc/motor_lock": "0", "cnc/step_freq": "10000"}}
|
|
diffs = baseline.check_fixed_against(ref, self.lines.append)
|
|
self.assertEqual(diffs, ["cnc/step_freq: boot=10000 constant=%s" % DEFAULT_TICK])
|
|
|
|
|
|
# -- the reference is taken after the controller applied its config -----
|
|
|
|
def test_wait_configured_returns_once_the_controller_wrote_its_config(self):
|
|
self._probe_state()
|
|
calls = {"n": 0}
|
|
|
|
def sleep(_s):
|
|
calls["n"] += 1
|
|
if calls["n"] == 3: # the controller's init writes land
|
|
# at the mode wait_controller_configured watches by default
|
|
for attr, val in baseline.configured_markers():
|
|
self._attr(attr, val)
|
|
ok = baseline.wait_controller_configured(
|
|
self.lines.append, {"controller": "running", "mode": "grbl", "motion": "verified"},
|
|
timeout=5, sleep=sleep)
|
|
self.assertTrue(ok)
|
|
self.assertTrue(any("controller configured" in l for l in self.lines))
|
|
self.assertGreaterEqual(calls["n"], 4) # 3 polls + the settle
|
|
|
|
def test_wait_configured_times_out_and_says_so(self):
|
|
self._probe_state()
|
|
t = {"now": 0.0}
|
|
real_time = baseline.time.time
|
|
baseline.time.time = lambda: t["now"]
|
|
try:
|
|
def sleep(s):
|
|
t["now"] += s
|
|
ok = baseline.wait_controller_configured(
|
|
self.lines.append, {"controller": "running", "mode": "grbl"}, timeout=2, sleep=sleep)
|
|
finally:
|
|
baseline.time.time = real_time
|
|
self.assertFalse(ok)
|
|
self.assertTrue(any("did not apply its config" in l for l in self.lines))
|
|
|
|
def test_wait_configured_is_a_noop_outside_grbl_mode(self):
|
|
self._probe_state()
|
|
ok = baseline.wait_controller_configured(
|
|
self.lines.append, {"controller": "running", "mode": "cloud"}, timeout=1,
|
|
sleep=lambda s: self.fail("slept in cloud mode"))
|
|
self.assertTrue(ok)
|
|
ok = baseline.wait_controller_configured(
|
|
self.lines.append, {"controller": "stopped", "mode": "grbl"}, timeout=1,
|
|
sleep=lambda s: self.fail("slept with the controller stopped"))
|
|
self.assertTrue(ok)
|
|
|
|
def test_preconfig_reference_is_recognized(self):
|
|
self.assertTrue(baseline.reference_preconfig(
|
|
{"sysfs": {"cnc/motor_lock": "0", "cnc/step_freq": "10000", "cnc/y_mode": "1"}}))
|
|
# The markers are read at the reference's own mode, so a dump taken
|
|
# on an x8 machine carries the setting that says so.
|
|
self.assertFalse(baseline.reference_preconfig(
|
|
{"sysfs": {"cnc/motor_lock": "8", "cnc/step_freq": "28160", "cnc/y_mode": "8"},
|
|
"forgectrl": {"/settings": {"xy_microsteps": "8"}}}))
|
|
# a genuinely different single constant is a machine fact, not pre-config
|
|
self.assertFalse(baseline.reference_preconfig(
|
|
{"sysfs": {"cnc/motor_lock": "8", "cnc/step_freq": "10000", "cnc/y_mode": "8"},
|
|
"forgectrl": {"/settings": {"xy_microsteps": "8"}}}))
|
|
self.assertFalse(baseline.reference_preconfig({"sysfs": {}}))
|
|
|
|
# -- the XY microstep mode drives the fixed values ------------------------
|
|
|
|
def test_the_fixed_values_follow_the_microstep_mode(self):
|
|
x8 = dict(baseline.fixed_sysfs(8))
|
|
self.assertEqual(x8, dict(baseline.FIXED_SYSFS))
|
|
x16 = dict(baseline.fixed_sysfs(16))
|
|
self.assertEqual((x16["cnc/x_mode"], x16["cnc/y_mode"], x16["cnc/step_freq"], x16["cnc/ramp_rate"]),
|
|
("16", "16", "56320", "250000"))
|
|
x32 = dict(baseline.fixed_sysfs(32))
|
|
self.assertEqual((x32["cnc/x_mode"], x32["cnc/step_freq"], x32["cnc/ramp_rate"]),
|
|
("32", "112640", "500000"))
|
|
# everything else is the same list, in the same order
|
|
self.assertEqual([a for a, _ in baseline.fixed_sysfs(32)], [a for a, _ in baseline.FIXED_SYSFS])
|
|
self.assertEqual(x32["cnc/x_decay"], x8["cnc/x_decay"])
|
|
|
|
def test_the_mode_is_read_from_the_settings_with_a_default(self):
|
|
# An unset or unusable key reads as the driver's default, x32.
|
|
self.assertEqual(baseline.xy_mode_of({"xy_microsteps": "16"}), 16)
|
|
self.assertEqual(baseline.xy_mode_of({"xy_microsteps": "32"}), 32)
|
|
self.assertEqual(baseline.xy_mode_of({"xy_microsteps": "8"}), 8)
|
|
self.assertEqual(baseline.xy_mode_of({"xy_microsteps": ""}), 32)
|
|
self.assertEqual(baseline.xy_mode_of({}), 32)
|
|
self.assertEqual(baseline.xy_mode_of(None), 32)
|
|
self.assertEqual(baseline.xy_mode_of({"xy_microsteps": "24"}), 32)
|
|
self.assertEqual(baseline.xy_mode_of({"xy_microsteps": "abc"}), 32)
|
|
self.assertEqual(baseline.ref_xy_mode({"forgectrl": {"/settings": {"xy_microsteps": "16"}}}), 16)
|
|
self.assertEqual(baseline.ref_xy_mode({"forgectrl": {"/settings": None}}), 32)
|
|
self.assertEqual(baseline.ref_xy_mode({}), 32)
|
|
|
|
def test_fixed_constants_are_checked_at_the_reference_mode(self):
|
|
ref = {"sysfs": {"cnc/x_mode": "16", "cnc/step_freq": "56320", "cnc/ramp_rate": "250000"},
|
|
"forgectrl": {"/settings": {"xy_microsteps": "16"}}}
|
|
self.assertEqual(baseline.check_fixed_against(ref, self.lines.append), [])
|
|
ref["forgectrl"]["/settings"]["xy_microsteps"] = "8"
|
|
diffs = baseline.check_fixed_against(ref, self.lines.append)
|
|
self.assertEqual(diffs, ["cnc/x_mode: boot=16 constant=8",
|
|
"cnc/step_freq: boot=56320 constant=28160",
|
|
"cnc/ramp_rate: boot=250000 constant=125000"])
|
|
|
|
def test_wait_configured_watches_the_mode_in_force(self):
|
|
self._probe_state()
|
|
calls = {"n": 0}
|
|
|
|
def sleep(_s):
|
|
calls["n"] += 1
|
|
if calls["n"] == 2: # an x16 controller's init writes land
|
|
self._attr("cnc/step_freq", "56320")
|
|
self._attr("cnc/y_mode", "16")
|
|
ok = baseline.wait_controller_configured(
|
|
self.lines.append, {"controller": "running", "mode": "grbl", "motion": "verified"},
|
|
timeout=5, sleep=sleep, xy_mode=16)
|
|
self.assertTrue(ok)
|
|
self.assertTrue(any("controller configured" in l for l in self.lines))
|
|
# a reference taken under x16 is not pre-config
|
|
self.assertFalse(baseline.reference_preconfig(
|
|
{"sysfs": {"cnc/step_freq": "56320", "cnc/y_mode": "16"},
|
|
"forgectrl": {"/settings": {"xy_microsteps": "16"}}}))
|
|
self.assertTrue(baseline.reference_preconfig(
|
|
{"sysfs": {"cnc/step_freq": "10000", "cnc/y_mode": "1"},
|
|
"forgectrl": {"/settings": {"xy_microsteps": "16"}}}))
|
|
|
|
def test_stale_preconfig_reference_is_retaken_on_a_fresh_boot(self):
|
|
# Both sides of BOOT_MAX_AGE_S, with the uptime held: read from the host, it can cross the limit
|
|
# between boot_reference()'s reading and the test's (it did, on a CI runner 600 s after its boot).
|
|
os.environ["FORGETEST_BOOT_ID"] = "test-boot-2"
|
|
real = baseline.uptime_s
|
|
try:
|
|
for up, fresh in ((baseline.BOOT_MAX_AGE_S - 60, True), (baseline.BOOT_MAX_AGE_S + 60, False)):
|
|
self.lines.clear()
|
|
path = os.path.join(self.tmp, "boot-test-boot-2.json")
|
|
with open(path, "w") as f:
|
|
json.dump({"ts": "old", "sysfs": {"cnc/motor_lock": "0", "cnc/step_freq": "10000",
|
|
"cnc/y_mode": "1"}}, f)
|
|
baseline.uptime_s = lambda up=up: up
|
|
ref = baseline.boot_reference(self.lines.append, self.tmp)
|
|
if fresh:
|
|
self.assertTrue(any("retaking" in l for l in self.lines), self.lines)
|
|
self.assertNotEqual(ref["ts"], "old")
|
|
self.assertEqual(ref["sysfs"]["cnc/motor_lock"], "0")
|
|
else:
|
|
# too old to retake: the stale reference stands, marked
|
|
self.assertTrue(any("predates the controller's config" in l for l in self.lines), self.lines)
|
|
self.assertEqual(ref["ts"], "old")
|
|
finally:
|
|
baseline.uptime_s = real
|
|
os.environ.pop("FORGETEST_BOOT_ID", None)
|
|
|
|
|
|
class ExtensionFreeTests(BaselineFixture, unittest.TestCase):
|
|
"""A campaign's machine is extension-free: what a test made is removed,
|
|
what is the operator's is reported and left alone."""
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.root = os.path.join(self.tmp, "ext")
|
|
for d in ("pkg", "keys", "data"):
|
|
os.makedirs(os.path.join(self.root, d))
|
|
# a stand-in for the host's command line: remove <id> takes the package away
|
|
self.forgeext = os.path.join(self.tmp, "forgeext")
|
|
with open(self.forgeext, "w") as f:
|
|
f.write('#!/bin/sh\n[ "$1" = remove ] && rm -rf "%s/pkg/$2" && echo \'{"ok": true}\'\n' % self.root)
|
|
os.chmod(self.forgeext, 0o755)
|
|
self.pool = {}
|
|
self._saved = (baseline.hw.EXT_ROOT, baseline.hw.FORGEEXT, baseline.hw.pool_pids, baseline.EXT_STOP_S)
|
|
baseline.hw.EXT_ROOT, baseline.hw.FORGEEXT = self.root, self.forgeext
|
|
baseline.hw.pool_pids = lambda: dict(self.pool)
|
|
baseline.EXT_STOP_S = 0.2
|
|
|
|
def tearDown(self):
|
|
baseline.hw.EXT_ROOT, baseline.hw.FORGEEXT, baseline.hw.pool_pids, baseline.EXT_STOP_S = self._saved
|
|
super().tearDown()
|
|
|
|
def _package(self, id_):
|
|
os.makedirs(os.path.join(self.root, "pkg", id_, "1.0.0"))
|
|
|
|
def _key(self, name):
|
|
with open(os.path.join(self.root, "keys", name), "w") as f:
|
|
f.write("x")
|
|
|
|
def test_an_installed_package_that_does_not_run_is_nobodys_leftover(self):
|
|
self._package("org.example.notify")
|
|
self._key("owner.pub")
|
|
self.assertEqual(self.bl().enforce("pre", captured=None), [])
|
|
|
|
def test_what_a_test_left_behind_is_removed_and_the_operators_is_not(self):
|
|
self._package("org.forgetest.reference")
|
|
self._package("org.example.notify")
|
|
self._key("forgetest-reference.pub")
|
|
self._key("owner.pub")
|
|
left = {x.item: x for x in self.bl().enforce("post", captured=None)}
|
|
self.assertEqual(sorted(left), ["ext.owner_key", "ext.package"])
|
|
self.assertEqual((left["ext.package"].found, left["ext.package"].action), ("org.forgetest.reference", "restored"))
|
|
self.assertEqual((left["ext.owner_key"].found, left["ext.owner_key"].action), ("forgetest-reference.pub", "restored"))
|
|
self.assertEqual(os.listdir(os.path.join(self.root, "pkg")), ["org.example.notify"])
|
|
self.assertEqual(os.listdir(os.path.join(self.root, "keys")), ["owner.pub"])
|
|
|
|
def test_a_package_that_cannot_be_removed_says_so(self):
|
|
self._package("org.forgetest.reference")
|
|
with open(self.forgeext, "w") as f:
|
|
f.write('#!/bin/sh\necho \'{"ok": false, "error": "the lock is held"}\'\nexit 1\n')
|
|
left = self.bl().enforce("post", captured=None)
|
|
self.assertEqual([x.item for x in left], ["ext.package"])
|
|
self.assertTrue(left[0].action.startswith("failed:") and "the lock is held" in left[0].action, left[0].action)
|
|
|
|
def test_running_extensions_are_reported_and_left_alone(self):
|
|
self._package("org.example.notify")
|
|
self.pool = {4242: 800}
|
|
left = self.bl().enforce("pre", captured=None)
|
|
self.assertEqual([(x.item, x.action) for x in left], [("ext.running", "unrestorable")])
|
|
self.assertIn("org.example.notify", left[0].found)
|
|
self.assertTrue(os.path.isdir(os.path.join(self.root, "pkg", "org.example.notify")))
|
|
|
|
|
|
class CloudOwnedSysfsTests(unittest.TestCase):
|
|
"""In cloud mode the cloud client configures the machine from the
|
|
pulse header it is playing, the lens included: z_mode comes from
|
|
ZSmd through gfhardware's set_mode_from_puls. Handing the GRBL values
|
|
back under it would be the baseline configuring another controller's
|
|
machine. In GRBL mode the pair is checked as before."""
|
|
|
|
def test_the_z_pair_is_the_cloud_clients(self):
|
|
for attr in ("head/z_current", "head/z_mode"):
|
|
self.assertIn(attr, baseline.GRBL_CONTROLLER_SYSFS)
|
|
|
|
def test_the_z_pair_is_still_checked_in_grbl_mode(self):
|
|
fixed = dict(baseline.fixed_sysfs())
|
|
self.assertEqual(fixed.get("head/z_current"), "1")
|
|
self.assertEqual(fixed.get("head/z_mode"), "1")
|
|
|
|
|
|
class PositionDeadbandTests(unittest.TestCase):
|
|
"""The counters count steps and a controller's own return lands within
|
|
a few hundredths of a millimeter, not on the step: a difference that
|
|
small is quantization, not a leftover. Found on the bench reference,
|
|
where motion.lid-cancel-home returned the head twice, each within
|
|
0.04 mm, and failed the hand-back on the four steps left over."""
|
|
|
|
def test_past_the_dead_band_is_a_leftover(self):
|
|
over = int(baseline.POSITION_DEADBAND_MM * baseline.XY_STEPS_PER_MM) + 2
|
|
self.assertFalse(baseline.position_quantized([0, 0, 0], [over, 0, 0]))
|
|
self.assertFalse(baseline.position_quantized([0, 0, 0], [0, over, 0]))
|
|
|
|
def test_an_unreadable_reading_is_never_quantization(self):
|
|
self.assertFalse(baseline.position_quantized(None, [0, 0, 0]))
|
|
self.assertFalse(baseline.position_quantized([0, 0, 0], None))
|
|
|
|
|
|
class BaselineModeTests(BaselineFixture, unittest.TestCase):
|
|
"""The baseline against a fake forgectrl: what the mode in force
|
|
owns, over the same fake sysfs tree."""
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
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 cloud(self):
|
|
self.fc.state["mode"] = {"mode": "cloud", "controller": "running", "pid": 7, "motion": "verified"}
|
|
self.fc.state["settings"]["controller_mode"] = "cloud"
|
|
|
|
def nohunt_marker(self, refuse=False):
|
|
"""The no-hunt marker under the fake tree; what each POST found.
|
|
The fake client takes the marker down as gfcloud does, first thing
|
|
at its start; with refuse the POST is refused and no client starts."""
|
|
baseline.NOHUNT_MARKER = os.path.join(self.tmp, "nohunt-marker")
|
|
self.fc.state["settings"].setdefault("cloud_enabled", "1")
|
|
seen = []
|
|
|
|
def on_post(path, form):
|
|
seen.append((path, dict(form), os.path.exists(baseline.NOHUNT_MARKER)))
|
|
if refuse:
|
|
return 409, {"error": "busy"}
|
|
if os.path.exists(baseline.NOHUNT_MARKER):
|
|
os.remove(baseline.NOHUNT_MARKER)
|
|
if path == "/mode":
|
|
self.fc.state["mode"] = dict(self.fc.state["mode"], mode=form["controller"], controller="running")
|
|
elif path == "/controller/start":
|
|
self.fc.state["mode"] = dict(self.fc.state["mode"], controller="running")
|
|
return None
|
|
self.fc.on_post = on_post
|
|
return seen
|
|
|
|
def test_a_refused_switch_leaves_no_marker_for_the_next_start(self):
|
|
seen = self.nohunt_marker(refuse=True)
|
|
ok, detail = self.bl().switch_mode("cloud")
|
|
self.assertFalse(ok)
|
|
self.assertEqual(seen, [("/mode", {"controller": "cloud"}, True)])
|
|
self.assertFalse(os.path.exists(baseline.NOHUNT_MARKER))
|
|
|
|
def test_a_switch_to_cloud_starts_the_client_without_the_hunt(self):
|
|
seen = self.nohunt_marker()
|
|
ok, detail = self.bl().switch_mode("cloud")
|
|
self.assertTrue(ok, detail)
|
|
self.assertEqual(seen, [("/mode", {"controller": "cloud"}, True)])
|
|
self.assertFalse(os.path.exists(baseline.NOHUNT_MARKER)) # one start, never left behind
|
|
|
|
def test_a_switch_to_cloud_turns_cloud_mode_on_when_it_is_off(self):
|
|
# A test that declares cloud mode gets it: cloud_enabled goes to 1
|
|
# first, with the typed phrase the daemon asks for and a log line,
|
|
# and only then the mode switch.
|
|
lines = []
|
|
seen = self.nohunt_marker()
|
|
self.fc.state["settings"]["cloud_enabled"] = "0"
|
|
ok, detail = baseline.Baseline(lines.append).switch_mode("cloud")
|
|
self.assertTrue(ok, detail)
|
|
self.assertEqual(seen[0], ("/settings", {"cloud_enabled": "1", "phrase": "I UNDERSTAND"}, False))
|
|
self.assertEqual(seen[1], ("/mode", {"controller": "cloud"}, True))
|
|
self.assertTrue(any("cloud mode turned on for the test" in ln for ln in lines))
|
|
|
|
def test_a_switch_to_grbl_sets_no_marker(self):
|
|
seen = self.nohunt_marker()
|
|
self.cloud()
|
|
ok, detail = self.bl().switch_mode("grbl")
|
|
self.assertFalse(ok) # the fake has no Grbl port: the switch itself happened
|
|
self.assertEqual(seen, [("/mode", {"controller": "grbl"}, False)])
|
|
self.assertFalse(os.path.exists(baseline.NOHUNT_MARKER))
|
|
|
|
def test_the_mode_handed_back_is_cloud_without_the_hunt(self):
|
|
seen = self.nohunt_marker()
|
|
self.cloud()
|
|
b = self.bl()
|
|
cap = b.capture() # found in cloud
|
|
self.fc.state["mode"] = dict(self.fc.state["mode"], mode="grbl") # the run left it in grbl
|
|
b.enforce("post", captured=cap)
|
|
self.assertEqual(seen[0], ("/mode", {"controller": "cloud"}, True))
|
|
self.assertFalse(os.path.exists(baseline.NOHUNT_MARKER))
|
|
|
|
def test_a_standby_cloud_client_is_started_without_the_hunt(self):
|
|
seen = self.nohunt_marker()
|
|
self.cloud()
|
|
b = self.bl()
|
|
cap = b.capture()
|
|
self.fc.state["mode"] = dict(self.fc.state["mode"], controller="standby")
|
|
b.enforce("post", captured=cap)
|
|
self.assertEqual(seen[0], ("/controller/start", {}, True))
|
|
self.assertFalse(os.path.exists(baseline.NOHUNT_MARKER))
|
|
|
|
def test_grbl_mode_restores_the_controller_values_and_the_lamp(self):
|
|
self._attr("cnc/step_freq", "10000")
|
|
self._attr("pic/lid_led", "77")
|
|
left = self.bl().enforce("pre", captured=None)
|
|
self.assertEqual(sorted(x.item for x in left), ["cnc/step_freq", "pic/lid_led"])
|
|
self.assertEqual(self._read("cnc/step_freq"), DEFAULT_TICK)
|
|
self.assertEqual(self._read("pic/lid_led"), "236")
|
|
|
|
def test_cloud_mode_leaves_the_clients_config_lamp_and_counters(self):
|
|
self.cloud()
|
|
self._attr("cnc/step_freq", "10000") # the cloud client's tick
|
|
self._attr("pic/x_step_current", "135")
|
|
self._attr("pic/lid_led", "77") # its lid-image level
|
|
b = self.bl()
|
|
cap = b.capture()
|
|
self.assertEqual(cap["mode"], "cloud")
|
|
self.assertNotIn("controller_mode", cap["settings"])
|
|
self._pos(-13096, -7400, 0) # the service re-zeroed and moved
|
|
self._attr("cnc/streaming", "1") # NOT the client's: still restored
|
|
left = b.enforce("post", captured=cap)
|
|
self.assertEqual([x.item for x in left], ["cnc/streaming"])
|
|
self.assertEqual(self._read("cnc/step_freq"), "10000")
|
|
self.assertEqual(self._read("pic/x_step_current"), "135")
|
|
self.assertEqual(self._read("pic/lid_led"), "77")
|
|
self.assertEqual(self._read("cnc/streaming"), "0")
|
|
self.assertEqual(self.fc.posts, []) # no mode switch, no settings write
|
|
|
|
def test_cloud_mode_still_relocks_the_latch(self):
|
|
self.cloud()
|
|
self._attr("cnc/interlock_circuit", "37") # bit 3 clear: unlocked
|
|
left = self.bl().enforce("post", captured=None)
|
|
self.assertEqual([x.item for x in left], ["laser_latch"])
|
|
self.assertEqual(self._read("cnc/laser_latch"), "1")
|
|
|
|
def test_undeclared_mode_change_is_handed_back_through_the_switch(self):
|
|
b = self.bl()
|
|
cap = b.capture() # found in grbl
|
|
self.assertEqual(cap["mode"], "grbl")
|
|
self.cloud() # the run left it in cloud, silently
|
|
left = b.enforce("post", captured=cap)
|
|
items = {x.item: x for x in left}
|
|
self.assertIn("mode", items)
|
|
self.assertEqual(items["mode"].action, "restored")
|
|
self.assertEqual(self.fc.posts, [("/mode", {"controller": "grbl"})])
|
|
self.assertEqual(self.fc.state["mode"]["mode"], "grbl")
|
|
|
|
def test_declared_mode_change_is_kept(self):
|
|
b = self.bl()
|
|
cap = b.capture()
|
|
self.cloud()
|
|
cap["mode"] = "cloud" # Context.mode_changed("cloud")
|
|
left = b.enforce("post", captured=cap)
|
|
self.assertNotIn("mode", [x.item for x in left])
|
|
self.assertEqual(self.fc.posts, [])
|
|
self.assertEqual(self.fc.state["mode"]["mode"], "cloud")
|
|
|
|
def test_controller_mode_setting_is_never_written_back_bare(self):
|
|
b = self.bl()
|
|
cap = b.capture()
|
|
self.assertNotIn("controller_mode", cap["settings"])
|
|
self.fc.state["settings"]["controller_mode"] = "cloud" # a switch persisted it
|
|
self.fc.state["mode"]["mode"] = "cloud"
|
|
cap["mode"] = "cloud"
|
|
b.enforce("post", captured=cap)
|
|
self.assertEqual([p for p, _ in self.fc.posts if p == "/settings"], [])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|