mirror of
https://github.com/openglow-org/forgefirm.git
synced 2026-10-16 01:29:55 -07:00
forgetest: nineteen tests a host test already proves leave the catalog
The catalog is for what needs the machine or an operator. Each test below asserted software rules that a host test in CI already proves, read assertion by assertion next to that host test before it left; the record of that reading is kept with the audit. Safety rules (each approved for deletion by the owner on that reading): - laser.disarm-in-hold: laser_lifecycle_test test_resume_after_grace. - laser.arm-wait-lid: laser_arm_test E-G, laser_lifecycle_test test_lid_open_in_wait. - laser.port-dark: ctlport_test test_port_dark (port jogs, a long jog and its cancel, four injected words, the panel ops: no FIRE tick under an open window with M3 modal). - motion.respawn-gate: super_test A and D. - motion.interlock-cancel-home: the lifecycle loop cases, laser_arm_test G. - motion.lid-policy-hold: laser_lifecycle_test test_lid_policy_hold. - cloud.lid-during-button-wait, cloud.dark-print, cloud.verdict-refuse, cloud.paused-lid-cancel: test_machine_lid_button.py. - cooling.fan-duty-readback: cool_fanwrite_test. - cooling.intake-one-fan: cool_intake_test A-E. - exthost.hold-pause-tier: holdkeep_test, run_test.py, cool_flow_test P, holds_test. Other rules: - homing.cloud-offsets: manual_home_test [own-offsets], test_gfhome_motion; the real session stays in cloud.mode-switch. - exthost.core-range: version_test, install_test. - exthost.operator-destinations: run_test.py, netrules_test, install_test, extpkg_test. - exthost.lifecycle: run_test.py's ext.shutdown read before the stop. - exthost.mcode: the null-sink mcode_test, forgectrl mcode_test, sdk_test, install_test. - motion.soft-limits: z_envelope_test's homed cases, which now assert no step before the alarm, and envelope_test. - cooling_intake.py, extcore.py, extdest.py, extlife.py and extmcode.py go whole; homeoff.py keeps only the camera-home helpers that cloud.mode-switch, setup.check-envelope and their meta-tests use. - The helpers only the removed tests used go with them. - exthost.armed-freeze still runs the cloud suite's dark-print helper, whose latch checks still run; its description says so. It now requires cooling.floor-and-warm-up, the dark print's own prerequisite, in place of the removed cloud.dark-print. - HOST_TESTED: forgectrl src/holds.* (holds_test) and src/mcode.* (mcode_test), whose only catalog home left with these tests. - The meta-tests that replayed the removed tests go: test_cooling_intake_suite.py, the cloud replay cases and their fixture. The one that catches a false PASS stays and runs the dark print as armed-freeze does: a latch open at the button fails. - Text that named a removed test (exthost.motion-job's description and comment, test_baseline's docstring) no longer does. Proof: the coverage lint on a manifest from the recipe pins reports 0 uncovered paths and 0 host-tested row problems over 111 tests (its one failure is setup.check-lens's src/lensfind.* entry, which the forgectrl pin bump clears); pyflakes adds no finding; the gate self-check refuses an empty artifact cleanly; the forgetest host suite passes under Linux (547 tests); the catalog loads with 111 tests after armed-freeze's description edit. Acceptance: the catalog hash moves, so the next image opens a new campaign. Every remaining test in laser.py, motion.py, cloud.py, cooling.py, exthost.py, homeoff.py and bedsize.py moves its fingerprint (their modules' shared text changed); exthost.armed-freeze moves by its requires.
This commit is contained in:
ScottW514
-41
@@ -1,41 +0,0 @@
|
||||
2026-08-17T09:44:00.417337+00:00 gfcloud[1927] INFO cnc:set_step_freq 28160
|
||||
2026-08-17T09:44:00.418627+00:00 gfcloud[1927] INFO cnc:set_x_decay 1
|
||||
2026-08-17T09:44:00.419858+00:00 gfcloud[1927] INFO cnc:set_x_mode 8
|
||||
2026-08-17T09:44:00.421083+00:00 gfcloud[1927] INFO cnc:set_x_current 135
|
||||
2026-08-17T09:44:00.422535+00:00 gfcloud[1927] INFO cnc:set_y_decay 1
|
||||
2026-08-17T09:44:00.423794+00:00 gfcloud[1927] INFO cnc:set_y_mode 8
|
||||
2026-08-17T09:44:00.425014+00:00 gfcloud[1927] INFO cnc:set_y_current 22
|
||||
2026-08-17T09:44:00.432591+00:00 gfcloud[1927] INFO machine:_run_loop starting run
|
||||
2026-08-17T09:44:00.434005+00:00 gfcloud[1927] INFO machine:_run_loop current state: MachineState.IDLE
|
||||
2026-08-17T09:44:00.438277+00:00 gfcloud[1927] INFO machine:_run_loop current state: MachineState.RUNNING
|
||||
2026-08-17T09:44:01.785230+00:00 gfcloud[1927] INFO machine:_run_loop current state: MachineState.IDLE
|
||||
2026-08-17T09:44:01.797728+00:00 gfcloud[1927] INFO machine:_run_loop finished run
|
||||
2026-08-17T09:44:01.800661+00:00 gfcloud[1927] INFO cnc:laser_latch 1
|
||||
2026-08-17T09:44:01.808504+00:00 gfcloud[1927] INFO machine:_motion_locked end positions (actual/expected): X (-10313/-10313), Y (-649/-649), Z (0/0)
|
||||
2026-08-17T09:44:01.809326+00:00 gfcloud[1927] INFO machine:_motion_locked motion bytes actual:34952, expected: 34952
|
||||
2026-08-17T09:44:01.810066+00:00 gfcloud[1927] INFO machine:_motion_locked start idle
|
||||
2026-08-17T09:44:01.811352+00:00 gfcloud[1927] INFO cnc:set_x_current 33
|
||||
2026-08-17T09:44:01.812874+00:00 gfcloud[1927] INFO cnc:set_y_current 5
|
||||
2026-08-17T09:44:01.820631+00:00 gfcloud[1927] INFO machine:_motion_locked end positions (-10313, -649, 0)
|
||||
2026-08-17T09:44:01.822062+00:00 gfcloud[1927] INFO machine:_motion end motion
|
||||
2026-08-17T09:44:01.822829+00:00 gfcloud[1927] INFO basemachine:_finish_action motion [1576550614]: finished with event ":completed"
|
||||
2026-08-17T09:44:01.824200+00:00 gfcloud[1927] INFO cnc:laser_latch 1
|
||||
2026-08-17T09:44:07.993647+00:00 gfcloud[1927] INFO gfuiservice:run service action request: print (ready)
|
||||
2026-08-17T09:44:07.998655+00:00 gfcloud[1927] INFO machine:_motion start motion
|
||||
2026-08-17T09:44:07.999643+00:00 gfcloud[1927] INFO gfuiservice:run print (ready) service action print
|
||||
2026-08-17T09:44:14.199666+00:00 gfcloud[1927] INFO cnc:laser_latch 0
|
||||
2026-08-17T09:44:14.208291+00:00 gfcloud[1927] INFO machine:_button_wait waiting for button
|
||||
2026-08-17T09:44:21.270846+00:00 gfcloud[1927] INFO machine:_switch_event lid opened
|
||||
2026-08-17T09:44:21.275703+00:00 gfcloud[1927] WARNING machine:_button_wait button wait lid opened - relocking the laser
|
||||
2026-08-17T09:44:21.278615+00:00 gfcloud[1927] INFO cnc:laser_latch 1
|
||||
2026-08-17T09:44:21.288305+00:00 gfcloud[1927] INFO machine:_return_home start return home
|
||||
2026-08-17T09:44:21.291373+00:00 gfcloud[1927] INFO machine:_return_home return home: already at the job start
|
||||
2026-08-17T09:44:21.292501+00:00 gfcloud[1927] INFO machine:_motion_locked start cool down
|
||||
2026-08-17T09:44:21.305578+00:00 gfcloud[1927] INFO machine:_motion_locked start idle
|
||||
2026-08-17T09:44:21.307322+00:00 gfcloud[1927] INFO cnc:set_x_current 33
|
||||
2026-08-17T09:44:21.309185+00:00 gfcloud[1927] INFO cnc:set_y_current 5
|
||||
2026-08-17T09:44:21.316340+00:00 gfcloud[1927] INFO machine:_motion_locked end positions (0, 0, 0)
|
||||
2026-08-17T09:44:21.318139+00:00 gfcloud[1927] INFO machine:_motion end motion
|
||||
2026-08-17T09:44:21.318860+00:00 gfcloud[1927] INFO basemachine:_finish_action print [1576550621]: finished with event ":cancelled"
|
||||
2026-08-17T09:44:21.320214+00:00 gfcloud[1927] INFO cnc:laser_latch 1
|
||||
2026-08-17T09:44:34.058320+00:00 gfcloud[1927] INFO machine:_switch_event lid closed
|
||||
@@ -531,9 +531,9 @@ class ExtensionFreeTests(BaselineTests):
|
||||
|
||||
class TransientNotLeftoverTests(BaselineTests):
|
||||
"""A leftover is what a run left behind, not the machine part-way
|
||||
through its own work. Found on the bench reference, where
|
||||
laser.arm-wait-lid passed every check it makes and failed its
|
||||
hand-back on the post-job smoke clear."""
|
||||
through its own work. Found on the bench reference, where a laser
|
||||
test passed every check it makes and failed its hand-back on the
|
||||
post-job smoke clear."""
|
||||
|
||||
def test_a_fading_button_led_is_not_a_leftover(self):
|
||||
# the trigger fades brightness toward target: target 0 with
|
||||
|
||||
@@ -69,6 +69,19 @@ def cut(lines, marker, count=1):
|
||||
raise KeyError(marker)
|
||||
|
||||
|
||||
def dark_print(ctx):
|
||||
"""The dark print as exthost.armed-freeze runs it: the cloud suite's
|
||||
dark_print_body between the offline service's setup and cleanup."""
|
||||
offset = cloud.enter_offline(ctx)
|
||||
job = cloud.offline_job(ctx, "dark.puls", seconds=30)
|
||||
off = cloud.Offline().__enter__()
|
||||
try:
|
||||
cloud.dark_print_body(ctx, ctx.evidence, off, job, offset, ctx.forgectrl)
|
||||
finally:
|
||||
off.__exit__(None, None, None)
|
||||
cloud.offline_cleanup(ctx)
|
||||
|
||||
|
||||
class Script:
|
||||
"""Answers every prompt with its first option; a hook per prompt or
|
||||
notice substring drives the fake machine and the log replay (a
|
||||
@@ -1040,30 +1053,7 @@ class CloudSuiteTests(unittest.TestCase):
|
||||
self.assertFails(cloud.lid_interlock_abort, "not edge-driven",
|
||||
hooks=self.abort_hooks(pre, lid_tail, ilk_stop, ilk_park, []))
|
||||
|
||||
def test_lid_during_button_wait_on_the_bench_excerpt(self):
|
||||
self.in_offline()
|
||||
lines = fixture("buttonwait")
|
||||
pre, rest = cut(lines, "waiting for button")
|
||||
# The bench excerpt is a print that fit the ring; the test now wants
|
||||
# one that did not, so the feeder is alive through the wait.
|
||||
long_job = ("2026-08-17T09:44:14.100000+00:00 gfcloud[1927] INFO machine:_feed_and_run "
|
||||
"job is longer than the ring: 33554432 bytes enqueued, feeding the rest as it plays")
|
||||
pre, rest = pre + [long_job, rest[0]], rest[1:]
|
||||
self.offline_print_hooks(pre)
|
||||
hooks = {"do NOT press it": lambda: (self.lid(False), self.append(rest, delay=0.05)),
|
||||
"Close the lid.": lambda: self.lid(True)}
|
||||
run = self.run_test(cloud.lid_during_button_wait, hooks=hooks, test_id="cloud.lid-during-button-wait")
|
||||
ev = run.evidence
|
||||
self.assertEqual([m["status"] for m in FakeOffline.sent], ["ready"])
|
||||
self.assertEqual(ev["runs_started_after_wait"], 0)
|
||||
self.assertIn(":cancelled", ev["log"]["print finished"])
|
||||
self.assertTrue(ev["latch_locked"])
|
||||
self.assertEqual(ev["program_total_after"], {"first": 0, "later": 0})
|
||||
self.assertEqual(ev["streaming_after"], 0)
|
||||
self.assertEqual(self.fc.posts, [])
|
||||
self.assertTrue(any("PASS: lid open at the button prompt" in l for l in run.lines))
|
||||
|
||||
# -- the cooling verdict: locked until the run, a hold that never clears --
|
||||
# -- the dark print exthost.armed-freeze runs: locked at the button --------
|
||||
def dark_parts(self):
|
||||
"""(the print prologue through the button prompt, the run and the
|
||||
print's end with no pause in it)."""
|
||||
@@ -1074,170 +1064,13 @@ class CloudSuiteTests(unittest.TestCase):
|
||||
run = [l for l in run if "paus" not in l and "resum" not in l]
|
||||
return pre, run
|
||||
|
||||
def test_dark_print_on_the_bench_excerpt(self):
|
||||
self.in_offline()
|
||||
self.fc.state["cool"].update({"verdict": "OK", "hold": False, "fire_ok": True, "armed": False})
|
||||
pre, run = self.dark_parts()
|
||||
self.offline_print_hooks(pre)
|
||||
|
||||
def press():
|
||||
# the run: the client unlocks the latch right before it, and
|
||||
# relocks when the job ends
|
||||
self._attr("cnc/interlock_circuit", "37")
|
||||
self.append(run[:1], delay=0.05)
|
||||
|
||||
def ended():
|
||||
self._attr("cnc/interlock_circuit", "45")
|
||||
self.append(run[1:], delay=0.0)
|
||||
threading.Timer(2.0, ended).start()
|
||||
hooks = {"press it. The print runs dark": press}
|
||||
run_ = self.run_test(cloud.dark_print, hooks=hooks, test_id="cloud.dark-print")
|
||||
ev = run_.evidence
|
||||
self.assertTrue(ev["latch_locked_at_button"])
|
||||
self.assertFalse(ev["latch_locked_in_run"])
|
||||
self.assertTrue(ev["latch_locked_after"])
|
||||
self.assertIn(":completed", ev["print finished"])
|
||||
self.assertTrue(any("PASS: locked at the button" in l for l in run_.lines))
|
||||
|
||||
def test_dark_print_fails_when_the_latch_is_open_at_the_button(self):
|
||||
self.in_offline()
|
||||
pre, run = self.dark_parts()
|
||||
self._attr("cnc/interlock_circuit", "37") # unlocked before the press
|
||||
self.offline_print_hooks(pre)
|
||||
hooks = {"press it. The print runs dark": lambda: self.append(run, delay=0.05)}
|
||||
self.assertFails(cloud.dark_print, "the latch is unlocked at the button wait", hooks=hooks)
|
||||
|
||||
def test_a_print_turned_away_before_the_button_fails_at_once_with_the_reason(self):
|
||||
"""The bench reference's own lines, 2026-09-20: the coolant read 27.4
|
||||
against the client's start ceiling of 27, and the test then sat out
|
||||
its two minutes and said only that the button wait never came."""
|
||||
self.in_offline()
|
||||
stamp = "2026-09-20T21:28:16.83+00:00 gfcloud[782] INFO "
|
||||
refused = [stamp + "gfuiservice:run service action request: print (ready)",
|
||||
stamp + "machine:_motion start motion",
|
||||
stamp + "machine:_safe_to_move machine temp is too high, temp: 27.4",
|
||||
stamp + "machine:_motion end motion",
|
||||
stamp + 'basemachine:_finish_action print [9005]: finished with event ":cancelled"']
|
||||
self.offline_print_hooks(refused)
|
||||
t0 = time.time()
|
||||
self.assertFails(cloud.dark_print, "turned the print away before the button wait: "
|
||||
"INFO machine:_safe_to_move machine temp is too high, temp: 27.4")
|
||||
self.assertLess(time.time() - t0, 30, "the refusal was not seen until the wait timed out")
|
||||
|
||||
def test_verdict_refuse_on_the_bench_excerpt(self):
|
||||
self.in_offline()
|
||||
self.fc.state["status"]["coolant"] = {"up_c": 21.5, "down_c": 21.4}
|
||||
self.fc.state["settings"].update({"cool_temp_min": "5", "cool_temp_start": "16",
|
||||
"cloud_hold_max_s": ""})
|
||||
self.fc.state["cool"].update({"verdict": "OK", "hold": False, "fire_ok": True, "armed": False})
|
||||
pre, run = self.dark_parts()
|
||||
stamp = "2026-08-17T09:44:14.%06d+00:00 gfcloud[1927] %s machine:_verdict_wait "
|
||||
wait = [stamp % (100000, "INFO") + "waiting on the cooling engine: WARMUP (WARM-UP: coolant "
|
||||
"21.5 C under the 29.5 C start gate - heater on, hold)"]
|
||||
refuse = [stamp % (200000, "WARNING") + "cooling hold held 60 s - relocking the laser",
|
||||
"2026-08-17T09:44:15.000000+00:00 gfcloud[1927] INFO basemachine:_finish_action print "
|
||||
"[9006]: finished with event \":cancelled\""]
|
||||
|
||||
def on_post(path, form):
|
||||
if path == "/settings" and form.get("cool_temp_start") not in (None, "0", "16"):
|
||||
self.fc.state["cool"].update({"verdict": "WARMUP", "hold": True, "fire_ok": False,
|
||||
"armed": True, "phase": "warm-up"})
|
||||
return None
|
||||
self.fc.on_post = on_post
|
||||
self.offline_print_hooks(pre)
|
||||
|
||||
def press():
|
||||
self.append(wait, delay=0.05)
|
||||
|
||||
def refused():
|
||||
self.fc.state["cool"].update({"verdict": "OK", "hold": False, "fire_ok": True,
|
||||
"armed": False, "phase": "idle"})
|
||||
self.append(refuse, delay=0.0)
|
||||
threading.Timer(2.5, refused).start()
|
||||
hooks = {"press it. Nothing runs": press}
|
||||
run_ = self.run_test(cloud.verdict_refuse, hooks=hooks, test_id="cloud.verdict-refuse")
|
||||
ev = run_.evidence
|
||||
self.assertEqual(ev["gate"], 29.5)
|
||||
self.assertTrue(ev["latch_locked_at_button"])
|
||||
self.assertEqual(ev["latch_samples"]["unlocked"], 0)
|
||||
self.assertGreaterEqual(ev["latch_samples"]["n"], 1)
|
||||
self.assertIn(":cancelled", ev["print finished"])
|
||||
posted = [f for p_, f in self.fc.posts if p_ == "/settings"]
|
||||
self.assertEqual(posted[0]["cool_temp_start"], "29.5")
|
||||
self.assertEqual(posted[0]["cloud_hold_max_s"], "60")
|
||||
self.assertEqual(posted[-1], {"cool_temp_min": "5", "cool_temp_start": "16", "cloud_hold_max_s": ""})
|
||||
self.assertTrue(any("PASS: held" in l for l in run_.lines))
|
||||
|
||||
def test_verdict_refuse_fails_when_the_latch_unlocks_during_the_hold(self):
|
||||
self.in_offline()
|
||||
self.fc.state["status"]["coolant"] = {"up_c": 21.5, "down_c": 21.4}
|
||||
self.fc.state["settings"].update({"cool_temp_min": "5", "cool_temp_start": "16"})
|
||||
self.fc.state["cool"].update({"verdict": "WARMUP", "hold": True, "fire_ok": False, "armed": True})
|
||||
pre, run = self.dark_parts()
|
||||
stamp = "2026-08-17T09:44:14.%06d+00:00 gfcloud[1927] %s machine:_verdict_wait "
|
||||
wait = [stamp % (100000, "INFO") + "waiting on the cooling engine: WARMUP (no reason given)"]
|
||||
refuse = [stamp % (200000, "WARNING") + "cooling hold held 60 s - relocking the laser",
|
||||
"2026-08-17T09:44:15.000000+00:00 gfcloud[1927] INFO basemachine:_finish_action print "
|
||||
"[9006]: finished with event \":cancelled\""]
|
||||
self.offline_print_hooks(pre)
|
||||
|
||||
def press():
|
||||
self.append(wait, delay=0.05)
|
||||
self._attr("cnc/interlock_circuit", "37") # unlocked under the hold
|
||||
threading.Timer(2.5, lambda: self.append(refuse, delay=0.0)).start()
|
||||
hooks = {"press it. Nothing runs": press}
|
||||
self.assertFails(cloud.verdict_refuse, "the latch unlocked during the hold", hooks=hooks)
|
||||
|
||||
# -- a paused print canceled by the lid, a running one by the app --------
|
||||
def cancel_parts(self):
|
||||
"""(print prologue, the pause lines, the lid stop + park + cancel,
|
||||
the same tail with the app's cancel as the trigger - what
|
||||
cloud.oversize-stream's ending looks like)."""
|
||||
lines = fixture("lidabort")
|
||||
pre, rest = cut(lines, "waiting for button")
|
||||
run_pre, rest = cut(rest, "machine:_run_loop starting run")
|
||||
pre, rest = pre + run_pre + [rest[0]], rest[1:]
|
||||
paused = ["2026-08-17T09:42:25.500000+00:00 gfcloud[1522] INFO machine:_run_loop "
|
||||
"button pressed mid-run; pausing",
|
||||
"2026-08-17T09:42:26.100000+00:00 gfcloud[1522] INFO machine:_run_loop "
|
||||
"paused at Position(x=41.2, y=17.0, z=0.0)"]
|
||||
app_cancel = [l.replace("lid opened mid-run; stopping motion",
|
||||
"action canceled mid-run; stopping motion") for l in rest]
|
||||
return pre, paused, rest, app_cancel
|
||||
|
||||
def test_paused_lid_cancel_on_the_bench_excerpt(self):
|
||||
self.in_offline()
|
||||
pre, paused, lid_tail, _app = self.cancel_parts()
|
||||
self.offline_print_hooks(pre)
|
||||
run = self.run_test(cloud.paused_lid_cancel,
|
||||
hooks={"the press pauses the print": lambda: self.append(paused, delay=0.05),
|
||||
"The print is paused: leave the lid open": lambda: (
|
||||
self.lid(False), self.append(lid_tail, delay=0.05)),
|
||||
"Close the lid.": lambda: self.lid(True)},
|
||||
test_id="cloud.paused-lid-cancel")
|
||||
ev = run.evidence
|
||||
self.assertEqual(len(FakeOffline.sent), 1) # one print, handed over the socket
|
||||
self.assertEqual(ev["paused"], {"button pressed mid-run; pausing": True, "paused at": True})
|
||||
self.assertIn(":cancelled", ev["lid_from_pause"]["print finished"])
|
||||
self.assertTrue(ev["lid_from_pause"]["return home complete"])
|
||||
self.assertEqual(ev["counters_after"], [0, 0, 3])
|
||||
self.assertTrue(ev["latch_locked_after"])
|
||||
self.assertFalse(ev["armed_after"])
|
||||
self.assertFalse(ev["button_dark"])
|
||||
self.assertEqual(self.fc.posts, [])
|
||||
self.assertTrue(any("PASS: a paused print canceled by the lid" in l for l in run.lines), run.lines)
|
||||
|
||||
def test_paused_lid_cancel_fails_when_the_paused_print_resumes_instead(self):
|
||||
# a lid that resumed (or was ignored) leaves the print ':completed'
|
||||
self.in_offline()
|
||||
pre, paused, lid_tail, _app = self.cancel_parts()
|
||||
lid_tail = [l.replace(':cancelled"', ':completed"') for l in lid_tail]
|
||||
self.offline_print_hooks(pre)
|
||||
self.assertFails(
|
||||
cloud.paused_lid_cancel, "the print did not end ':cancelled'",
|
||||
hooks={"the press pauses the print": lambda: self.append(paused, delay=0.05),
|
||||
"The print is paused: leave the lid open": lambda: (
|
||||
self.lid(False), self.append(lid_tail, delay=0.05))})
|
||||
self.assertFails(dark_print, "the latch is unlocked at the button wait", hooks=hooks)
|
||||
|
||||
def test_print_finish_is_the_prints_not_another_actions(self):
|
||||
# a motion that completes before the print must not satisfy the print's finish
|
||||
|
||||
@@ -1,176 +0,0 @@
|
||||
# Copyright 2026 514 LLC d/b/a OpenGlow
|
||||
# Written by Scott Wiederhold
|
||||
# https://community.openglow.org
|
||||
# SPDX-License-Identifier: MIT
|
||||
|
||||
"""cooling.intake-one-fan replayed host-side under the real runner
|
||||
Context against a scripted engine: the fake reads cool_intake_fans and
|
||||
the floors at every M8 (the engine reloads at run start), publishes
|
||||
intake_2 not fitted with no floor when one intake fan is set, holds every
|
||||
fan in grace for the configured seconds, then judges the bench readings
|
||||
against the floors. The negative engines are the two ways the change
|
||||
could go wrong: the setting ignored (intake 2 still judged) and the
|
||||
setting taking intake 1 out with it."""
|
||||
import threading
|
||||
import time
|
||||
import unittest
|
||||
|
||||
import helpers
|
||||
from forgetest.runner import Context, Failed, Run
|
||||
from forgetest.suite import cooling, cooling_intake
|
||||
from test_cooling_suite import FakeGrbl
|
||||
|
||||
|
||||
class IntakeOneFanTests(unittest.TestCase):
|
||||
READINGS = {"exhaust": 11640, "intake_1": 4157, "intake_2": 4158, "air_assist": 11050, "purge": 625}
|
||||
FLOORS = {"exhaust": ("cool_tach_exhaust_min_rpm", 6400.0, 20000.0),
|
||||
"intake_1": ("cool_tach_intake_min_rpm", 2290.0, 20000.0),
|
||||
"intake_2": ("cool_tach_intake_min_rpm", 2290.0, 20000.0),
|
||||
"air_assist": ("cool_tach_air_assist_min_rpm", 6000.0, 30000.0),
|
||||
"purge": ("cool_purge_min_current", 300.0, 1023.0)}
|
||||
GATE_OF = {"exhaust": "exhaust", "intake_1": "intake", "intake_2": "intake",
|
||||
"air_assist": "air_assist", "purge": "purge"}
|
||||
|
||||
def setUp(self):
|
||||
self.fc = helpers.FakeForgectrl().start()
|
||||
self.grbl = FakeGrbl()
|
||||
self.saved = (cooling.VERDICT_WAIT_S, cooling.SESSION_END_WAIT_S, cooling_intake.INTAKE_TRIP_WAIT_S)
|
||||
cooling.VERDICT_WAIT_S = 4
|
||||
cooling.SESSION_END_WAIT_S = 3
|
||||
cooling_intake.INTAKE_TRIP_WAIT_S = 4
|
||||
self.fc.state["status"] = dict(self.fc.state["status"], gates_off=[])
|
||||
self.fc.state["cool"] = {"phase": "idle", "verdict": "OK", "fire_ok": False, "hold": False,
|
||||
"gates_off": [], "fan_gates": {}}
|
||||
for key, _d, _h in self.FLOORS.values():
|
||||
self.fc.state["settings"].setdefault(key, "")
|
||||
self.fc.state["settings"].setdefault("cool_fan_grace_s", "")
|
||||
self.fc.state["settings"].setdefault("cool_intake_fans", "")
|
||||
self.honors_setting = True # the engine takes intake 2 out with one intake fan
|
||||
self.drops_both = False # the engine takes intake 1 out with it
|
||||
self.grace_scale = 0.1 # seconds of fake grace per configured second
|
||||
self._describe()
|
||||
self.grbl.on_command = self._engine
|
||||
self.fc.on_post = self._on_post
|
||||
|
||||
def tearDown(self):
|
||||
cooling.VERDICT_WAIT_S, cooling.SESSION_END_WAIT_S, cooling_intake.INTAKE_TRIP_WAIT_S = self.saved
|
||||
self.grbl.close()
|
||||
self.fc.stop()
|
||||
|
||||
def setting(self, key, default):
|
||||
v = self.fc.state["settings"].get(key) or ""
|
||||
return float(v) if v else default
|
||||
|
||||
def _describe(self):
|
||||
gates = {}
|
||||
for fan, (key, default, hi) in self.FLOORS.items():
|
||||
v = self.setting(key, default)
|
||||
gates[key] = {"gate": self.GATE_OF[fan], "def": default, "lo": 0.0, "hi": hi,
|
||||
"band": [default * 0.9, default * 1.1], "off": "low", "value": v,
|
||||
"state": "off" if v <= 0 else "ok"}
|
||||
self.fc.state["settings"]["gates"] = gates
|
||||
|
||||
def _on_post(self, path, form):
|
||||
if path != "/settings":
|
||||
return None
|
||||
self.fc.state["settings"].update(form)
|
||||
self._describe()
|
||||
return (200, self.fc.state["settings"])
|
||||
|
||||
def _engine(self, line):
|
||||
self._describe()
|
||||
cool = self.fc.state["cool"]
|
||||
if line == "M9":
|
||||
def end():
|
||||
time.sleep(0.3)
|
||||
cool["phase"] = "idle"
|
||||
cool.update(verdict="OK", fire_ok=True, hold=False, reason="")
|
||||
for g in cool.get("fan_gates", {}).values():
|
||||
if g["state"] != "not fitted":
|
||||
g["state"] = "idle"
|
||||
threading.Thread(target=end, daemon=True).start()
|
||||
return
|
||||
if line != "M8":
|
||||
return
|
||||
cool["phase"] = "run"
|
||||
one = self.fc.state["settings"].get("cool_intake_fans") == "1" and self.honors_setting
|
||||
out = {"intake_2"} | ({"intake_1"} if self.drops_both else set()) if one else set()
|
||||
floors = {fan: 0.0 if fan in out else self.setting(key, default)
|
||||
for fan, (key, default, _h) in self.FLOORS.items()}
|
||||
cool.update(verdict="OK", fire_ok=True, hold=False, resume_ok=True, reason="", gates_off=[],
|
||||
fan_gates={f: {"reading": self.READINGS[f], "floor": floors[f],
|
||||
"state": "not fitted" if f in out else "grace"} for f in floors})
|
||||
grace = self.setting("cool_fan_grace_s", 15.0) * self.grace_scale
|
||||
|
||||
def judge():
|
||||
time.sleep(grace)
|
||||
if cool["phase"] != "run":
|
||||
return
|
||||
for f in floors:
|
||||
if f not in out:
|
||||
cool["fan_gates"][f]["state"] = "ok" if self.READINGS[f] >= floors[f] else "under"
|
||||
time.sleep(0.3)
|
||||
if cool["phase"] != "run":
|
||||
return
|
||||
for f in floors:
|
||||
if f not in out and floors[f] > self.READINGS[f]:
|
||||
cool["fan_gates"][f]["state"] = "TRIPPED"
|
||||
if cool["verdict"] != "AIRFLOW":
|
||||
cool.update(verdict="AIRFLOW", fire_ok=False, hold=True, resume_ok=False,
|
||||
reason="AIRFLOW: %s %d under the %d floor for 3 s - hold, no resume this job"
|
||||
% (f, self.READINGS[f], floors[f]))
|
||||
threading.Thread(target=judge, daemon=True).start()
|
||||
|
||||
def run_test(self):
|
||||
run = Run("test", "cooling.intake-one-fan", "t")
|
||||
ctx = Context(run, None, helpers.make_test("cooling.intake-one-fan", []))
|
||||
cooling_intake.intake_one_fan(ctx)
|
||||
return run
|
||||
|
||||
def settings_posts(self):
|
||||
return [f for p, f in self.fc.posts if p == "/settings"]
|
||||
|
||||
def restored(self):
|
||||
return {"cool_intake_fans": "", "cool_tach_intake_min_rpm": "", "cool_fan_grace_s": ""}
|
||||
|
||||
def test_one_fan_then_intake_1_trip_then_restore_pass(self):
|
||||
run = self.run_test()
|
||||
ev = run.evidence
|
||||
self.assertEqual(ev["one_fan"]["intake_2"]["state"], "not fitted")
|
||||
self.assertEqual(ev["one_fan"]["intake_1"]["state"], "ok")
|
||||
self.assertEqual(ev["intake_1_trip"]["intake_1"]["state"], "TRIPPED")
|
||||
self.assertEqual(ev["intake_1_trip"]["intake_2"]["state"], "not fitted")
|
||||
self.assertTrue(all(g["state"] == "ok" for g in ev["restored"].values()))
|
||||
posts = self.settings_posts()
|
||||
self.assertEqual(posts[0], {"cool_intake_fans": "1", "cool_fan_grace_s": "8"})
|
||||
self.assertEqual(posts[1], {"cool_tach_intake_min_rpm": "20000.0"})
|
||||
self.assertEqual(posts[-1], self.restored())
|
||||
self.assertEqual(self.grbl.commands.count("M8"), 3)
|
||||
self.assertEqual(self.grbl.commands.count("M9"), 3)
|
||||
|
||||
def test_an_engine_that_ignores_the_setting_fails_and_restores(self):
|
||||
self.honors_setting = False
|
||||
with self.assertRaises(Failed) as cm:
|
||||
self.run_test()
|
||||
self.assertIn("intake_2 is not published not fitted", str(cm.exception))
|
||||
self.assertEqual(self.settings_posts()[-1], self.restored())
|
||||
self.assertEqual(self.fc.state["cool"]["verdict"], "OK")
|
||||
|
||||
def test_an_engine_that_drops_intake_1_too_fails_and_restores(self):
|
||||
self.drops_both = True
|
||||
with self.assertRaises(Failed) as cm:
|
||||
self.run_test()
|
||||
self.assertIn("intake_1 is not judged ok at the intake floor", str(cm.exception))
|
||||
self.assertEqual(self.settings_posts()[-1], self.restored())
|
||||
|
||||
def test_a_machine_already_at_one_intake_fan_is_refused_untouched(self):
|
||||
self.fc.state["settings"]["cool_intake_fans"] = "1"
|
||||
with self.assertRaises(Failed) as cm:
|
||||
self.run_test()
|
||||
self.assertIn("already 1", str(cm.exception))
|
||||
self.assertEqual(self.settings_posts(), [])
|
||||
self.assertEqual(self.grbl.commands.count("M8"), 0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user