vh/forgefirm
1
0
Fork 0
mirror of https://github.com/openglow-org/forgefirm.git synced 2026-10-19 02:51:42 -07:00
Code Issues Packages Projects Releases Wiki Activity

forgetest: a failed press is recorded as the fixture's only when its request went out

Browse Source 
_press_taken recorded the press as the fixture's for every FixtureError
that arrived with the button out. That included failures in which the
box made no press for the request: its refusal (a 401, or the 409 of a
jumper pulled after the probe), a failed mDNS lookup, a refused
connect. When the person at the machine pressed meanwhile, or the arm
wait ended for another reason, the evidence and the exported artifact
gave the box a press it never made, and press_now returned True.

FixtureError now carries `sent`: True when the failure came after the
request went out (a timeout or a reset once the request was sent,
which urllib raises bare; a failed connect or send is a URLError), so
the box may have pressed. The one retry after a fresh lookup keeps it,
also when the lookup itself fails. A refusal and a failed lookup leave
it False. With the button out, _press_taken still puts up no notice,
and it records the press as the fixture's only for a sent error;
otherwise the record stays the operator's, and press_now returns
whether the fixture pressed. No judge reads `by`, and no energy path
changes.

Proof: forgetest-ci "Unit tests" (`python -B -m unittest discover -s
tests -v` in forgetest): test_fixture.py runs 22 tests, OK on Ubuntu
24.04 with python 3.12, and on Windows with python 3.14 with
test_ready_reads_a_short_press_as_an_edge skipped (no os.mkfifo).
test_a_failure_after_the_request_went_out_says_so drives the real
client against the fake box: a press whose reply comes after the
client's timeout is sent, with one press, also when the refresh lookup
after it finds nothing; the jumper-out refusal, a failed lookup and a
refused connect are not sent, with no press.
test_the_arm_notice_comes_down_when_the_press_is_taken gains the leg
where the box refuses while the button goes out: press_now returns
False with no notice and the record the operator's, and arm_press
leaves the same. test_server.py (10), test_operator.py (17),
test_queue.py (15), test_mode.py (9), test_setup_dark.py (21) and
test_setup_burns.py (24) pass on both. Negative controls on scratch
copies: _press_taken giving every failure to the box fails the refusal
leg (press_now True); `sent` never set fails the first client leg;
`sent` set for a URLError too fails the refused-connect leg; the retry
that drops the flag fails the retry leg. pyflakes reports no undefined
name.

