From 6e71ac86f0882ae903892ba3a795c4a87c7312cd Mon Sep 17 00:00:00 2001 From: ScottW514 Date: Sat, 26 Sep 2026 18:35:53 -0400 Subject: [PATCH] forgetest: exthost.sender-keep-out, camera.lid-privacy for local viewers, and ctlport_test's keep-out - suite/extsender.py (new module, so no other test's fingerprint moves): exthost.sender-keep-out. A reference package whose service claims the Grbl sender out as it starts, installed with the sender.keep_out grant (and refused without it), on an idle GRBL-mode machine with no sender connected. It checks the holder and the ext: lease of the extension kind, the host's claim file fresh past forgectrl's grace, a sender from the machine's LAN address turned away with the message and one from the loopback admitted, the operator's release (lease gone, network welcomed, the package reading released), a new claim after a restart, and the stale end with the notice after a disable; everything is put back as found. - camera.lid-privacy: the head camera answers the panel's token and the extension host's header with the lid open, and refuses the cloud client's form (no token, no header) and a stream for no local viewer; the lid camera stays refused to everyone. covers gains src/auth.*. - scripts/bench/ctlport_test.py: test_sender_out on the null-sink controller build (Sender(host=) and closed, own_address(), generation()). Proof: forgetest's unit tests pass; the coverage lint passes against the working trees' file lists. Against the current pins it reports src/senderout.* as selecting nothing until forgectrl's pin moves to the revision that adds it, which lands with this. On the bench reference with the suite mounted: exthost.sender-keep-out PASS (first run found the check racing the host's first write; the test now waits for the file and holds past the grace), and camera.lid-privacy passed every check of its body (its hand-back failed only on the controller gated by the privacy advisory's Revision 5 before it was accepted). ctlport_test.py passes in the driver's CI. --- forgetest/forgetest/suite/__init__.py | 1 + forgetest/forgetest/suite/camera.py | 42 +++- forgetest/forgetest/suite/extsender.py | 286 +++++++++++++++++++++++++ scripts/bench/ctlport_test.py | 125 ++++++++++- 4 files changed, 442 insertions(+), 12 deletions(-) create mode 100644 forgetest/forgetest/suite/extsender.py diff --git a/forgetest/forgetest/suite/__init__.py b/forgetest/forgetest/suite/__init__.py index c9d9225..4ebf6d4 100644 --- a/forgetest/forgetest/suite/__init__.py +++ b/forgetest/forgetest/suite/__init__.py @@ -32,3 +32,4 @@ from . import extwizard # noqa: F401,E402 from . import updlock # noqa: F401,E402 from . import homeoff # noqa: F401,E402 from . import bedsize # noqa: F401,E402 +from . import extsender # noqa: F401,E402 diff --git a/forgetest/forgetest/suite/camera.py b/forgetest/forgetest/suite/camera.py index 13431d4..a62ab34 100644 --- a/forgetest/forgetest/suite/camera.py +++ b/forgetest/forgetest/suite/camera.py @@ -320,17 +320,22 @@ def frame_health(ctx): ctx.check(restarts == 0, "the capture stream was restarted %d time(s)", restarts) -@test("camera.lid-privacy", title="Cameras capture only with the lid closed", subsystem="camera", +@test("camera.lid-privacy", title="Cameras see the room only with the lid closed", subsystem="camera", kind="operator", est_min=3, - covers=_PRIVACY_COVERS, requires=["forgectrl.panel-serves"], actions=["lid"], + covers=_PRIVACY_COVERS + [("forgectrl", "src/auth.*")], requires=["forgectrl.panel-serves"], + actions=["lid"], steps=["Start with the lid closed; you will be told to open it, then close it again.", "Nothing moves and the laser is not involved."], - description="The privacy gate: with the lid open neither camera captures. A running stream " - "stops within a frame or so of the lid opening, /cam/status reports capture as " - "not allowed, and both the snapshot and the stream are refused with 409 and a " - "reason naming the lid. Closing the lid restores all of it. This is what stops " - "the machine - and in cloud mode the Glowforge service - from imaging the room " - "through an open lid.") + description="The privacy gate: with the lid open the lid camera does not capture, and the head " + "camera captures only for a local viewer. A running lid-camera stream stops within a " + "frame or so of the lid opening, /cam/status reports capture as not allowed, and the " + "lid camera's snapshot and stream are refused with 409 and a reason naming the lid, " + "to the panel too. The head camera, which looks down at the bed, still answers the " + "panel (its token) and the extension host (its client header from this host), while " + "the same snapshot and stream asked without either - the cloud client's form - are " + "refused for the lid. Closing the lid restores all of it. This is what stops the " + "machine - and in cloud mode the Glowforge service - from imaging the room through " + "an open lid.") def lid_privacy(ctx): import urllib.error import urllib.request @@ -411,9 +416,26 @@ def lid_privacy(ctx): ctx.check(b"lid" in data.lower(), "the refusal should name the lid: %r", data[:120]) ctx.check(data[:2] != b"\xff\xd8", "a JPEG was returned with the lid open") + # The head camera looks at the bed: a local viewer still has it, the cloud client's form does not. st, data = fc.get("/cam/snapshot", params={"cam": "head", "res": "half"}, raw=True) - ctx.log("lid open: head snapshot -> %s", st) - ctx.check(st == 409, "the head camera should be refused too, got %s", st) + ctx.log("lid open: the panel's head snapshot -> %s", st) + ctx.check(st == 200 and data[:2] == b"\xff\xd8", "the panel's head snapshot with the lid open -> %s", st) + st, data = fc.get("/cam/snapshot", params={"cam": "head", "res": "half"}, raw=True, auth=False, + headers={"X-ForgeFIRM-Client": "extension-host"}) + ctx.log("lid open: the extension host's head snapshot -> %s", st) + ctx.check(st == 200 and data[:2] == b"\xff\xd8", "the extension host's head snapshot with the lid open -> %s", st) + st, data = fc.get("/cam/snapshot", params={"cam": "head", "res": "half"}, raw=True, auth=False) + ev["head_cloud_form_lid_open"] = [st, (data or b"")[:120].decode("utf-8", "replace")] + ctx.log("lid open: head snapshot without a token or the header -> %s %s", st, (data or b"")[:120]) + ctx.check(st == 409 and b"lid" in (data or b"").lower(), + "the head camera without a token or the header should be refused for the lid, got %s", st) + try: + hreq = urllib.request.Request(fc.base + "/cam/stream?cam=head", headers={"Host": fc.host_header()}) + with urllib.request.urlopen(hreq, timeout=10) as r: + ctx.fail("the head stream opened with the lid open, for no local viewer (%s)", r.status) + except urllib.error.HTTPError as e: + ctx.log("lid open: head stream for no local viewer -> %s", e.code) + ctx.check(e.code == 409, "the head stream for no local viewer should be refused with 409, got %s", e.code) try: with urllib.request.urlopen(req, timeout=10) as r: diff --git a/forgetest/forgetest/suite/extsender.py b/forgetest/forgetest/suite/extsender.py new file mode 100644 index 0000000..37e4ed4 --- /dev/null +++ b/forgetest/forgetest/suite/extsender.py @@ -0,0 +1,286 @@ +# Copyright 2026 514 LLC d/b/a OpenGlow +# Written by Scott Wiederhold +# https://community.openglow.org +# SPDX-License-Identifier: MIT + +"""An extension package keeps the Grbl sender out, by the operator's grant. + +Its own module rather than a test in exthost.py, for the reason extcore.py +gives: a test added there moves the fingerprint of every test in that +module. The package is the reference package's id and key with a service +that claims the sender out as it starts and says how its claim stands +every half second, so exthost's put-back takes it away like the others. +""" + +import json +import os +import socket +import time + +from ..catalog import test +from .exthost import (EXT_ROOT, FWUP, REF_ID, REF_KEY, _as_found, _forgeext, _put_back, _svc, _tree, _until, + _write) +from .setup import SAFETY_PHRASE, read_file, record_path, request + +CLAIM_DIR = "/run/forgefirm/sender-out" +GRBL_STATE = "/run/forgefirm/grbl.state" +KEEP_OUT_MSG = b"[MSG:The machine is in use: senders are kept out for now]" + +# The service: it claims the sender out once, as it starts, writes the +# machine's answer, and then writes how its claim stands, twice a second. +KEEPER_SERVICE = r''' +import json, os, socket, time +data = os.environ["FFX_DATA"] + + +def api(method, path, body=None): + s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + s.settimeout(10) + try: + s.connect(os.environ["FFX_API"]) + payload = b"" if body is None else json.dumps(body).encode() + head = "%s %s HTTP/1.1\r\nHost: forgeext\r\n" % (method, path) + if body is not None: + head += "Content-Type: application/json\r\nContent-Length: %d\r\n" % len(payload) + s.sendall(head.encode() + b"\r\n" + payload) + buf = b"" + while True: + c = s.recv(65536) + if not c: + break + buf += c + top, _, content = buf.partition(b"\r\n\r\n") + return int(top.split()[1]), json.loads(content or b"null") + finally: + s.close() + + +def write(name, doc): + p = os.path.join(data, name) + with open(p + ".new", "w") as f: + json.dump(doc, f) + os.rename(p + ".new", p) + + +status, doc = api("POST", "/v0/sender", {"out": True}) +write("claim.json", {"status": status, "doc": doc}) +print("keeper up: %s %s" % (status, doc), flush=True) +while True: + try: + write("state.json", dict(zip(("status", "doc"), api("GET", "/v0/sender")))) + except OSError: + pass + time.sleep(0.5) +''' + + +def _pack_keeper(work): + """The reference package's id and key, with the service above: (archive, public key).""" + import io + import subprocess + import tarfile + manifest = {"manifest": 1, "id": REF_ID, "name": "forgetest sender keeper", "version": "1.0.0", + "author": "forgetest", "license": "MIT", "api": "0.1", "runtime": "python", + "service": {"exec": "bin/keeper.py"}, "capabilities": ["sender.keep_out"]} + payload = os.path.join(work, "payload.tar.gz") + with tarfile.open(payload, "w:gz") as t: + for name, text, mode in (("manifest.json", json.dumps(manifest), 0o644), ("bin/keeper.py", KEEPER_SERVICE, 0o755)): + info = tarfile.TarInfo(name) + data = text.encode() + info.size, info.mode = len(data), mode + t.addfile(info, io.BytesIO(data)) + conf = os.path.join(work, "fwup.conf") + _write(conf, 'meta-product = "ForgeFIRM extension"\nmeta-description = "%s"\nmeta-version = "1.0.0"\n' + 'meta-platform = "forgefirm-ext"\nfile-resource payload.tar.gz {\n host-path = "%s"\n}\n' + % (REF_ID, payload)) + key = os.path.join(work, REF_KEY) + raw, signed = os.path.join(work, "raw.ffx"), os.path.join(work, "keeper.ffx") + for cmd in ([FWUP, "-g", "-o", key], [FWUP, "-c", "-f", conf, "-o", raw], + [FWUP, "-S", "-s", key + ".priv", "-i", raw, "-o", signed]): + subprocess.run(cmd, check=True, capture_output=True, timeout=60, cwd=work) + return signed, key + ".pub" + + +def _lan_address(): + """This machine's own address on the network: a sender that connects to it comes from the network + as the controller sees it, and not from the loopback.""" + u = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: + u.connect(("192.0.2.1", 9)) + return u.getsockname()[0] + except OSError: + return "" + finally: + u.close() + + +def _sender(addr, secs=2.0): + """A Grbl sender that connects and listens: (what it read, whether the controller closed it).""" + s = socket.create_connection((addr, 23), timeout=5) + s.settimeout(0.3) + got, closed, end = b"", False, time.time() + secs + try: + while time.time() < end: + try: + c = s.recv(4096) + except socket.timeout: + continue + if not c: + closed = True + break + got += c + finally: + s.close() + return got, closed + + +def _read_json(path): + try: + with open(path) as f: + return json.load(f) + except (OSError, ValueError): + return None + + +@test("exthost.sender-keep-out", title="An extension package keeps the Grbl sender out, by the operator's grant", + subsystem="exthost", kind="auto", hardware="api", est_min=4, + covers=[("forgectrl", "src/senderout.*"), ("forgectrl", "src/lease.*"), ("forgectrl", "src/grblport.*"), + ("forgectrl", "src/main.c"), ("forgectrl", "src/status.*"), ("forgeext", "src/api.*"), + ("forgeext", "src/run.*"), ("forgeext", "src/machine.*"), ("forgeext", "src/holdkeep.*"), + ("forgeext", "src/caps.*"), ("grblhal-glowforge", "src/ctlport.*"), ("grblhal-glowforge", "src/serial.*")], + requires=["exthost.service"], + description="A package the operator granted sender.keep_out keeps the Grbl sender out while it uses the " + "machine: the test installs a reference package whose service asks POST /v0/sender " + "{\"out\": true} as it starts, on an idle machine with no sender connected. The machine " + "grants it: /status names the package in sender_out and holds the machine lease as " + "ext:, of the extension kind; the host keeps the claim fresh under " + "/run/forgefirm/sender-out; a Grbl sender that connects from the network (the machine's " + "own LAN address) reads the controller's message and is closed at once, and one from " + "this host still connects. The operator lets the sender back in from the panel " + "(POST /motion/sender out=0): the holder and the lease go, a sender from the network is " + "welcomed again, and the package's service reads released. Turned off and on, the " + "package's service claims again; turned off with its claim standing, forgectrl ends the " + "claim within seconds by itself, with the notice that names it, and lets senders in; the " + "notice is cleared from the panel. The package, the key, the setting, and the setup " + "record are put back as found.") +def sender_keep_out(ctx): + fc = ctx.forgectrl + ev = ctx.evidence + ctx.check(fc.wait_idle(timeout=30, abort=ctx.aborted), "machine not idle: settings are locked") + st, m0 = fc.get("/mode") + ctx.check(st == 200 and isinstance(m0, dict) and m0.get("mode") == "grbl" and m0.get("controller") == "running", + "start this test in grbl mode with the controller running (now %s)", m0) + grbl = _read_json(GRBL_STATE) or {} + ctx.check(not (grbl.get("sender") or {}).get("connected"), + "a Grbl sender is connected: close it first (this test would disconnect it)") + lan = _lan_address() + ctx.check(lan and not lan.startswith("127."), "the machine has no address on the network: %r", lan) + prior = fc.settings().get("ext_enabled") or "" + raw = read_file(record_path()) + found_tree = _tree(EXT_ROOT) + dir_mode = os.stat(os.path.dirname(EXT_ROOT)).st_mode & 0o7777 + st, body, hdrs = request(fc.base, "GET", "/advisories/extensions", headers={"Host": fc.host_header()}) + etag = hdrs.get("etag") + import shutil + import tempfile + work = tempfile.mkdtemp(prefix="forgetest-sender.") + owner_key = os.path.join(EXT_ROOT, "keys", REF_KEY + ".pub") + data = os.path.join(EXT_ROOT, "data", REF_ID) + claim_file = os.path.join(CLAIM_DIR, REF_ID + ".json") + + def sender_out(): + return fc.status().get("sender_out") or {} + + def holder(): + return (sender_out().get("holder") or {}).get("id") + + try: + got, closed = _sender(lan) + ev["before"] = [got[:80].decode(errors="replace"), closed] + ctx.check(b"Grbl" in got and not closed, "before: a sender from the network is not welcomed: %r", got[:80]) + + archive, pub = _pack_keeper(work) + shutil.copy(pub, owner_key) + os.chmod(owner_key, 0o644) + r = _forgeext("install", archive, "--consent-community") + ctx.check(r.get("ok") is False and "sender.keep_out" in (r.get("error") or ""), + "an install without the operator's grant is refused -> %s", r.get("error")) + r = _forgeext("install", archive, "--consent-community", "--grant", "sender.keep_out") + ctx.check(r.get("ok") is True, "the install with the grant -> %s", r.get("error")) + st, reply = fc.post("/settings", data={"ext_enabled": "1", "advisory": etag, "phrase": SAFETY_PHRASE}) + ctx.check(st == 200, "ext_enabled=1 -> %s %r", st, reply) + up = _until(ctx, lambda: _svc(REF_ID).get("state") == "running" and holder() == REF_ID, 60) + claim = _read_json(os.path.join(data, "claim.json")) + ev["claim"] = claim + ctx.log("the service's claim: %s; sender_out %s", claim, sender_out()) + ctx.check(up and (claim or {}).get("status") == 200 and (claim.get("doc") or {}).get("out") is True, + "the service's claim was not granted: %s, %s", claim, sender_out()) + + doc = fc.status() + ev["kept"] = {"sender_out": doc.get("sender_out"), "lease": (doc.get("lease") or {}).get("holder")} + lease = (doc.get("lease") or {}).get("holder") or {} + ctx.check(lease.get("owner") == "ext:" + REF_ID and lease.get("kind") == "extension", + "the lease is not the package's: %s", lease) + def fresh(): + k = _read_json(claim_file) or {} + return k if (k.get("id") == REF_ID and k.get("raised") is True and + time.clock_gettime(time.CLOCK_MONOTONIC) - float(k.get("ts_mono") or 0) < 2.0) else None + # The host writes the claim at its next turn; past forgectrl's 3 s grace the claim stands only + # because the host keeps it fresh. + kept = _until(ctx, fresh, 5) + ev["claim_file"] = kept + ctx.check(kept, "the host does not write the claim: %s", _read_json(claim_file)) + time.sleep(5.0) + ctx.check(holder() == REF_ID and fresh(), "past the grace, the claim did not stand fresh: %s, %s", + sender_out(), _read_json(claim_file)) + got, closed = _sender(lan) + ev["kept_out"] = [got[:120].decode(errors="replace"), closed] + ctx.check(closed and KEEP_OUT_MSG in got and b"Grbl" not in got, + "a sender from the network was not turned away: %r closed=%s", got[:120], closed) + got, closed = _sender("127.0.0.1", 1.0) + ev["loopback"] = [got[:80].decode(errors="replace"), closed] + ctx.check(b"Grbl" in got, "a sender from this host was not let in: %r", got[:80]) + + # The operator lets it back in, from the panel. + st, reply = fc.post("/motion/sender", data={"out": "0"}) + ev["released"] = [st, reply] + ctx.check(st == 200 and holder() is None, "the operator's release -> %s %s", st, reply) + ctx.check(not ((fc.status().get("lease") or {}).get("holder")), "the lease stayed after the release") + got, closed = _sender(lan) + ctx.check(b"Grbl" in got and not closed, "released, a sender from the network is not welcomed: %r", got[:80]) + saw = _until(ctx, lambda: ((_read_json(os.path.join(data, "state.json")) or {}).get("doc") or {}).get("released"), 10) + ctx.check(saw is True, "the package's service did not read released: %s", + _read_json(os.path.join(data, "state.json"))) + + # Off and on: the operator's release lasts until the package stops claiming, and the + # service claims again as it starts. + ctx.check(_forgeext("disable", REF_ID).get("ok") is True, "disable") + ctx.check(_until(ctx, lambda: not os.path.exists(claim_file), 15), "disabled, the host kept the claim file") + time.sleep(3.0) # past forgectrl's freshness window, so the release is forgotten + ctx.check(_forgeext("enable", REF_ID).get("ok") is True, "enable") + again = _until(ctx, lambda: holder() == REF_ID, 60) + ctx.check(again, "enabled again, the service's claim was not granted: %s", sender_out()) + + # Turned off with its claim standing: forgectrl ends it by itself. + t0 = time.time() + ctx.check(_forgeext("disable", REF_ID).get("ok") is True, "disable with the claim standing") + gone = _until(ctx, lambda: holder() is None, 15) + ev["stale_s"] = round(time.time() - t0, 1) + notice = sender_out().get("notice") or {} + ev["notice"] = notice + ctx.log("the claim ended %.1f s after the disable; notice %s", time.time() - t0, notice) + ctx.check(gone is True and notice.get("id") == REF_ID and notice.get("why") == "stopped", + "a claim its keeper stopped keeping did not end with the notice: %s", sender_out()) + ctx.check(not os.path.exists(claim_file), "the claim file stayed after the service stopped") + got, closed = _sender(lan) + ctx.check(b"Grbl" in got and not closed, "after it ended, a sender from the network is not welcomed: %r", got[:80]) + st, reply = fc.post("/motion/sender", data={"notice": "clear"}) + ctx.check(st == 200 and not sender_out().get("notice"), "the notice was not cleared -> %s %s", st, reply) + finally: + if holder(): + fc.post("/motion/sender", data={"out": "0"}) + if sender_out().get("notice"): + fc.post("/motion/sender", data={"notice": "clear"}) + _put_back(ctx, fc, work, prior, etag, raw, dir_mode) + ctx.check(holder() is None, "the sender is still kept out at the end") + _as_found(ctx, fc, prior, raw, dir_mode, found_tree) diff --git a/scripts/bench/ctlport_test.py b/scripts/bench/ctlport_test.py index d95efbf..2e1e1b2 100644 --- a/scripts/bench/ctlport_test.py +++ b/scripts/bench/ctlport_test.py @@ -37,6 +37,14 @@ job after it. one FIRE tick. A jog block carries the modal spindle state, so the stream's mask on jogging is the only thing that keeps a jog dark, and this is what holds the port to it + 10. "sender out" keeps senders from the network out, and only on an idle + machine with a quiet sender: it is refused while a sender line runs + (a dwell), within 2 s of the sender's last line, and while a program + moves; granted, it drops the sender after a message line, turns every + sender from the network away the same way without touching the + session, still lets one from this host in, outlives the port's client, + and "sender in" ends it. A sender from the network is one that + connects to this host's own address rather than the loopback Usage: ctlport_test.py [path/to/grblHAL_glowforge] """ @@ -65,14 +73,15 @@ class Sender: """The scripted sender: every line it receives is classified, and the responses (ok, error:N) are counted apart from everything else.""" - def __init__(self, port, eol="\n"): + def __init__(self, port, eol="\n", host="127.0.0.1"): self.eol = eol self.responses = [] # "ok" / "error:N", in order self.other = [] # messages, reports, the banner self.lock = threading.Lock() + self.closed = False # the controller closed the socket for _ in range(50): try: - self.sock = socket.create_connection(("127.0.0.1", port), timeout=1) + self.sock = socket.create_connection((host, port), timeout=1) break except OSError: time.sleep(0.1) @@ -92,8 +101,10 @@ class Sender: except socket.timeout: continue except OSError: + self.closed = True return if not data: + self.closed = True return buf += data while b"\n" in buf: @@ -633,6 +644,115 @@ def test_dead_man(s): fail("[dead-man] a new client inherited the old jog") +def own_address(): + """This host's own address on the network, which a connection to it + comes from: a sender from the network, as the controller sees it.""" + u = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: + u.connect(("192.0.2.1", 9)) # TEST-NET: routes, sends nothing + addr = u.getsockname()[0] + except OSError: + addr = "" + finally: + u.close() + if not addr or addr.startswith("127."): + fail("[sender-out] this host has no address but the loopback; the test needs one") + return addr + + +def generation(s): + with open(os.path.join(s.workdir, "grbl.state")) as f: + return json.load(f)["sender"]["generation"] + + +def wait_closed(sender, what, timeout=3.0): + end = time.time() + timeout + while time.time() < end and not sender.closed: + time.sleep(0.05) + if not sender.closed: + fail("[sender-out] %s was not closed" % what) + + +def test_sender_out(): + s = Session() + try: + own = own_address() + if s.port.state().get("sender_out") is not False: + fail("[sender-out] the state does not say the sender is let in") + net = Sender(PORT, host=own) # displaces the session's own sender + net.wait_state("Idle") + + # A line that runs: a dwell is Idle to the core, and still the sender's. + net.sock.sendall(b"G4 P3" + net.eol.encode()) + time.sleep(1.0) + got = s.port.request("sender out") + if got != "busy:sender": + fail("[sender-out] during the sender's dwell: %r, wanted busy:sender" % got) + end = time.time() + 5 + while net.count() < 1 and time.time() < end: + time.sleep(0.05) + # A sender that just finished a line is not done with the machine. + if net.send("G21") != "ok": + fail("[sender-out] G21 was not answered ok") + got = s.port.request("sender out") + if got != "busy:sender": + fail("[sender-out] within 2 s of the sender's last line: %r, wanted busy:sender" % got) + # A program that moves. + net.sock.sendall(b"G91 G1 X20 F300" + net.eol.encode()) + time.sleep(0.8) + got = s.port.request("sender out") + if got != "busy:state": + fail("[sender-out] while a program moves: %r, wanted busy:state" % got) + net.wait_state("Idle") + print("PASS [sender-out]: refused during a dwell, within 2 s of a line, and while a program moves") + + time.sleep(2.3) + gen = generation(s) + got = s.port.request("sender out") + if got != "ok": + fail("[sender-out] an idle machine and a quiet sender: %r, wanted ok" % got) + wait_closed(net, "the connected sender") + if not net.saw("[MSG:The machine is in use"): + fail("[sender-out] the dropped sender was not told why: %r" % net.other[-3:]) + st = s.port.state() + if st.get("sender_out") is not True or st.get("sender") is not False: + fail("[sender-out] the state after it: %r" % st) + print("PASS [sender-out]: granted, it dropped the sender after a message line") + + gen = generation(s) + late = Sender(PORT, host=own) + wait_closed(late, "a sender from the network") + if not late.saw("[MSG:The machine is in use") or late.count(): + fail("[sender-out] the turned-away sender read %r" % late.other) + time.sleep(0.3) + if generation(s) != gen or s.port.state().get("sender") is not False: + fail("[sender-out] turning a sender away changed the session") + local = Sender(PORT) + if local.closed or local.send("G21") != "ok": + fail("[sender-out] a sender on this host was not let in") + local.close() + print("PASS [sender-out]: a sender from the network is turned away and the session is " + "untouched; one from this host connects") + + s.port.close() + s.port = PortClient(s.path) + if s.port.state().get("sender_out") is not True: + fail("[sender-out] it did not outlive the port's client") + if s.port.request("sender out") != "ok": + fail("[sender-out] asked again, it was not ok") + if s.port.request("sender in") != "ok": + fail("[sender-out] sender in was not ok") + if s.port.state().get("sender_out") is not False: + fail("[sender-out] sender in did not end it") + again = Sender(PORT, host=own) + if again.closed or again.send("G21") != "ok": + fail("[sender-out] after sender in, a sender from the network was not let in") + again.close() + print("PASS [sender-out]: it outlives the port's client, and sender in lets the network back in") + finally: + s.close() + + def main(): if not os.path.exists(BIN): fail("no controller binary at %s" % BIN) @@ -653,6 +773,7 @@ def main(): test_poll_lf() test_poll_crlf() test_port_dark() + test_sender_out() print("ALL PASS")