#!/usr/bin/env python3
"""
What is on screen after watching one episode of a show, and closing the player.
Playing an episode used to close the show's detail modal, so the next episode
meant opening the show again and picking the season again, every time. The
modal now stays open under the player. That is a claim about three components
at once — `PosterGrid` deciding whether to close it, `GroupPage` mounting the
player beside it, and the stylesheet deciding which of two `.video-overlay`s is
on top — so this renders the shipped `GroupPage` against a stub node and walks
it as a reader would, reading back what is on the page after each step.
A film goes through the same modal and must still close it: it has nothing left
to pick from.
video_series_probe.py
"""
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 = 8761
RECORDS = []
socketserver.TCPServer.allow_reuse_address = True
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.encode(), "text/html; charset=utf-8")
elif path == "/case":
self._send(FRAME.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()
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)
deadline = time.time() + 90
while not RECORDS and time.time() < deadline:
time.sleep(0.2)
proc.terminate()
proc.wait(timeout=20)
if not RECORDS:
print("the page never reported", file=sys.stderr)
return 1
print(json.dumps(RECORDS[0]))
return 0
if __name__ == "__main__":
raise SystemExit(main())