#!/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"] 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/file?t=x&v=1", "startAt": 0}: 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/file?t=x&v=2", "startAt": 0}: 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 check_music(engine: str, m: dict) -> list[str]: if "error" in m: return [f"{engine}: music: {m['error']}"] bad = [] f = m["first"]["file"] if not (f["type"] == "audio/mpeg" and f["title"] == "One" and f["artist"] == "Some Band" and f["album"] == "An Album" and f["cover"] and m["first"]["coverFetched"]): bad.append(f"the first track left without its album's cover: {m['first']}") third = m["third"] if not (third["file"]["title"] == "Three" and third["file"]["cover"] and third["mbCover"]): bad.append(f"the node's cover did not reach the television: {third}") if "t1" in third["metaAsked"] or "t2" in third["metaAsked"]: bad.append(f"the node was asked for a cover a track already had: {third['metaAsked']}") if m["first"]["connect"] != {"deviceId": "tv1", "mediaUrl": "http://tv/file?t=x&v=1", "startAt": 0}: bad.append(f"connect: {m['first']['connect']}") if not m["barButton"]: bad.append("the music bar's cast button does not say it is casting") if not m["volumeDisabled"]: bad.append("the volume slider still acts while casting") if m["seek"] != 30: bad.append(f"seek: {m['seek']}") s = m["second"] if s["file"]["title"] != "Two" or s["title"] != "Two" or s["reload"] != { "mediaUrl": "http://tv/file?t=x&v=2", "startAt": 0}: bad.append(f"the end of a track did not move the queue on: {s}") if m["filesAfterSettling"] != 2: bad.append(f"a finished track moved the queue on more than once: {m['filesAfterSettling']}") if m["afterStop"]["disconnects"] < 1 or m["afterStop"]["barCasting"]: bad.append(f"stopping: {m['afterStop']}") return [f"{engine}: music: {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") music = browse(engine, "?music") if cast is None or plain is None or music is None: failures.append(f"{engine}: the page never reported") continue print(f"{engine}:", json.dumps({"photos-videos": cast, "music": music}, indent=1)) failures += check(engine, cast, plain) failures += check_music(engine, music) 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())