#!/usr/bin/env python3 """Range-capable loopback fixtures for the real native network playback check.""" from __future__ import annotations import argparse import json import re import subprocess import threading from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import parse_qs, unquote, urlsplit class _Gate: def __init__(self, total: int): self.total = total self.entered = threading.Event() self.release = threading.Event() self.claimed = False self.waiting = 0 self.requests = 0 self.ranges = 0 self.bytes = 0 class StreamingServer: """Context-managed server; safe to reuse for actual resolver/playback smoke checks.""" headers = { "av.mp4": "combined", "streaming.mp4": "combined", "frame-step.mp4": "combined", "video.mp4": "video", "frame-step-video.mp4": "video", "audio.wav": "audio", "frame-step-audio.m4a": "audio", "network-vod.m3u8": "combined", **{f"network-vod-{index}.ts": "combined" for index in range(4)}, } def __init__(self, fixtures: str | Path): self.fixtures = Path(fixtures).resolve() self.lock = threading.Lock() self.gates: dict[str, _Gate] = {} self.stats: dict[str, dict[str, int]] = {} self.stopping = threading.Event() owner = self class Handler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" def log_message(self, *_args): pass # Media URLs/headers never appear in logs. def do_GET(self): try: owner._get(self) except (BrokenPipeError, ConnectionResetError, ConnectionAbortedError): pass # The engine intentionally cancels active media connections. self.http = ThreadingHTTPServer(("127.0.0.1", 0), Handler) self.http.daemon_threads = True self.thread = threading.Thread(target=self.http.serve_forever, daemon=True) self.base_url = f"http://127.0.0.1:{self.http.server_port}" def __enter__(self): self.thread.start() return self def __exit__(self, *_args): self.stopping.set() with self.lock: for gate in self.gates.values(): gate.release.set() self.http.shutdown() self.http.server_close() self.thread.join() @staticmethod def _reply(request, status: int, body: bytes, content_type="application/json"): request.send_response(status) request.send_header("Content-Type", content_type) request.send_header("Content-Length", str(len(body))) request.send_header("Connection", "close") request.end_headers() request.wfile.write(body) request.close_connection = True @staticmethod def _range(value: str | None, size: int): if value is None: return 0, size - 1 match = re.fullmatch(r"bytes=(\d*)-(\d*)", value) if not match or not any(match.groups()): return None first, last = match.groups() if not first: suffix = int(last) return (max(0, size - suffix), size - 1) if suffix else None start = int(first) end = min(int(last), size - 1) if last else size - 1 return (start, end) if 0 <= start <= end < size else None def _get(self, request): parsed = urlsplit(request.path) parts = unquote(parsed.path).strip("/").split("/") if parts == ["control", "stats"]: with self.lock: body = json.dumps(self.stats).encode() self._reply(request, 200, body) return if len(parts) == 3 and parts[0] == "control" and parts[1] in ("status", "release"): with self.lock: gate = self.gates.get(parts[2]) if gate is None: body = {"entered": False, "released": False, "waiting": 0, "requests": 0, "ranges": 0, "bytes": 0, "total": 0} else: if parts[1] == "release": gate.release.set() body = {"entered": gate.entered.is_set(), "released": gate.release.is_set(), "waiting": gate.waiting, "requests": gate.requests, "ranges": gate.ranges, "bytes": gate.bytes, "total": gate.total} self._reply(request, 200, json.dumps(body).encode()) return mode = parts[0] token = None if len(parts) == 3 and mode in ("stall", "stall-open"): token, name = parts[1:] if not re.fullmatch(r"[A-Za-z0-9_-]+", token): self._reply(request, 400, b"Invalid gate token", "text/plain") return elif len(parts) == 2 and mode in ("media", "no-range", "failure"): name = parts[1] else: self._reply(request, 404, b"Unknown fixture endpoint", "text/plain") return if name not in self.headers or not (self.fixtures / name).is_file(): self._reply(request, 404, b"Fixture not found", "text/plain") return with self.lock: stats = self.stats.setdefault(name, {"requests": 0, "ranges": 0, "bytes": 0, "denied": 0}) stats["requests"] += 1 if request.headers.get("X-Terebi-Fixture") != self.headers[name]: stats["denied"] += 1 denied = True else: denied = False if denied: self._reply(request, 403, b"Required fixture header missing or incorrect", "text/plain") return if mode == "failure": self._reply(request, 503, b"Controlled fixture extraction failure", "text/plain") return path = self.fixtures / name size = path.stat().st_size range_header = request.headers.get("Range") if mode != "no-range" else None byte_range = self._range(range_header, size) if byte_range is None: request.send_response(416) request.send_header("Content-Range", f"bytes */{size}") request.send_header("Content-Length", "0") request.send_header("Connection", "close") request.end_headers() request.close_connection = True return start, end = byte_range gate = None gate_at = 0 should_gate = False if token is not None: query = parse_qs(parsed.query) try: gate_at = int(query.get("after", ["131072"])[0]) if mode == "stall" else start except ValueError: self._reply(request, 400, b"Invalid gate offset", "text/plain") return with self.lock: gate = self.gates.setdefault(token, _Gate(size)) gate.requests += 1 gate.ranges += int(range_header is not None) # Only the original prefix connection is gated: a seek can create a new # range request while the old read is still blocked and never released. should_gate = not gate.claimed and start <= gate_at <= end if should_gate: gate.claimed = True with self.lock: stats["ranges"] += int(range_header is not None) request.send_response(206 if range_header is not None else 200) request.send_header("Accept-Ranges", "none" if mode == "no-range" else "bytes") content_types = {".wav": "audio/wav", ".m3u8": "application/vnd.apple.mpegurl", ".ts": "video/mp2t"} request.send_header("Content-Type", content_types.get(path.suffix, "video/mp4")) request.send_header("Content-Length", str(end - start + 1)) if range_header is not None: request.send_header("Content-Range", f"bytes {start}-{end}/{size}") request.send_header("Connection", "close") request.end_headers() request.wfile.flush() request.close_connection = True cursor = start with path.open("rb") as media: media.seek(start) while cursor <= end and not self.stopping.is_set(): if should_gate and cursor == gate_at: with self.lock: gate.waiting += 1 gate.entered.set() try: gate.release.wait() # No timer silently unblocks a cancellation check. finally: with self.lock: gate.waiting -= 1 should_gate = False amount = min(16384, end - cursor + 1) if should_gate: amount = min(amount, gate_at - cursor) chunk = media.read(amount) if not chunk: break request.wfile.write(chunk) request.wfile.flush() cursor += len(chunk) with self.lock: stats["bytes"] += len(chunk) if gate is not None: gate.bytes += len(chunk) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("binary", type=Path) parser.add_argument("fixtures", type=Path) args = parser.parse_args() required = ("av.mp4", "video.mp4", "audio.wav", "frame-step-video.mp4", "frame-step-audio.m4a", "streaming.mp4", "network-vod.m3u8") missing = [name for name in required if not (args.fixtures / name).is_file()] if missing: parser.error("Missing streaming fixtures: " + ", ".join(missing)) with StreamingServer(args.fixtures) as server: try: result = subprocess.run([str(args.binary.resolve()), server.base_url, str(args.fixtures.resolve())], timeout=180, check=False) except subprocess.TimeoutExpired: raise SystemExit("FAIL: native network check exceeded its infrastructure deadline") raise SystemExit(result.returncode) if __name__ == "__main__": main()