From 25cc42b879c59c1805c24f2a4c42e2d15af6dbbc Mon Sep 17 00:00:00 2001 From: ScottW514 Date: Sun, 20 Sep 2026 07:14:04 -0400 Subject: [PATCH] forgetest: events.stream The acceptance test for forgectrl's GET /events. The daemon counts event streams per peer address, so the test is several peers at once: it binds its client sockets to 127.0.0.2 through 127.0.0.5, which are all this host. Three streams must each get 200, text/event-stream, and the hello event; a fourth address must get 503 with the reason while GET /settings still answers; a second stream from the first address must be served, and the older one must get bye and the end of its response; POST /motion/release and /motion/energize must arrive on a stream that was open all along as motors.released and motors.energized, in order, with ids that count up; and with every stream closed a place must come back within 25 s, since a closed client is only noticed at the daemon's next write to it (its keep-alive). The SSE client was run against a host-built daemon before any bench time was spent, which found two defects in it: its reads blocked for the socket timeout, not the window the caller asked for, and a replaced stream ends with the last chunk of the response, not with the socket closing. Passes on the bench reference: the fourth stream refused in words, the replacement told and ended, alarm, motors.released, and motors.energized as ids 1 to 3, and a place back 3 s after every stream closed. It covers forgectrl's src/events.*, src/main.c, src/status.*, and src/grblport.*. --- forgetest/forgetest/suite/forgectrl.py | 139 +++++++++++++++++++++++++ 1 file changed, 139 insertions(+) diff --git a/forgetest/forgetest/suite/forgectrl.py b/forgetest/forgetest/suite/forgectrl.py index 76b22f1..4ba53d9 100644 --- a/forgetest/forgetest/suite/forgectrl.py +++ b/forgetest/forgetest/suite/forgectrl.py @@ -8,6 +8,7 @@ import json import os import socket import time +import urllib.parse from ..catalog import test from .. import hw @@ -417,3 +418,141 @@ def panel_serves(ctx): st, text = fc.get("/curve/ladder.gcode", raw=True) ctx.check(st == 200 and b"S1000" in (text or b"") and b"M5" in (text or b""), "GET /curve/ladder.gcode -> %s without the ladder", st) + + +class _EventStream: + """GET /events over a raw socket bound to a chosen loopback source + address. Every address in 127.0.0.0/8 is this host, so the test can be + several peers at once, and the daemon counts streams per peer address.""" + + def __init__(self, base, source, timeout=5.0): + u = urllib.parse.urlsplit(base) + self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self.sock.bind((source, 0)) + self.sock.settimeout(timeout) + self.sock.connect((u.hostname or "127.0.0.1", u.port or 80)) + self.sock.sendall(b"GET /events HTTP/1.1\r\nHost: %s\r\nAccept: text/event-stream\r\n" + b"Connection: close\r\n\r\n" % (u.netloc or "127.0.0.1").encode()) + self.buf = b"" + self.eof = False # the response is over: the last chunk, or the socket closed + head = self._until(b"\r\n\r\n") + self.status = int(head.split(b" ", 2)[1]) if head.startswith(b"HTTP/") else 0 + self.head = head.decode("latin-1", "replace") + + def _fill(self): + try: + chunk = self.sock.recv(4096) + except socket.timeout: + return False + if not chunk: + self.eof = True + return False + self.buf += chunk + return True + + def _until(self, mark): + while mark not in self.buf: + if not self._fill(): + break + i = self.buf.find(mark) + if i < 0: + out, self.buf = self.buf, b"" + return out + out, self.buf = self.buf[:i], self.buf[i + len(mark):] + return out + + def text(self, seconds): + """Everything the stream says in the next `seconds` (chunk framing + and all: the checks look for event names inside it).""" + end = time.time() + seconds + self.sock.settimeout(0.2) # the window is the caller's, not one long read + while time.time() < end and not self.eof: + self._fill() + if self.buf.endswith(b"\r\n0\r\n\r\n"): + self.eof = True + out, self.buf = self.buf, b"" + return out.decode("utf-8", "replace") + + def close(self): + try: + self.sock.close() + except OSError: + pass + + +@test("events.stream", title="The event stream: edges arrive, and the cap holds", + subsystem="forgectrl", kind="auto", mode="grbl", est_min=2, + covers=[("forgectrl", "src/events.*"), ("forgectrl", "src/main.c"), ("forgectrl", "src/status.*"), + ("forgectrl", "src/grblport.*")], + description="GET /events from three loopback source addresses: each gets 200, " + "text/event-stream, and the hello event. A fourth address gets 503 with the " + "reason, and GET /settings still answers while the three are held. A second " + "stream from the first address is served, and the older one gets bye and the " + "end of its response. An edge arrives on an open stream: POST /motion/release " + "and /motion/energize show up as motors.released and motors.energized, with " + "ids that count up (X and Y lose their reference, as after any release). With " + "every stream closed, the places come back within two keep-alive intervals.") +def events_stream(ctx): + ev = ctx.evidence + fc = hw.Forgectrl() + streams = [] + try: + for src in ("127.0.0.2", "127.0.0.3", "127.0.0.4"): + s = _EventStream(fc.base, src) + streams.append(s) + hello = s.text(1.0) + ctx.check(s.status == 200 and "text/event-stream" in s.head and "event: hello" in hello, + "GET /events from %s -> %s, %r", src, s.status, hello[:120]) + fourth = _EventStream(fc.base, "127.0.0.5") + body = fourth.text(1.0) + fourth.close() + ev["fourth"] = [fourth.status, body.strip()[-120:]] + ctx.log("the fourth stream -> %s %s", fourth.status, ev["fourth"][1]) + ctx.check(fourth.status == 503 and "every event stream is taken" in body, + "the fourth stream -> %s %r, expected 503 and the reason", fourth.status, body[:160]) + st, _body = fc.get("/settings") + ctx.check(st == 200, "GET /settings -> %s with three event streams held", st) + + # One per address, by replacement. + newer = _EventStream(fc.base, "127.0.0.2") + hello = newer.text(1.0) + ctx.check(newer.status == 200 and "event: hello" in hello, + "a second stream from one address -> %s %r", newer.status, hello[:120]) + bye = streams[0].text(2.0) + ev["replaced"] = {"bye": "event: bye" in bye, "ended": streams[0].eof} + ctx.check("event: bye" in bye and "replaced" in bye, "the older stream was not told: %r", bye[-160:]) + ctx.check(streams[0].eof, "the older stream was not ended") + streams[0].close() + streams[0] = newer + + # An edge, on a stream that has been open all along. + st, body = fc.post("/motion/release") + ctx.check(st == 200, "POST /motion/release -> %s %s", st, body) + ctx.sleep(1.0) + st, body = fc.post("/motion/energize") + ctx.check(st == 200, "POST /motion/energize -> %s %s", st, body) + text = streams[1].text(2.0) + ev["edges"] = [l for l in text.splitlines() if l.startswith(("id:", "event:"))] + ctx.log("the stream carried: %s", ev["edges"]) + i_rel, i_en = text.find("event: motors.released"), text.find("event: motors.energized") + ctx.check(0 <= i_rel < i_en, "the release and the energize did not arrive in order: %r", text[-300:]) + ids = [int(l.split(":")[1]) for l in text.splitlines() if l.startswith("id:")] + ctx.check(ids and ids == sorted(ids) and len(set(ids)) == len(ids), "the ids do not count up: %s", ids) + finally: + for s in streams: + s.close() + if fc.status().get("motors_released"): + fc.post("/motion/energize") + + # A closed client is noticed at the daemon's next write to it. + def reopened(): + s = _EventStream(fc.base, "127.0.0.5") + try: + return s.status == 200 + finally: + s.close() + took = ctx.wait_for(reopened, 25, poll=1.0) + ev["place_back_s"] = took + ctx.check(took is not None, "no stream could be opened 25 s after every stream was closed") + ctx.log("PASS: three streams, the fourth refused in words with the settings route answering, a " + "replacement told and ended, two edges in order, and a place back after %.0f s", took)