diff --git a/forgetest/forgetest/suite/cooling.py b/forgetest/forgetest/suite/cooling.py index fbeb4e3..aed880e 100644 --- a/forgetest/forgetest/suite/cooling.py +++ b/forgetest/forgetest/suite/cooling.py @@ -178,6 +178,7 @@ def fans_quiet(ctx): GATE_KEYS = ("cool_temp_max", "cool_temp_resume") VERDICT_WAIT_S = 20 # the engine reloads settings at run start and ticks at 1 Hz +SESSION_END_WAIT_S = 15 # M9 -> the engine's phase leaves "run" (1 Hz reports, 1 Hz ticks) GATE_LOG_LINES = "400" # how far back the run-start gate lines can sit in the forgectrl log @@ -200,9 +201,25 @@ def _set_gates(ctx, fc, values): return body +def _session_ended(ctx, fc, what): + """After M9 the GRBL client's next report ends the engine's run + session; the client reports at 1 Hz and the engine samples at 1 Hz, + so an M8 sent inside that window is not a new session and nothing is + re-read. Wait until the phase has left "run".""" + t0 = time.time() + while time.time() - t0 < SESSION_END_WAIT_S: + ctx.sleep(1) + if _cool(fc).get("phase") != "run": + return True + ctx.log("%s: the engine is still in phase run %d s after M9", what, SESSION_END_WAIT_S) + return False + + def _run_session(ctx, g, fc, until, what): """M8 opens a run session (the engine re-reads its settings there and - ticks the gates at 1 Hz); wait for `until(cool)` to hold, then M9.""" + ticks the gates at 1 Hz); wait for `until(cool)` to hold, then M9 and + wait for the session to end, so the next M8 is a new one.""" + ctx.check(_cool(fc).get("phase") != "run", "%s: a run session is already open", what) g.command("M8") try: t0 = time.time() @@ -217,6 +234,7 @@ def _run_session(ctx, g, fc, until, what): return c finally: g.command("M9") + _session_ended(ctx, fc, what) def _tail_has(fc, needle): @@ -256,10 +274,10 @@ def gate_off(ctx): ctx.check(isinstance(top, (int, float)) and isinstance(bottom, (int, float)), "no range in the reply: %s", g_default) restored = False - try: - with ctx.grbl() as grbl: - st = grbl.status_report() - ctx.check(st["state"].startswith("Idle"), "controller is %s", st["state"]) + with ctx.grbl() as grbl: + st = grbl.status_report() + ctx.check(st["state"].startswith("Idle"), "controller is %s", st["state"]) + try: # Leg 1: a ceiling the coolant is already over. Legal, outside # the band (warned), and it must trip at the next run start. @@ -301,10 +319,20 @@ def gate_off(ctx): ctx.check(c.get("verdict") == "OK" and c.get("gates_off") == [], "engine did not return to OK with no gate off after the restore: %s", c) ctx.log("restored ceiling %s reports state %s", val, state) - finally: - if not restored: - st, body = fc.post("/settings", params=orig) - ctx.log("restore on failure: POST /settings %s -> %s", orig, st) + finally: + if not restored: + # The engine reads settings at run start only: restoring + # the file is not enough, a hold taken against the test's + # ceiling would stand until the operator's next job. + st, body = fc.post("/settings", params=orig) + ctx.log("restore on failure: POST /settings %s -> %s", orig, st) + try: + c = _run_session(ctx, grbl, fc, + lambda c: c.get("verdict") == "OK" and not c.get("gates_off"), + "restore on failure") + ctx.log("restore on failure: engine %s gates_off %s", c.get("verdict"), c.get("gates_off")) + except Exception as e: # the original failure is the one to report + ctx.log("restore on failure: run session did not complete (%s)", e) after = fc.settings() ctx.check(all(after.get(k, "") == orig[k] for k in GATE_KEYS), "settings not restored: %s", {k: after.get(k) for k in GATE_KEYS}) diff --git a/forgetest/tests/test_cooling_suite.py b/forgetest/tests/test_cooling_suite.py index 050286c..256232e 100644 --- a/forgetest/tests/test_cooling_suite.py +++ b/forgetest/tests/test_cooling_suite.py @@ -224,8 +224,9 @@ class GateOffTests(unittest.TestCase): def setUp(self): self.fc = helpers.FakeForgectrl().start() self.grbl = FakeGrbl() - self.saved = cooling.VERDICT_WAIT_S + self.saved = (cooling.VERDICT_WAIT_S, cooling.SESSION_END_WAIT_S) cooling.VERDICT_WAIT_S = 3 + cooling.SESSION_END_WAIT_S = 3 self.fc.state["status"] = dict(self.fc.state["status"], coolant={"down_c": 22.4, "up_c": 22.3, "pump": True, "tec": False}, gates_off=[]) @@ -235,12 +236,13 @@ class GateOffTests(unittest.TestCase): self.log_line = True # the engine writes its run-start line self.report_off = True # the engine reports the off gate self.trips = True # the engine trips a low ceiling + self.sessions = 0 # run sessions the engine saw (M8 with the phase not run) self._describe() self.grbl.on_command = self._engine self.fc.on_post = self._on_post def tearDown(self): - cooling.VERDICT_WAIT_S = self.saved + cooling.VERDICT_WAIT_S, cooling.SESSION_END_WAIT_S = self.saved self.grbl.close() self.fc.stop() @@ -267,11 +269,25 @@ class GateOffTests(unittest.TestCase): return (200, self.fc.state["settings"]) def _engine(self, line): + """M8 opens a run session: the engine re-reads the ceiling and + ticks the gate. M9 ends it a report period later (the phase + leaves run); a hold taken against the old ceiling stands until + the next session re-reads.""" self._describe() + cool = self.fc.state["cool"] + if line == "M9": + def end(): + time.sleep(0.3) + cool["phase"] = "smoke" + time.sleep(0.2) + cool["phase"] = "idle" + threading.Thread(target=end, daemon=True).start() + return if line != "M8": return + self.sessions += 1 + cool["phase"] = "run" v = self.ceiling() - cool = self.fc.state["cool"] off = v >= self.TOP if off and self.log_line: self.fc.state["logs_tail"]["text"] += ( @@ -314,6 +330,28 @@ class GateOffTests(unittest.TestCase): self.assertIn("did not trip", str(cm.exception)) self.assertEqual(self.settings_posts()[-1], {"cool_temp_max": "", "cool_temp_resume": ""}) self.assertEqual(self.fc.state["settings"]["cool_temp_max"], "") + # The restore cycles a run session so the engine re-reads the + # restored values; the bench is not left holding on the test's. + self.assertEqual(self.grbl.commands.count("M8"), 2) + self.assertEqual(self.grbl.commands.count("M9"), 2) + self.assertEqual(self.fc.state["cool"]["verdict"], "OK") + + def test_every_session_waits_for_the_previous_one_to_end(self): + """Each M8 must find the engine out of phase run, or the engine + never re-reads: the bench failure behind this case sent M9 and + the next M8 300 ms apart and the 1 Hz report pipeline swallowed + the session end.""" + seen = [] + inner = self._engine + + def engine(line): + if line == "M8": + seen.append(self.fc.state["cool"]["phase"]) + inner(line) + self.grbl.on_command = engine + self.run_test() + self.assertEqual(seen, ["idle", "idle", "idle"]) + self.assertEqual(self.sessions, 3) def test_an_engine_that_hides_the_off_gate_fails(self): self.report_off = False