199 lines
11 KiB
Python
199 lines
11 KiB
Python
import json
|
|
import os
|
|
from pathlib import Path
|
|
import plistlib
|
|
import selectors
|
|
import shutil
|
|
import shlex
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import uuid
|
|
|
|
from streaming import StreamingServer
|
|
|
|
|
|
def check_cli_instances(root, app, fixtures):
|
|
cli = root / "terebi"
|
|
shutil.copy2(app_source.parent / "terebi", cli)
|
|
extractor = shutil.which("yt-dlp")
|
|
runtime = shutil.which("deno") or shutil.which("node")
|
|
canonical = "https://www.youtube.com/watch?v=aqz-KE-bpKQ"
|
|
|
|
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 wait_states(count, ready, message):
|
|
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")}
|
|
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 state in states.values()):
|
|
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}")
|
|
def settings(state):
|
|
return (state["showBufferInfo"], state["automaticQuality"], state["snapToEdges"])
|
|
|
|
with StreamingServer(fixtures) as server:
|
|
if extractor and runtime:
|
|
metadata = root / "youtube.json"
|
|
metadata.write_text(json.dumps({
|
|
"id": "aqz-KE-bpKQ", "title": "Independent YouTube fixture", "duration": 120,
|
|
"is_live": False, "webpage_url": canonical, "extractor": "youtube",
|
|
"extractor_key": "Youtube", "formats": [{
|
|
"format_id": "180", "height": 180, "width": 320,
|
|
"url": server.base_url + "/media/cache.mp4", "protocol": "http", "ext": "mp4",
|
|
"vcodec": "mpeg4", "acodec": "aac", "dynamic_range": "SDR",
|
|
"http_headers": {"X-Terebi-Fixture": "combined"}}]}))
|
|
wrapper = root / "yt-dlp"
|
|
wrapper.write_text("#!/bin/bash\nset -eu\nargs=(\"$@\")\nexec " + shlex.quote(extractor)
|
|
+ " --load-info-json " + shlex.quote(str(metadata))
|
|
+ ' "${args[@]:0:${#args[@]}-2}"\n')
|
|
wrapper.chmod(0o755)
|
|
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))
|
|
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"] for state in states.values()),
|
|
"cold default CLI opens one actual player window")
|
|
first_pid = next(iter(first))
|
|
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,
|
|
"warm default CLI reuses the existing process without replacing media")
|
|
|
|
config.parent.mkdir(parents=True)
|
|
config_text = '"show_buffer_info" = true # optional display\n automatic_quality = true\nsnap_to_edges = false\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}")
|
|
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("show_buffer_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")
|
|
|
|
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,
|
|
"default media delivery still 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 = map(lambda path: Path(path).resolve(), sys.argv[1:])
|
|
with tempfile.TemporaryDirectory(prefix="terebi-startup-") as directory:
|
|
root = Path(directory)
|
|
app = root / "terebi.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_path.write_bytes(plistlib.dumps(info))
|
|
shutil.copy2(library, app / "Contents/Frameworks/libterebi-player.dylib")
|
|
pipe = root / "result.fifo"
|
|
os.mkfifo(pipe)
|
|
descriptor = os.open(pipe, os.O_RDONLY | os.O_NONBLOCK)
|
|
launch = None
|
|
try:
|
|
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_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)
|