the page that calls its own service
\n" - - -def _pack_caller(work): - """The reference package's id and key, with a page and the service above: (archive, public key).""" - import io - import subprocess - import tarfile - manifest = {"manifest": 1, "id": REF_ID, "name": "forgetest page call", "version": "1.0.0", - "author": "forgetest", "license": "MIT", "api": "0.1", "runtime": "python", - "service": {"exec": "bin/caller.py"}, "capabilities": ["ui", "machine.read"]} - 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/caller.py", CALL_SERVICE, 0o755), ("ui/index.html", CALL_UI, 0o644)): - 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, "caller.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" - - -@test("exthost.page-call", title="A package's page asks its own service, through the panel", - subsystem="exthost", kind="auto", hardware="api", est_min=4, - covers=[("forgeext", "src/call.*"), ("forgeext", "src/run.*"), ("forgeext", "src/sandbox.*"), - ("forgeext", "src/main.c"), ("forgectrl", "src/extpkg.*"), ("forgectrl", "src/main.c")], - requires=["exthost.service"], - description="A package with a page and a service may have the one ask the other. The host binds the " - "service's call socket in /run/forgefirm/ext/call, a directory that is root's alone " - "(0700), as root's 0600 socket, and hands the service only the listening end, at " - "descriptor 4 (FFX_CALL_FD=4), with nothing at 3 or 5. POST /ext/call relays a call " - "through the host's command line: a POST's JSON object reaches the service and its " - "answer comes back as its status and its JSON; a GET carries no body; the service's own " - "refusal is its status and its words, not the relay's; a service may ask the machine " - "while it answers. The form is held before anything runs: a method that is not GET or " - "POST, a GET with a body, and a body that is not a JSON object are 400, and a path with " - ".. in it and a package that is not installed are the host's refusals, 409 in its words. " - "The command line answers alike. Disabled, the package's service stops, a call is refused " - "in words, and the socket's name is gone; enabled again, the service answers again. A " - "call to a frozen service and the relay's bridge in a browser are not this test's: " - "forgeext's sdk_test and the frame-isolation harness hold those. The package, the key, " - "the setting, and the setup record are put back as found.") -def page_call(ctx): - import shutil - import tempfile - fc = ctx.forgectrl - ev = ctx.evidence - ctx.check(fc.wait_idle(timeout=30, abort=ctx.aborted), "machine not idle: settings are locked") - 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") - work = tempfile.mkdtemp(prefix="forgetest-call.") - owner_key = os.path.join(EXT_ROOT, "keys", REF_KEY + ".pub") - sock = os.path.join(CALL_DIR, REF_ID + ".sock") - report = os.path.join(EXT_ROOT, "data", REF_ID, "report.json") - - def call(method, path, body=None, id_=REF_ID): - form = {"id": id_, "method": method, "path": path} - if body is not None: - form["body"] = body - return fc.post("/ext/call", data=form) - - try: - archive, pub = _pack_caller(work) - shutil.copy(pub, owner_key) - os.chmod(owner_key, 0o644) - r = _forgeext("install", archive, "--consent-community") - ctx.check(r.get("ok") is True, "the install of a package with a page and a service -> %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 os.path.exists(report), 60) - ctx.check(up, "the service did not come up: %s", _svc(REF_ID)) - - # What the service was handed, from inside, and the socket from outside. - with open(report) as f: - inside = json.load(f) - ev["inside"] = inside - ctx.log("inside: %s", inside) - ctx.check(inside.get("call_env") == "4", "FFX_CALL_FD is %r, not 4", inside.get("call_env")) - ctx.check(inside.get("fds") == {"3": "closed", "4": "socket", "5": "closed"}, - "its descriptors 3, 4, 5: %s", inside.get("fds")) - ctx.check(inside.get("listening") == 1, "descriptor 4 is not a listening socket: %s", inside.get("listening")) - d, s = os.lstat(CALL_DIR), os.lstat(sock) - ev["socket"] = {"dir": [oct(d.st_mode & 0o7777), d.st_uid], "sock": [oct(s.st_mode & 0o7777), s.st_uid]} - ctx.check(stat.S_ISDIR(d.st_mode) and d.st_uid == 0 and d.st_mode & 0o7777 == 0o700, - "%s is not root's 0700 directory: %s", CALL_DIR, ev["socket"]["dir"]) - ctx.check(stat.S_ISSOCK(s.st_mode) and s.st_uid == 0 and s.st_mode & 0o777 == 0o600, - "%s is not root's 0600 socket: %s", sock, ev["socket"]["sock"]) - - # Calls through the relay, as the panel's bridge makes them. - st_, doc = call("POST", "/echo", '{"a": 1, "b": "two"}') - ev["post"] = [st_, doc] - ctx.log("POST /ext/call /echo -> %s %s", st_, doc) - ctx.check(st_ == 200 and (doc or {}).get("status") == 200 - and doc.get("body") == {"method": "POST", "path": "/echo", "body": {"a": 1, "b": "two"}}, - "a POST -> %s %s", st_, doc) - st_, doc = call("GET", "/echo") - ev["get"] = [st_, doc] - ctx.check(st_ == 200 and (doc or {}).get("body") == {"method": "GET", "path": "/echo", "body": {}}, - "a GET -> %s %s", st_, doc) - st_, doc = call("GET", "/refuse") - ev["refuse"] = [st_, doc] - ctx.check(st_ == 200 and (doc or {}).get("status") == 409 and doc.get("body") == {"error": "not now"}, - "the service's own refusal -> %s %s", st_, doc) - st_, doc = call("GET", "/me") - ev["me"] = [st_, doc] - ctx.check(st_ == 200 and (doc or {}).get("body") == REF_ID, "a service asking the machine while it answers " - "-> %s %s", st_, doc) - for args, want, words in ((("PUT", "/echo"), 400, "a call is GET or POST"), - (("GET", "/echo", "{}"), 400, "a GET call has no body"), - (("POST", "/echo", "[1]"), 400, "a call's body is a JSON object"), - (("GET", "/../etc"), 409, "a call's path starts with"), - (("GET", "/echo", None, "org.forgetest.nothere"), 409, "that package is not installed")): - st_, why = call(*args) - ev.setdefault("refusals", []).append([list(args), st_, why]) - ctx.check(st_ == want and words in json.dumps(why), "%s -> %s %s, wanted %s %s", args, st_, why, want, words) - - # The command line answers alike. - r = _forgeext("call", REF_ID, "POST", "/echo", '{"from": "the console"}') - ev["cli"] = r - ctx.check(r.get("ok") is True and r.get("status") == 200 - and (r.get("body") or {}).get("body") == {"from": "the console"}, "forgeext call -> %s", r) - - # Disabled: no service, no socket, a refusal in words; and back. - r = _forgeext("disable", REF_ID) - ctx.check(r.get("ok") is True, "disable -> %s", r.get("error")) - st_, why = call("GET", "/echo") - ev["disabled"] = [st_, why] - ctx.check(st_ == 409 and "disabled" in json.dumps(why), "a call to a disabled package -> %s %s", st_, why) - gone = _until(ctx, lambda: not os.path.exists(sock) and _svc(REF_ID).get("state") != "running", 30) - ctx.check(gone, "disabled, the socket's name is still there or the service still runs: %s", _svc(REF_ID)) - r = _forgeext("enable", REF_ID) - ctx.check(r.get("ok") is True, "enable -> %s", r.get("error")) - back = _until(ctx, lambda: _svc(REF_ID).get("state") == "running" and os.path.exists(sock), 60) - ctx.check(back, "enabled again, the service did not come back: %s", _svc(REF_ID)) - st_, doc = _until(ctx, lambda: (lambda a: a if a[0] == 200 else None)(call("GET", "/echo")), 20) or (0, None) - ctx.check(st_ == 200 and (doc or {}).get("status") == 200, "enabled again, it answers -> %s %s", st_, doc) - finally: - _put_back(ctx, fc, work, prior, etag, raw, dir_mode) - ctx.check(not os.path.exists(sock), "the call socket's name is left after the package went") - _as_found(ctx, fc, prior, raw, dir_mode, found_tree) diff --git a/forgetest/forgetest/suite/exthost.py b/forgetest/forgetest/suite/exthost.py index b1ba1d8..c9c14ea 100644 --- a/forgetest/forgetest/suite/exthost.py +++ b/forgetest/forgetest/suite/exthost.py @@ -763,49 +763,6 @@ def _tree(root): return sorted(out) -EVENTS_MAX_STREAMS = 3 # forgectrl's own cap (src/events.h) -EVENTS_HOST_HEADER = "X-ForgeFIRM-Client: extension-host" - - -def _stream(addr, host_client=False, keep=False, source=None): - """One GET /events straight at forgectrl's read-only listener: (status, - the words of a refusal, the socket). The socket is left open when keep, - so the caller can hold a stream. `source` binds the address the - connection comes from, because the cap counts peer addresses and the - kernel would otherwise give every loopback connection 127.0.0.1.""" - import socket - c = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - c.settimeout(8) - try: - if source: - c.bind((source, 0)) - c.connect((addr, 80)) - req = "GET /events HTTP/1.0\r\nHost: %s\r\nAccept: text/event-stream\r\n%s\r\n" % ( - addr, EVENTS_HOST_HEADER + "\r\n" if host_client else "") - c.sendall(req.encode()) - head = b"" - while b"\r\n\r\n" not in head and len(head) < 8192: - k = c.recv(4096) - if not k: - break - head += k - top, _, rest = head.partition(b"\r\n\r\n") - code = int(top.split()[1]) if top.split()[1:] else 0 - why = "" - if code != 200: - try: - why = (json.loads(rest or b"{}") or {}).get("error") or "" - except ValueError: - why = rest.decode("utf-8", "replace")[:120] - if not keep or code != 200: - c.close() - c = None - return code, why, c - except OSError as e: - c.close() - return 0, str(e), None - - REF_UI = ("the reference package's own page
\n") @@ -1355,164 +1312,6 @@ def armed_freeze(ctx): REQUIRED_HOLDS = EXT_ROOT + "/required-holds" -# ------------------------------------------------- the operator's door - -@test("exthost.events", title="The machine's events reach a package, over one subscription", - subsystem="exthost", kind="operator", hardware="takeover", est_min=6, - covers=[("forgeext", "src/evfeed.*"), ("forgeext", "src/api.*"), ("forgeext", "src/run.*"), - ("forgectrl", "src/events.*"), ("forgectrl", "src/main.c")], - requires=["exthost.service"], 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 reference package is installed with the events capability. The host holds one " - "subscription to forgectrl's stream for every package that wants events, and none when " - "no package does: with the service running the status file says the stream is connected " - "and one service wants it, and with extensions off it says neither. The package polls " - "POST /v0/events over its API socket. A poll with nothing to say comes back at its own " - "deadline and not before, with an empty list and the place it already had; the lid " - "opened and closed on the fixture reaches the package as lid events, in order and " - "numbered, within a second, so a waiting poll is woken and not left to time out. The " - "host's slot is its own: with the three ordinary streams taken, the host's subscription " - "is still there and a LAN client that asks for the host's slot by name is refused, " - "because only a loopback peer may claim it. Nothing moves and nothing fires: the lid is " - "the only thing touched. Everything is put back as exthost.service puts it back.") -def events(ctx): - import shutil - import tempfile - fc = ctx.forgectrl - ev = ctx.evidence - ctx.check(len(_host_pids()) == 1, "the extension host is not one running process: %s", _host_pids()) - ctx.check(fc.wait_idle(timeout=30, abort=ctx.aborted), "machine not idle: settings are locked") - prior = fc.settings().get("ext_enabled") or "" - raw = read_file(record_path()) - ctx.check(raw, "no setup record at %s", record_path()) - ctx.check(not os.path.exists(SAFE_FILE), "%s exists: the machine is in safe mode", SAFE_FILE) - 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") - ctx.check(st == 200 and etag, "GET /advisories/extensions -> %s", st) - work = tempfile.mkdtemp(prefix="forgetest-ffx.") - owner_key = os.path.join(EXT_ROOT, "keys", REF_KEY + ".pub") - - def said(): - """What the package has written of the events it polled for.""" - try: - return json.loads(_read(os.path.join(EXT_ROOT, "data", REF_ID, "events.json"))) - except (OSError, ValueError): - return {} - - def feed(): - return _host_status().get("events") or {} - - try: - archive, pub = _pack_reference(work, lan_ip(), more_caps=["events"]) - shutil.copy(pub, owner_key) - os.chmod(owner_key, 0o644) - r = _forgeext("install", archive, "--consent-community") - ctx.check(r.get("ok") is True, "the install -> %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 over the advisory -> %s %r", st, reply) - x = _until(ctx, lambda: _svc(REF_ID) if _svc(REF_ID).get("state") == "running" else None, 90, poll=0.5) - ctx.check(x, "the service is not running: %s", _svc(REF_ID)) - - # One subscription, and only while a package wants it. - f = _until(ctx, lambda: feed() if feed().get("connected") else None, 30) - ev["feed_with_a_reader"] = f - ctx.log("the host's subscription: %s", f) - ctx.check(f and f.get("connected") is True and f.get("wanted") == 1, - "with one package that wants events the host says %s", f) - - # A poll that has nothing to say waits for its own deadline. The - # first carries no place, so the wait is for one that does. - def polled(): - x = said() - return x if [q for q in x.get("polls") or [] if q.get("placed")] else None - - p = _until(ctx, polled, 40) - ctx.check(p, "the package has not polled with a place of its own: %s", said()) - if not p: - p = said() - ev["first_polls"] = (p.get("polls") or [])[:4] - ctx.check(all(q.get("status") == 200 for q in p.get("polls") or []), - "a poll was refused: %s", ev["first_polls"]) - # The first poll carries no place: it asks where the present is and - # is answered at once, by design. Only a poll that named its place - # and had nothing to be told waits for its deadline. - first = (p.get("polls") or [])[0] - ctx.check(first.get("placed") is False and first.get("took", 9) < 1.0, - "the first poll, which asks only where the present is, did not come back at once: %s", first) - quiet = [q for q in p.get("polls") or [] if q.get("got") == 0 and q.get("placed")] - ctx.check(quiet and all(q.get("took", 0) >= 1.8 for q in quiet), - "a poll with nothing to say came back early: %s", quiet[:4]) - place = p.get("place") - ctx.check(isinstance(place, int), "the package has no place in the stream: %r", place) - - # An edge on the machine reaches it, and wakes a waiting poll. - began = time.time() - ctx.act("lid", "open", text="The package is to be told of it.") - got = _until(ctx, lambda: said() if "lid" in (said().get("seen") or []) else None, 20, poll=0.25) - took = round(time.time() - began, 2) - ev["lid_event"] = {"after_s": took, "seen": (got or {}).get("seen"), "place": (got or {}).get("place")} - ctx.log("the lid reached the package after %.2f s: %s", took, ev["lid_event"]["seen"]) - ctx.check(got, "the lid did not reach the package: %s", said()) - ctx.check(got and got.get("place", 0) > place, "the package's place did not move: %r -> %r", - place, (got or {}).get("place")) - ctx.act("lid", "close", text="The lid ends as it was found.") - ctx.check(_until(ctx, lambda: (said().get("seen") or []).count("lid") >= 2, 20, poll=0.25), - "the lid closing did not reach the package: %s", said().get("seen")) - ev["polls_after"] = (said().get("polls") or [])[-4:] - woken = [q for q in said().get("polls") or [] if q.get("got") and q.get("placed")] - ctx.check(woken and min(q.get("took", 99) for q in woken) < 1.5, - "no poll was woken by an event: every one ran to its deadline: %s", ev["polls_after"]) - - # The host's slot is outside the cap, and only a loopback peer may claim it. - ctx.check(fc.settings().get("panel_open_reads") in (None, "", "1"), - "panel_open_reads is %r: this test reads /events without a session", - fc.settings().get("panel_open_reads")) - held = [] - try: - # Three addresses, because the cap is one stream per address, - # and the whole of 127/8 is this host: each stream comes from - # one of them, or the kernel would send all three from - # 127.0.0.1 and they would replace each other. - for i in range(EVENTS_MAX_STREAMS): - code, why, sock = _stream("127.0.0.1", keep=True, source="127.0.0.%d" % (i + 2)) - held.append((code, why, sock)) - ev["ordinary_streams"] = [h[0] for h in held] - ctx.check(all(h[0] == 200 for h in held), "the three ordinary streams: %s", - [(h[0], h[1]) for h in held]) - code, why, _sock = _stream("127.0.0.1", source="127.0.0.9") - ev["fourth_ordinary_stream"] = [code, why] - ctx.check(code == 503 and "every event stream is taken" in (why or ""), - "with the three taken a fourth ordinary stream got %s %r", code, why) - ctx.check(feed().get("connected") is True, "the host's stream went with the three being taken: %s", feed()) - code, why, _sock = _stream(lan_ip(), host_client=True) - ev["lan_claims_the_host_slot"] = [code, why] - ctx.log("a LAN client asking for the host's slot -> %s %s", code, why) - ctx.check(code == 503 and "every event stream is taken" in (why or ""), - "a LAN client that asked for the host's slot got %s %r", code, why) - ctx.check(feed().get("connected") is True, "the host lost its stream to a LAN client: %s", feed()) - finally: - for h in held: - if h[2]: - try: - h[2].close() - except OSError: - pass - - # No package wants events: the subscription is let go. - st, reply = fc.post("/settings", params={"ext_enabled": ""}) - ctx.check(st == 200, "ext_enabled='' -> %s %r", st, reply) - f = _until(ctx, lambda: feed() if not feed().get("connected") else None, 30) - ev["feed_with_no_reader"] = f - ctx.check(f is not None and not f.get("connected") and not f.get("wanted"), - "with no package wanting events the host still holds a stream: %s", feed()) - finally: - _put_back(ctx, fc, work, prior, etag, raw, dir_mode) - _as_found(ctx, fc, prior, raw, dir_mode, found_tree) - - @test("exthost.motion-jog", title="A package jogs the machine, bounded and dark", subsystem="exthost", kind="auto", hardware="takeover", mode="grbl", est_min=7, covers=[("forgeext", "src/api.*"), ("forgeext", "src/machine.*"), ("forgeext", "src/run.*"), diff --git a/forgetest/forgetest/suite/extsender.py b/forgetest/forgetest/suite/extsender.py deleted file mode 100644 index 37e4ed4..0000000 --- a/forgetest/forgetest/suite/extsender.py +++ /dev/null @@ -1,286 +0,0 @@ -# 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: