feat: report authoritative playback progress from shared backend
This commit is contained in:
+271
-18
@@ -1,5 +1,6 @@
|
||||
import json
|
||||
import os
|
||||
import math
|
||||
from pathlib import Path
|
||||
import plistlib
|
||||
import selectors
|
||||
@@ -20,6 +21,7 @@ def check_cli_instances(root, app, fixtures):
|
||||
extractor = shutil.which("yt-dlp")
|
||||
runtime = shutil.which("deno") or shutil.which("node")
|
||||
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))
|
||||
@@ -31,48 +33,131 @@ def check_cli_instances(root, app, fixtures):
|
||||
raise RuntimeError(f"CLI {arguments} returned {result.returncode}, expected {expected_code}; {result.stdout}{result.stderr}")
|
||||
return result
|
||||
|
||||
def wait_states(count, ready, message):
|
||||
def process_diagnostics(pid):
|
||||
try:
|
||||
return (root / f"stderr-{pid}.log").read_text(errors="replace")
|
||||
except FileNotFoundError:
|
||||
return "<not published>"
|
||||
|
||||
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 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}")
|
||||
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 input_request(pid, count, ready, message, events=None, media=None):
|
||||
request = {"id": uuid.uuid4().hex, "events": events or []}
|
||||
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 wait_states(count, lambda states: states[pid]["requestID"] == request["id"] and ready(states[pid]),
|
||||
message)
|
||||
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]
|
||||
fields = {"version", "event", "media_url", "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"] == 1
|
||||
and record["event"] in ("progress", "ended", "closed")
|
||||
and record["media_url"] 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):
|
||||
write_request(pid, action="quit")
|
||||
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}"]
|
||||
if 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}")
|
||||
|
||||
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"}}]}))
|
||||
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" {shlex.quote(url)}) info={shlex.quote(str(metadata))} ;;")
|
||||
wrapper = root / "yt-dlp"
|
||||
wrapper.write_text("#!/bin/bash\nset -eu\nargs=(\"$@\")\nexec " + shlex.quote(extractor)
|
||||
+ " --load-info-json " + shlex.quote(str(metadata))
|
||||
wrapper.write_text("#!/bin/bash\nset -eu\nargs=(\"$@\")\n"
|
||||
'case "${args[${#args[@]}-1]}" in\n' + "\n".join(choices)
|
||||
+ '\n *) echo "Fixture extraction deliberately failed" >&2; exit 1 ;;\nesac\nexec '
|
||||
+ shlex.quote(extractor) + ' --load-info-json "$info"'
|
||||
+ ' "${args[@]:0:${#args[@]}-2}"\n')
|
||||
wrapper.chmod(0o755)
|
||||
runtime_settings = {"directory": str(root), "path": str(root) + os.pathsep + os.environ.get("PATH", "")}
|
||||
@@ -186,6 +271,174 @@ def check_cli_instances(root, app, fixtures):
|
||||
else:
|
||||
print("SKIP independent YouTube playback: yt-dlp and a supported JS runtime are required")
|
||||
|
||||
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")
|
||||
quit_instance(second_reporter, total)
|
||||
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")
|
||||
|
||||
Reference in New Issue
Block a user