#!/usr/bin/env python3
"""
Can the reader scroll up in the chat, and does the panel sit still when nobody
is touching it?
`scroll_probe.py` answers "is the document taller than the window". This one
mounts **the real `ChatPanel`** — the shipped module, not a model of it — in a
browser, drives it the way a person does, and reports where the list ends up.
It exists because the defect it was written for is invisible to every other kind
of test here. `fit()` set the panel's height, read the document's overflow back
and subtracted it, so the document alternately did and did not overflow the
window; the page scrollbar appeared and vanished with it, `visualViewport` fired
`resize` at each pass, and `fit()` is bound to that event. It therefore re-ran
about 120 times a second for the life of the panel, re-pinning the list to the
bottom every time — which undid each attempt to scroll up *inside the same
frame*, before the `scroll` event that would have recorded it was delivered. The
source reads as correct: every pin is guarded by "only if the reader is at the
bottom", and the reader never got to stop being at the bottom.
chat_scroll_probe.py
Prints JSON: an `idle` block (viewport-resize firings on a page nobody touches)
and a `steps` list (scroll position after each stage of a reading session).
"""
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 = 8747
RECORDS = []
socketserver.TCPServer.allow_reuse_address = True
# A bare document fires a couple of these while it settles. Anything above this
# is the panel driving itself.
IDLE_RESIZE_CEILING = 20
PAGE = r"""
a group
a description
"""
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:
# Real time, not `--virtual-time-budget`: the defect is a feedback
# loop between layout and an event, and a virtual clock does not
# run it.
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())