Acceptance: every catalog test's fingerprint moves: runner.py is a
shared judge, in every implementation hash (43 of 43 move); fixture.py
is in none. Against the campaign this series already needs, that adds
no run: 850498f moved all 43 once. tests/ are in none.
This commit is contained in:
ScottW514
2026-10-16 23:36:59 -04:00
parent e046c534c2
commit 49f087f376
3 changed files with 103 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
forgetest/tests/test_fixture.py
+66 -3
View File
@@ -40,6 +40,8 @@ class FakeFixture:
self.button_enabled = button_enabled
self.calls = []
self.presses = 0 # presses the device performed
self.reply_delay = 0.0 # seconds between a press and its reply
self.replied = 0 # delayed replies the device is done with (sent or lost)
srv = self
class H(BaseHTTPRequestHandler):
@@ -95,6 +97,14 @@ class FakeFixture:
threading.Timer(0.05, lambda: srv.state.__setitem__("button", "idle")).start()
s = self._state()
s["pulse_ms"] = 200
if srv.reply_delay:
time.sleep(srv.reply_delay) # the reply comes after the client gave up
try:
self._json(200, s)
except OSError:
pass
srv.replied += 1
return None
return self._json(200, s)
if self.path == "/release":
srv.state = {"lid": "closed", "interlock": "closed", "button": "idle"}
@@ -109,6 +119,11 @@ class FakeFixture:
self.httpd.shutdown()
self.httpd.server_close()
def wait_replied(self, n, timeout=5.0):
deadline = time.time() + timeout
while self.replied < n and time.time() < deadline:
time.sleep(0.02)
class ConfigTests(unittest.TestCase):
def setUp(self):
@@ -155,11 +170,11 @@ class ClientTests(unittest.TestCase):
self.resolved.append(hostname)
return "127.0.0.1"
def client(self, **over):
def client(self, timeout=3.0, resolver=None, **over):
cfg = {"hostname": "forgefixture", "key": KEY, "channels": ["lid", "interlock", "button"],
"port": self.dev.port}
cfg.update(over)
return fx.Fixture(cfg, resolver=self.resolver)
return fx.Fixture(cfg, timeout=timeout, resolver=resolver or self.resolver)
def test_status_covers_act_release(self):
f = self.client()
@@ -196,6 +211,35 @@ class ClientTests(unittest.TestCase):
f.act("button", "press")
self.assertIn("jumper", str(cm.exception))
def test_a_failure_after_the_request_went_out_says_so(self):
# The press made and its reply lost (it comes after the client's
# timeout): sent, one press. A refusal, a failed lookup and a
# refused connect: not sent, no press. The one retry keeps it.
self.dev.reply_delay = 0.6
with self.assertRaises(fx.FixtureError) as cm:
self.client(timeout=0.2, ip="127.0.0.1").act("button", "press")
self.assertTrue(cm.exception.sent)
self.dev.wait_replied(1)
self.assertEqual(self.dev.presses, 1)
answers = ["127.0.0.1"] # the refresh lookup after the timeout finds nothing
with self.assertRaises(fx.FixtureError) as cm:
self.client(timeout=0.2, resolver=lambda h: answers.pop(0) if answers else None).act("button", "press")
self.assertTrue(cm.exception.sent)
self.dev.wait_replied(2)
self.assertEqual(self.dev.presses, 2)
self.dev.reply_delay = 0.0
self.dev.button_enabled = False
with self.assertRaises(fx.FixtureError) as cm:
self.client(ip="127.0.0.1").act("button", "press")
self.assertFalse(cm.exception.sent)
with self.assertRaises(fx.FixtureError) as cm:
self.client(resolver=lambda h: None).act("button", "press")
self.assertFalse(cm.exception.sent)
with self.assertRaises(fx.FixtureError) as cm:
self.client(timeout=0.5, ip="127.0.0.1", port=1).act("button", "press")
self.assertFalse(cm.exception.sent)
self.assertEqual(self.dev.presses, 2)
def test_a_wrong_key_is_refused(self):
f = self.client(key="wrong")
with self.assertRaises(fx.FixtureError) as cm:
@@ -631,7 +675,7 @@ class RoutingTests(unittest.TestCase):
# press was taken, and no notice asks for a second one.
def pressed_then_failed(channel, state, ms=None):
lit["v"] = False # the controller takes the press
raise fx.FixtureError("timed out")
raise fx.FixtureError("timed out", sent=True)
self.stub.act = pressed_then_failed
lit["v"] = True
self.assertTrue(ctx.press_now())
@@ -645,6 +689,25 @@ class RoutingTests(unittest.TestCase):
time.sleep(0.1)
self.assertEqual(ctx.evidence["actions"][-1]["by"], "fixture")
self.assertIsNone(run.notice)
# A refusal (or a request that never reached the box) while the
# button went out: a press at the machine was taken, so no notice
# goes up either, and the record does not give it to the box.
def refused_while_pressed(channel, state, ms=None):
lit["v"] = False # a person's press is taken
raise fx.FixtureError("button press -> 409 button disabled: the enable jumper is out")
self.stub.act = refused_while_pressed
lit["v"] = True
self.assertFalse(ctx.press_now())
self.assertIsNone(run.notice)
self.assertEqual(ctx.evidence["actions"][-1]["by"], "operator")
lit["v"] = True
self.assertTrue(ctx.arm_press())
deadline = time.time() + 2
while "fixture_error" not in ctx.evidence["actions"][-1] and time.time() < deadline:
time.sleep(0.02)
time.sleep(0.1)
self.assertEqual(ctx.evidence["actions"][-1]["by"], "operator")
self.assertIsNone(run.notice)
finally:
runner_mod.hw.button_lit = saved