#!/usr/bin/env python3 """ A group owner's settings: who was invited, and which other nodes asked to host. Renders the shipped `GroupSettingsPanel` with `fetch` stubbed: one member, one unanswered invitation, one node asking to host and one already approved. Then clicks Approve on the request and reports what reached the hub. group_hosts_probe.py [--engine chrome|firefox] Prints JSON. """ import argparse import http.server import json import socketserver import subprocess import sys import tempfile import threading import time from pathlib import Path STATIC = Path(__file__).resolve().parents[2] / "src" / "meshbay_hub" / "static" PORT = 8773 RECORDS = [] FINISHED = threading.Event() socketserver.TCPServer.allow_reuse_address = True PAGE = r"""

__HOLD__""" HOLD_TAG = '

' HOLD = "" class H(http.server.BaseHTTPRequestHandler): def log_message(self, *a): pass def do_POST(self): length = int(self.headers.get("Content-Length") or 0) body = self.rfile.read(length) if self.path == "/log": RECORDS.append(json.loads(body.decode())) FINISHED.set() self.send_response(204) self.end_headers() def _send(self, body: bytes, ctype: str) -> None: self.send_response(200) self.send_header("Content-Type", ctype) self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def do_GET(self): path = self.path.split("?")[0] if path == "/hold": FINISHED.wait(60) self._send(b"", "image/gif") elif path == "/": self._send(PAGE.replace("__HOLD__", HOLD).encode(), "text/html; charset=utf-8") else: asset = (STATIC / path.lstrip("/")).resolve() if not str(asset).startswith(str(STATIC)) or not asset.is_file(): self.send_response(404) self.end_headers() return self._send(asset.read_bytes(), "text/css" if asset.suffix == ".css" else "text/javascript" if asset.suffix == ".js" else "application/octet-stream") # Launchers and profile rule: see sticky_header_probe.py. ENGINES = { "chrome": lambda profile: [ "google-chrome", "--headless=new", "--disable-gpu", "--no-sandbox", f"--user-data-dir={profile}", "--window-size=1100,900"], "firefox": lambda profile: [ "firefox", "--headless", "--profile", profile, "--screenshot", str(Path(profile) / "shot.png"), "--window-size", "1100,900"], } def main() -> int: global HOLD ap = argparse.ArgumentParser() ap.add_argument("--engine", choices=sorted(ENGINES), default="chrome") args = ap.parse_args() HOLD = HOLD_TAG if args.engine == "firefox" else "" parent = None if args.engine == "firefox": snap = Path.home() / "snap" / "firefox" / "common" parent = str(snap if snap.is_dir() else Path.home()) with socketserver.ThreadingTCPServer(("127.0.0.1", PORT), H) as srv: threading.Thread(target=srv.serve_forever, daemon=True).start() with tempfile.TemporaryDirectory(ignore_cleanup_errors=True, prefix="meshbay-probe-", dir=parent) as profile: proc = subprocess.Popen(ENGINES[args.engine](profile) + [f"http://127.0.0.1:{PORT}/"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) for _ in range(300): if RECORDS: break time.sleep(0.1) proc.terminate() try: proc.wait(timeout=10) except subprocess.TimeoutExpired: proc.kill() proc.wait() if not RECORDS: print(json.dumps({"error": "no measurement"}), file=sys.stderr) return 1 print(json.dumps(dict(RECORDS[0], engine=args.engine), indent=1)) return 0 if __name__ == "__main__": raise SystemExit(main())