#!/usr/bin/env python3
"""
An invitation link, opened in the real application.
`test_invite_link_client.py` runs the link's functions one by one. What only the
running application can show is how they meet the router, the sign-in state and
the hub calls: that the code is out of the address before anything routes on
it, that a signed-out reader is sent to register with the invitation kept, and
that a signed-in reader is shown the invitation, joins with one click and lands
on the group — with the code never in a request to the hub.
Loads the shipped `app.js` in a real browser with `fetch` stubbed, twice:
signed_out — a link, no session
signed_in — the same link, a session; then the Join button is clicked
invite_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 = 8771
RECORDS = []
socketserver.TCPServer.allow_reuse_address = True
GROUP = "0f8fad5b-d9cb-469f-a165-70867728950e"
TICKET = "AbCdEfGhIjKlMnOpQr-_12"
NODE = "A" * 43
CODE = "K7P2-9WQX"
LINK = f"#/invite?v=1&g={GROUP}&t={TICKET}&n={NODE}&c={CODE}"
PAGE = r"""
""".replace("__GROUP__", GROUP).replace("__LINK__", LINK).replace("__CODE__", CODE)
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("signed_out"), _run("signed_in")]
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())