import json import os import math from pathlib import Path import plistlib import selectors import shutil import subprocess import sys import tempfile import time import uuid from urllib.parse import urlsplit from streaming import StreamingServer, resolved_fixtures class StartupStreamingServer(StreamingServer): def __init__(self, fixtures): super().__init__(fixtures) self.subtitle_requests = [0, 0] def _get(self, request): parsed = urlsplit(request.path) if parsed.path in ("/startup-0.vtt", "/startup-1.vtt"): index = int(parsed.path[len("/startup-")]) if parsed.query != self.resolved_token or request.headers.get("X-Subtitle") != f"track-{index}": self._reply(request, 403, b"Missing subtitle credentials") return with self.lock: self.subtitle_requests[index] += 1 body = f"WEBVTT\n\n00:00:00.000 --> 00:01:00.000\nIndependent subtitle {index}\n".encode() self._reply(request, 200, body, "text/vtt") return super()._get(request) def check_overlay_hover(root, app, fixtures): result_path = root / "overlay-hover.json" diagnostics = root / "overlay-hover.log" result = subprocess.run( ["open", "-n", "-W", "--stderr", str(diagnostics), "-a", str(app), str(fixtures / "cache.mp4"), "--args", "--no-config", "--check-overlay-hover", str(result_path)], capture_output=True, text=True, timeout=25, ) report = json.loads(result_path.read_text()) if result_path.exists() else {} if result.returncode != 0 or not report.get("passed") or report.get("steps") != 7: log = diagnostics.read_text(errors="replace") if diagnostics.exists() else "" raise RuntimeError(f"native overlay hover failed: {report}; {result.stdout}{result.stderr}{log}") print("PASS native overlay hover: video, title, playback, top-right and queue controls stay visible; " "window exit hides and reentry reveals controls") def check_window_drag(root, app, fixtures): result_path = root / "window-drag.json" diagnostics = root / "window-drag.log" result = subprocess.run( ["open", "-n", "-W", "--stderr", str(diagnostics), "-a", str(app), str(fixtures / "cache.mp4"), "--args", "--no-config", "--check-window-drag", str(result_path), str(fixtures)], capture_output=True, text=True, timeout=30, ) report = json.loads(result_path.read_text()) if result_path.exists() else {} if result.returncode != 0 or not report.get("passed"): log = diagnostics.read_text(errors="replace") if diagnostics.exists() else "" raise RuntimeError(f"native window drag failed: {report}; {result.stdout}{result.stderr}{log}") print("PASS native window dragging: " + str(report)) def check_queue_drag(root, app, fixtures): result_path = root / "queue-drag.json" diagnostics = root / "queue-drag.log" result = subprocess.run( ["open", "-n", "-W", "--stderr", str(diagnostics), "-a", str(app), str(fixtures / "cache.mp4"), "--args", "--no-config", "--check-queue-drag", str(result_path)], capture_output=True, text=True, timeout=30, ) report = json.loads(result_path.read_text()) if result_path.exists() else {} if result.returncode != 0 or not report.get("passed"): log = diagnostics.read_text(errors="replace") if diagnostics.exists() else "" raise RuntimeError(f"native queue drag failed: {report}; {result.stdout}{result.stderr}{log}") print("PASS native queue: drag down/up without changing playback, release to nearest slot, single click plays without reordering") def check_window_layout(root, app, fixtures): result_path = root / "window-layout.json" diagnostics = root / "window-layout.log" result = subprocess.run( ["open", "-g", "-n", "-W", "--stderr", str(diagnostics), "-a", str(app), str(fixtures / "cache.mp4"), "--args", "--no-config", "--check-window-layout", str(result_path)], capture_output=True, text=True, timeout=40, ) log = diagnostics.read_text(errors="replace") if diagnostics.exists() else "" report = json.loads(result_path.read_text()) if result_path.exists() else {} expected_animations = 0 if report.get("reduceMotion") else 16 expected_queue_animations = 0 if report.get("reduceMotion") else 9 if (result.returncode != 0 or not report.get("passed") or report.get("steps") != 36 or report.get("animatedTransitions") != expected_animations or report.get("animatedQueueTransitions") != expected_queue_animations): raise RuntimeError(f"native window layout failed: {report}; {result.stdout}{result.stderr}{log}") print(f"PASS native SwiftUI window layout: {report['animatedTransitions']} compact and " f"{report['animatedQueueTransitions']} queue animations, stable video size, responsive reversals, " "screen-edge fitting, fullscreen fixed bounds, minimum sizes and exact normal geometry restored") def check_startup_errors(root, app): config = root / "invalid-startup.toml" config.write_text("[ui]\nbuffer_info = [\n") for name, arguments, title, detail in [ ("configuration", ["--config-file", config.as_uri()], "Could not load Terebi configuration", str(config)), ("progress", ["--no-config", "--progress-file", str(root / "missing-parent/progress.jsonl")], "Could not initialize playback progress reporting", "progress"), ]: report_path = root / f"startup-error-{name}.json" diagnostics = root / f"startup-error-{name}.log" result = subprocess.run( ["open", "-n", "-W", "--stderr", str(diagnostics), "-a", str(app), "--args", "--check-startup-error", str(report_path), *arguments], capture_output=True, text=True, timeout=25, ) report = json.loads(report_path.read_text()) if report_path.exists() else {} log = diagnostics.read_text(errors="replace") if diagnostics.exists() else "" if (result.returncode != 0 or not report.get("passed") or title not in report.get("text", []) or detail not in "\n".join(report.get("text", [])) or detail not in log): raise RuntimeError(f"native {name} startup alert failed: {report}; " f"open status {result.returncode}; {result.stdout}{result.stderr}{log}") print("PASS native startup errors: configuration and progress failures show focused critical alerts with diagnostics") def check_cli_instances(root, app, fixtures): cli = root / "terebi" cli.symlink_to(app / "Contents/MacOS/terebi-cli") extractor = app / "Contents/MacOS/yt-dlp-offline" (app / "Contents/MacOS/yt-dlp").rename(extractor) runtime = app / "Contents/MacOS/deno" canonical = "https://www.youtube.com/watch?v=aqz-KE-bpKQ" reported_urls = [f"https://www.youtube.com/watch?v=progress00{index}" for index in range(1, 5)] config = root / "xdg/terebi/config.toml" environment = dict(os.environ, XDG_CONFIG_HOME=str(config.parent.parent)) def invoke(arguments, expected_code=0): result = subprocess.run([str(cli), *arguments], capture_output=True, text=True, timeout=20, env=environment) if result.returncode != expected_code: raise RuntimeError(f"CLI {arguments} returned {result.returncode}, expected {expected_code}; {result.stdout}{result.stderr}") return result def process_diagnostics(pid): try: return (root / f"stderr-{pid}.log").read_text(errors="replace") except FileNotFoundError: return "" def wait_states(count, ready, message, allow_errors=()): deadline = time.monotonic() + 30 states = {} while time.monotonic() < deadline: states = {int(path.stem.split("-")[1]): json.loads(path.read_text()) for path in root.glob("state-*.json")} for pid in states: try: os.kill(pid, 0) except ProcessLookupError: raise RuntimeError(f"{message}: GUI PID {pid} exited unexpectedly; " f"last state: {states[pid]}; diagnostics: {process_diagnostics(pid)}") complete = len(states) == count and all( set(state["pids"]) == set(states) and state["windows"] == 1 for state in states.values()) if any(state["error"] for pid, state in states.items() if pid not in allow_errors): raise RuntimeError(f"{message}: native playback error: {states}") if complete and ready(states): return states time.sleep(0.02) # Infrastructure deadline; assertions use native events/PTS, not wall dates. raise RuntimeError(f"{message}: native state deadline: {states}; " f"process diagnostics: { {pid: process_diagnostics(pid) for pid in states} }") def settings(state): return (state["showBufferInfo"], state["automaticQuality"], state["snapToEdges"]) def write_request(pid, events=None, media=None, action=None, **arguments): request = {"id": uuid.uuid4().hex, "events": events or [], **arguments} if media is not None: request["media"] = str(media) if action is not None: request["action"] = action path = root / f"keys-{pid}.json" pending = path.with_suffix(".tmp") pending.write_text(json.dumps(request)) pending.replace(path) return request["id"] def input_request(pid, count, ready, message, events=None, media=None, action=None, allow_error=False, **arguments): request_id = write_request(pid, events=events, media=media, action=action, **arguments) return wait_states(count, lambda states: states[pid]["requestID"] == request_id and ready(states[pid]), f"{message} (expected request {request_id})", allow_errors=(pid,) if allow_error else ()) def read_reports(path, allowed): try: lines = path.read_bytes().split(b"\n")[:-1] # The writer may still be filling the trailing line. except FileNotFoundError: return [] records = [json.loads(line) for line in lines if line] miruro = isinstance(allowed, list) identity_field = "media_identity" if miruro else "media_url" fields = {"version", "event", identity_field, "position_seconds", "duration_seconds", "completion_eligible"} for record in records: duration = record.get("duration_seconds") position = record.get("position_seconds") valid = (set(record) == fields and record["version"] == (2 if miruro else 1) and record["event"] in ("progress", "ended", "closed") and record[identity_field] in allowed and type(position) in (int, float) and math.isfinite(position) and position >= 0 and (duration is None or (type(duration) in (int, float) and math.isfinite(duration) and duration > 0)) and type(record["completion_eligible"]) is bool and (not record["completion_eligible"] or duration is not None)) if not valid: raise RuntimeError(f"invalid or mixed native progress stream {path}: {record}") return records def wait_report(path, allowed, ready, message): deadline = time.monotonic() + 30 records = [] while time.monotonic() < deadline: records = read_reports(path, allowed) if ready(records): return records time.sleep(0.02) raise RuntimeError(f"{message}: native report deadline {path}: {records}") def at_position(state, position): return (state["status"] == 2 and not state["busy"] and not state["seekPending"] and abs(state["position"] - position) < 0.5) def progress_at(record, url, position): return (record["event"] == "progress" and record["media_url"] == url and abs(record["position_seconds"] - position) < 0.5) def watch_fact(record): return (record["completion_eligible"] and record["duration_seconds"] is not None and record["position_seconds"] / record["duration_seconds"] >= 0.95) def quit_instance(pid, count, events=None): if events is None: write_request(pid, action="quit") else: write_request(pid, events=events) deadline = time.monotonic() + 30 while time.monotonic() < deadline: survivors = [json.loads(path.read_text()) for path in root.glob("state-*.json") if path.stem != f"state-{pid}"] exited = False if count == 1: try: os.kill(pid, 0) except ProcessLookupError: exited = True if exited or (survivors and all(pid not in state["pids"] for state in survivors)): (root / f"state-{pid}.json").unlink() (root / f"keys-{pid}.json").unlink() return wait_states(count - 1, lambda _: True, "normal quit unregisters only the targeted GUI") time.sleep(0.02) raise RuntimeError(f"normal quit did not unregister GUI PID {pid}") def check_resolved_instances(server, count): originals = wait_states(count, lambda _: True, "existing players are ready before resolved launches") states = originals resolved_pids = [] resolved_reports = [] identities = [] for index in range(2): source = root / f"resolved source {index}.json" report = root / f"resolved progress {index}.jsonl" identity = {"provider": "miruro", "media_id": "startup_opaque", "episode": index + 1} source.write_text(json.dumps({ "version": 1, "url": server.base_url + "/resolved/encrypted/master.m3u8?" + server.resolved_token, "headers": {"X-Terebi-Fixture": "combined", "Authorization": "Bearer " + server.resolved_secret}, "identity": identity, "title": f"Independent resolved episode {index + 1}", "subtitles": [{"url": server.base_url + f"/startup-{index}.vtt?" + server.resolved_token, "format": "vtt", "language": "en", "label": f"Track {index}", "default": True, "headers": {"X-Subtitle": f"track-{index}"}}] })) source.chmod(0o600) previous = set(states) invoke(["--new-instance", "--source-file", str(source), "--progress-file", str(report)]) states = wait_states(count + index + 1, lambda snapshots: any( pid not in previous and state["media"] == f"Independent resolved episode {index + 1}" and state["status"] == 3 and state["hasVideo"] and state["hasAudio"] and state["canSeek"] and not state["busy"] and not state["seekPending"] and state["subtitleLabels"] == [f"Track {index}"] and state["selectedSubtitle"] == 0 and state["subtitleText"] == f"Independent subtitle {index}" and not state["subtitleError"] for pid, state in snapshots.items()), "actual source-file CLI reaches registered GUI HLS and VTT playback") pid = next(pid for pid in states if pid not in previous) states = input_request(pid, count + index + 1, lambda state: state["status"] == 2, "resolved GUI pauses independently after decoded startup", action="pause") states = input_request(pid, count + index + 1, lambda state: state["selectedSubtitle"] == -1 and not state["subtitleText"], "visible native Off item disables rendered captions", action="caption", label="Off") states = input_request(pid, count + index + 1, lambda state: state["selectedSubtitle"] == 0 and state["subtitleText"] == f"Independent subtitle {index}", "visible native track item restores rendered captions", action="caption", label=f"Track {index} (en)") wait_report(report, [identity], lambda rows: any( row["event"] == "progress" and row["completion_eligible"] for row in rows), "resolved GUI owns v2 stable-identity reporting after CLI exit") resolved_pids.append(pid) resolved_reports.append(report) identities.append(identity) for pid, original in originals.items(): if states[pid]["media"] != original["media"] or settings(states[pid]) != settings(original): raise RuntimeError("resolved launches disturbed an existing player") with server.lock: stats = server.resolved_stats.get("encrypted", {}) if any(server.subtitle_requests[index] == 0 for index in range(2)) or any( stats.get(kind, {}).get("requests", 0) == 0 or stats[kind]["denied"] for kind in ("master", "variant", "key", "segment")): raise RuntimeError("actual CLI playback lost complete HLS URLs, credentials, or independent VTT headers") for index in reversed(range(2)): pid = resolved_pids[index] quit_instance(pid, count + index + 1) wait_report(resolved_reports[index], [identities[index]], lambda rows: bool(rows) and rows[-1]["event"] == "closed", "resolved GUI quit drains only its stable episode identity") report_bytes = resolved_reports[index].read_bytes() if not report_bytes.endswith(b"\n") or any(secret in report_bytes for secret in ( server.resolved_token.encode(), server.resolved_secret.encode(), server.base_url.encode())): raise RuntimeError("resolved report is incomplete or exposes transport secrets") wait_states(count, lambda _: True, "original players survive independent resolved GUI quits") print("PASS actual resolved CLI handoff: two isolated GUI episodes, authenticated encrypted HLS, independent default VTT, v2 progress and quit") def check_stream_queue_instance(): # DEBUG-queue-startup: distinguish queue startup from ordinary empty startup. invoke(["--new-instance", "--no-config"]) empty_states = wait_states(1, lambda _: True, "diagnostic empty startup creates its window") quit_instance(next(iter(empty_states)), 1) wait_states(0, lambda _: True, "diagnostic empty startup quits") session = (root / "stream-queue-session").resolve() session.mkdir(mode=0o700) paths = [session / "first.jsonl", session / "second.jsonl"] urls = [reported_urls[0], canonical] # Deliberately not alphabetical URL order. manifest = session / "queue.json" manifest.write_text(json.dumps({"version": 1, "items": [ {"url": url, "progress_file": str(path)} for url, path in zip(urls, paths)]})) invoke(["--new-instance", "--no-config", "--queue-file", str(manifest)]) states = wait_states(1, lambda snapshots: all( state["queueURLs"] == urls and state["queueIndex"] == 0 and not state["busy"] and state["status"] == 3 and state["hasVideo"] and state["position"] > 0.2 for state in snapshots.values()), "actual queue CLI delivers ordered native stream playback") pid = next(iter(states)) if paths[1].exists(): raise RuntimeError("unselected queue entry must not create or write its report") input_request(pid, 1, lambda state: state["queueIndex"] == 1 and not state["busy"] and state["status"] == 3, "native queue manual next", action="queueAdvance", offset=1) input_request(pid, 1, lambda state: state["queueIndex"] == 0 and not state["busy"] and state["status"] == 3, "native queue previous revisits owned report", action="queueAdvance", offset=-1) states = input_request(pid, 1, lambda state: state["status"] == 2 and not state["busy"], "pause queue first item", action="pause") target = states[pid]["duration"] - 0.3 input_request(pid, 1, lambda state: at_position(state, target), "acknowledged queue end seek", action="seek", position=target) input_request(pid, 1, lambda state: state["queueIndex"] == 1 and not state["busy"] and state["status"] == 3 and state["queueURLs"] == urls, "natural end advances without losing supplied queue", action="play") quit_instance(pid, 1) for path, url in zip(paths, urls): rows = wait_report(path, {url}, lambda rows: bool(rows) and rows[-1]["event"] == "closed", "queue quit drains independent stable report") if len(rows) < 4: raise RuntimeError(f"queue revisit must append without clobbering earlier records: {path}: {rows}") wait_states(0, lambda _: True, "queue GUI quit leaves no stale instance") print("PASS actual queue CLI: ordered lazy streams, manual next/previous, natural end advance and private report revisits") resolved_fixtures(fixtures) with StartupStreamingServer(fixtures) as server: if extractor and runtime: choices = [] for index, url in enumerate([canonical, *reported_urls[:3]]): metadata = root / f"youtube-{index}.json" metadata.write_text(json.dumps({ "id": url.split("v=")[1], "title": "Independent YouTube fixture" if index == 0 else f"Reported YouTube fixture {index}", "duration": 120, "is_live": False, "webpage_url": url, "extractor": "youtube", "extractor_key": "Youtube", "formats": [{ "format_id": str(height), "height": height, "width": height * 16 // 9, "url": server.base_url + "/media/cache.mp4", "protocol": "http", "ext": "mp4", "vcodec": "mpeg4", "acodec": "aac", "dynamic_range": "SDR", "http_headers": {"X-Terebi-Fixture": "combined"}} for height in (180, 360)]})) choices.append(f" if (!strcmp(url, {json.dumps(url)})) info = {json.dumps(str(metadata))};") wrapper = app / "Contents/MacOS/yt-dlp" source = root / "offline-extractor.c" source.write_text( '#include \n#include \n#include \n' 'int main(int argc, char **argv) {\n' ' if (argc < 3) return 1;\n' ' const char *url = argv[argc - 1];\n char *info = NULL;\n' + "\n".join(choices) + '\n if (!info) { fputs("Fixture extraction deliberately failed\\n", stderr); return 1; }\n' + f' char *args[argc + 2]; args[0] = {json.dumps(str(extractor))};\n' + ' args[1] = "--load-info-json"; args[2] = info;\n' + ' for (int i = 1; i < argc - 2; ++i) args[i + 2] = argv[i];\n' + ' args[argc] = NULL;\n execv(args[0], args); perror("execv"); return 1;\n}\n') subprocess.run(["xcrun", "clang", "-O2", str(source), "-o", str(wrapper)], check=True) subprocess.run(["codesign", "--force", "--sign", "-", str(wrapper)], check=True) runtime_settings = {"directory": str(root), "path": str(root) + os.pathsep + os.environ.get("PATH", "")} (app / "Contents/Resources/cli-check.json").write_text(json.dumps(runtime_settings)) subprocess.run(["codesign", "--force", "--sign", "-", str(app)], check=True) if os.environ.get("TEREBI_CHECK_STREAM_QUEUE_ONLY") == "1": if not extractor or not runtime: raise RuntimeError("stream queue CLI smoke requires bundled extractor and runtime fixtures") check_stream_queue_instance() return if os.environ.get("TEREBI_CHECK_RESOLVED_ONLY") == "1": check_resolved_instances(server, 0) return if extractor and runtime: check_stream_queue_instance() movie = fixtures / "cache.mp4" audio = root / "audio with spaces.wav" shutil.copy2(fixtures / "audio.wav", audio) invoke([str(movie)]) first = wait_states(1, lambda states: all(state["media"] == movie.name and state["status"] == 3 and state["position"] > 0 and state["hasVideo"] and state["active"] and state["keyWindow"] and state["mainWindow"] for state in states.values()), "cold default CLI opens and focuses one actual player window before synthetic input") first_pid = next(iter(first)) renderer = first[first_pid]["window"] if (not renderer.get("rendererAttached") or renderer.get("rendererHidden") or abs(renderer["rendererWidth"] - renderer["videoWidth"]) > 1 or abs(renderer["rendererHeight"] - renderer["videoHeight"]) > 1): raise RuntimeError(f"playing renderer must fill its visible native surface: {renderer}") opened = input_request(first_pid, 1, lambda state: state["showsQueue"] and state["window"].get("videoOwnsMouse") is True and state["window"].get("rendererAttached") is True and state["window"].get("rendererHidden") is False and abs(state["window"]["videoWidth"] - first[first_pid]["window"]["videoWidth"]) < 1, "queue preserves video size and mouse target", action="queue", visible=True) window = opened[first_pid]["window"] if (not window["resizable"] or window["aspectWidth"] != 0 or window["aspectHeight"] != 0 or window["resizeIncrementWidth"] <= 0 or window["resizeIncrementHeight"] <= 0 or window["minWidth"] > 660 or window["minHeight"] > 300 or window["maxWidth"] <= window["width"] + 80 or window["maxHeight"] <= window["height"] + 80 or not window["controlsOwnMouse"] or not window["sidebarOwnsMouse"] or not window["edgesAvoidDragSurface"] or window["surfaceMovesOnMouseDown"]): raise RuntimeError(f"queue must preserve native resize limits and independent mouse targets: {window}") input_request(first_pid, 1, lambda state: not state["showsQueue"] and abs(state["window"]["videoWidth"] - first[first_pid]["window"]["videoWidth"]) < 1, "queue closes without changing video size", action="queue", visible=False) print(f"PASS native queue window: independent video/controls/sidebar/edge targets, " f"resizable min {window['minWidth']}x{window['minHeight']}, unconstrained maxima and cleared aspect") if os.environ.get("TEREBI_CHECK_WINDOW_ONLY") == "1": return if settings(first[first_pid]) != (False, False, True) or config.parent.exists(): raise RuntimeError(f"missing config must use defaults without creating a file: {first}") invoke([]) wait_states(1, lambda states: states[first_pid]["media"] == movie.name and states[first_pid]["active"] and states[first_pid]["keyWindow"] and states[first_pid]["mainWindow"], "warm default CLI focuses the existing process without replacing media") input_request(first_pid, 1, lambda state: state["status"] == 2, "default Space pauses once", events=[{"key": " ", "code": 49}]) input_request(first_pid, 1, lambda state: state["status"] == 2, "held Space cannot oscillate", events=[{"key": " ", "code": 49, "repeat": True}]) input_request(first_pid, 1, lambda state: state["status"] == 3, "same Space resumes", events=[{"key": " ", "code": 49}]) input_request(first_pid, 1, lambda state: state["status"] == 3, "rapid pause/play preserves latest intent", events=[{"key": " ", "code": 49}, {"key": " ", "code": 49}]) input_request(first_pid, 1, lambda state: state["compact"], "P enters the same compact window", events=[{"key": "p", "code": 35}]) input_request(first_pid, 1, lambda state: state["compact"], "held P cannot exit compact mode", events=[{"key": "p", "code": 35, "repeat": True}]) input_request(first_pid, 1, lambda state: not state["compact"], "same P restores normal mode", events=[{"key": "p", "code": 35}]) config.parent.mkdir(parents=True) config_text = ('[playback]\nautomatic_quality = true\n[ui]\n"buffer_info" = true # optional display\n' 'compact_snap_to_edges = false\n[keys]\nplay = ["cmd+8", "cmd+x", "x"]\n' 'pause = ["cmd+8", "cmd+x", "x"]\nshow_buffer_info = "b"\nhide_buffer_info = "b"\n') config.write_text(config_text) invoke(["--new-instance"]) second = wait_states(2, lambda states: any(state["media"] == "" and state["status"] == 0 and settings(state) == (True, True, False) for pid, state in states.items() if pid != first_pid), "empty independent CLI creates exactly one extra player") empty_pid = next(pid for pid in second if pid != first_pid) if settings(second[first_pid]) != (False, False, True) or config.read_text() != config_text: raise RuntimeError(f"config loading changed an existing session or wrote the file: {second}") input_request(empty_pid, 2, lambda state: state["status"] == 3, "configured player loads real video", media=movie) input_request(empty_pid, 2, lambda state: state["status"] == 2, "configured X pauses actual playback", events=[{"key": "x"}]) input_request(empty_pid, 2, lambda state: state["status"] == 2, "old K binding is removed", events=[{"key": "k", "code": 40}]) input_request(empty_pid, 2, lambda state: state["status"] == 2 and state["editorText"] == "x", "typing X reaches the editor without controlling playback", events=[{"key": "x", "editing": True}]) input_request(empty_pid, 2, lambda state: state["status"] == 2, "configured Cmd+8 leaves text editing alone", events=[{"key": "8", "code": 28, "flags": 1 << 20, "editing": True}]) input_request(empty_pid, 2, lambda state: state["status"] == 3, "configured Cmd+X resumes actual playback", events=[{"key": "x", "flags": 1 << 20}]) input_request(empty_pid, 2, lambda state: state["status"] == 3, "held modified playback chord does not toggle", events=[{"key": "x", "flags": 1 << 20, "repeat": True}]) input_request(empty_pid, 2, lambda state: state["status"] == 2, "same Cmd+X pauses actual playback", events=[{"key": "x", "flags": 1 << 20}]) input_request(empty_pid, 2, lambda state: state["status"] == 3, "configured native menu Cmd+8 resumes", events=[{"key": "8", "code": 28, "flags": 1 << 20, "menu": True}]) input_request(empty_pid, 2, lambda state: state["status"] == 2, "native Pause item applies current state", events=[{"select": "Pause"}]) input_request(empty_pid, 2, lambda state: state["status"] == 3, "native Play item resumes the same player", events=[{"select": "Play"}]) input_request(empty_pid, 2, lambda state: not state["showBufferInfo"], "paired B hides buffer info", events=[{"key": "b", "code": 11}]) input_request(empty_pid, 2, lambda state: state["showBufferInfo"], "same B shows buffer info", events=[{"key": "b", "code": 11}]) input_request(empty_pid, 2, lambda state: state["status"] == 0, "configured close-file menu restores empty player", events=[{"key": "w", "code": 13, "flags": (1 << 20) | (1 << 17), "menu": True}]) invoke(["--new-instance", str(audio)]) third = wait_states(3, lambda states: any(state["media"] == audio.name and state["status"] == 3 and state["hasAudio"] and not state["hasVideo"] for pid, state in states.items() if pid not in second), "independent local media reaches only the freshly launched process") if third[first_pid]["media"] != movie.name or third[first_pid]["status"] != 3 or third[empty_pid]["media"]: raise RuntimeError(f"independent launch disturbed an existing player: {third}") config.write_text("[ui]\nbuffer_info = [\n") failure = invoke(["--new-instance"], expected_code=78) if str(config) not in failure.stderr: raise RuntimeError(f"configuration failure omitted its path: {failure.stderr}") invoke(["--no-config"]) unchanged = wait_states(3, lambda _: True, "bypass reuses an existing process without resetting its settings") if any(settings(unchanged[pid]) != settings(third[pid]) for pid in third): raise RuntimeError(f"warm bypass changed an existing session: {third}; {unchanged}") invoke(["--no-config", "--new-instance", str(movie)]) bypass = wait_states(4, lambda states: any(state["media"] == movie.name and state["status"] == 3 and settings(state) == (False, False, True) for pid, state in states.items() if pid not in third), "fresh bypass skips invalid config and plays with built-in settings") invoke(["--no-config", "--new-instance", "--help"]) wait_states(4, lambda _: True, "help skips invalid config without starting another process") config.write_text(config_text) count = 4 if extractor and runtime: invoke(["--new-instance", "--url", canonical]) fourth = wait_states(5, lambda states: any( state["media"] == "Independent YouTube fixture" and state["status"] == 3 and state["position"] > 0 and state["hasAudio"] and state["hasVideo"] and settings(state) == (True, True, False) for pid, state in states.items() if pid not in bypass), "independent YouTube playback reaches only one new registered process") if fourth[first_pid]["media"] != movie.name or fourth[first_pid]["status"] != 3 or fourth[empty_pid]["media"]: raise RuntimeError(f"independent YouTube launch disturbed existing playback: {fourth}") count = 5 else: print("SKIP independent YouTube playback: yt-dlp and a supported JS runtime are required") check_resolved_instances(server, count) existing_report = root / "existing progress.jsonl" sentinel = b"existing destination must remain byte-for-byte unchanged\n" existing_report.write_bytes(sentinel) rejection = invoke(["--new-instance", "--url", reported_urls[0], "--progress-file", str(existing_report)], expected_code=73) if str(existing_report) not in rejection.stderr or existing_report.read_bytes() != sentinel: raise RuntimeError(f"existing report rejection lost its path or altered data: {rejection.stderr}") missing_target = root / "missing progress target.jsonl" dangling_report = root / "dangling progress.jsonl" dangling_report.symlink_to(missing_target) rejection = invoke(["--new-instance", "--url", reported_urls[0], "--progress-file", str(dangling_report)], expected_code=73) if str(dangling_report) not in rejection.stderr or not dangling_report.is_symlink() or missing_target.exists(): raise RuntimeError(f"dangling report rejection altered its destination: {rejection.stderr}") originals = wait_states(count, lambda _: True, "report preflight failures never launch a GUI") if extractor and runtime: report_paths = [root / f"progress session {index} {uuid.uuid4().hex}.jsonl" for index in (1, 2)] reporting_pids = [] allowed_urls = ({reported_urls[0], reported_urls[2]}, {reported_urls[1]}) states = originals for index, path in enumerate(report_paths): previous_pids = set(states) # Deliberately put the space-containing absolute report path after --url. invoke(["--new-instance", "--url", reported_urls[index], "--progress-file", str(path)]) states = wait_states(count + index + 1, lambda snapshots: any( pid not in previous_pids and state["media"] == f"Reported YouTube fixture {index + 1}" and state["status"] == 3 and state["position"] > 0 and state["hasVideo"] and state["canSeek"] and not state["seekPending"] for pid, state in snapshots.items()), "reporting CLI returned before actual registered GUI playback observations") pid = next(pid for pid in states if pid not in previous_pids) reporting_pids.append(pid) wait_report(path, allowed_urls[index], lambda records: any( record["event"] == "progress" and record["media_url"] == reported_urls[index] and record["completion_eligible"] and record["duration_seconds"] > 0 for record in records), "registered GUI emits an initial actual decoded report") total = count + 2 first_reporter, second_reporter = reporting_pids first_path, second_path = report_paths for pid, original in originals.items(): if states[pid]["media"] != original["media"] or settings(states[pid]) != settings(original): raise RuntimeError(f"reporting launches changed an original nonreporting player: {original}; {states[pid]}") before_pause = len(read_reports(first_path, allowed_urls[0])) paused = input_request(first_reporter, total, lambda state: state["status"] == 2, "reporting pause is acknowledged by the actual backend", action="pause") paused_position = paused[first_reporter]["position"] wait_report(first_path, allowed_urls[0], lambda records: any( progress_at(record, reported_urls[0], paused_position) for record in records[before_pause:]), "acknowledged playing-to-paused transition reports native position") input_request(second_reporter, total, lambda state: state["status"] == 2, "second reporting stream pauses independently", action="pause") bookmark = paused[first_reporter]["duration"] * 0.4 input_request(first_reporter, total, lambda state: at_position(state, bookmark), "paused seek reaches an acknowledged decoded position", action="seek", position=bookmark) wait_report(first_path, allowed_urls[0], lambda records: any( progress_at(record, reported_urls[0], bookmark) for record in records), "seek report contains acknowledged native PTS, not just a requested target") before_quality = len(read_reports(first_path, allowed_urls[0])) wait_states(total, lambda snapshots: snapshots[first_reporter]["canChangeQuality"], "manual quality control becomes available after the acknowledged seek") input_request(first_reporter, total, lambda state: state["youtubeHeight"] == 180 and at_position(state, bookmark), "actual quality reopen preserves the paused acknowledged timeline", action="quality", height=180) quality_records = wait_report(first_path, allowed_urls[0], lambda records: any( progress_at(record, reported_urls[0], bookmark) for record in records[before_quality:]), "same-identity quality reopen emits its acknowledged initial record") if any(record["event"] == "closed" for record in quality_records): raise RuntimeError(f"quality reopen incorrectly closed its reporting identity: {quality_records}") input_request(first_reporter, total, lambda state: state["media"] == "Reported YouTube fixture 3" and state["status"] == 3 and not state["busy"] and not state["seekPending"], "new YouTube identity replaces only its reporting player", action="youtube", url=reported_urls[2]) wait_report(first_path, allowed_urls[0], lambda records: ( any(record["event"] == "closed" and record["media_url"] == reported_urls[0] and abs(record["position_seconds"] - bookmark) < 0.5 for record in records) and any(record["event"] == "progress" and record["media_url"] == reported_urls[2] for record in records)), "replacement closes cached old identity and begins only the new one") input_request(first_reporter, total, lambda state: state["status"] == 0, "semantic close invokes actual native media close", action="close") closed = wait_report(first_path, allowed_urls[0], lambda records: any( record["event"] == "closed" and record["media_url"] == reported_urls[2] for record in records), "explicit close reports the replacement identity") before_resume = len(closed) input_request(first_reporter, total, lambda state: state["media"] == "Reported YouTube fixture 1" and state["status"] == 3 and not state["busy"] and not state["seekPending"] and abs(state["position"] - bookmark) < 1, "native history resumes a previously acknowledged position", action="youtube", url=reported_urls[0]) resumed = wait_report(first_path, allowed_urls[0], lambda records: any( record["event"] == "progress" and record["media_url"] == reported_urls[0] for record in records[before_resume:]), "first ready report follows the backend's initial resume acknowledgement") initial_resume = next(record for record in resumed[before_resume:] if record["event"] == "progress" and record["media_url"] == reported_urls[0]) if abs(initial_resume["position_seconds"] - bookmark) >= 1: raise RuntimeError(f"initial resumed report exposed an unacknowledged zero timeline: {initial_resume}") input_request(first_reporter, total, lambda state: state["status"] == 2, "resumed reporting stream can pause", action="pause") crossing = paused[first_reporter]["duration"] * 0.96 input_request(first_reporter, total, lambda state: at_position(state, crossing), "seek acknowledges a watch-fact crossing", action="seek", position=crossing) wait_report(first_path, allowed_urls[0], lambda records: any( record["media_url"] == reported_urls[0] and watch_fact(record) for record in records), "watch fact requires eligible actual decoded ratio, not launcher success") input_request(first_reporter, total, lambda state: state["status"] == 3, "native play continues from the acknowledged crossing", action="play") ended = wait_states(total, lambda snapshots: snapshots[first_reporter]["status"] == 4, "actual playback reaches its decoded end") if ended[first_reporter]["position"] / ended[first_reporter]["duration"] < 0.95: raise RuntimeError(f"native ended timeline unexpectedly reset before reporting: {ended[first_reporter]}") wait_report(first_path, allowed_urls[0], lambda records: any( record["event"] == "ended" and record["media_url"] == reported_urls[0] and watch_fact(record) for record in records), "ended report retains actual final position rather than bookmark-reset zero") input_request(first_reporter, total, lambda state: state["media"] == audio.name and state["status"] == 3 and state["hasAudio"] and not state["hasVideo"], "local audio replacement remains playable without reporting local media", media=audio) local_boundary = wait_report(first_path, allowed_urls[0], lambda records: ( records[-1]["event"] == "closed" and records[-1]["media_url"] == reported_urls[0] and watch_fact(records[-1])) if records else False, "local replacement closes the final YouTube snapshot") input_request(first_reporter, total, lambda state: state["status"] == 2, "unreported local pause uses the same real model", action="pause") input_request(first_reporter, total, lambda state: bool(state["error"]) and not state["busy"], "fixture extraction failure reaches the actual GUI error path", action="youtube", url=reported_urls[3], allow_error=True) failed_boundary = read_reports(first_path, allowed_urls[0]) if any(record["event"] != "closed" or record["media_url"] != reported_urls[0] for record in failed_boundary[len(local_boundary):]): raise RuntimeError(f"local media or failed resolution emitted a progress observation: {failed_boundary}") input_request(first_reporter, total, lambda state: state["media"] == "Reported YouTube fixture 3" and state["status"] == 3 and not state["busy"] and not state["seekPending"] and not state["error"], "successful resolution recovers after an unreported failed open", action="youtube", url=reported_urls[2], allow_error=True) wait_report(first_path, allowed_urls[0], lambda records: any( record["event"] == "progress" and record["media_url"] == reported_urls[2] for record in records[len(failed_boundary):]), "recovery reports only successfully opened identity") second_position = states[second_reporter]["duration"] * 0.2 input_request(second_reporter, total, lambda state: at_position(state, second_position), "second file receives only its own acknowledged seek", action="seek", position=second_position) wait_report(second_path, allowed_urls[1], lambda records: any( progress_at(record, reported_urls[1], second_position) for record in records), "second progress stream never receives first player's media identity") input_request(second_reporter, total, lambda state: state["editorText"] == "q", "plain q remains text input rather than Quit", events=[{"key": "q", "code": 12, "editing": True}]) quit_instance(second_reporter, total, events=[ {"key": "q", "code": 12, "flags": 1 << 20, "editing": True, "queued": True}, ]) final_second = wait_report(second_path, allowed_urls[1], lambda records: ( records[-1]["event"] == "closed" and abs(records[-1]["position_seconds"] - second_position) < 0.5) if records else False, "normal GUI quit drains the second player's final acknowledged snapshot") quit_instance(first_reporter, total - 1) final_first = wait_report(first_path, allowed_urls[0], lambda records: ( records[-1]["event"] == "closed" and records[-1]["media_url"] == reported_urls[2]) if records else False, "normal GUI quit drains the first player's current identity") if not final_first or not final_second or any(not path.read_bytes().endswith(b"\n") for path in report_paths): raise RuntimeError("normal quit left an incomplete JSONL record") remaining = wait_states(count, lambda _: True, "original nonreporting GUIs survive both reporting quits") for pid, original in originals.items(): if remaining[pid]["media"] != original["media"] or settings(remaining[pid]) != settings(original): raise RuntimeError(f"reporting actions changed an unrelated session: {original}; {remaining[pid]}") print("PASS real progress launch: absolute spaced paths after --url, two isolated GUI streams, decoded initial/pause/seek/resume/quality/replacement/end/quit, eligible watch fact, local and failed-open exclusions") else: print("SKIP reporting GUI playback: yt-dlp and a supported JS runtime are required") before = wait_states(count, lambda _: True, "all independent processes remain registered") invoke(["--new-instance", "--help"]) wait_states(count, lambda _: True, "help never creates an extra application") replacement = root / "replacement.mp4" shutil.copy2(fixtures / "av.mp4", replacement) invoke([str(replacement)]) after = wait_states(count, lambda states: sum(state["media"] == replacement.name for state in states.values()) == 1 and all( state["active"] and state["keyWindow"] and state["mainWindow"] for state in states.values() if state["media"] == replacement.name), "default document delivery focuses and reuses exactly one existing process") changed = [pid for pid in before if before[pid]["media"] != after[pid]["media"]] if len(changed) != 1: raise RuntimeError(f"default media open changed multiple players: {before}; {after}") if config.read_text() != config_text: raise RuntimeError("native app startup modified config.toml") print("PASS actual CLI instances: native startup config/defaults/invalid-file error/bypass/read-only settings, cold/default reuse, independent empty/local/YouTube, one window per PID, targeted media, original playback retained, help, warm replacement") app_source, library, fixtures, dependencies_root = map(lambda path: Path(path).resolve(), sys.argv[1:]) with tempfile.TemporaryDirectory(prefix="terebi-startup-") as directory: root = Path(directory) app = root / "Terebi Startup Check.app" shutil.copytree(app_source, app) identifier = "local.personal.terebi.startup-check." + uuid.uuid4().hex info_path = app / "Contents/Info.plist" info = plistlib.loads(info_path.read_bytes()) info["CFBundleIdentifier"] = identifier info["CFBundleName"] = "Terebi Startup Check" info["CFBundleDisplayName"] = "Terebi Startup Check" info_path.write_bytes(plistlib.dumps(info)) shutil.copy2(library, app / "Contents/Frameworks/libterebi-player.dylib") subprocess.run([sys.executable, str(Path(__file__).resolve().parent.parent / "Scripts/bundle.py"), "--app", str(app), "--dependencies", str(dependencies_root)], check=True) pipe = root / "result.fifo" os.mkfifo(pipe) descriptor = os.open(pipe, os.O_RDONLY | os.O_NONBLOCK) launch = None try: if os.environ.get("TEREBI_CHECK_STREAM_QUEUE_ONLY") == "1": check_cli_instances(root, app, fixtures) sys.exit(0) with selectors.DefaultSelector() as selector: selector.register(descriptor, selectors.EVENT_READ) launch = subprocess.Popen( ["open", "-g", "-n", "-W", str(app), "--args", str(pipe), str(fixtures)], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) if not selector.select(timeout=30): raise RuntimeError("packaged app did not report its startup state") result = json.loads(os.read(descriptor, 4096)) stdout, stderr = launch.communicate(timeout=10) if launch.returncode != 0 or not result["passed"] or result["bundleIdentifier"] != identifier: raise RuntimeError(f"packaged startup failed: {result}; {stdout}{stderr}") print("PASS packaged bootstrap: registered app PID/identity, in-process player load, actual playback and Now Playing") check_startup_errors(root, app) check_overlay_hover(root, app, fixtures) if os.environ.get("TEREBI_CHECK_HOVER_ONLY") == "1": sys.exit(0) check_window_drag(root, app, fixtures) if os.environ.get("TEREBI_CHECK_DRAG_ONLY") != "1": check_queue_drag(root, app, fixtures) if os.environ.get("TEREBI_CHECK_QUEUE_ONLY") != "1": check_window_layout(root, app, fixtures) if os.environ.get("TEREBI_CHECK_LAYOUT_ONLY") != "1": check_cli_instances(root, app, fixtures) finally: os.close(descriptor) subprocess.run( ["xcrun", "swift", "-e", 'import AppKit; for app in NSWorkspace.shared.runningApplications { ' 'if app.bundleIdentifier == CommandLine.arguments[1] { app.forceTerminate() } }', identifier], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False, ) if launch is not None and launch.poll() is None: launch.terminate() launch.communicate(timeout=5) subprocess.run( ["/System/Library/Frameworks/CoreServices.framework/Frameworks/LaunchServices.framework/Support/lsregister", "-u", str(app)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False, ) for domain in [identifier, identifier + ".history"]: subprocess.run(["defaults", "delete", domain], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False)