#!/usr/bin/env python3 """ A group link, `#/name@owner[/path]`, opened in the real application. `test_group_link.py` runs `group-link.js` on its own; this is where it meets the router, the account's group list and the sign-in state. Loads the shipped `app.js` in a real browser with `fetch` stubbed (no node answers, so a group page goes as far as "offline"), once per case: handle — `#/demo@someowner`: the group page, the address left as it is uuid — `#/group/`: the same page, the address showing the handle file — `#/demo@someowner/`: the page, the path kept in the address until the index can say what it is (no node here: never) unknown — `#/demo@stranger1`: not among the account's groups signed_out — `#/demo@someowner/` with no session: the sign-in form, and the address untouched under it group_link_probe.py Prints JSON: one object per case. """ 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 = 8774 RECORDS = [] socketserver.TCPServer.allow_reuse_address = True GROUP = "0f8fad5b-d9cb-469f-a165-70867728950e" FILE = "backup/city_2015/backup/IMG_0001.JPG" LINKS = { "handle": "#/demo@someowner", "uuid": f"#/group/{GROUP}", "file": f"#/demo@someowner/{FILE}", "unknown": "#/demo@stranger1", "signed_out": f"#/demo@someowner/{FILE}", } CASES = list(LINKS) PAGE = r"""

""".replace("__GROUP__", GROUP).replace("__LINKS__", json.dumps(LINKS)) class H(http.server.SimpleHTTPRequestHandler): 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())) 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 == "/": self._send(PAGE.encode(), "text/html; charset=utf-8") return 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 ctype = "text/javascript" if asset.suffix in (".js", ".mjs") else ( "application/wasm" if asset.suffix == ".wasm" else "application/octet-stream") self._send(asset.read_bytes(), ctype) def _run(case: str) -> dict | None: before = len(RECORDS) with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as profile: proc = subprocess.Popen( ["google-chrome", "--headless=new", "--disable-gpu", "--no-sandbox", f"--user-data-dir={profile}", f"http://127.0.0.1:{PORT}/?case={case}"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) for _ in range(300): if len(RECORDS) > before: break time.sleep(0.1) proc.terminate() try: proc.wait(timeout=10) except subprocess.TimeoutExpired: proc.kill() proc.wait() return RECORDS[before] if len(RECORDS) > before else None def main() -> int: with socketserver.TCPServer(("127.0.0.1", PORT), H) as srv: threading.Thread(target=srv.serve_forever, daemon=True).start() results = [_run(case) for case in CASES] if not all(results): print(json.dumps({"error": "no measurement", "got": results}), file=sys.stderr) return 1 print(json.dumps(results, indent=1)) return 0 if __name__ == "__main__": raise SystemExit(main())