#!/usr/bin/env python3
"""
Does sending a chat message come back?
Mounts **the real `ChatPanel` over the real `MeshBayTransport`** — both shipped
modules, neither a model of the other — and types a message into the composer
the way a person does. Only the DataChannel is replaced, by a stand-in that
answers what the node answers.
It exists because the defect it was written for is invisible to a structural
test and to `chat_scroll_probe.py` alike: nothing in `chat-app.js` is wrong,
and the transport routes every message it knows how to route. The node's reply
to a chat message is a bare `{"type": "ack"}` that names no request, so it fell
through to `_dispatch`'s arrival-order guess and was handed to whichever
request happened to be waiting — a `media_meta_req` from the Videos tab, say,
which the node never answered because it refused the file_id with a bare
`error` that named no request either. The chat send then waited out its own 30s
timeout with the composer disabled, so the tab looked frozen and the message
never appeared, while the node had stored it all along.
chat_send_probe.py
Prints JSON: `steps`, the state of the panel at each stage, and `log`, what the
transport sent and how the deliberately-unanswered request ended up.
"""
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 = 8755
RECORDS = []
socketserver.TCPServer.allow_reuse_address = True
PAGE = r"""
a group
"""
class H(http.server.BaseHTTPRequestHandler):
def log_message(self, *a):
pass
def do_POST(self):
RECORDS.append(json.loads(
self.rfile.read(int(self.headers["Content-Length"])).decode()))
self.send_response(204)
self.end_headers()
def do_GET(self):
if self.path == "/":
body, ctype = PAGE.encode(), "text/html; charset=utf-8"
else:
path = (STATIC / self.path.lstrip("/")).resolve()
if not str(path).startswith(str(STATIC)) or not path.is_file():
self.send_response(404)
self.end_headers()
return
body = path.read_bytes()
ctype = ("text/css" if path.suffix == ".css"
else "text/javascript" if path.suffix == ".js"
else "application/octet-stream")
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 main() -> int:
with socketserver.TCPServer(("127.0.0.1", PORT), H) as srv:
threading.Thread(target=srv.serve_forever, daemon=True).start()
with tempfile.TemporaryDirectory() as profile:
proc = subprocess.Popen(
["google-chrome", "--headless=new", "--disable-gpu", "--no-sandbox",
f"--user-data-dir={profile}", "--window-size=1100,800",
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()
if not RECORDS:
print(json.dumps({"error": "no measurement"}), file=sys.stderr)
return 1
print(json.dumps(RECORDS[0], indent=1))
return 0
if __name__ == "__main__":
raise SystemExit(main())