#!/usr/bin/env python3 """ Which tab does a group actually open on? The landing tab is chosen from a preference before the node has said which applications the group runs, so the two can disagree — the group has Chat disabled while 'chat' is the default, or the reader's preferred app is one this group does not offer. Both used to land on a tab that rendered *nothing*: no panel, no tab shown active, and nothing on screen to say why. That is not visible in the source, where the tab is a plain string and the app list arrives from a handshake several `await`s later. So this renders the real `GroupPage` — the shipped module — against a stub node that answers with a chosen `enabled_apps`, and reports the tab that ended up active. group_tab_probe.py Prints JSON: one entry per case, with the tab bar as rendered and whether a panel was drawn under it. """ 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 = 8748 RECORDS = [] socketserver.TCPServer.allow_reuse_address = True # name -> (the reader's default_tab preference, what the group runs) CASES = [ ("every app, default chat", "chat", None), ("chat disabled", "chat", ["files", "video"]), ("preferred app absent", "music", ["chat", "files"]), ("preferred app present", "video", ["chat", "files", "video"]), ("preference names nothing real", "nonsense", ["files", "music"]), ] FRAME = r"""

""" PAGE = r"""

""" class H(http.server.BaseHTTPRequestHandler): def log_message(self, *a): pass def do_POST(self): length = int(self.headers.get("Content-Length") or 0) if self.path == "/log": RECORDS.append(json.loads(self.rfile.read(length).decode())) else: self.rfile.read(length) 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 % {"count": len(CASES)}).encode(), "text/html; charset=utf-8") elif path == "/case": index = int(self.path.split("n=")[1]) _, pref, enabled = CASES[index] self._send((FRAME % {"index": index, "pref": json.dumps(pref), "enabled": json.dumps(enabled)}).encode(), "text/html; charset=utf-8") elif path == "/v1/groups/g1/nodes": self._send(b'{"nodes": [{"node_id": "n1"}]}', "application/json") 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") def main() -> int: with socketserver.TCPServer(("127.0.0.1", PORT), H) as srv: threading.Thread(target=srv.serve_forever, daemon=True).start() # ignore_cleanup_errors: Chrome's children (zygote, renderer, gpu) # outlive terminate() on the parent by a moment and go on writing into # the profile. rmtree then walks a directory that gains a file between # its readdir and its rmdir and raises "Directory not empty" -- which # failed the probe, which failed every test in the file, intermittently # and for a reason nowhere near the chat code they were testing. A few # bytes left in a throwaway profile are harmless; failing the run is not. 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}", "--window-size=1100,900", 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 by_case = {r["case"]: r for r in RECORDS[0]} print(json.dumps( [dict(name=CASES[i][0], preference=CASES[i][1], enabled=CASES[i][2], **by_case[i]) for i in sorted(by_case)], indent=1)) return 0 if __name__ == "__main__": raise SystemExit(main())