86 lines
4.5 KiB
Python
86 lines
4.5 KiB
Python
"""Local-only resolved HLS/VTT + native model smoke; no live requests."""
|
|
import http.server
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
from urllib.parse import urlsplit
|
|
|
|
binary = str(Path(sys.argv[1]).resolve())
|
|
fixtures = Path(sys.argv[2]).resolve()
|
|
seen = []
|
|
errors = []
|
|
|
|
class Handler(http.server.BaseHTTPRequestHandler):
|
|
def log_message(self, *_):
|
|
pass
|
|
|
|
def do_GET(self):
|
|
path = urlsplit(self.path).path
|
|
seen.append(path)
|
|
alternate = path.startswith("/alternate/")
|
|
captions = path.startswith("/captions/")
|
|
if alternate or captions:
|
|
path = "/" + path.split("/", 2)[2]
|
|
expected = "alternate-private" if alternate else "subtitle-private" if captions or path == "/english.jpeg" else "stream-private"
|
|
if self.headers.get("X-Resource") != expected:
|
|
errors.append("wrong resource headers: " + path)
|
|
self.send_error(403)
|
|
return
|
|
if path == "/master.m3u8":
|
|
if urlsplit(self.path).query != "token=fixture":
|
|
errors.append("master token lost")
|
|
data = b'#EXTM3U\n#EXT-X-STREAM-INF:BANDWIDTH=500000\nvariant.m3u8?token=variant\n'
|
|
elif path == "/variant.m3u8":
|
|
if urlsplit(self.path).query != "token=variant":
|
|
errors.append("variant token lost")
|
|
data = (fixtures / "resolved-vod.m3u8").read_bytes()
|
|
elif path == "/english.jpeg":
|
|
data = (b'WEBVTT\n\n00:00.000 --> 00:10.000\nAlternate English cue\n' if alternate else
|
|
b'WEBVTT\n\n00:00.000 --> 00:02.000\nFirst English cue\n\n00:02.000 --> 00:04.000\nSecond English cue\n')
|
|
elif path.startswith("/resolved-vod-") and path.endswith(".ts"):
|
|
data = (fixtures / path[1:]).read_bytes()
|
|
else:
|
|
self.send_error(404)
|
|
return
|
|
self.send_response(200)
|
|
self.send_header("Content-Length", str(len(data)))
|
|
self.send_header("Content-Type", "text/vtt" if path.endswith(".jpeg") else "application/octet-stream")
|
|
self.end_headers()
|
|
self.wfile.write(data)
|
|
|
|
subprocess.run(["ffmpeg", "-hide_banner", "-loglevel", "error", "-y", "-i", str(fixtures / "av.mp4"),
|
|
"-c:v", "libx264", "-g", "24", "-keyint_min", "24", "-sc_threshold", "0", "-c:a", "aac",
|
|
"-hls_time", "1", "-hls_playlist_type", "vod", "-hls_segment_filename", str(fixtures / "resolved-vod-%d.ts"),
|
|
str(fixtures / "resolved-vod.m3u8")], check=True)
|
|
server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
try:
|
|
base = f"http://127.0.0.1:{server.server_port}"
|
|
with tempfile.TemporaryDirectory(prefix="terebi-resolved-") as temp:
|
|
source = Path(temp) / "source.json"
|
|
progress = Path(temp) / "progress.jsonl"
|
|
source.write_text(json.dumps({"version": 1, "url": base + "/master.m3u8?token=fixture",
|
|
"headers": {"X-Resource": "stream-private"}, "identity": {"provider": "miruro", "media_id": "fixture-ID", "episode": 1},
|
|
"title": "Local resolved fixture", "subtitles": [{"url": base + "/english.jpeg", "format": "vtt", "language": "en",
|
|
"label": "English", "default": True, "headers": {"X-Resource": "subtitle-private"}}]}))
|
|
source.chmod(0o600)
|
|
subprocess.run([binary, str(source), str(progress)], check=True, timeout=100)
|
|
records = [json.loads(line) for line in progress.read_text().splitlines()]
|
|
assert records and all(record["version"] == 2 and record["media_identity"] ==
|
|
{"provider": "miruro", "media_id": "fixture-ID", "episode": 1} for record in records), "stable episode progress identity"
|
|
assert any(record["completion_eligible"] for record in records), "acknowledged video progress eligible"
|
|
assert "stream-private" not in progress.read_text() and "token=" not in progress.read_text(), "progress has no transport secrets"
|
|
assert not errors, errors
|
|
assert {"/master.m3u8", "/variant.m3u8", "/english.jpeg"}.issubset(seen), "master, nested playlist and external VTT fetched"
|
|
assert {"/alternate/master.m3u8", "/alternate/english.jpeg"}.issubset(seen), "alternate video host and independent caption provider fetched"
|
|
assert any(path.endswith(".ts") for path in seen), "media segment fetched"
|
|
print("PASS: model HLS/VTT playback, independent resource headers, tokens, stable progress")
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
thread.join()
|