From 243bcf017b4957237b3ddf556bd0a37296eb8798 Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Mon, 5 Oct 2026 01:16:12 +0200 Subject: feat(cast): photos on the TV, and a television chosen once for the session A cast button in Videos' toolbar, at the top of Photos, in an album's bar and in the lightbox, in a group and in Search alike. A television chosen there is kept for the session: a film opened plays on it with the player as its remote from the start, and a photo opened in the lightbox is shown on it, scaled to 1920x1080, upright, as JPEG. The lightbox gains a slideshow. The relay serves one photo at /image behind the stream's token, on the desktop and on Android; the shell, not the page, decides that the receiver loads it as a picture. Co-Authored-By: Claude Opus 5.5 --- .../tests/harness/cast_session_probe.py | 393 +++++++++++++++++++++ 1 file changed, 393 insertions(+) create mode 100644 packages/meshbay-hub/tests/harness/cast_session_probe.py (limited to 'packages/meshbay-hub/tests/harness') diff --git a/packages/meshbay-hub/tests/harness/cast_session_probe.py b/packages/meshbay-hub/tests/harness/cast_session_probe.py new file mode 100644 index 0000000..2904860 --- /dev/null +++ b/packages/meshbay-hub/tests/harness/cast_session_probe.py @@ -0,0 +1,393 @@ +#!/usr/bin/env python3 +""" +A television chosen once, then photos and a film sent to it. + +Casting a photo or opening a film on a television chosen elsewhere is a claim +about four pieces at once: the cast button (cast-session.js), the lightbox +that follows it, the player that opens as a remote, and the native bridge the +page talks to. This renders the shipped `GroupPage` against a stub node and a +stub bridge that records what the page asks of it, and walks it as a reader +would: open an album, choose the television, page on, start a slideshow, +then open a film from Videos and stop casting from its remote. + +A second case renders the same page with no bridge at all — a browser — where +no cast button may appear. + + cast_session_probe.py Chrome, then Firefox +""" + +import http.server +import json +import shutil +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 = 8763 +RECORDS = [] +FINISHED = threading.Event() +socketserver.TCPServer.allow_reuse_address = True + +FRAME = r""" + + +

+ +""" + +# Firefox headless runs a page only while taking its screenshot, which is taken +# once the page has loaded: an image that does not answer until the case has +# reported keeps it loading (the same hold as invitation_probe.py). +HOLD_TAG = '

' + +PAGE = r""" + + +{hold}""" + + +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())) + FINISHED.set() + 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, _, query = self.path.partition("?") + if path == "/hold": + FINISHED.wait(90) + self._send(b"", "image/gif") + elif path == "/": + q = f"?{query.replace('engine=firefox', '').strip('&')}" if query else "" + hold = HOLD_TAG if "engine=firefox" in query else "" + page = PAGE.replace("{query}", q if q != "?" else "").replace("{hold}", hold) + 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 browse(engine: str, query: str) -> dict | None: + RECORDS.clear() + FINISHED.clear() + sep = "&" if query else "?" + url = f"http://127.0.0.1:{PORT}/{query}{sep + 'engine=firefox' if engine == 'firefox' else ''}" + # Firefox is a snap here, and needs its profile under $HOME (invitation_probe.py). + snap = Path.home() / "snap" / "firefox" / "common" + parent = (str(snap if snap.is_dir() else Path.home()) if engine == "firefox" else None) + with tempfile.TemporaryDirectory(ignore_cleanup_errors=True, prefix="meshbay-probe-", + dir=parent) as profile: + if engine == "chrome": + cmd = ["google-chrome", "--headless=new", "--disable-gpu", "--no-sandbox", + f"--user-data-dir={profile}", "--window-size=1100,900", url] + else: + cmd = ["firefox", "--headless", "--profile", profile, "--window-size=1100,900", + "--screenshot", str(Path(profile) / "shot.png"), url] + proc = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + deadline = time.time() + 90 + while not RECORDS and time.time() < deadline: + time.sleep(0.2) + proc.terminate() + try: + proc.wait(timeout=20) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait() + return RECORDS[0] if RECORDS else None + + +def check(engine: str, cast: dict, plain: dict) -> list[str]: + bad = [] + if "error" in cast: + return [f"{engine}: {cast['error']}"] + if not cast["landingButton"]: + bad.append("no cast button on the Photos landing") + if not cast["albumButton"]: + bad.append("no cast button in the album bar") + first = cast["first"] + if first["count"] != "1 / 3": + bad.append(f"choosing a television did not open the first photo: {first['count']}") + img = first["image"]["arg"] + if not (img["type"] == "image/jpeg" and img["jpeg"] and img["width"] == 1440 + and img["height"] == 1080): + bad.append(f"the photo did not leave as a 1440x1080 JPEG: {img}") + if first["connect"] != {"deviceId": "tv1", "mediaUrl": "http://tv/image?t=x&v=1"}: + bad.append(f"connect: {first['connect']}") + if not first["lightboxCasting"]: + bad.append("the lightbox's cast button does not say it is casting") + if cast["second"]["count"] != "2 / 3" or cast["second"]["reload"] != { + "mediaUrl": "http://tv/image?t=x&v=2"}: + bad.append(f"next photo: {cast['second']}") + if cast["slideshow"]["count"] != "3 / 3" or cast["slideshow"]["images"] != 3: + bad.append(f"slideshow: {cast['slideshow']}") + if not cast["videosCasting"]: + bad.append("Videos does not show the television chosen in Photos") + if cast["remote"]["device"] != "Salon" or cast["remote"]["pickerOpen"]: + bad.append(f"the film did not open as the television's remote: {cast['remote']}") + if cast["afterStop"]["videosCasting"] or cast["afterStop"]["disconnects"] < 1: + bad.append(f"stopping from the remote: {cast['afterStop']}") + if "error" in plain: + bad.append(f"no bridge: {plain['error']}") + elif plain["landingButton"] or plain["albumButton"] or plain["videosButton"]: + bad.append(f"a cast button without a bridge: {plain}") + return [f"{engine}: {b}" for b in bad] + + +def main() -> int: + failures = [] + with socketserver.ThreadingTCPServer(("127.0.0.1", PORT), H) as srv: + threading.Thread(target=srv.serve_forever, daemon=True).start() + for engine in ("chrome", "firefox"): + if shutil.which("google-chrome" if engine == "chrome" else engine) is None: + print(f"{engine}: not installed, skipped") + continue + cast = browse(engine, "") + plain = browse(engine, "?nobridge") + if cast is None or plain is None: + failures.append(f"{engine}: the page never reported") + continue + print(f"{engine}:", json.dumps(cast, indent=1)) + failures += check(engine, cast, plain) + for f in failures: + print("FAIL", f, file=sys.stderr) + if not failures: + print("OK") + return 1 if failures else 0 + + +if __name__ == "__main__": + sys.exit(main()) -- cgit v1.2.3