diff --git a/deploy/bootstrap.sh b/deploy/bootstrap.sh new file mode 100755 index 0000000..4d6a221 --- /dev/null +++ b/deploy/bootstrap.sh @@ -0,0 +1,46 @@ +#!/bin/bash +# Everything that does NOT need root. Run as susan, before deploy.sh. +# +# /opt/ytstream/deploy/bootstrap.sh +set -euo pipefail + +STATE=/var/lib/ytstream +VENV=$STATE/venv +REPO=/opt/ytstream + +say() { printf '\n\033[1m==> %s\033[0m\n' "$1"; } + +if [[ ! -d $STATE ]]; then + echo "$STATE does not exist yet — run 'sudo $REPO/deploy/deploy.sh' first," >&2 + echo "or create it with: sudo install -d -o susan -g automation -m 0770 $STATE" >&2 + exit 1 +fi + +say "Creating the virtualenv" +python3 -m venv "$VENV" +"$VENV/bin/pip" install --quiet --upgrade pip +"$VENV/bin/pip" install --quiet -e "$REPO" + +say "Verifying yt-dlp and the POT plugin" +# The plugin is what makes YouTube reachable at all. A venv without it looks fine +# until the first playback fails with a bare 403. +"$VENV/bin/yt-dlp" --version +"$VENV/bin/python3" - <<'PY' +import importlib.util +missing = [name for name in ("yt_dlp_plugins",) if importlib.util.find_spec(name) is None] +print("POT plugin:", "MISSING" if missing else "present") +raise SystemExit(1 if missing else 0) +PY + +say "Starting the POT provider container if it is not already up" +if ! curl -sf --max-time 5 http://127.0.0.1:4416/ping >/dev/null 2>&1; then + docker run -d --name bgutil-provider --restart unless-stopped \ + -p 4416:4416 brainicism/bgutil-ytdlp-pot-provider || true +else + echo " already responding on 127.0.0.1:4416" +fi + +say "Initialising the database" +"$VENV/bin/ytstream" status || true + +say "Done — now run 'sudo $REPO/deploy/deploy.sh'" diff --git a/deploy/crontab.fragment b/deploy/crontab.fragment new file mode 100644 index 0000000..4448f11 --- /dev/null +++ b/deploy/crontab.fragment @@ -0,0 +1,15 @@ +# ytstream — add these to susan's crontab (`crontab -e`). +# +# HC_API_URL is already set at the top of susan's crontab; these follow the +# existing one-UUID-per-job convention. `sg mediaserver` guarantees new files are +# group-owned by mediaserver even if the invoking shell's primary group differs. +# +# Get fresh UUIDs from hc.jihakuz.xyz before enabling these. + +# Sync subscriptions, poll, materialise, reap. Hourly. Runs at :23 to stay clear +# of youtube-automate's :17 entry during the overlap period. +23 * * * * runitor -uuid REPLACE-WITH-UUID -- sg mediaserver "/usr/local/bin/ytstream run" + +# Keep yt-dlp current — this is the thing that keeps playback working as YouTube +# changes. Mondays 04:50, after youtube-automate's 04:40 slot. +50 4 * * 1 runitor -uuid REPLACE-WITH-UUID -- /opt/ytstream/deploy/update-ytdlp.sh diff --git a/deploy/deploy.sh b/deploy/deploy.sh new file mode 100755 index 0000000..4d1ae35 --- /dev/null +++ b/deploy/deploy.sh @@ -0,0 +1,89 @@ +#!/bin/bash +# Root-requiring installation steps for ytstream. +# +# susan has no passwordless sudo, so everything needing root is collected here for +# the operator to run in one go: +# +# sudo /opt/ytstream/deploy/deploy.sh +# +# Everything that does NOT need root — the venv, the database, the POT provider +# container, subscriptions, the API key — is handled by `bootstrap.sh` and the +# application itself. Run bootstrap.sh (as susan) first. +set -euo pipefail + +REPO=/opt/ytstream +STATE=/var/lib/ytstream +VENV=$STATE/venv +HOSTNAME_=tube.jihakuz.xyz + +if [[ $EUID -ne 0 ]]; then + echo "This script needs root. Run: sudo $0" >&2 + exit 1 +fi + +say() { printf '\n\033[1m==> %s\033[0m\n' "$1"; } + +say "Creating $STATE" +# root-owned directory, group-writable by `automation` so susan's cron job and the +# admin server can both write the database. +install -d -o susan -g automation -m 0770 "$STATE" + +say "Installing the /usr/local/bin shim" +cat > /usr/local/bin/ytstream </dev/null; then + echo " $HOSTNAME_ is already served by an existing vhost." + echo " Repoint its proxy_pass to http://127.0.0.1:8086 by hand, then:" + echo " nginx -t && systemctl reload nginx" + echo " (Deliberately not edited automatically — that block also serves" + echo " other names and owns the TLS certificate.)" +else + echo " No existing vhost found. Install one proxying to 127.0.0.1:8086" + echo " and run: certbot --nginx -d $HOSTNAME_" +fi + +say "Done" +cat <<'EOF' +Remaining steps, all as susan and none needing root: + + ytstream set-password # admin UI login + ytstream set-jellyfin-key # verified against the live server + ytstream set-youtube-key # verified against the live API + ytstream add-source @cflux1030 # the mirrored account + ytstream sync # queues the subscriptions + ytstream pending # review them + ytstream approve --all # or approve a subset by id + ytstream run # first real cycle + ytstream doctor # confirm everything is wired up + +Then add the cron entries from deploy/crontab.fragment to susan's crontab. +EOF diff --git a/deploy/update-ytdlp.sh b/deploy/update-ytdlp.sh new file mode 100755 index 0000000..46a7d19 --- /dev/null +++ b/deploy/update-ytdlp.sh @@ -0,0 +1,23 @@ +#!/bin/bash +# Keep yt-dlp current inside ytstream's venv, then confirm the install still works. +# +# This is the single most important recurring job: YouTube changes its player and +# streaming protocol frequently, and a stale yt-dlp means every playback fails. +set -euo pipefail + +VENV=/var/lib/ytstream/venv + +before=$("$VENV/bin/yt-dlp" --version 2>/dev/null || echo "none") +"$VENV/bin/pip" install --quiet --upgrade "yt-dlp[default]" +after=$("$VENV/bin/yt-dlp" --version) + +echo "yt-dlp: $before -> $after" + +if [[ "$before" != "$after" ]]; then + # A new yt-dlp can change format availability, so restart the proxy to drop + # any cached extraction state and re-run the checks. + systemctl restart ytstream-proxy.service 2>/dev/null \ + || echo "could not restart the proxy (needs root); do it by hand" +fi + +exec "$VENV/bin/ytstream" doctor diff --git a/deploy/ytstream-admin.service b/deploy/ytstream-admin.service new file mode 100644 index 0000000..762e532 --- /dev/null +++ b/deploy/ytstream-admin.service @@ -0,0 +1,34 @@ +[Unit] +Description=ytstream admin server +Documentation=file:///opt/ytstream/plan.md +After=network-online.target +Wants=network-online.target + +[Service] +Type=simple +# See plan.md §9 — the same ownership rules as the proxy, because subscribing a +# channel through the UI writes into the media tree. +User=susan +Group=mediaserver +UMask=0002 + +Environment=PATH=/var/lib/ytstream/venv/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin +Environment=PYTHONUNBUFFERED=1 + +WorkingDirectory=/opt/ytstream +# Port 8086: youtube-automate holds 8085 for as long as the two run side by side. +ExecStart=/var/lib/ytstream/venv/bin/ytstream serve --host 127.0.0.1 --port 8086 + +Restart=always +RestartSec=5 + +NoNewPrivileges=true +PrivateTmp=true +ProtectSystem=full +ProtectControlGroups=true +ProtectKernelTunables=true +RestrictSUIDSGID=true +ReadWritePaths=/var/lib/ytstream /disks/Plex/_ytstream + +[Install] +WantedBy=multi-user.target diff --git a/deploy/ytstream-proxy.service b/deploy/ytstream-proxy.service new file mode 100644 index 0000000..ee145c6 --- /dev/null +++ b/deploy/ytstream-proxy.service @@ -0,0 +1,46 @@ +[Unit] +Description=ytstream just-in-time YouTube streaming proxy +Documentation=file:///opt/ytstream/plan.md +After=network-online.target docker.service +Wants=network-online.target +# The bgutil POT provider runs in Docker on 127.0.0.1:4416 and yt-dlp cannot +# fetch anything without it. +Requires=docker.service + +[Service] +Type=simple +# Group=mediaserver and UMask=0002 are load-bearing: Jellyfin reaches the media +# tree only through the mediaserver group. See plan.md §9. +User=susan +Group=mediaserver +UMask=0002 + +# LOAD-BEARING. The only yt-dlp with the bgutil POT plugin is the one in this +# venv. /usr/local/bin/yt-dlp is a 2023.11.16 binary that cannot talk to YouTube +# at all, and if it wins the PATH race every playback fails with no clear reason. +Environment=PATH=/var/lib/ytstream/venv/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin +Environment=PYTHONUNBUFFERED=1 + +WorkingDirectory=/opt/ytstream +ExecStart=/var/lib/ytstream/venv/bin/python3 /opt/ytstream/proxy/ytstream_proxy.py \ + --host 127.0.0.1 --port 8099 \ + --work /dev/shm/ytstream \ + --cache-gb 8 \ + --max-pipelines 2 \ + --max-retries 2 \ + --max-starts 20 --starts-window 3600 + +Restart=always +RestartSec=5 + +NoNewPrivileges=true +PrivateTmp=true +ProtectSystem=full +ProtectControlGroups=true +ProtectKernelTunables=true +RestrictSUIDSGID=true +# /dev/shm is the cache; nothing else needs to be writable. +ReadWritePaths=/dev/shm + +[Install] +WantedBy=multi-user.target diff --git a/proxy/ytstream_proxy.py b/proxy/ytstream_proxy.py new file mode 100644 index 0000000..c8d4a0d --- /dev/null +++ b/proxy/ytstream_proxy.py @@ -0,0 +1,755 @@ +#!/usr/bin/env python3 +""" +ytstream -- just-in-time YouTube streaming proxy for Jellyfin. + +Serves one endpoint per video: + + GET /watch/ muxed avc1+mp4a MP4, range-capable + GET /healthz JSON status (sessions, cache use, counters) + +A `.strm` file whose contents are `http://127.0.0.1:8099/watch/` plays +that video without any of its bytes ever having been stored on disk beforehand. + +ffmpeg reads only local FIFOs; every piece of YouTube protocol handling stays +inside yt-dlp. See FINDINGS.md for what has and has not been verified. + +Safety rails, because a Jellyfin library scan can ask for every episode at once: + + --no-fetch never start a pipeline. Log the request and return 503. + Use this for a first library scan: it reveals whether + Jellyfin probes .strm targets with zero YouTube traffic. + --max-pipelines cap on CONCURRENT pipelines (default 2). Excess -> 503. + --max-starts cap on TOTAL cold starts per window (default 20/hour). This + is the one that bounds a runaway metadata refresh: concurrency + alone only slows a 1249-episode churn down, it does not stop + it. A person watching podcasts starts a handful of videos an + hour; a refresh storm hits the cap in seconds. + --cache-gb tmpfs budget; least-recently-used complete files are evicted. + +Two serving modes, as in the PoC: + + default wait for the mux to finish, then serve. Correct duration, ranges + and seeking. Costs time-to-first-byte (~60x realtime pull, so a + 46-minute video is ready in about 50s). + --growing serve while writing. Low TTFB, but a probe of a partially written + fragmented MP4 reports only the duration written so far. ffmpeg + ignores both mvhd.duration and an injected mehd box, so this + cannot be fixed in the container. +""" + +import argparse +import collections +import json +import os +import re +import shutil +import subprocess +import sys +import threading +import time +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +POT_ARGS = "youtubepot-bgutilhttp:base_url=http://127.0.0.1:4416" +# player_client: `mweb` is deliberately NOT included. Measured 2026-08-12: +# * mweb formats 403 on every attempt, both directly and via --load-info-json, +# even though they are the only ones carrying a PO token in the URL. +# * mweb is also the ONLY source of the DRC and dubbed-language variants that +# caused the two original picker bugs. Dropping it removes both hazards +# before the picker ever sees them. +# * `default` resolves to android_vr for avc1+mp4a, which works without a PO +# token. web / ios / web_safari / tv are SABR-only and yield no usable +# formats at all. Left as `default` rather than pinned to `android_vr` so a +# yt-dlp update can follow YouTube if android_vr stops working. +CLIENT_ARGS = "youtube:player_client=default" +MAX_HEIGHT = 720 +STALL_TIMEOUT = 45.0 +VIDEO_ID_RE = re.compile(r"^[A-Za-z0-9_-]{11}$") + +_log_lock = threading.Lock() +_access_log = None + + +def log(msg): + print(f"[ytstream] {msg}", file=sys.stderr, flush=True) + + +def access(msg): + """Append to the access log. This is the evidence trail for the scan-probe + question, so it is written unbuffered and never dropped.""" + line = f"{time.strftime('%Y-%m-%dT%H:%M:%S')} {msg}" + with _log_lock: + print(line, file=sys.stderr, flush=True) + if _access_log: + with open(_access_log, "a") as f: + f.write(line + "\n") + + +# -------------------------------------------------------------------------- +# Extraction and format selection +# -------------------------------------------------------------------------- + +def extract_info(url, work): + """One extraction. Both downloads reuse it via --load-info-json, which is + verified not to re-extract.""" + info_path = os.path.join(work, "info.json") + cmd = [ + "yt-dlp", "-J", "--no-warnings", + "--extractor-args", CLIENT_ARGS, + "--extractor-args", POT_ARGS, + url, + ] + t0 = time.monotonic() + out = subprocess.run(cmd, capture_output=True, text=True) + if out.returncode != 0: + raise RuntimeError(f"yt-dlp -J failed: {out.stderr.strip()[:500]}") + info = json.loads(out.stdout) + with open(info_path, "w") as f: + json.dump(info, f) + log(f"extracted {info.get('id')} in {time.monotonic() - t0:.1f}s -- " + f"{info.get('title')!r}") + return info, info_path + + +def pick_formats(info): + """Prefer avc1 video and the original-language mp4a audio so the mux is a + pure copy and Jellyfin can direct-play.""" + fmts = info.get("formats", []) + + def usable(f): + return f.get("url") and f.get("protocol", "").startswith("http") + + def is_drc(f): + # YouTube advertises DRC (dynamic-range-compressed) audio variants that + # carry the same abr as their plain counterparts but 403 on download. + # Verified against NH2MhBFQm9w: 140-drc -> 403, 140 -> fine. + return "-drc" in (f.get("format_id") or "") or "DRC" in (f.get("format_note") or "") + + vids = [ + f for f in fmts + if usable(f) + and f.get("vcodec") not in (None, "none") + and f.get("acodec") in (None, "none") + and (f.get("height") or 0) <= MAX_HEIGHT + ] + auds = [ + f for f in fmts + if usable(f) + and f.get("acodec") not in (None, "none") + and f.get("vcodec") in (None, "none") + ] + + # Ranking notes, all learned the hard way against real videos: + # + # * language_preference outranks everything. Multi-language uploads expose + # 140-0..140-N with IDENTICAL abr, so max() would otherwise tie-break on + # list order and silently pick whichever dub YouTube listed first -- + # German audio on an English podcast (verified: J9O3sxoMs5U). The + # original track carries language_preference 10, dubs carry -1. + # * non-DRC next, because DRC variants 403. Language correctness ranks + # above this deliberately: a 403 is a loud failure, wrong-language audio + # is a silent one that would ship. + # * only then codec and bitrate. + def langpref(f): + return f.get("language_preference") or 0 + + def vkey(f): + return ( + not is_drc(f), + f.get("vcodec", "").startswith("avc1"), + f.get("height") or 0, + f.get("tbr") or 0, + ) + + def akey(f): + return ( + langpref(f), + not is_drc(f), + f.get("acodec", "").startswith("mp4a"), + f.get("abr") or 0, + ) + + if not vids or not auds: + raise RuntimeError( + "no separate video+audio pair; formats present: " + ", ".join( + f"{f.get('format_id')}({f.get('vcodec')}/{f.get('acodec')})" + for f in fmts[:15])) + + v = max(vids, key=vkey) + a = max(auds, key=akey) + log(f" video {v['format_id']} {v.get('vcodec')} {v.get('height')}p | " + f"audio {a['format_id']} {a.get('acodec')} lang={a.get('language')}") + if not v.get("vcodec", "").startswith("avc1"): + log(" WARNING: no h264 at this height -- Jellyfin may transcode") + if len({f.get("language") for f in auds}) > 1 and langpref(a) <= 0: + log(f" WARNING: multi-language upload and the chosen track " + f"({a.get('language')}) is not the original -- expect a dub") + return v["format_id"], a["format_id"] + + +# -------------------------------------------------------------------------- +# Producer: two yt-dlp -> FIFOs -> ffmpeg -c copy -> fragmented MP4 +# -------------------------------------------------------------------------- + +def start_producer(info_path, vfmt, afmt, work): + vfifo = os.path.join(work, "v.fifo") + afifo = os.path.join(work, "a.fifo") + for p in (vfifo, afifo): + if os.path.exists(p): + os.unlink(p) + os.mkfifo(p) + + out_path = os.path.join(work, "out.mp4") + done_path = out_path + ".done" + err_path = out_path + ".err" + procs = [] + + def feed(fmt, fifo, tag): + # Opening a FIFO for write blocks until a reader attaches, so ffmpeg + # must already be starting -- it is, just below. + fh = open(fifo, "wb") + p = subprocess.Popen( + [ + "yt-dlp", "--load-info-json", info_path, + "-f", fmt, "-o", "-", "--quiet", "--no-warnings", + "--extractor-args", CLIENT_ARGS, + "--extractor-args", POT_ARGS, + ], + stdout=fh, stderr=subprocess.PIPE, + ) + procs.append((tag, p)) + fh.close() + + ff = subprocess.Popen( + [ + "ffmpeg", "-y", "-loglevel", "error", + "-i", vfifo, "-i", afifo, + "-c", "copy", + "-movflags", "frag_keyframe+empty_moov+default_base_moof", + "-f", "mp4", out_path, + ], + stderr=subprocess.PIPE, + ) + + threading.Thread(target=feed, args=(vfmt, vfifo, "video"), daemon=True).start() + threading.Thread(target=feed, args=(afmt, afifo, "audio"), daemon=True).start() + + def reap(): + rc = ff.wait() + err = ff.stderr.read().decode(errors="replace").strip() + problems = [] + if rc != 0: + problems.append(f"ffmpeg exited {rc}: {err[:400]}") + for tag, p in procs: + if p.poll() not in (0, None): + problems.append( + f"yt-dlp[{tag}] exited {p.returncode}: " + f"{p.stderr.read().decode(errors='replace')[:300]}") + if problems: + # A single 403 on either stream kills the run and there is no retry + # yet; record it so /healthz and the caller can see why. + with open(err_path, "w") as f: + f.write("\n".join(problems)) + log(f"producer FAILED in {work}: {problems[0]}") + else: + log(f"mux complete: {out_path} {os.path.getsize(out_path)} bytes") + open(done_path, "w").close() + + threading.Thread(target=reap, daemon=True).start() + return out_path, done_path, err_path + + +# -------------------------------------------------------------------------- +# Session manager +# -------------------------------------------------------------------------- + +class Session: + """One video's pipeline and its output file. + + `final` is the terminal signal: set once the session has either produced a + complete file or given up after retries. It is deliberately separate from + the producer's own done-marker, because a failed attempt writes that marker + too and a waiting request must not mistake a retry for a finished file. + """ + + def __init__(self, video_id, work): + self.video_id = video_id + self.work = work + self.out_path = os.path.join(work, "out.mp4") + self.started = time.time() + self.last_used = time.time() + self.final = threading.Event() + self.failed = None + self.attempts = 0 + self.readers = 0 + + @property + def complete(self): + """True only when the file is finished AND good.""" + return self.final.is_set() and not self.failed + + def size(self): + try: + return os.path.getsize(self.out_path) + except OSError: + return 0 + + +class Manager: + def __init__(self, work_root, max_pipelines, cache_bytes, no_fetch, growing, + max_retries=2, max_starts=20, starts_window=3600.0): + self.work_root = work_root + self.max_pipelines = max_pipelines + self.max_retries = max_retries + self.max_starts = max_starts + self.starts_window = starts_window + # Timestamps of cold starts, pruned to the window. Cache hits are not + # recorded: re-watching or resuming must never be rate limited. + self.start_log = collections.deque() + self.cache_bytes = cache_bytes + self.no_fetch = no_fetch + self.growing = growing + self.lock = threading.Lock() + self.sessions = {} + self.counters = {"requests": 0, "started": 0, "reused": 0, + "refused_nofetch": 0, "refused_busy": 0, + "refused_ratelimit": 0, "retried": 0, "failed": 0, + "evicted": 0} + + def _active(self): + return [s for s in self.sessions.values() if not s.complete] + + def get(self, video_id): + """Return (session, error_string). Never raises.""" + with self.lock: + self.counters["requests"] += 1 + s = self.sessions.get(video_id) + if s is not None: + s.last_used = time.time() + self.counters["reused"] += 1 + return s, None + + if self.no_fetch: + self.counters["refused_nofetch"] += 1 + return None, "no-fetch mode: refusing to start a pipeline" + + if len(self._active()) >= self.max_pipelines: + self.counters["refused_busy"] += 1 + return None, (f"at pipeline cap ({self.max_pipelines}); " + f"refusing to start another") + + # Cold-start budget. Refusing produces a short retry burst from + # libavformat and then aborts whatever refresh triggered it, which + # is the intended protective outcome: an aborted refresh costs some + # re-derivable metadata, a runaway one costs the whole catalogue in + # YouTube traffic. + now = time.monotonic() + while self.start_log and now - self.start_log[0] > self.starts_window: + self.start_log.popleft() + if len(self.start_log) >= self.max_starts: + self.counters["refused_ratelimit"] += 1 + oldest = self.starts_window - (now - self.start_log[0]) + log(f"RATE LIMIT: {len(self.start_log)} cold starts in the last " + f"{self.starts_window / 60:.0f}min, refusing {video_id}. " + f"Budget frees in {oldest:.0f}s. If this was a library " + f"refresh, that refresh is being stopped on purpose.") + return None, (f"cold-start budget exhausted " + f"({self.max_starts} per " + f"{self.starts_window / 60:.0f}min)") + self.start_log.append(now) + + work = os.path.join(self.work_root, video_id) + shutil.rmtree(work, ignore_errors=True) + os.makedirs(work, exist_ok=True) + s = Session(video_id, work) + self.sessions[video_id] = s + self.counters["started"] += 1 + + threading.Thread(target=self._run, args=(s,), daemon=True).start() + return s, None + + def _run(self, s): + """Run the pipeline, retrying on failure with a fresh extraction. + + Intermittent 403s do happen -- observed on both the video and the audio + stream of videos that succeeded minutes earlier, clustered after heavy + use, so most likely transient rate limiting. A single one used to kill + playback outright. Each retry re-extracts, because the resolved URLs and + their PO token binding are the most likely thing to have gone stale. + + Only retried while nothing has been served yet (size == 0). If bytes + already went out we cannot rewind under a reader. + """ + url = f"https://www.youtube.com/watch?v={s.video_id}" + last = None + for attempt in range(1, self.max_retries + 2): + s.attempts = attempt + try: + shutil.rmtree(s.work, ignore_errors=True) + os.makedirs(s.work, exist_ok=True) + info, info_path = extract_info(url, s.work) + vfmt, afmt = pick_formats(info) + _out, done_path, err_path = start_producer( + info_path, vfmt, afmt, s.work) + + while not os.path.exists(done_path): + time.sleep(0.2) + if not os.path.exists(err_path): + last = None + break + with open(err_path) as f: + last = f.read().strip() + except Exception as e: # noqa: BLE001 - report anything + last = str(e)[:500] + + if s.size() > 0: + log(f"{s.video_id}: attempt {attempt} failed after serving " + f"bytes -- not retrying") + break + if attempt <= self.max_retries: + with self.lock: + self.counters["retried"] += 1 + log(f"{s.video_id}: attempt {attempt} failed " + f"({(last or '')[:90]}) -- retrying") + time.sleep(2.0 * attempt) + + s.failed = last + if last: + with self.lock: + self.counters["failed"] += 1 + log(f"session {s.video_id} gave up after {s.attempts} attempt(s)") + s.final.set() + self.evict() + + def evict(self): + """Drop least-recently-used complete sessions until under budget.""" + with self.lock: + done = [s for s in self.sessions.values() + if s.complete and s.readers == 0] + total = sum(s.size() for s in self.sessions.values()) + for s in sorted(done, key=lambda x: x.last_used): + if total <= self.cache_bytes: + break + total -= s.size() + self.sessions.pop(s.video_id, None) + shutil.rmtree(s.work, ignore_errors=True) + self.counters["evicted"] += 1 + log(f"evicted {s.video_id}") + + def status(self): + with self.lock: + return { + "mode": "growing" if self.growing else "wait-for-complete", + "no_fetch": self.no_fetch, + "max_pipelines": self.max_pipelines, + "max_starts": self.max_starts, + "starts_in_window": len(self.start_log), + "starts_window_min": round(self.starts_window / 60), + "cache_gb": round(self.cache_bytes / 2**30, 2), + "cache_used_gb": round( + sum(s.size() for s in self.sessions.values()) / 2**30, 3), + "counters": dict(self.counters), + "sessions": [ + {"video_id": s.video_id, + "complete": s.complete, + "bytes": s.size(), + "readers": s.readers, + "age_s": round(time.time() - s.started, 1), + "attempts": s.attempts, + "error": s.failed} + for s in self.sessions.values() + ], + } + + +# -------------------------------------------------------------------------- +# HTTP +# -------------------------------------------------------------------------- + +def make_handler(mgr, wait_timeout): + + class Handler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def log_message(self, *a): + pass # we do our own logging + + # -- range helpers (verified by test_range.py) ---------------------- + + def _size(self, path): + return os.path.getsize(path) if os.path.exists(path) else 0 + + def _wait_for_bytes(self, path, offset, complete): + deadline = time.monotonic() + STALL_TIMEOUT + while True: + size = self._size(path) + if size > offset or complete(): + return size + if time.monotonic() > deadline: + raise TimeoutError + time.sleep(0.05) + + def _parse_range(self, rng, total): + """Parse a single byte range against a known total size. + + Returns (start, end) inclusive, "unsatisfiable", or None when there + is no usable range and the whole body should be sent. Multi-range + requests fall into the None case: answering with the whole body is + legal and beats mis-serving one part. + """ + if not rng or not rng.startswith("bytes=") or "," in rng: + return None + first, sep, last = rng[6:].strip().partition("-") + if not sep: + return None + try: + if not first: # bytes=-N -> final N bytes + n = int(last) + return (max(0, total - n), total - 1) if n > 0 else "unsatisfiable" + start = int(first) + end = int(last) if last else total - 1 + except ValueError: + return None + if start >= total or start > end: + return "unsatisfiable" + return start, min(end, total - 1) + + # -- routing -------------------------------------------------------- + + def do_HEAD(self): + self._route(True) + + def do_GET(self): + self._route(False) + + def _route(self, head_only): + rng = self.headers.get("Range", "-") + ua = (self.headers.get("User-Agent") or "-")[:60] + access(f"{self.command} {self.path} range={rng} ua={ua!r}") + + if self.path == "/healthz": + body = json.dumps(mgr.status(), indent=2).encode() + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + if not head_only: + self.wfile.write(body) + return + + if not self.path.startswith("/watch/"): + self._fail(404, "not found") + return + + video_id = self.path[len("/watch/"):].split("?")[0] + if not VIDEO_ID_RE.match(video_id): + self._fail(400, "bad video id") + return + + sess, refusal = mgr.get(video_id) + if refusal: + access(f" -> 503 {refusal}") + self._fail(503, refusal, retry_after=30) + return + + if not mgr.growing: + # Correct duration and working seeks require a finished file. + if not sess.final.is_set(): + access(f" -> waiting for {video_id} to finish muxing") + if not sess.final.wait(wait_timeout): + self._fail(504, "mux did not finish in time") + return + if sess.failed: + access(f" -> 502 producer failed: {sess.failed[:120]}") + self._fail(502, f"producer failed: {sess.failed[:200]}") + return + else: + # Growing mode still needs the first bytes to exist. Retries + # happen underneath while size is still 0, so wait on both. + deadline = time.monotonic() + STALL_TIMEOUT + while sess.size() == 0 and not sess.final.is_set(): + if time.monotonic() > deadline: + self._fail(504, "producer wrote nothing") + return + time.sleep(0.1) + if sess.failed and sess.size() == 0: + self._fail(502, f"producer failed: {sess.failed[:200]}") + return + + with mgr.lock: + sess.readers += 1 + sess.last_used = time.time() + try: + self._serve(sess, head_only) + finally: + with mgr.lock: + sess.readers -= 1 + sess.last_used = time.time() + + def _fail(self, code, msg, retry_after=None): + body = (msg + "\n").encode() + try: + self.send_response(code) + if retry_after: + self.send_header("Retry-After", str(retry_after)) + self.send_header("Content-Type", "text/plain") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + except (BrokenPipeError, ConnectionResetError): + pass + + # -- body ----------------------------------------------------------- + + def _serve(self, sess, head_only): + path = sess.out_path + complete = lambda: sess.complete # noqa: E731 + + # Ranges are only honoured once the file is complete; while it is + # still growing there is no reliable time-to-byte mapping into a + # fragmented MP4, so we present a non-seekable stream instead. + chunked = not complete() + start, end, is_range = 0, None, False + if not chunked: + spec = self._parse_range(self.headers.get("Range"), + self._size(path)) + if spec == "unsatisfiable": + total = self._size(path) + self.send_response(416) + self.send_header("Content-Range", f"bytes */{total}") + self.send_header("Content-Length", "0") + self.end_headers() + return + if spec: + start, end = spec + is_range = True + + try: + self._wait_for_bytes(path, start, complete) + except TimeoutError: + self._fail(504, "producer stalled") + return + + if chunked: + self.send_response(200) + self.send_header("Transfer-Encoding", "chunked") + else: + total = self._size(path) + if end is None: + end = total - 1 + if is_range: + self.send_response(206) + self.send_header("Content-Range", f"bytes {start}-{end}/{total}") + else: + self.send_response(200) + self.send_header("Content-Length", str(end - start + 1)) + self.send_header("Accept-Ranges", "bytes") + self.send_header("Content-Type", "video/mp4") + self.end_headers() + if head_only: + return + + sent = start + # None while chunked: the growing case has no known end and stops + # when the producer does. A bounded range must send exactly the + # bytes it promised in Content-Length, no more. + remaining = None if end is None else end - start + 1 + try: + with open(path, "rb") as f: + f.seek(start) + while True: + if remaining is not None and remaining <= 0: + break + buf = f.read(65536 if remaining is None + else min(65536, remaining)) + if buf: + if chunked: + self.wfile.write(b"%X\r\n" % len(buf)) + self.wfile.write(buf) + self.wfile.write(b"\r\n") + else: + self.wfile.write(buf) + sent += len(buf) + if remaining is not None: + remaining -= len(buf) + continue + if complete() and sent >= self._size(path): + break + try: + self._wait_for_bytes(path, sent, complete) + except TimeoutError: + break + if chunked: + self.wfile.write(b"0\r\n\r\n") + except (BrokenPipeError, ConnectionResetError): + # Normal: a probe reads the header then hangs up. + pass + + return Handler + + +# -------------------------------------------------------------------------- + +def main(): + ap = argparse.ArgumentParser( + description="just-in-time YouTube streaming proxy for Jellyfin") + ap.add_argument("--host", default="127.0.0.1") + ap.add_argument("--port", type=int, default=8099) + ap.add_argument("--work", default="/dev/shm/ytstream") + ap.add_argument("--cache-gb", type=float, default=8.0) + ap.add_argument("--max-pipelines", type=int, default=2) + ap.add_argument("--wait-timeout", type=float, default=300.0, + help="how long a request may block waiting for the mux") + ap.add_argument("--max-starts", type=int, default=20, + help="max cold starts per window; bounds a runaway library " + "refresh (default 20)") + ap.add_argument("--starts-window", type=float, default=3600.0, + help="rate-limit window in seconds (default 3600)") + ap.add_argument("--max-retries", type=int, default=2, + help="retries after a failed pipeline, each with a fresh " + "extraction (intermittent 403s do happen)") + ap.add_argument("--no-fetch", action="store_true", + help="never start a pipeline; log and 503. Use for a first " + "library scan to detect probing with no YouTube traffic") + ap.add_argument("--growing", action="store_true", + help="serve while still writing (low TTFB, wrong duration)") + ap.add_argument("--access-log", default=None) + args = ap.parse_args() + + if not shutil.which("yt-dlp") or not shutil.which("ffmpeg"): + sys.exit("need yt-dlp and ffmpeg on PATH") + + ver = subprocess.run(["yt-dlp", "--version"], capture_output=True, text=True) + version = ver.stdout.strip() + if version < "2025": + log(f"WARNING: yt-dlp {version} looks far too old, and a POT plugin is " + f"required. Expected the automation venv " + f"(/var/lib/youtube-automate/venv/bin) on PATH.") + else: + log(f"yt-dlp {version}") + + global _access_log + _access_log = args.access_log + + os.makedirs(args.work, exist_ok=True) + mgr = Manager(args.work, args.max_pipelines, + int(args.cache_gb * 2**30), args.no_fetch, args.growing, + args.max_retries, args.max_starts, args.starts_window) + + if args.no_fetch: + log("NO-FETCH MODE: every /watch request will be logged and refused. " + "No YouTube traffic will be generated.") + if args.growing: + log("growing mode: probes will see a partial duration") + + srv = ThreadingHTTPServer((args.host, args.port), + make_handler(mgr, args.wait_timeout)) + log(f"listening on http://{args.host}:{args.port} " + f"(/watch/, /healthz)") + try: + srv.serve_forever() + except KeyboardInterrupt: + log("shutting down") + + +if __name__ == "__main__": + main() diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..367fcdc --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,33 @@ +[project] +name = "ytstream" +version = "1.0.0" +description = "A just-in-time YouTube library for Jellyfin. Stores no video bytes." +requires-python = ">=3.11" + +# Pins that matter, inherited from youtube-automate's specs.md §3: +# - the [default] extra ships yt-dlp-ejs, the JS challenge solver, which is +# mandatory. Only the proxy uses yt-dlp; cataloguing runs on the Data API. +# - curl-cffi must stay below 0.16 or yt-dlp rejects it as unsupported +# - the bgutil plugin must match the POT container tag +dependencies = [ + "yt-dlp[default]", + "bgutil-ytdlp-pot-provider==1.3.1", + "curl-cffi<0.16", +] + +[project.optional-dependencies] +dev = ["pytest>=8"] + +[project.scripts] +ytstream = "ytstream.cli:main" + +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[tool.setuptools] +packages = ["ytstream", "ytstream.web"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +addopts = "-q" diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..2ddc3a6 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,202 @@ +"""Test fixtures. + +Every path the application uses is redirected into a tmpdir. No test touches the +network, a real yt-dlp, a real Jellyfin, or the real media tree — the YouTube API +client is always a stub (see `FakeApi`), because the point of a test suite here is +to pin down behaviour that only shows up on the failure paths: a 403, an empty +response, a video that ages out and must not come back. +""" + +from __future__ import annotations + +import os +import tempfile +from pathlib import Path + +# config resolves its paths at import time, so the environment has to be set +# before anything from the package is imported. +_SANDBOX = Path(tempfile.mkdtemp(prefix="yts-tests-")) +os.environ.setdefault("YTS_STATE_DIR", str(_SANDBOX / "state")) +os.environ.setdefault("YTS_MEDIA_ROOT", str(_SANDBOX / "media")) +os.environ.setdefault("YTS_DB_PATH", str(_SANDBOX / "state" / "ytstream.db")) +os.environ.setdefault("YTS_LOCK_PATH", str(_SANDBOX / "state" / "run.lock")) +os.environ.setdefault("YTS_VENV_BIN", str(_SANDBOX / "venv" / "bin")) + +import pytest # noqa: E402 + +from ytstream import api, config, db, util, videos # noqa: E402 +from ytstream.settings import Settings # noqa: E402 + +FIXTURES = Path(__file__).parent / "fixtures" + +# A real-looking channel id: UC + 22 chars. +CHANNEL_ID = "UCW7jUEpYT_t0Gsf632d6_wQ" + + +@pytest.fixture() +def media_root(tmp_path, monkeypatch): + """Point the media root at a per-test tmpdir.""" + root = tmp_path / "media" + root.mkdir(parents=True) + monkeypatch.setattr(config, "MEDIA_ROOT", root) + return root + + +@pytest.fixture() +def conn(tmp_path): + connection = db.connect(tmp_path / "ytstream.db") + yield connection + connection.close() + + +@pytest.fixture() +def settings(conn): + settings = Settings(conn) + # Tests that reach the API go through FakeApi, but the code refuses to call + # out at all without a key, so give it one that is never used for real. + settings.set("youtube_api_key", "test-key") + return settings + + +@pytest.fixture() +def channel(conn): + """One subscribed channel, returned as a row.""" + return add_channel(conn, CHANNEL_ID, "clabretro", "clabretro") + + +@pytest.fixture() +def no_network(monkeypatch): + """Fail loudly if anything tries to open a socket. + + Belt and braces: a test that accidentally hits the network would pass locally + and fail in a different week for reasons nobody could reproduce. + """ + import urllib.request + + def forbidden(*args, **kwargs): + raise AssertionError("test attempted a network call") + + monkeypatch.setattr(urllib.request, "urlopen", forbidden) + + +# --------------------------------------------------------------------- helpers + + +def add_channel(conn, channel_id: str, title: str, dir_name: str, **kwargs): + fields = {"source": "youtube", "backfilled": 1, "uploads_playlist": "UULF"} + fields.update(kwargs) + columns = ", ".join(fields) + marks = ", ".join("?" * len(fields)) + with conn: + conn.execute( + f"INSERT INTO channel (channel_id, handle, title, description, dir_name, " + f"added_at, {columns}) VALUES (?, ?, ?, ?, ?, ?, {marks})", + (channel_id, f"@{dir_name}", title, "A channel", dir_name, + util.utcnow_iso(), *fields.values()), + ) + return conn.execute( + "SELECT * FROM channel WHERE channel_id = ?", (channel_id,) + ).fetchone() + + +def add_video(conn, channel_pk, video_id, **kwargs): + """Insert a video row with sensible defaults.""" + defaults = { + "title": f"Video {video_id}", + "upload_date": "2026-08-01", + "state": videos.LISTED, + "discovery_source": videos.SOURCE_UULF, + "duration": 900, + } + defaults.update(kwargs) + videos.insert(conn, channel_pk=channel_pk, video_id=video_id, **defaults) + return videos.get(conn, video_id) + + +def feed_bytes(name: str) -> bytes: + return (FIXTURES / name).read_bytes() + + +def make_feed(entries: list[dict], *, playlist_published: str = "2019-01-01T00:00:00+00:00") -> bytes: + """Build an Atom feed shaped like YouTube's. + + Includes the feed-level that is NOT an entry, because scraping + timestamps instead of walking atom:entry picks it up and yields nonsense + upload rates — a mistake made once while measuring. + """ + items = "".join( + f""" + + yt:video:{e['video_id']} + {e['video_id']} + {e.get('title', 'Untitled')} + {e['published']} + """ + for e in entries + ) + return f""" + + Videos + {playlist_published}{items} +""".encode() + + +class FakeApi: + """Stand-in for `api.Api` with scriptable responses. + + Counts calls so tests can assert the batching actually batches — the whole + quota argument in the plan rests on 50 ids per call, and a regression to one + call per video would be silent and expensive. + """ + + def __init__(self, *, subs=None, uploads=None, durations=None, channel=None, + raises=None): + self._subs = subs + self._uploads = uploads or [] + self._durations = durations or {} + self._channel = channel + self._raises = raises + self.calls = 0 + self.duration_calls = 0 + self.subscription_calls = 0 + + def subscriptions(self, channel_id): + self.subscription_calls += 1 + self.calls += 1 + if isinstance(self._raises, Exception): + raise self._raises + return list(self._subs or []) + + def uploads(self, channel_id, *, kind="UULF", since=None, limit=None, + page_token=None): + self.calls += 1 + produced = 0 + for entry, token in self._uploads: + if since is not None and entry["published"] < since: + return + yield entry, token + produced += 1 + if limit is not None and produced >= limit: + return + + def durations(self, video_ids): + self.duration_calls += 1 + self.calls += 1 + return {vid: self._durations[vid] for vid in video_ids + if vid in self._durations} + + def channel(self, channel_id): + self.calls += 1 + return self._channel + + def resolve_handle(self, handle): + self.calls += 1 + return self._channel + + +def patch_api(monkeypatch, module, fake: FakeApi): + """Make `module.api.Api(...)` return `fake` regardless of arguments.""" + monkeypatch.setattr(module.api, "Api", lambda *a, **kw: fake) + return fake diff --git a/tests/fixtures/uulf_feed.xml b/tests/fixtures/uulf_feed.xml new file mode 100644 index 0000000..8598f2b --- /dev/null +++ b/tests/fixtures/uulf_feed.xml @@ -0,0 +1,37 @@ + + + yt:playlist:UULFW7jUEpYT_t0Gsf632d6_wQ + Uploads from clabretro + + yt:video:08Ajr5fP52I + 08Ajr5fP52I + UCW7jUEpYT_t0Gsf632d6_wQ + Learning to Design 3D Prints + 2026-08-07T15:00:11+00:00 + + Tinkercad & a cheap printer. Part 1/3 <of a series>. + + + + yt:video:8k8nAQq0s_s + 8k8nAQq0s_s + UCW7jUEpYT_t0Gsf632d6_wQ + Trying to use a Nortel PBX: part two + 2026-08-02T14:30:00+00:00 + + Telephony experiments. + + + + yt:video:vcYYcQyecNQ + vcYYcQyecNQ + UCW7jUEpYT_t0Gsf632d6_wQ + IBM Director on an xSeries 346 from 2004 + 2026-06-17T12:00:00+00:00 + + Old enterprise management software. + + + diff --git a/tests/test_api.py b/tests/test_api.py new file mode 100644 index 0000000..3763f2f --- /dev/null +++ b/tests/test_api.py @@ -0,0 +1,329 @@ +"""The YouTube Data API client, with no YouTube involved. + +Two things here earn their keep beyond ordinary coverage: + +* **Error classification.** Three distinct operator mistakes all arrive as HTTP + 403 `forbidden`, and the only reliable signal is `error.details[].reason`. That + cost three attempts to diagnose by hand once; it should cost nobody a second one. +* **`totalResults` is not the list length.** It reported 127 against 119 actually + returned, because terminated and private channels still count as subscriptions. +""" + +from __future__ import annotations + +import io +import json +import urllib.error + +import pytest + +from ytstream import api + + +# ------------------------------------------------------------------- durations + + +@pytest.mark.parametrize("text,expected", [ + ("PT1H2M3S", 3723), + ("PT21M", 1260), + ("PT45S", 45), + ("P1DT2H", 93600), + ("P1W", 604800), + ("PT0S", 0), # live and upcoming videos report this + ("", None), + ("garbage", None), + ("1H2M", None), +]) +def test_parse_duration(text, expected): + assert api.parse_duration(text) == expected + + +def test_parse_published_keeps_the_exact_string(): + """The date drives naming; the exact stamp is kept because approximate_date + is wrong by up to two days and episode numbers cannot be re-derived later.""" + day, exact = api.parse_published("2026-08-11T16:32:10Z") + assert day.isoformat() == "2026-08-11" + assert exact == "2026-08-11T16:32:10Z" + + +def test_parse_published_normalises_to_utc(): + day, _ = api.parse_published("2026-08-11T23:30:00-08:00") + assert day.isoformat() == "2026-08-12" + + +def test_parse_published_tolerates_rubbish(): + assert api.parse_published("") == (None, None) + assert api.parse_published("not a date")[0] is None + + +# ------------------------------------------------------------------- playlists + + +def test_uploads_playlist_id(): + assert api.uploads_playlist_id("UCjCJ2LaOIsPzOoXUTMDI3wg") == "UULFjCJ2LaOIsPzOoXUTMDI3wg" + assert api.uploads_playlist_id("UCjCJ2LaOIsPzOoXUTMDI3wg", "UU") == "UUjCJ2LaOIsPzOoXUTMDI3wg" + + +def test_uploads_playlist_id_rejects_non_channel_ids(): + with pytest.raises(ValueError): + api.uploads_playlist_id("PLnotachannel") + + +# -------------------------------------------------------- error classification + + +def _http_error(code, body): + return urllib.error.HTTPError( + "https://example/", code, "Forbidden", {}, + io.BytesIO(json.dumps(body).encode()), + ) + + +def _classify(code, body): + return api.Api("k")._classify(_http_error(code, body)) + + +def test_service_disabled_is_not_configured(): + """The API is not enabled on the project. Carries an activation URL.""" + error = _classify(403, {"error": { + "code": 403, + "message": "YouTube Data API v3 has not been used in project 510818173753 " + "before or it is disabled.", + "errors": [{"reason": "accessNotConfigured"}], + "details": [{"reason": "SERVICE_DISABLED", + "metadata": {"activationUrl": "https://console/enable"}}], + }}) + assert isinstance(error, api.NotConfigured) + + +def test_key_service_blocked_is_not_configured(): + """A different mistake entirely — the key's own API restrictions — and it + reports only `forbidden` in errors[].""" + error = _classify(403, {"error": { + "code": 403, + "message": "Requests to this API youtube method " + "youtube.api.v3.V3DataVideoService.List are blocked.", + "errors": [{"reason": "forbidden"}], + "details": [{"reason": "API_KEY_SERVICE_BLOCKED"}], + }}) + assert isinstance(error, api.NotConfigured) + + +def test_subscription_forbidden_is_its_own_type(): + """Must never be mistaken for an empty subscription list.""" + error = _classify(403, {"error": { + "code": 403, + "message": "The requester is not allowed to access the requested subscriptions.", + "errors": [{"reason": "subscriptionForbidden"}], + }}) + assert isinstance(error, api.SubscriptionsPrivate) + assert not isinstance(error, api.NotConfigured) + + +def test_transient_server_error_is_plain_api_error(): + error = _classify(500, {"error": { + "code": 500, "message": "Backend Error", + "errors": [{"reason": "backendError"}], + }}) + assert type(error) is api.ApiError + + +def test_quota_exceeded_is_plain_api_error(): + """Retrying tomorrow helps, so it must not be classed as misconfiguration.""" + error = _classify(403, {"error": { + "code": 403, "message": "The request cannot be completed because you have " + "exceeded your quota.", + "errors": [{"reason": "quotaExceeded"}], + }}) + assert type(error) is api.ApiError + + +def test_unparseable_body_still_yields_an_error(): + exc = urllib.error.HTTPError( + "https://example/", 502, "Bad Gateway", {}, io.BytesIO(b"nope") + ) + error = api.Api("k")._classify(exc) + assert isinstance(error, api.ApiError) + assert error.status == 502 + + +def test_missing_key_raises_before_any_request(monkeypatch): + import urllib.request + + monkeypatch.setattr(urllib.request, "urlopen", lambda *a, **k: + (_ for _ in ()).throw(AssertionError("should not be called"))) + with pytest.raises(api.NotConfigured): + api.Api("").durations(["dQw4w9WgXcQ"]) + + +# -------------------------------------------------------------- request shapes + + +class Recorder: + """Captures the URLs requested and replays canned pages.""" + + def __init__(self, pages): + self.pages = list(pages) + self.urls = [] + + def __call__(self, request, timeout=None): + self.urls.append(request.full_url) + payload = json.dumps(self.pages.pop(0)).encode() + + class Response(io.BytesIO): + def __enter__(self_inner): + return self_inner + + def __exit__(self_inner, *exc): + return False + + return Response(payload) + + +@pytest.fixture() +def recorder(monkeypatch): + def install(pages): + rec = Recorder(pages) + monkeypatch.setattr(api.urllib.request, "urlopen", rec) + return rec + return install + + +def test_subscriptions_pages_to_exhaustion(recorder): + rec = recorder([ + {"pageInfo": {"totalResults": 127}, + "items": [{"snippet": {"title": f"C{i}", + "resourceId": {"channelId": f"UC{i:022d}"}}} + for i in range(50)], + "nextPageToken": "T2"}, + {"pageInfo": {"totalResults": 127}, + "items": [{"snippet": {"title": "Last", + "resourceId": {"channelId": "UC" + "z" * 22}}}]}, + ]) + + found = api.Api("k").subscriptions("UCbrother") + + assert len(found) == 51 + assert len(rec.urls) == 2 + assert "pageToken=T2" in rec.urls[1] + + +def test_subscriptions_returns_the_fetched_list_not_total_results(recorder): + """Measured: totalResults said 127, pagination returned 119.""" + recorder([{"pageInfo": {"totalResults": 127}, + "items": [{"snippet": {"title": "One", + "resourceId": {"channelId": "UC" + "a" * 22}}}]}]) + + found = api.Api("k").subscriptions("UCbrother") + + assert len(found) == 1 + + +def test_subscriptions_deduplicates(recorder): + recorder([{"items": [ + {"snippet": {"title": "Dup", "resourceId": {"channelId": "UC" + "a" * 22}}}, + {"snippet": {"title": "Dup", "resourceId": {"channelId": "UC" + "a" * 22}}}, + ]}]) + assert len(api.Api("k").subscriptions("UCbrother")) == 1 + + +def test_subscriptions_skips_entries_without_a_channel_id(recorder): + recorder([{"items": [ + {"snippet": {"title": "Broken", "resourceId": {}}}, + {"snippet": {"title": "Fine", "resourceId": {"channelId": "UC" + "a" * 22}}}, + ]}]) + assert [entry["title"] for entry in api.Api("k").subscriptions("UCb")] == ["Fine"] + + +def test_durations_batches_fifty_ids_per_call(recorder): + """The whole quota argument rests on this. One call per video would be a + silent 50x regression.""" + ids = [f"vid{i:08d}" for i in range(120)] + rec = recorder([ + {"items": [{"id": vid, "contentDetails": {"duration": "PT10M"}} + for vid in ids[start:start + 50]]} + for start in (0, 50, 100) + ]) + + found = api.Api("k").durations(ids) + + assert len(found) == 120 + assert len(rec.urls) == 3 + + +def test_durations_flags_livestreams(recorder): + recorder([{"items": [ + {"id": "live0000001", "contentDetails": {"duration": "PT0S"}, + "liveStreamingDetails": {"actualStartTime": "2026-08-01T00:00:00Z"}}, + {"id": "vod00000001", "contentDetails": {"duration": "PT30M"}}, + ]}]) + + found = api.Api("k").durations(["live0000001", "vod00000001"]) + + assert found["live0000001"]["is_live"] is True + assert found["vod00000001"]["is_live"] is False + + +def test_uploads_stops_at_the_since_date(recorder): + recorder([{"items": [ + {"contentDetails": {"videoId": "new00000001", + "videoPublishedAt": "2026-08-10T00:00:00Z"}}, + {"contentDetails": {"videoId": "old00000001", + "videoPublishedAt": "2026-01-01T00:00:00Z"}}, + {"contentDetails": {"videoId": "old00000002", + "videoPublishedAt": "2025-01-01T00:00:00Z"}}, + ], "nextPageToken": "T2"}]) + + from datetime import date + + got = list(api.Api("k").uploads("UC" + "a" * 22, since=date(2026, 8, 1))) + + assert [entry["video_id"] for entry, _ in got] == ["new00000001"] + + +def test_uploads_honours_the_limit(recorder): + recorder([{"items": [ + {"contentDetails": {"videoId": f"vid{i:08d}", + "videoPublishedAt": "2026-08-10T00:00:00Z"}} + for i in range(10) + ]}]) + + got = list(api.Api("k").uploads("UC" + "a" * 22, limit=3)) + + assert len(got) == 3 + + +def test_uploads_skips_entries_with_no_publish_date(recorder): + """A private or deleted video keeps its playlist slot but loses its date. + Skipping is right; stopping would truncate the backfill at that point.""" + recorder([{"items": [ + {"contentDetails": {"videoId": "priv0000001"}}, + {"contentDetails": {"videoId": "good0000001", + "videoPublishedAt": "2026-08-10T00:00:00Z"}}, + ]}]) + + got = list(api.Api("k").uploads("UC" + "a" * 22)) + + assert [entry["video_id"] for entry, _ in got] == ["good0000001"] + + +def test_uploads_yields_the_next_page_token_for_resumability(recorder): + recorder([{"items": [ + {"contentDetails": {"videoId": "vid00000001", + "videoPublishedAt": "2026-08-10T00:00:00Z"}} + ], "nextPageToken": "CARRY_ON"}]) + + # Take only the first entry: the generator would otherwise follow the token + # on to a page the recorder has not been given, which is correct paging + # behaviour and covered elsewhere. + _, token = next(api.Api("k").uploads("UC" + "a" * 22)) + + assert token == "CARRY_ON" + + +def test_call_counter_tracks_quota(recorder): + recorder([{"items": []}, {"items": []}]) + client = api.Api("k") + client.durations(["a"]) + client.durations(["b"]) + assert client.calls == 2 diff --git a/tests/test_auth.py b/tests/test_auth.py new file mode 100644 index 0000000..1b9e27d --- /dev/null +++ b/tests/test_auth.py @@ -0,0 +1,203 @@ +"""Password hashing, session cookies, CSRF tokens and login throttling.""" + +import time + +import pytest + +from ytstream.web import auth + + +class TestPasswords: + def test_round_trip(self): + stored = auth.hash_password("correct horse battery staple") + assert auth.verify_password(stored, "correct horse battery staple") + + def test_wrong_password_rejected(self): + stored = auth.hash_password("secret") + assert not auth.verify_password(stored, "Secret") + assert not auth.verify_password(stored, "") + + def test_salt_makes_hashes_unique(self): + assert auth.hash_password("same") != auth.hash_password("same") + + def test_hash_is_not_the_plaintext(self): + assert "secret" not in auth.hash_password("secret") + + def test_empty_stored_hash_rejects_everything(self): + assert not auth.verify_password("", "anything") + + def test_malformed_stored_hash_does_not_raise(self): + for junk in ("nonsense", "scrypt$bad", "a$b$c$d$e$f", "scrypt$x$y$z$q$r"): + assert auth.verify_password(junk, "anything") is False + + def test_unicode_password(self): + stored = auth.hash_password("pässwörd🎬") + assert auth.verify_password(stored, "pässwörd🎬") + + +class TestSessions: + def test_issue_and_verify(self): + secret = auth.new_secret() + token = auth.issue_session(secret) + assert auth.verify_session(secret, token) + + def test_a_different_secret_rejects(self): + token = auth.issue_session(auth.new_secret()) + assert not auth.verify_session(auth.new_secret(), token) + + def test_tampered_payload_rejected(self): + secret = auth.new_secret() + token = auth.issue_session(secret) + payload, signature = token.split(".", 1) + assert not auth.verify_session(secret, f"{payload}x.{signature}") + + def test_tampered_signature_rejected(self): + secret = auth.new_secret() + payload, _ = auth.issue_session(secret).split(".", 1) + assert not auth.verify_session(secret, f"{payload}.deadbeef") + + def test_garbage_rejected(self): + secret = auth.new_secret() + for junk in ("", "no-dot", "a.b.c", "...."): + assert auth.verify_session(secret, junk) is False + + def test_expires_after_a_year(self): + secret = auth.new_secret() + issued = time.time() - auth.SESSION_MAX_AGE - 10 + token = auth.issue_session(secret, issued_at=issued) + assert not auth.verify_session(secret, token) + + def test_still_valid_just_inside_a_year(self): + secret = auth.new_secret() + issued = time.time() - auth.SESSION_MAX_AGE + 60 + token = auth.issue_session(secret, issued_at=issued) + assert auth.verify_session(secret, token) + + def test_a_token_from_the_future_is_rejected(self): + secret = auth.new_secret() + token = auth.issue_session(secret, issued_at=time.time() + 3600) + assert not auth.verify_session(secret, token) + + +class TestCookie: + def test_carries_the_hardening_flags(self): + header = auth.cookie_header("abc") + for flag in ("HttpOnly", "Secure", "SameSite=Lax", "Path=/"): + assert flag in header + assert f"Max-Age={auth.SESSION_MAX_AGE}" in header + + def test_secure_can_be_omitted_for_local_http_testing(self): + assert "Secure" not in auth.cookie_header("abc", secure=False) + + def test_clear_cookie_expires_immediately(self): + assert "Max-Age=0" in auth.clear_cookie_header() + + +class TestCookieParsing: + """Regression tests for a real failure: http.cookies.SimpleCookie silently + drops everything after a value it dislikes, which made valid sessions + invisible and bounced users back to the login page with no error.""" + + def test_finds_our_cookie_alone(self): + assert auth.cookie_value("yta_session=abc") == "abc" + + def test_finds_it_after_a_neighbour(self): + assert auth.cookie_value("sessionid=xyz; yta_session=abc") == "abc" + + def test_finds_it_before_a_neighbour(self): + assert auth.cookie_value("yta_session=abc; sessionid=xyz") == "abc" + + @pytest.mark.parametrize( + "neighbour", + [ + 'prefs={"a":1}', # JSON value — what actually broke it + "junk=[1,2,3]", + "weird=a b c", + "empty=", + "novalue", + "quoted=\"has spaces\"", + "path=/a/b/c", + "colons=a:b:c", + "comma=a,b", + ], + ) + def test_survives_hostile_neighbours(self, neighbour): + assert auth.cookie_value(f"{neighbour}; yta_session=abc") == "abc" + assert auth.cookie_value(f"yta_session=abc; {neighbour}") == "abc" + + def test_strips_surrounding_quotes(self): + assert auth.cookie_value('yta_session="abc"') == "abc" + + def test_tolerates_whitespace(self): + assert auth.cookie_value(" yta_session = abc ") == "abc" + + def test_absent_cookie_returns_empty(self): + assert auth.cookie_value("sessionid=xyz") == "" + + def test_empty_header_returns_empty(self): + assert auth.cookie_value("") == "" + assert auth.cookie_value(None) == "" + + def test_does_not_match_a_name_that_merely_contains_ours(self): + assert auth.cookie_value("not_yta_session=nope") == "" + + def test_real_token_round_trips_through_the_header(self): + secret = auth.new_secret() + token = auth.issue_session(secret) + header = f'prefs={{"theme":"dark"}}; yta_session={token}; other=1' + assert auth.verify_session(secret, auth.cookie_value(header)) + + +class TestCsrf: + def test_token_verifies(self): + secret, session = auth.new_secret(), auth.issue_session(auth.new_secret()) + token = auth.csrf_token(secret, session) + assert auth.verify_csrf(secret, session, token) + + def test_token_is_bound_to_the_session(self): + secret = auth.new_secret() + one = auth.issue_session(secret, issued_at=1000) + two = auth.issue_session(secret, issued_at=2000) + assert not auth.verify_csrf(secret, two, auth.csrf_token(secret, one)) + + def test_empty_token_rejected(self): + secret, session = auth.new_secret(), "sess" + assert not auth.verify_csrf(secret, session, "") + + def test_wrong_token_rejected(self): + secret, session = auth.new_secret(), "sess" + assert not auth.verify_csrf(secret, session, "deadbeef") + + +class TestThrottle: + def test_allows_up_to_the_limit(self): + throttle = auth.LoginThrottle(max_failures=3, lockout=60) + for _ in range(2): + throttle.record_failure("1.2.3.4", now=1000) + assert not throttle.locked("1.2.3.4", now=1000) + + def test_locks_after_the_limit(self): + throttle = auth.LoginThrottle(max_failures=3, lockout=60) + for _ in range(3): + throttle.record_failure("1.2.3.4", now=1000) + assert throttle.locked("1.2.3.4", now=1000) + + def test_lock_expires(self): + throttle = auth.LoginThrottle(max_failures=3, lockout=60) + for _ in range(3): + throttle.record_failure("1.2.3.4", now=1000) + assert not throttle.locked("1.2.3.4", now=1061) + + def test_success_clears_the_counter(self): + throttle = auth.LoginThrottle(max_failures=3, lockout=60) + for _ in range(2): + throttle.record_failure("1.2.3.4", now=1000) + throttle.record_success("1.2.3.4") + assert not throttle.locked("1.2.3.4", now=1000) + + def test_addresses_are_tracked_separately(self): + throttle = auth.LoginThrottle(max_failures=2, lockout=60) + for _ in range(2): + throttle.record_failure("1.1.1.1", now=1000) + assert throttle.locked("1.1.1.1", now=1000) + assert not throttle.locked("2.2.2.2", now=1000) diff --git a/tests/test_channels.py b/tests/test_channels.py new file mode 100644 index 0000000..b0afb29 --- /dev/null +++ b/tests/test_channels.py @@ -0,0 +1,151 @@ +"""Channel subscribe/unsubscribe, and the API-only sync path.""" + +from __future__ import annotations + +import pytest + +from ytstream import api, channels, config + +from conftest import CHANNEL_ID, FakeApi, add_channel, add_video + + +def test_normalise_url_accepts_every_reference_form(): + assert channels.normalise_url(CHANNEL_ID).endswith("/channel/" + CHANNEL_ID) + assert channels.normalise_url("@clabretro") == "https://www.youtube.com/@clabretro" + assert channels.normalise_url("clabretro") == "https://www.youtube.com/@clabretro" + assert channels.normalise_url("https://youtube.com/x") == "https://youtube.com/x" + assert channels.normalise_url("youtube.com/x") == "https://youtube.com/x" + + +def test_normalise_url_rejects_nonsense(): + with pytest.raises(channels.ResolutionError): + channels.normalise_url("") + with pytest.raises(channels.ResolutionError): + channels.normalise_url("not a handle!!") + + +def test_uulf_playlist_id(): + assert channels.uulf_playlist_id(CHANNEL_ID) == "UULF" + CHANNEL_ID[2:] + + +# ---------------------------------------------------------- subscribe_from_sync + + +@pytest.fixture() +def api_channel(monkeypatch): + """Patch channels' lazily-imported api module.""" + fake = FakeApi(channel={"channel_id": "UC" + "a" * 22, "title": "Alpha Channel", + "description": "About alpha", "handle": "@alpha", + "avatar_url": None}) + monkeypatch.setattr(api, "Api", lambda *a, **kw: fake) + return fake + + +def test_subscribe_from_sync_never_calls_yt_dlp(conn, settings, media_root, + api_channel, monkeypatch): + """119 channels' worth of yt-dlp resolution is the request burst this design + exists to avoid.""" + monkeypatch.setattr(channels.ytdlp, "run_json", + lambda *a, **k: pytest.fail("yt-dlp must not be called")) + + row = channels.subscribe_from_sync(conn, settings, "UC" + "a" * 22, "Alpha") + + assert row["title"] == "Alpha Channel" + assert row["source"] == "youtube" + + +def test_subscribe_from_sync_uses_the_api_description(conn, settings, media_root, + api_channel): + row = channels.subscribe_from_sync(conn, settings, "UC" + "a" * 22, "Alpha") + assert row["description"] == "About alpha" + + +def test_subscribe_from_sync_is_idempotent(conn, settings, media_root, api_channel): + first = channels.subscribe_from_sync(conn, settings, "UC" + "a" * 22, "Alpha") + second = channels.subscribe_from_sync(conn, settings, "UC" + "a" * 22, "Alpha") + assert first["id"] == second["id"] + assert conn.execute("SELECT COUNT(*) FROM channel").fetchone()[0] == 1 + + +def test_subscribe_from_sync_survives_a_metadata_failure(conn, settings, media_root, + monkeypatch): + """A channel we cannot describe is still a channel we can mirror.""" + class Failing(FakeApi): + def channel(self, channel_id): + raise api.ApiError(500, "backendError", "boom") + + monkeypatch.setattr(api, "Api", lambda *a, **kw: Failing()) + + row = channels.subscribe_from_sync(conn, settings, "UC" + "a" * 22, "Fallback Name") + + assert row["title"] == "Fallback Name" + + +def test_subscribe_from_sync_makes_no_directory(conn, settings, media_root, + monkeypatch): + """The tree appears with the first episode, so a channel with nothing inside + the window leaves no empty series behind.""" + class NoArt(FakeApi): + def channel(self, channel_id): + return {"channel_id": channel_id, "title": "Quiet", "description": "", + "handle": None, "avatar_url": None} + + monkeypatch.setattr(api, "Api", lambda *a, **kw: NoArt()) + + row = channels.subscribe_from_sync(conn, settings, "UC" + "a" * 22, "Quiet") + + assert not (media_root / row["dir_name"]).exists() + + +def test_directory_names_are_deduplicated(conn, settings, media_root, monkeypatch): + """Two channels can legitimately share a title; dir_name is UNIQUE.""" + class Same(FakeApi): + def channel(self, channel_id): + return {"channel_id": channel_id, "title": "Same Name", + "description": "", "handle": None, "avatar_url": None} + + monkeypatch.setattr(api, "Api", lambda *a, **kw: Same()) + + first = channels.subscribe_from_sync(conn, settings, "UC" + "a" * 22, "Same Name") + second = channels.subscribe_from_sync(conn, settings, "UC" + "b" * 22, "Same Name") + + assert first["dir_name"] == "Same Name" + assert second["dir_name"] == "Same Name (2)" + + +# ------------------------------------------------------------------ unsubscribe + + +def test_unsubscribe_removes_the_tree_and_the_rows(conn, media_root, channel): + tree = media_root / "clabretro" + (tree / "Season 2026").mkdir(parents=True) + (tree / "tvshow.nfo").write_text("") + add_video(conn, channel["id"], "vid00000001") + + title = channels.unsubscribe(conn, channel["id"]) + + assert title == "clabretro" + assert not tree.exists() + assert conn.execute("SELECT COUNT(*) FROM channel").fetchone()[0] == 0 + assert conn.execute("SELECT COUNT(*) FROM video").fetchone()[0] == 0 + + +def test_unsubscribe_rejects_an_unknown_id(conn): + with pytest.raises(LookupError): + channels.unsubscribe(conn, 999) + + +def test_lookup_helpers(conn, channel): + assert channels.get(conn, channel["id"])["title"] == "clabretro" + assert channels.get_by_channel_id(conn, CHANNEL_ID)["id"] == channel["id"] + assert channels.get(conn, 999) is None + assert channels.get_by_channel_id(conn, "UCnope") is None + + +def test_all_channels_is_sorted_case_insensitively(conn): + add_channel(conn, "UC" + "a" * 22, "zebra", "zebra") + add_channel(conn, "UC" + "b" * 22, "Apple", "Apple") + + titles = [row["title"] for row in channels.all_channels(conn)] + + assert titles == ["Apple", "zebra"] diff --git a/tests/test_discovery.py b/tests/test_discovery.py new file mode 100644 index 0000000..03b0307 --- /dev/null +++ b/tests/test_discovery.py @@ -0,0 +1,391 @@ +"""Discovery: feed parsing, the API backfill, and duration enrichment.""" + +from __future__ import annotations + +from datetime import date, timedelta + +import pytest + +from ytstream import api, discovery, util, videos + +from conftest import FakeApi, add_video, make_feed, patch_api + + +def _entry(video_id, published, title="A video"): + return {"video_id": video_id, "published": published, "title": title} + + +@pytest.fixture() +def offline(monkeypatch): + """No feed fetches and no API calls unless a test asks for them.""" + monkeypatch.setattr(discovery, "fetch_feed", + lambda *a, **k: pytest.fail("unexpected feed fetch")) + patch_api(monkeypatch, discovery, FakeApi()) + + +# --------------------------------------------------------------- feed parsing + + +def test_parse_entries_reads_only_atom_entries(): + """The feed-level is the playlist's creation date, sometimes years + old. Treating it as a video produced a nonsense 0.01/day upload rate once.""" + payload = make_feed( + [_entry("vid00000001", "2026-08-11T10:00:00+00:00")], + playlist_published="2019-03-01T00:00:00+00:00", + ) + + entries = discovery.parse_entries(payload) + + assert len(entries) == 1 + assert entries[0]["published"] == date(2026, 8, 11) + + +def test_parse_entries_keeps_the_exact_timestamp(): + payload = make_feed([_entry("vid00000001", "2026-08-11T16:32:10+00:00")]) + assert discovery.parse_entries(payload)[0]["published_at"].startswith( + "2026-08-11T16:32:10" + ) + + +def test_parse_entries_skips_undated_entries(): + payload = make_feed([_entry("vid00000001", "not-a-date"), + _entry("vid00000002", "2026-08-11T10:00:00+00:00")]) + assert [e["video_id"] for e in discovery.parse_entries(payload)] == ["vid00000002"] + + +def test_parse_entries_rejects_garbage(): + with pytest.raises(discovery.FeedUnavailable): + discovery.parse_entries(b"', "Say hello now"), + ("path\\to\\thing", "path to thing"), + ("what? really * | yes", "what really yes"), + (" .leading and trailing. ", "leading and trailing"), + ("line one\nline two", "line one line two"), + ("tabs\tand\r\nnewlines", "tabs and newlines"), + ], + ) + def test_removes_illegal_characters(self, raw, expected): + assert naming.sanitize_component(raw) == expected + + def test_empty_input_is_empty(self): + assert naming.sanitize_component("") == "" + assert naming.sanitize_component(None) == "" + + def test_title_that_is_only_illegal_characters_collapses_to_empty(self): + assert naming.sanitize_component("///???") == "" + + def test_truncates_to_120_characters(self): + long = "word " * 60 + result = naming.sanitize_component(long) + assert len(result) <= naming.MAX_TITLE_LEN + + def test_truncation_prefers_a_word_boundary(self): + text = "alpha bravo charlie delta echo foxtrot golf hotel india juliet " * 3 + result = naming.sanitize_component(text) + assert not result.endswith(" ") + # Should not cut mid-word when a boundary is available late enough. + assert result == result.rstrip() + assert " " in result + + def test_hard_cuts_when_no_late_word_boundary_exists(self): + text = "a" + "b" * 400 + result = naming.sanitize_component(text) + assert len(result) == naming.MAX_TITLE_LEN + + +class TestEpisodeNumbering: + @pytest.mark.parametrize( + "day, ordinal, expected", + [ + (date(2026, 8, 11), 0, 8110), + (date(2026, 8, 11), 1, 8111), + (date(2026, 8, 12), 0, 8120), + (date(2026, 1, 1), 0, 1010), + (date(2026, 12, 31), 0, 12310), + (date(2026, 12, 31), 9, 12319), + ], + ) + def test_episode_number(self, day, ordinal, expected): + assert naming.episode_number(day, ordinal) == expected + + def test_ordinal_is_clamped_at_nine(self): + assert naming.episode_number(date(2026, 8, 11), 12) == 8119 + + def test_negative_ordinal_clamps_to_zero(self): + assert naming.episode_number(date(2026, 8, 11), -3) == 8110 + + def test_numbers_sort_chronologically_across_the_year(self): + days = [date(2026, 1, 1), date(2026, 6, 15), date(2026, 8, 11), date(2026, 12, 31)] + numbers = [naming.episode_number(day, 0) for day in days] + assert numbers == sorted(numbers) + + def test_episode_range_covers_ten_slots(self): + low, high = naming.episode_range(date(2026, 8, 11)) + assert (low, high) == (8110, 8119) + + def test_season_is_the_upload_year(self): + assert naming.season_for(date(2026, 8, 11)) == 2026 + + +class TestParseUploadDate: + def test_accepts_ytdlp_compact_form(self): + assert naming.parse_upload_date("20260811") == date(2026, 8, 11) + + def test_accepts_iso_form(self): + assert naming.parse_upload_date("2026-08-11") == date(2026, 8, 11) + + def test_accepts_iso_timestamp(self): + assert naming.parse_upload_date("2026-08-11T12:00:00+00:00") == date(2026, 8, 11) + + def test_passes_through_a_date(self): + assert naming.parse_upload_date(date(2026, 8, 11)) == date(2026, 8, 11) + + def test_rejects_nonsense(self): + with pytest.raises(ValueError): + naming.parse_upload_date("not a date") + + +class TestBasename: + def test_includes_video_id_for_uniqueness(self): + stem = naming.basename("clabretro", 2026, 8110, "A Title", "dQw4w9WgXcQ") + assert stem == "clabretro - S2026E8110 - A Title [dQw4w9WgXcQ]" + + def test_episode_is_unpadded_to_match_the_nfo(self): + stem = naming.basename("c", 2026, 8110, "t", "id") + assert "S2026E8110" in stem + assert "E08110" not in stem + + def test_falls_back_to_video_id_when_the_title_sanitises_away(self): + stem = naming.basename("c", 2026, 8110, "///", "dQw4w9WgXcQ") + assert stem.endswith("dQw4w9WgXcQ [dQw4w9WgXcQ]") + + def test_two_videos_same_title_differ_by_id(self): + one = naming.basename("c", 2026, 8110, "Same", "aaaaaaaaaaa") + two = naming.basename("c", 2026, 8111, "Same", "bbbbbbbbbbb") + assert one != two + + +class TestChannelDirName: + def test_sanitises_the_title(self): + assert naming.channel_dir_name("Tom / Jerry", "UCabc") == "Tom Jerry" + + def test_falls_back_to_channel_id_when_title_is_unusable(self): + assert naming.channel_dir_name("???", "UCabc") == "UCabc" diff --git a/tests/test_nfo.py b/tests/test_nfo.py new file mode 100644 index 0000000..47ddccc --- /dev/null +++ b/tests/test_nfo.py @@ -0,0 +1,94 @@ +import xml.etree.ElementTree as ET + +from ytstream import nfo + +HOSTILE = ( + "Ampersands & angle and \"quotes\"\n" + "control chars: \x00\x07\x1b\n" + "emoji 🎬 and em-dash — and links https://example.com/?a=1&b=2" +) + + +class TestEpisodeNfo: + def build(self, **overrides): + kwargs = dict( + title="Video Title", + show_title="Some Channel", + season=2026, + episode=8110, + plot="A plot.", + aired="2026-08-11", + duration_seconds=762, + video_id="dQw4w9WgXcQ", + ) + kwargs.update(overrides) + return nfo.episode_nfo(**kwargs) + + def test_is_well_formed_xml(self): + root = ET.fromstring(self.build()) + assert root.tag == "episodedetails" + + def test_hostile_description_still_parses(self): + root = ET.fromstring(self.build(plot=HOSTILE)) + plot = root.findtext("plot") + assert "&" in plot and "" in plot + assert "🎬" in plot + + def test_control_characters_are_stripped(self): + plot = ET.fromstring(self.build(plot=HOSTILE)).findtext("plot") + for bad in ("\x00", "\x07", "\x1b"): + assert bad not in plot + + def test_newlines_are_preserved(self): + plot = ET.fromstring(self.build(plot="one\ntwo")).findtext("plot") + assert plot == "one\ntwo" + + def test_runtime_is_rounded_minutes(self): + assert ET.fromstring(self.build(duration_seconds=762)).findtext("runtime") == "13" + + def test_short_video_still_gets_at_least_one_minute(self): + assert ET.fromstring(self.build(duration_seconds=20)).findtext("runtime") == "1" + + def test_runtime_omitted_when_duration_unknown(self): + assert ET.fromstring(self.build(duration_seconds=None)).find("runtime") is None + + def test_unique_id_marks_youtube_as_default(self): + unique = ET.fromstring(self.build()).find("uniqueid") + assert unique.get("type") == "youtube" + assert unique.get("default") == "true" + assert unique.text == "dQw4w9WgXcQ" + + def test_season_and_episode_are_present(self): + root = ET.fromstring(self.build()) + assert root.findtext("season") == "2026" + assert root.findtext("episode") == "8110" + + def test_title_keeps_characters_that_the_filename_strips(self): + root = ET.fromstring(self.build(title="Hermitcraft S11#11: Expanding Business")) + assert root.findtext("title") == "Hermitcraft S11#11: Expanding Business" + + def test_empty_plot_does_not_break(self): + assert ET.fromstring(self.build(plot=None)).find("plot") is not None + + +class TestTvshowNfo: + def test_well_formed_and_carries_channel_id(self): + root = ET.fromstring(nfo.tvshow_nfo("clabretro", HOSTILE, "UCabc123")) + assert root.tag == "tvshow" + assert root.findtext("title") == "clabretro" + assert root.findtext("studio") == "YouTube" + assert root.find("uniqueid").text == "UCabc123" + + +class TestWrite: + def test_write_is_atomic_and_leaves_no_temp_file(self, tmp_path): + target = tmp_path / "sub" / "tvshow.nfo" + nfo.write(target, b"") + assert target.read_bytes() == b"" + assert list(tmp_path.rglob("*.tmp")) == [] + + def test_overwrites_existing(self, tmp_path): + target = tmp_path / "tvshow.nfo" + nfo.write(target, b"") + nfo.write(target, b"") + assert target.read_bytes() == b"" diff --git a/tests/test_proxy.py b/tests/test_proxy.py new file mode 100644 index 0000000..4cc04de --- /dev/null +++ b/tests/test_proxy.py @@ -0,0 +1,354 @@ +"""The streaming proxy: HTTP range serving, and the safety rails. + +Ported from the standalone `proxy/test_range.py` and `proxy/test_limits.py` so that +one `pytest` run covers everything. These now drive the real `make_handler(mgr, …)` +rather than the PoC's `make_handler(path, done)`, which means the routing, the +video-id validation and the wait-for-complete logic are exercised too. + +No network, no yt-dlp, no ffmpeg: the pipeline runner is stubbed and the output +file is synthetic. +""" + +from __future__ import annotations + +import http.client +import importlib.util +import json +import os +import threading +import time +from http.server import ThreadingHTTPServer +from pathlib import Path + +import pytest + +PROXY_PATH = Path(__file__).resolve().parent.parent / "proxy" / "ytstream_proxy.py" + + +def _load_proxy(): + spec = importlib.util.spec_from_file_location("ytstream_proxy", PROXY_PATH) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +proxy = _load_proxy() + +# Position-identifiable body, so a wrong range returns visibly wrong bytes rather +# than merely the wrong length. +BODY = bytes(range(256)) * 400 +TOTAL = len(BODY) +VIDEO_ID = "dQw4w9WgXcQ" + + +class StubSession: + """A Session whose output file already exists.""" + + def __init__(self, path: Path, *, finished: bool = True): + self.video_id = VIDEO_ID + self.out_path = str(path) + self.final = threading.Event() + self.failed = None + self.readers = 0 + self.last_used = time.time() + if finished: + self.final.set() + + @property + def complete(self): + return self.final.is_set() and not self.failed + + def size(self): + try: + return os.path.getsize(self.out_path) + except OSError: + return 0 + + +class StubManager: + def __init__(self, session, *, growing=False): + self.session = session + self.growing = growing + self.lock = threading.Lock() + + def get(self, video_id): + return self.session, None + + def status(self): + return {"mode": "stub", "cache_used_gb": 0.0} + + +@pytest.fixture() +def complete_server(tmp_path): + """A server over a finished file, with ranges honoured.""" + path = tmp_path / "out.mp4" + path.write_bytes(BODY) + session = StubSession(path, finished=True) + yield from _serve(StubManager(session)) + + +@pytest.fixture() +def growing_server(tmp_path): + """A server over a file still being written.""" + path = tmp_path / "out.mp4" + path.write_bytes(BODY[:1000]) + session = StubSession(path, finished=False) + manager = StubManager(session, growing=True) + + def finish(): + time.sleep(0.2) + with open(path, "ab") as handle: + handle.write(BODY[1000:]) + session.final.set() + + threading.Thread(target=finish, daemon=True).start() + yield from _serve(manager) + + +def _serve(manager): + server = ThreadingHTTPServer(("127.0.0.1", 0), proxy.make_handler(manager, 30)) + threading.Thread(target=server.serve_forever, daemon=True).start() + try: + yield server.server_address[1] + finally: + server.shutdown() + server.server_close() + + +def request(port, headers=None, method="GET", path=f"/watch/{VIDEO_ID}"): + conn = http.client.HTTPConnection("127.0.0.1", port, timeout=20) + conn.request(method, path, headers=headers or {}) + response = conn.getresponse() + body = response.read() + head = dict(response.getheaders()) + conn.close() + return response.status, head, body + + +# ------------------------------------------------------------- complete file + + +def test_bounded_range(complete_server): + status, head, body = request(complete_server, {"Range": "bytes=1000-1999"}) + assert status == 206 + assert len(body) == 1000 + assert body == BODY[1000:2000] + assert head["Content-Range"] == f"bytes 1000-1999/{TOTAL}" + assert head["Content-Length"] == "1000" + + +def test_open_ended_range(complete_server): + status, head, body = request(complete_server, {"Range": "bytes=102000-"}) + assert status == 206 + assert body == BODY[102000:] + assert head["Content-Range"] == f"bytes 102000-{TOTAL - 1}/{TOTAL}" + + +def test_suffix_range(complete_server): + """bytes=-N means the final N bytes, not the first N.""" + status, head, body = request(complete_server, {"Range": "bytes=-100"}) + assert status == 206 + assert len(body) == 100 + assert body == BODY[-100:] + assert head["Content-Range"] == f"bytes {TOTAL - 100}-{TOTAL - 1}/{TOTAL}" + + +def test_single_byte_range(complete_server): + status, _, body = request(complete_server, {"Range": "bytes=5-5"}) + assert status == 206 + assert body == BODY[5:6] + + +def test_end_beyond_eof_is_clamped(complete_server): + status, head, body = request(complete_server, + {"Range": f"bytes=102000-{TOTAL + 5000}"}) + assert status == 206 + assert body == BODY[102000:] + assert head["Content-Range"] == f"bytes 102000-{TOTAL - 1}/{TOTAL}" + + +def test_range_past_eof_is_416(complete_server): + """And it must carry Content-Range: bytes */total, or clients retry forever.""" + status, head, _ = request(complete_server, {"Range": f"bytes={TOTAL + 10}-"}) + assert status == 416 + assert head["Content-Range"] == f"bytes */{TOTAL}" + + +def test_no_range_serves_the_whole_body(complete_server): + status, head, body = request(complete_server) + assert status == 200 + assert body == BODY + assert head["Accept-Ranges"] == "bytes" + + +def test_multi_range_falls_back_to_the_whole_body(complete_server): + """Legal, and beats mis-serving one part of a multipart response.""" + status, _, body = request(complete_server, {"Range": "bytes=0-99,200-299"}) + assert status == 200 + assert body == BODY + + +@pytest.mark.parametrize("header", ["bytes=", "bytes=abc-def", "items=0-99", "0-99"]) +def test_unparseable_range_serves_the_whole_body(complete_server, header): + status, _, body = request(complete_server, {"Range": header}) + assert status == 200 + assert len(body) == TOTAL + + +def test_head_returns_headers_and_no_body(complete_server): + status, head, body = request(complete_server, method="HEAD") + assert status == 200 + assert body == b"" + assert head["Content-Length"] == str(TOTAL) + + +# ------------------------------------------------------------- growing file + + +def test_growing_file_is_chunked_and_tracks_to_eof(growing_server): + """Ranges cannot be honoured mid-write: there is no reliable time-to-byte + mapping into a fragmented MP4 yet, so it becomes a non-seekable stream.""" + status, head, body = request(growing_server, {"Range": "bytes=500-999"}) + assert status == 200 + assert head.get("Transfer-Encoding") == "chunked" + assert body == BODY + + +# ------------------------------------------------------------------- routing + + +def test_unknown_path_is_404(complete_server): + status, _, _ = request(complete_server, path="/nope") + assert status == 404 + + +def test_malformed_video_id_is_400(complete_server): + status, _, _ = request(complete_server, path="/watch/short") + assert status == 400 + + +@pytest.mark.parametrize("bad", ["../../etc/passwd", "abcdefghij", "abcdefghijkl"]) +def test_video_id_must_be_exactly_eleven_safe_chars(complete_server, bad): + status, _, _ = request(complete_server, path=f"/watch/{bad}") + assert status in (400, 404) + + +def test_healthz_returns_json(complete_server): + status, head, body = request(complete_server, path="/healthz") + assert status == 200 + assert head["Content-Type"] == "application/json" + assert json.loads(body)["mode"] == "stub" + + +# ------------------------------------------------- safety rails on the manager + + +def vid(n): + return f"vid{n:08d}" + + +@pytest.fixture() +def manager_factory(tmp_path): + made = [] + + def build(**kwargs): + options = dict(max_pipelines=99, cache_bytes=10 ** 12, no_fetch=False, + growing=False, max_retries=0, max_starts=3, + starts_window=3600.0) + options.update(kwargs) + work = tmp_path / f"work{len(made)}" + work.mkdir() + + class Recording(proxy.Manager): + """Records starts instead of spawning a pipeline.""" + + def __init__(self, *args, **kw): + super().__init__(*args, **kw) + self.ran = [] + + def _run(self, session): + self.ran.append(session.video_id) + session.final.set() # completes instantly, never stays active + + manager = Recording(str(work), **options) + made.append(manager) + return manager + + return build + + +def test_cold_start_budget_is_enforced(manager_factory): + manager = manager_factory(max_starts=3) + results = [manager.get(vid(i)) for i in range(5)] + + started = [r for r in results if r[0] is not None] + refused = [r for r in results if r[0] is None] + + assert len(started) == 3 + assert len(refused) == 2 + assert all("cold-start budget" in r[1] for r in refused) + assert manager.counters["refused_ratelimit"] == 2 + assert len(manager.ran) == 3 + + +def test_cache_hits_are_never_rate_limited(manager_factory): + """Re-watching an already-fetched video must keep working with the budget + spent, or a metadata refresh would break normal playback for an hour.""" + manager = manager_factory(max_starts=3) + for i in range(5): + manager.get(vid(i)) + + session, refusal = manager.get(vid(0)) + + assert session is not None and refusal is None + assert manager.counters["reused"] >= 1 + assert len(manager.ran) == 3 + + +def test_start_budget_window_prunes(manager_factory): + manager = manager_factory(max_starts=2) + manager.get(vid(10)) + manager.get(vid(11)) + assert manager.get(vid(12))[0] is None + + # Age the recorded starts past the window. + manager.start_log = type(manager.start_log)(t - 3601 for t in manager.start_log) + + assert manager.get(vid(13))[0] is not None + assert len(manager.start_log) == 1 + + +def test_concurrency_cap(tmp_path): + class Blocking(proxy.Manager): + """Pipelines that never finish, so sessions stay active.""" + + def _run(self, session): + pass + + work = tmp_path / "blocking" + work.mkdir() + manager = Blocking(str(work), max_pipelines=2, cache_bytes=10 ** 12, + no_fetch=False, growing=False, max_retries=0, + max_starts=99, starts_window=3600.0) + + results = [manager.get(vid(20 + i)) for i in range(4)] + ok = [r for r in results if r[0] is not None] + refused = [r for r in results if r[0] is None] + + assert len(ok) == 2 + assert len(refused) == 2 + assert all("pipeline cap" in r[1] for r in refused) + # A busy refusal must not consume start budget, or a burst of concurrent + # requests would exhaust the hourly allowance without fetching anything. + assert manager.counters["refused_busy"] == 2 + assert len(manager.start_log) == 2 + + +def test_no_fetch_mode_refuses_everything(manager_factory): + manager = manager_factory(no_fetch=True, max_starts=99) + session, refusal = manager.get(vid(30)) + + assert session is None + assert "no-fetch" in refusal + assert manager.ran == [] + assert len(manager.start_log) == 0 diff --git a/tests/test_reap.py b/tests/test_reap.py new file mode 100644 index 0000000..8e258d7 --- /dev/null +++ b/tests/test_reap.py @@ -0,0 +1,201 @@ +"""Retention: the window, the min-keep floor, and the tombstones. + +These are the tests that matter most in this suite. Retention is the only part of +ytstream that deletes things, and two of its rules exist because of measured +behaviour rather than taste: + +* `min_keep_videos` exists because 52 of 117 real channels upload nothing in 30 + days and would otherwise be empty, flickering Jellyfin series. +* the `aged_out` tombstone exists because without it the poller re-materialises + everything the sweep just deleted, forever. +""" + +from __future__ import annotations + +from datetime import timedelta + +import pytest + +from ytstream import discovery, reap, strm, util, videos + +from conftest import add_channel, add_video + + +def _old(days: int) -> str: + return (util.today() - timedelta(days=days)).isoformat() + + +@pytest.fixture() +def materialised(conn, settings, media_root, channel): + """Put videos on disk with a spread of ages. Returns their ids, newest first.""" + counter = {"n": 0} + + def build(ages: list[int], chan=None): + chan = chan or channel + ids = [] + for age in ages: + video_id = f"vid{counter['n']:08d}" + counter["n"] += 1 + add_video(conn, chan["id"], video_id, upload_date=_old(age)) + strm.materialise(conn, settings, chan, videos.get(conn, video_id)) + ids.append(video_id) + return ids + + return build + + +def test_video_inside_the_window_is_kept(conn, settings, materialised): + materialised([1]) + assert reap.candidates(conn, settings) == [] + + +def test_video_outside_the_window_is_deleted(conn, settings, materialised, media_root): + settings.set("min_keep_videos", "0") + [video_id] = materialised([99]) + + path = media_root / videos.get(conn, video_id)["rel_path"] + assert path.exists() + + result = reap.run(conn, settings) + + assert result["aged_out"] == 1 + assert not path.exists() + assert videos.get(conn, video_id)["state"] == videos.AGED_OUT + + +def test_deleting_removes_the_sidecars_too(conn, settings, materialised, media_root): + settings.set("min_keep_videos", "0") + [video_id] = materialised([99]) + strm_path = media_root / videos.get(conn, video_id)["rel_path"] + stem = strm_path.name[: -len(".strm")] + nfo_path = strm_path.with_name(stem + ".nfo") + assert nfo_path.exists() + + reap.run(conn, settings) + + assert not nfo_path.exists() + assert not strm_path.exists() + + +def test_min_keep_videos_protects_the_newest_regardless_of_age( + conn, settings, materialised +): + """The measured common case: a channel whose only videos are all ancient.""" + settings.set("min_keep_videos", "5") + materialised([400, 500, 600, 700, 800]) + + assert reap.candidates(conn, settings) == [] + assert reap.run(conn, settings)["aged_out"] == 0 + + +def test_min_keep_videos_releases_once_enough_newer_ones_exist( + conn, settings, materialised +): + """Six old videos with a floor of five: exactly the oldest one goes.""" + settings.set("min_keep_videos", "5") + ids = materialised([100, 200, 300, 400, 500, 600]) + oldest = ids[-1] + + assert reap.run(conn, settings)["aged_out"] == 1 + + assert videos.get(conn, oldest)["state"] == videos.AGED_OUT + for kept in ids[:-1]: + assert videos.get(conn, kept)["state"] == videos.MATERIALISED + + +def test_min_keep_floor_counts_per_channel_not_globally( + conn, settings, media_root, channel, materialised +): + other = add_channel(conn, "UCzzzzzzzzzzzzzzzzzzzzzz", "Other", "Other") + settings.set("min_keep_videos", "2") + materialised([300, 400], chan=channel) + materialised([300, 400], chan=other) + + # Two channels, two videos each, floor of two: nothing is eligible. If the + # floor were global, two of the four would be deleted. + assert reap.candidates(conn, settings) == [] + + +def test_channel_retention_override_beats_the_global_setting( + conn, settings, media_root, channel, materialised +): + settings.set("min_keep_videos", "0") + settings.set("retention_days", "365") + materialised([90]) + assert reap.candidates(conn, settings) == [] + + with conn: + conn.execute("UPDATE channel SET retention_days = 30 WHERE id = ?", + (channel["id"],)) + assert len(reap.candidates(conn, settings)) == 1 + + +def test_aged_out_row_is_a_tombstone_the_poller_will_not_revive( + conn, settings, media_root, channel, materialised +): + """The failure this prevents: sweep deletes, poll re-adds, forever.""" + settings.set("min_keep_videos", "0") + [video_id] = materialised([99]) + reap.run(conn, settings) + assert videos.get(conn, video_id)["state"] == videos.AGED_OUT + + # The video is still in the feed — YouTube has no idea we deleted it. + entry = {"video_id": video_id, "title": "Video", "published": util.today(), + "published_at": None} + outcome = discovery._record( + conn, channel, entry, videos.SOURCE_UULF, util.today() - timedelta(days=30) + ) + + assert outcome == "known" + assert videos.get(conn, video_id)["state"] == videos.AGED_OUT + + +def test_rescan_revives_skipped_old_but_never_aged_out( + conn, settings, media_root, channel, materialised +): + settings.set("min_keep_videos", "0") + [gone] = materialised([99]) + reap.run(conn, settings) + add_video(conn, channel["id"], "skippedold1", upload_date=_old(40), + state=videos.SKIPPED_OLD) + + settings.set("retention_days", "365") + revived = discovery.rescan_channel(conn, settings, channel) + + assert revived == 1 + assert videos.get(conn, "skippedold1")["state"] == videos.LISTED + assert videos.get(conn, gone)["state"] == videos.AGED_OUT + + +def test_empty_season_directory_is_pruned(conn, settings, materialised, media_root): + settings.set("min_keep_videos", "0") + [video_id] = materialised([99]) + season_dir = (media_root / videos.get(conn, video_id)["rel_path"]).parent + assert season_dir.is_dir() + + reap.run(conn, settings) + + assert not season_dir.exists() + # The channel directory survives: it still holds tvshow.nfo and artwork, and + # an active subscription should not vanish from Jellyfin between uploads. + assert season_dir.parent.is_dir() + + +def test_prune_never_climbs_past_the_media_root(media_root): + nested = media_root / "Chan" / "Season 2026" + nested.mkdir(parents=True) + removed = util.prune_empty_dirs(nested, media_root) + assert removed == 2 + assert media_root.is_dir() + + +def test_row_without_rel_path_still_gets_a_tombstone(conn, settings, channel): + """A video whose files vanished underneath us must not be retried forever.""" + settings.set("min_keep_videos", "0") + add_video(conn, channel["id"], "orphan0001", upload_date=_old(99), + state=videos.MATERIALISED) + row = videos.get(conn, "orphan0001") + assert row["rel_path"] is None + + assert reap.delete_video(conn, row) is False + assert videos.get(conn, "orphan0001")["state"] == videos.AGED_OUT diff --git a/tests/test_runner.py b/tests/test_runner.py new file mode 100644 index 0000000..5a7336e --- /dev/null +++ b/tests/test_runner.py @@ -0,0 +1,232 @@ +"""Orchestration: ordering, the lock, and what turns the cron check red.""" + +from __future__ import annotations + +import pytest + +from ytstream import config, discovery, jellyfin, reap, runner, subsync, videos + +from conftest import add_video + + +@pytest.fixture() +def quiet_jellyfin(monkeypatch): + """Jellyfin refresh is best effort; record calls instead of making them.""" + calls = [] + monkeypatch.setattr(jellyfin.Jellyfin, "refresh", lambda self: calls.append(1)) + return calls + + +@pytest.fixture() +def no_thumbs(monkeypatch): + from ytstream import strm + + monkeypatch.setattr(strm, "fetch_thumbnail", lambda *a, **k: False) + + +# ----------------------------------------------------------------------- lock + + +def test_lock_is_exclusive(tmp_path): + path = tmp_path / "run.lock" + with runner.exclusive_lock(path): + with pytest.raises(runner.AlreadyRunning): + with runner.exclusive_lock(path): + pass # pragma: no cover + + +def test_lock_is_released_after_use(tmp_path): + path = tmp_path / "run.lock" + with runner.exclusive_lock(path): + pass + with runner.exclusive_lock(path): + pass + + +def test_lock_is_released_even_when_the_body_raises(tmp_path): + path = tmp_path / "run.lock" + with pytest.raises(ValueError): + with runner.exclusive_lock(path): + raise ValueError("boom") + with runner.exclusive_lock(path): + pass + + +# --------------------------------------------------------------- materialising + + +def test_materialise_all_writes_everything_listed(conn, settings, media_root, + channel, no_thumbs): + for index in range(3): + add_video(conn, channel["id"], f"vid{index:08d}") + + stats = runner.materialise_all(conn, settings) + + assert stats["materialised"] == 3 + assert stats["errors"] == 0 + assert videos.queue_depth(conn) == 0 + + +def test_tvshow_nfo_is_written_once_per_channel(conn, settings, media_root, + channel, no_thumbs): + for index in range(3): + add_video(conn, channel["id"], f"vid{index:08d}") + + stats = runner.materialise_all(conn, settings) + + assert stats["shows"] == 1 + assert (media_root / "clabretro" / "tvshow.nfo").exists() + + +def test_channel_with_nothing_in_the_window_creates_no_directory( + conn, settings, media_root, channel, no_thumbs +): + """52 of 117 measured channels are in this state, and an empty series in + Jellyfin looks like a bug rather than a quiet channel.""" + add_video(conn, channel["id"], "vid00000001", state=videos.SKIPPED_OLD) + + runner.materialise_all(conn, settings) + + assert not (media_root / "clabretro").exists() + + +def test_materialise_all_honours_the_limit(conn, settings, media_root, channel, + no_thumbs): + for index in range(5): + add_video(conn, channel["id"], f"vid{index:08d}") + stats = runner.materialise_all(conn, settings, limit=2) + assert stats["materialised"] == 2 + + +def test_materialise_skips_a_video_whose_channel_vanished(conn, settings, + media_root, channel, + no_thumbs): + add_video(conn, channel["id"], "vid00000001") + # Simulate the channel being unsubscribed between claiming and writing, with + # foreign keys off so the row survives to be found. + conn.execute("PRAGMA foreign_keys = OFF") + with conn: + conn.execute("DELETE FROM channel WHERE id = ?", (channel["id"],)) + + stats = runner.materialise_all(conn, settings) + + assert stats["materialised"] == 0 + assert stats["errors"] == 0 + + +# ------------------------------------------------------------------ full cycle + + +def test_run_order_is_sync_poll_materialise_reap(conn, settings, media_root, + channel, monkeypatch, + quiet_jellyfin): + order = [] + monkeypatch.setattr(subsync, "sync_all", + lambda *a: order.append("sync") or {"added": 0, "queued": 0, + "removed": 0, "refused": 0, + "sources": 0, + "pending_removal": 0}) + monkeypatch.setattr(discovery, "poll_all", + lambda *a: order.append("poll") or {"queued": 0}) + monkeypatch.setattr(runner, "materialise_all", + lambda *a, **k: order.append("materialise") or + {"materialised": 0, "shows": 0, "errors": 0}) + monkeypatch.setattr(reap, "run", + lambda *a: order.append("reap") or {"aged_out": 0}) + + runner.run(conn, settings) + + assert order == ["sync", "poll", "materialise", "reap"] + + +def test_single_channel_run_skips_the_sync(conn, settings, channel, monkeypatch, + quiet_jellyfin): + """A targeted run is an operator action, not a mirror pass.""" + monkeypatch.setattr(subsync, "sync_all", + lambda *a: pytest.fail("sync should not run")) + monkeypatch.setattr(discovery, "poll_all", lambda *a: {"queued": 0}) + monkeypatch.setattr(reap, "run", lambda *a: {"aged_out": 0}) + + result = runner.run(conn, settings, channel_pk=channel["id"]) + + assert "sync" not in result + + +def test_run_records_last_run_at(conn, settings, channel, monkeypatch, + quiet_jellyfin): + monkeypatch.setattr(subsync, "sync_all", lambda *a: {"added": 0, "queued": 0, + "removed": 0, "refused": 0, + "sources": 0, + "pending_removal": 0}) + monkeypatch.setattr(discovery, "poll_all", lambda *a: {"queued": 0}) + monkeypatch.setattr(reap, "run", lambda *a: {"aged_out": 0}) + + runner.run(conn, settings) + + assert settings.raw("last_run_at") + + +def test_jellyfin_is_refreshed_only_when_the_tree_changed( + conn, settings, media_root, channel, monkeypatch, quiet_jellyfin, no_thumbs +): + monkeypatch.setattr(subsync, "sync_all", lambda *a: {"added": 0, "queued": 0, + "removed": 0, "refused": 0, + "sources": 0, + "pending_removal": 0}) + monkeypatch.setattr(discovery, "poll_all", lambda *a: {"queued": 0}) + + runner.run(conn, settings) + assert quiet_jellyfin == [] + + add_video(conn, channel["id"], "vid00000001") + runner.run(conn, settings) + assert len(quiet_jellyfin) == 1 + + +# ------------------------------------------------------------------ exit codes + + +def test_refused_sync_turns_the_check_red(): + """A silently-broken mirror is the worst outcome available: nothing looks + wrong until someone asks why a channel never appeared.""" + assert runner.exit_code({"sync": {"refused": 1}}) == 1 + + +def test_materialise_errors_turn_the_check_red(): + assert runner.exit_code({"materialise": {"errors": 2}}) == 1 + + +def test_poll_failures_alone_do_not_turn_the_check_red(): + """Two of 119 measured channels fail permanently — terminated or private.""" + assert runner.exit_code({"poll": {"failed": 2}}) == 0 + + +def test_clean_run_is_zero(): + assert runner.exit_code({"sync": {"refused": 0}, "poll": {"failed": 0}, + "materialise": {"errors": 0}}) == 0 + + +# ------------------------------------------------------------------- summarise + + +def test_summarise_mentions_a_refusal(): + text = runner.summarise({ + "sync": {"added": 0, "queued": 0, "removed": 0, "refused": 1}, + "poll": {"channels": 5, "queued": 0}, + "materialise": {"materialised": 0}, + "reap": {"aged_out": 0}, + }) + assert "SYNC_REFUSED" in text + + +def test_summarise_is_a_single_line(): + text = runner.summarise({ + "sync": {"added": 1, "queued": 2, "removed": 3, "refused": 0}, + "poll": {"channels": 119, "queued": 4, "shorts": 5, "live": 6, "failed": 2}, + "materialise": {"materialised": 7, "errors": 0}, + "reap": {"aged_out": 8}, + }) + assert "\n" not in text + assert "channels=119" in text + assert "materialised=7" in text + assert "aged_out=8" in text diff --git a/tests/test_settings.py b/tests/test_settings.py new file mode 100644 index 0000000..d65610a --- /dev/null +++ b/tests/test_settings.py @@ -0,0 +1,125 @@ +"""Typed settings accessors and form validation.""" + +import pytest + +from ytstream import settings as settings_module +from ytstream.settings import DEFAULTS, Settings, validate, validate_all + + +class TestAccessors: + def test_missing_key_returns_the_default(self, settings): + assert settings.get_int("retention_days") == 30 + assert settings.get_int("min_keep_videos") == 5 + assert settings.get_str("proxy_base_url") == "http://127.0.0.1:8099" + + def test_unknown_key_never_raises(self, settings): + assert settings.get_str("no_such_key") == "" + assert settings.get_int("no_such_key") == 0 + + def test_set_then_get(self, settings): + settings.set("retention_days", "30") + assert settings.get_int("retention_days") == 30 + + def test_set_overwrites(self, settings): + settings.set("retention_days", "30") + settings.set("retention_days", "45") + assert settings.get_int("retention_days") == 45 + + def test_corrupt_integer_falls_back_to_the_default(self, settings): + settings.set("retention_days", "not a number") + assert settings.get_int("retention_days") == 30 + + def test_reads_are_not_cached(self, conn, settings): + """Rotating the API key must be nothing more than saving a new value — + no restart of the hourly job or the admin server.""" + settings.set("youtube_api_key", "first") + assert settings.get_str("youtube_api_key") == "first" + + # A second Settings object standing in for the other process. + other = Settings(conn) + other.set("youtube_api_key", "second") + + assert settings.get_str("youtube_api_key") == "second" + + def test_all_editable_covers_every_default(self, settings): + assert set(settings.all_editable()) == set(DEFAULTS) + + def test_secrets_are_not_editable(self): + for key in settings_module.SECRET_KEYS: + assert key not in settings_module.EDITABLE + + def test_both_api_keys_are_masked(self): + assert "youtube_api_key" in settings_module.MASKED_KEYS + assert "jellyfin_api_key" in settings_module.MASKED_KEYS + + def test_download_era_settings_are_gone(self): + for gone in ("disk_cap_gb", "write_subs", "sub_langs", "sponsorblock_mark", + "max_attempts", "backfill_days"): + assert gone not in DEFAULTS + + +class TestValidation: + @pytest.mark.parametrize( + "key, value", + [ + ("retention_days", "30"), + ("min_keep_videos", "0"), + ("min_keep_videos", "5"), + ("backfill_max_videos", "300"), + ("max_height", "1080"), + ("min_duration_seconds", "120"), + ("subsync_max_new", "25"), + ("subsync_missing_threshold", "3"), + ("jellyfin_url", "http://127.0.0.1:8096"), + ("pot_provider_url", "https://example.com:4416"), + ("proxy_base_url", "http://127.0.0.1:8099"), + ("youtube_api_key", ""), + ], + ) + def test_accepts_good_values(self, key, value): + ok, _ = validate(key, value) + assert ok + + @pytest.mark.parametrize( + "key, value", + [ + ("retention_days", "abc"), + ("retention_days", "0"), + ("retention_days", "-5"), + ("min_keep_videos", "-1"), + ("max_height", "10"), + ("min_duration_seconds", "-1"), + ("jellyfin_url", "not-a-url"), + ("jellyfin_url", "ftp://host/"), + ("jellyfin_url", "http://"), + ("proxy_base_url", "nonsense"), + ], + ) + def test_rejects_bad_values(self, key, value): + ok, message = validate(key, value) + assert not ok + assert message + + def test_zero_missing_threshold_is_rejected(self): + """Zero would unsubscribe on the first absent response, which is exactly + the failure the threshold exists to prevent.""" + ok, message = validate("subsync_missing_threshold", "0") + assert not ok + assert message + + def test_validate_all_reports_each_bad_field(self): + errors = validate_all( + {"retention_days": "abc", "max_height": "1080", "jellyfin_url": "nope"} + ) + assert set(errors) == {"retention_days", "jellyfin_url"} + + def test_validate_all_ignores_unknown_keys(self): + assert validate_all({"not_a_setting": "x"}) == {} + + def test_empty_api_keys_are_acceptable(self): + assert validate("jellyfin_api_key", "")[0] + assert validate("youtube_api_key", "")[0] + + def test_whitespace_is_tolerated(self): + ok, _ = validate("retention_days", " 21 ") + assert ok diff --git a/tests/test_strm.py b/tests/test_strm.py new file mode 100644 index 0000000..8498867 --- /dev/null +++ b/tests/test_strm.py @@ -0,0 +1,260 @@ +"""Materialising: .strm contents, NFO sidecars, naming, and idempotency.""" + +from __future__ import annotations + +from datetime import date + +import pytest + +from ytstream import config, strm, videos + +from conftest import add_video + + +@pytest.fixture() +def video(conn, channel): + return add_video(conn, channel["id"], "dQw4w9WgXcQ", + title="A Video: With/Punctuation", + upload_date="2026-08-12", duration=1337) + + +@pytest.fixture() +def no_thumbs(monkeypatch): + """Thumbnails are a network fetch; every test here runs without one.""" + monkeypatch.setattr(strm, "fetch_thumbnail", lambda *a, **k: False) + + +def test_strm_contains_only_the_proxy_url(conn, settings, media_root, channel, + video, no_thumbs): + result = strm.materialise(conn, settings, channel, video) + + path = media_root / result["rel_path"] + assert path.read_text() == "http://127.0.0.1:8099/watch/dQw4w9WgXcQ" + # No trailing newline, and nothing else in the file. + assert path.read_bytes() == b"http://127.0.0.1:8099/watch/dQw4w9WgXcQ" + + +def test_strm_url_follows_the_configured_base(conn, settings, media_root, channel, + video, no_thumbs): + settings.set("proxy_base_url", "http://127.0.0.1:9999/") + result = strm.materialise(conn, settings, channel, video) + assert (media_root / result["rel_path"]).read_text().startswith( + "http://127.0.0.1:9999/watch/" + ) + + +def test_layout_matches_the_naming_scheme(conn, settings, media_root, channel, + video, no_thumbs): + result = strm.materialise(conn, settings, channel, video) + + # Season = upload year, episode = MMDD*10 + ordinal. + assert result["season"] == 2026 + assert result["episode"] == 8120 + assert result["rel_path"] == ( + "clabretro/Season 2026/" + "clabretro - S2026E8120 - A Video With Punctuation [dQw4w9WgXcQ].strm" + ) + + +def test_nfo_is_written_alongside(conn, settings, media_root, channel, video, + no_thumbs): + result = strm.materialise(conn, settings, channel, video) + nfo_path = (media_root / result["rel_path"]).with_suffix(".nfo") + + text = nfo_path.read_text() + + assert "2026" in text + assert "8120" in text + assert "2026-08-12" in text + # durationinseconds is what stops a .strm episode showing a zero runtime + # before it has ever been played. + assert "1337" in text + assert 'type="youtube"' in text + + +def test_nfo_has_no_streamdetails(conn, settings, media_root, channel, video, + no_thumbs): + """Pre-seeding them was measured to change nothing about Jellyfin probing.""" + result = strm.materialise(conn, settings, channel, video) + text = (media_root / result["rel_path"]).with_suffix(".nfo").read_text() + assert "streamdetails" not in text + assert "fileinfo" not in text + + +def test_materialise_is_idempotent(conn, settings, media_root, channel, video, + no_thumbs): + first = strm.materialise(conn, settings, channel, video) + before = (media_root / first["rel_path"]).read_bytes() + + second = strm.materialise( + conn, settings, channel, videos.get(conn, "dQw4w9WgXcQ") + ) + + assert second["rel_path"] == first["rel_path"] + assert (media_root / second["rel_path"]).read_bytes() == before + + +def test_row_is_marked_materialised(conn, settings, media_root, channel, video, + no_thumbs): + strm.materialise(conn, settings, channel, video) + row = videos.get(conn, "dQw4w9WgXcQ") + assert row["state"] == videos.MATERIALISED + assert row["rel_path"] + assert row["materialised_at"] + + +def test_episode_ordinals_increment_within_a_day(conn, settings, media_root, + channel, no_thumbs): + for index in range(3): + row = add_video(conn, channel["id"], f"vid{index:08d}", + upload_date="2026-08-12") + result = strm.materialise(conn, settings, channel, row) + assert result["episode"] == 8120 + index + + +def test_ordinals_are_stable_when_an_earlier_video_ages_out( + conn, settings, media_root, channel, no_thumbs +): + """Aged-out rows keep their episode number, so later ordinals never shift.""" + first = add_video(conn, channel["id"], "vid00000001", upload_date="2026-08-12") + strm.materialise(conn, settings, channel, first) + videos.mark_aged_out(conn, "vid00000001") + + second = add_video(conn, channel["id"], "vid00000002", upload_date="2026-08-12") + result = strm.materialise(conn, settings, channel, second) + + assert result["episode"] == 8121 + + +def test_title_falls_back_to_the_video_id(conn, settings, media_root, channel, + no_thumbs): + """Backfilled rows carry no title until the feed supplies one.""" + row = add_video(conn, channel["id"], "vid00000001", title="", + upload_date="2026-08-12") + result = strm.materialise(conn, settings, channel, row) + assert "vid00000001" in result["rel_path"] + + +def test_write_show_creates_tvshow_nfo(media_root, channel): + strm.write_show(channel) + text = (media_root / "clabretro" / "tvshow.nfo").read_text() + assert "clabretro" in text + assert channel["channel_id"] in text + + +def test_show_nfo_uses_the_channel_title_not_a_video_title(conn, media_root, channel): + """The channel/video join has `title` on both sides, and reading the wrong one + renamed every series after whichever episode happened to be first.""" + add_video(conn, channel["id"], "vid00000001", title="Some Episode Title") + strm.write_show(channel) + text = (media_root / "clabretro" / "tvshow.nfo").read_text() + assert "Some Episode Title" not in text + + +def test_remove_deletes_strm_and_sidecars(conn, settings, media_root, channel, + video, no_thumbs): + result = strm.materialise(conn, settings, channel, video) + strm_path = media_root / result["rel_path"] + stem = strm_path.name[: -len(".strm")] + thumb = strm_path.with_name(stem + "-thumb.jpg") + thumb.write_bytes(b"x" * 2000) + + removed = strm.remove(videos.get(conn, "dQw4w9WgXcQ")) + + assert removed == 3 # .strm, .nfo, -thumb.jpg + assert not strm_path.exists() + assert not thumb.exists() + + +def test_remove_leaves_files_it_does_not_own(conn, settings, media_root, channel, + video, no_thumbs): + result = strm.materialise(conn, settings, channel, video) + season_dir = (media_root / result["rel_path"]).parent + stranger = season_dir / "someone-elses-file.txt" + stranger.write_text("not ours") + + strm.remove(videos.get(conn, "dQw4w9WgXcQ")) + + assert stranger.exists() + + +def test_remove_channel_tree(conn, media_root, channel): + tree = media_root / "clabretro" + (tree / "Season 2026").mkdir(parents=True) + (tree / "tvshow.nfo").write_text("") + + assert strm.remove_channel_tree(channel) is True + assert not tree.exists() + + +def test_remove_channel_tree_refuses_the_media_root(conn, media_root): + from conftest import add_channel + + row = add_channel(conn, "UC" + "q" * 22, "Blank", "") + keep = media_root / "keep" + keep.mkdir() + + assert strm.remove_channel_tree(row) is False + assert keep.exists() + + +def test_fetch_thumbnail_rejects_the_grey_placeholder(monkeypatch, tmp_path): + """YouTube serves a tiny placeholder rather than a 404 for missing maxres.""" + import urllib.request + + class Response: + status = 200 + + def __init__(self, payload): + self.payload = payload + + def read(self): + return self.payload + + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + calls = [] + + def fake_open(request, timeout=None): + calls.append(request.full_url) + # maxres returns the placeholder; hq returns something real. + return Response(b"x" * 120 if "maxres" in request.full_url else b"y" * 5000) + + monkeypatch.setattr(urllib.request, "urlopen", fake_open) + destination = tmp_path / "out-thumb.jpg" + + assert strm.fetch_thumbnail("dQw4w9WgXcQ", destination) is True + assert destination.read_bytes() == b"y" * 5000 + assert len(calls) == 2 + + +def test_fetch_thumbnail_is_skipped_when_one_already_exists(monkeypatch, tmp_path): + import urllib.request + + monkeypatch.setattr(urllib.request, "urlopen", + lambda *a, **k: pytest.fail("should not fetch")) + existing = tmp_path / "out-thumb.jpg" + existing.write_bytes(b"cached") + + assert strm.fetch_thumbnail("dQw4w9WgXcQ", existing) is True + + +def test_untitled_video_does_not_get_its_id_written_back_as_a_title( + conn, settings, media_root, channel, no_thumbs +): + """Writing the fallback back to the database makes the row look titled, which + permanently disables the title repair in discovery._record. That shipped once + and left five of twenty episodes named after their video ids.""" + row = add_video(conn, channel["id"], "vid00000001", title="", + upload_date="2026-08-12") + + result = strm.materialise(conn, settings, channel, row) + + # The filename falls back to the id... + assert "vid00000001]" in result["rel_path"] + # ...but the row stays untitled, so a later feed poll can still repair it. + assert videos.get(conn, "vid00000001")["title"] == "" diff --git a/tests/test_subsync.py b/tests/test_subsync.py new file mode 100644 index 0000000..812d621 --- /dev/null +++ b/tests/test_subsync.py @@ -0,0 +1,366 @@ +"""Subscription mirroring, and above all its refusals. + +The sync is authoritative in both directions and the removal half deletes a +channel's whole tree, so most of what needs pinning down here is what it does +with *bad* data. Every one of these failure modes looks identical to "he +unsubscribed from everything" on the wire: + + 403 subscriptionForbidden he re-ticked the privacy box + network error susan's link dropped + 200 with zero items could be true, could be a broken response + +None of them may delete anything. The tests below are the reason that claim can +be made with a straight face. +""" + +from __future__ import annotations + +import pytest + +from ytstream import api, subsync, videos + +from conftest import CHANNEL_ID, FakeApi, add_channel, add_video, patch_api + +BROTHER = "UCPcTWaLV8zwx4WP4QExHj4Q" + + +@pytest.fixture() +def source(conn): + key = subsync.add_source(conn, channel_id=BROTHER, label="C Flux") + with conn: + conn.execute("UPDATE source SET imported = 1 WHERE key = ?", (key,)) + return subsync.get_source(conn, key) + + +@pytest.fixture() +def fresh_source(conn): + """A source that has never imported — the first-sync path.""" + key = subsync.add_source(conn, channel_id=BROTHER, label="C Flux") + return subsync.get_source(conn, key) + + +def sub(channel_id, title): + return {"channel_id": channel_id, "title": title} + + +def _fake(monkeypatch, **kwargs): + return patch_api(monkeypatch, subsync, FakeApi(**kwargs)) + + +# ------------------------------------------------------------------- additions + + +def test_first_sync_queues_everything_and_adds_nothing( + conn, settings, fresh_source, monkeypatch +): + """119 subscriptions would trip any cap, so day one is approval-only.""" + _fake(monkeypatch, subs=[sub(f"UC{i:022d}", f"Chan {i}") for i in range(40)]) + + stats = subsync.sync_source(conn, settings, fresh_source) + + assert stats["added"] == 0 + assert stats["queued"] == 40 + assert conn.execute("SELECT COUNT(*) FROM channel").fetchone()[0] == 0 + assert len(subsync.pending(conn)) == 40 + # And it does not queue them again on the next pass. + assert subsync.get_source(conn, fresh_source["key"])["imported"] == 1 + + +def test_second_sync_adds_within_the_cap(conn, settings, source, monkeypatch): + settings.set("subsync_max_new", "25") + _fake(monkeypatch, subs=[sub("UCaaaaaaaaaaaaaaaaaaaaaa", "Alpha")], + channel={"channel_id": "UCaaaaaaaaaaaaaaaaaaaaaa", "title": "Alpha", + "description": "", "handle": None, "avatar_url": None}) + + stats = subsync.sync_source(conn, settings, source) + + assert stats["added"] == 1 + row = conn.execute("SELECT * FROM channel").fetchone() + assert row["title"] == "Alpha" + assert row["source"] == "youtube" + + +def test_burst_over_the_cap_adds_nothing_and_queues_all( + conn, settings, source, monkeypatch +): + settings.set("subsync_max_new", "3") + _fake(monkeypatch, subs=[sub(f"UC{i:022d}", f"Chan {i}") for i in range(10)]) + + stats = subsync.sync_source(conn, settings, source) + + assert stats["added"] == 0 + assert stats["queued"] == 10 + assert conn.execute("SELECT COUNT(*) FROM channel").fetchone()[0] == 0 + + +def test_rejected_channels_are_never_queued_again( + conn, settings, fresh_source, monkeypatch +): + entries = [sub("UCaaaaaaaaaaaaaaaaaaaaaa", "Alpha")] + _fake(monkeypatch, subs=entries) + subsync.sync_source(conn, settings, fresh_source) + + queued = subsync.pending(conn) + assert len(queued) == 1 + subsync.resolve(conn, [queued[0]["id"]], "rejected") + + source = subsync.get_source(conn, fresh_source["key"]) + stats = subsync.sync_source(conn, settings, source) + + assert stats["queued"] == 0 + assert stats["added"] == 0 + assert subsync.pending(conn) == [] + + +def test_approving_subscribes(conn, settings, fresh_source, monkeypatch): + _fake(monkeypatch, subs=[sub("UCaaaaaaaaaaaaaaaaaaaaaa", "Alpha")], + channel={"channel_id": "UCaaaaaaaaaaaaaaaaaaaaaa", "title": "Alpha", + "description": "Desc", "handle": "@alpha", "avatar_url": None}) + subsync.sync_source(conn, settings, fresh_source) + queued = subsync.pending(conn) + + stats = subsync.approve(conn, settings, [row["id"] for row in queued]) + + assert stats == {"added": 1, "failed": 0} + assert conn.execute("SELECT COUNT(*) FROM channel").fetchone()[0] == 1 + assert subsync.pending(conn) == [] + + +# -------------------------------------------------------------------- refusals + + +def test_private_subscriptions_change_nothing(conn, settings, source, monkeypatch): + add_channel(conn, CHANNEL_ID, "Existing", "Existing") + _fake(monkeypatch, raises=api.SubscriptionsPrivate( + 403, "subscriptionForbidden", "not allowed")) + + stats = subsync.sync_source(conn, settings, source) + + assert stats["refused"] + assert "private" in stats["refused"] + assert stats["removed"] == 0 + assert conn.execute("SELECT COUNT(*) FROM channel").fetchone()[0] == 1 + # The channel's miss counter must not move either, or three consecutive + # outages would delete the library without a single healthy response. + assert conn.execute("SELECT missing_syncs FROM channel").fetchone()[0] == 0 + + +def test_empty_response_is_treated_as_suspect(conn, settings, source, monkeypatch): + """A genuinely empty list and a broken one are indistinguishable, so assume + the harmless reading.""" + add_channel(conn, CHANNEL_ID, "Existing", "Existing") + _fake(monkeypatch, subs=[]) + + stats = subsync.sync_source(conn, settings, source) + + assert stats["refused"] + assert stats["removed"] == 0 + assert conn.execute("SELECT COUNT(*) FROM channel").fetchone()[0] == 1 + + +def test_api_not_configured_changes_nothing(conn, settings, source, monkeypatch): + add_channel(conn, CHANNEL_ID, "Existing", "Existing") + _fake(monkeypatch, raises=api.NotConfigured(403, "forbidden", "blocked")) + + stats = subsync.sync_source(conn, settings, source) + + assert stats["refused"] + assert conn.execute("SELECT COUNT(*) FROM channel").fetchone()[0] == 1 + + +def test_network_error_changes_nothing(conn, settings, source, monkeypatch): + add_channel(conn, CHANNEL_ID, "Existing", "Existing") + _fake(monkeypatch, raises=api.ApiError(0, "network", "connection reset")) + + stats = subsync.sync_source(conn, settings, source) + + assert stats["refused"] + assert conn.execute("SELECT COUNT(*) FROM channel").fetchone()[0] == 1 + + +def test_refusal_is_recorded_on_the_source(conn, settings, source, monkeypatch): + _fake(monkeypatch, raises=api.ApiError(0, "network", "boom")) + subsync.sync_source(conn, settings, source) + + row = subsync.get_source(conn, source["key"]) + assert row["last_sync_ok"] == 0 + assert row["consecutive_failures"] == 1 + assert "boom" in row["last_error"] + + +def test_three_outages_in_a_row_still_delete_nothing( + conn, settings, source, monkeypatch +): + """The threshold counts absences from healthy responses, not failures.""" + add_channel(conn, CHANNEL_ID, "Existing", "Existing") + _fake(monkeypatch, raises=api.ApiError(0, "network", "down")) + + for _ in range(5): + subsync.sync_source(conn, settings, subsync.get_source(conn, source["key"])) + + assert conn.execute("SELECT COUNT(*) FROM channel").fetchone()[0] == 1 + + +# -------------------------------------------------------------------- removals + + +def test_absence_counts_up_but_does_not_delete_below_the_threshold( + conn, settings, source, monkeypatch +): + settings.set("subsync_missing_threshold", "3") + add_channel(conn, CHANNEL_ID, "Doomed", "Doomed") + _fake(monkeypatch, subs=[sub("UCaaaaaaaaaaaaaaaaaaaaaa", "Alpha")]) + settings.set("subsync_max_new", "0") # keep the addition path out of this + + for expected in (1, 2): + stats = subsync.sync_source( + conn, settings, subsync.get_source(conn, source["key"]) + ) + assert stats["removed"] == 0 + assert stats["pending_removal"] == 1 + assert conn.execute( + "SELECT missing_syncs FROM channel WHERE dir_name = 'Doomed'" + ).fetchone()[0] == expected + + +def test_deletion_happens_on_the_threshold_sync( + conn, settings, source, media_root, monkeypatch +): + settings.set("subsync_missing_threshold", "3") + settings.set("subsync_max_new", "0") + doomed = add_channel(conn, CHANNEL_ID, "Doomed", "Doomed") + tree = media_root / "Doomed" + (tree / "Season 2026").mkdir(parents=True) + (tree / "tvshow.nfo").write_text("") + add_video(conn, doomed["id"], "vid00000001") + _fake(monkeypatch, subs=[sub("UCaaaaaaaaaaaaaaaaaaaaaa", "Alpha")]) + + for _ in range(3): + stats = subsync.sync_source( + conn, settings, subsync.get_source(conn, source["key"]) + ) + + assert stats["removed"] == 1 + assert conn.execute("SELECT COUNT(*) FROM channel").fetchone()[0] == 0 + assert not tree.exists() + # The videos went with it, via ON DELETE CASCADE. + assert conn.execute("SELECT COUNT(*) FROM video").fetchone()[0] == 0 + + +def test_reappearing_resets_the_counter(conn, settings, source, monkeypatch): + settings.set("subsync_missing_threshold", "3") + settings.set("subsync_max_new", "0") + add_channel(conn, CHANNEL_ID, "Flaky", "Flaky") + + _fake(monkeypatch, subs=[sub("UCaaaaaaaaaaaaaaaaaaaaaa", "Alpha")]) + subsync.sync_source(conn, settings, subsync.get_source(conn, source["key"])) + assert conn.execute("SELECT missing_syncs FROM channel").fetchone()[0] == 1 + + _fake(monkeypatch, subs=[sub(CHANNEL_ID, "Flaky")]) + subsync.sync_source(conn, settings, subsync.get_source(conn, source["key"])) + + assert conn.execute("SELECT missing_syncs FROM channel").fetchone()[0] == 0 + + +def test_manual_channels_are_never_removed( + conn, settings, source, media_root, monkeypatch +): + settings.set("subsync_missing_threshold", "1") + settings.set("subsync_max_new", "0") + add_channel(conn, CHANNEL_ID, "Pinned", "Pinned", source="manual") + _fake(monkeypatch, subs=[sub("UCaaaaaaaaaaaaaaaaaaaaaa", "Alpha")]) + + for _ in range(3): + stats = subsync.sync_source( + conn, settings, subsync.get_source(conn, source["key"]) + ) + + assert stats["removed"] == 0 + assert conn.execute("SELECT COUNT(*) FROM channel").fetchone()[0] == 1 + assert conn.execute("SELECT missing_syncs FROM channel").fetchone()[0] == 0 + + +def test_threshold_of_one_deletes_on_the_first_absence( + conn, settings, source, monkeypatch +): + """Configurable so tests need not loop; the default stays at 3.""" + settings.set("subsync_missing_threshold", "1") + settings.set("subsync_max_new", "0") + add_channel(conn, CHANNEL_ID, "Doomed", "Doomed") + _fake(monkeypatch, subs=[sub("UCaaaaaaaaaaaaaaaaaaaaaa", "Alpha")]) + + stats = subsync.sync_source(conn, settings, source) + + assert stats["removed"] == 1 + + +def test_zero_threshold_is_clamped_to_one(conn, settings, source, monkeypatch): + """A stored 0 must not mean "delete before any absence is confirmed".""" + settings.set("subsync_missing_threshold", "0") + settings.set("subsync_max_new", "0") + add_channel(conn, CHANNEL_ID, "Doomed", "Doomed") + _fake(monkeypatch, subs=[sub("UCaaaaaaaaaaaaaaaaaaaaaa", "Alpha")]) + + stats = subsync.sync_source(conn, settings, source) + + # Clamped to 1, so the first absence is enough — but it took one absence, + # not zero, and the row was actually observed missing. + assert stats["removed"] == 1 + assert stats["seen"] == 1 + + +# --------------------------------------------------------------------- general + + +def test_unsafe_channel_dir_is_not_deleted(conn, settings, media_root): + """A blank dir_name must never resolve the delete to the media root.""" + row = add_channel(conn, CHANNEL_ID, "Bad", " ") + (media_root / "keepme").mkdir() + + from ytstream import strm + + assert strm.remove_channel_tree(row) is False + assert (media_root / "keepme").exists() + + +def test_one_unresolvable_channel_does_not_abort_the_sync( + conn, settings, source, monkeypatch +): + class Exploding(FakeApi): + def channel(self, channel_id): + if channel_id.endswith("bad"): + raise api.ApiError(500, "backendError", "boom") + return {"channel_id": channel_id, "title": "Fine", "description": "", + "handle": None, "avatar_url": None} + + monkeypatch.setattr( + subsync.channels, "subscribe_from_sync", + lambda conn, settings, cid, title: (_ for _ in ()).throw(RuntimeError("no")) + if cid.endswith("bad") else add_channel(conn, cid, title, title), + ) + patch_api(monkeypatch, subsync, + FakeApi(subs=[sub("UC" + "a" * 19 + "bad", "Bad"), + sub("UC" + "b" * 22, "Good")])) + + stats = subsync.sync_source(conn, settings, source) + + assert stats["added"] == 1 + assert conn.execute("SELECT COUNT(*) FROM channel").fetchone()[0] == 1 + + +def test_sync_all_aggregates_and_counts_refusals(conn, settings, source, monkeypatch): + _fake(monkeypatch, raises=api.ApiError(0, "network", "boom")) + totals = subsync.sync_all(conn, settings) + assert totals["sources"] == 1 + assert totals["refused"] == 1 + + +def test_disabled_source_is_skipped(conn, settings, source, monkeypatch): + with conn: + conn.execute("UPDATE source SET enabled = 0") + fake = _fake(monkeypatch, subs=[sub("UCaaaaaaaaaaaaaaaaaaaaaa", "Alpha")]) + + totals = subsync.sync_all(conn, settings) + + assert totals["sources"] == 0 + assert fake.subscription_calls == 0 diff --git a/tests/test_videos.py b/tests/test_videos.py new file mode 100644 index 0000000..4bd79b6 --- /dev/null +++ b/tests/test_videos.py @@ -0,0 +1,209 @@ +"""The video state machine and episode numbering.""" + +from __future__ import annotations + +from datetime import date + +import pytest + +from ytstream import videos + +from conftest import add_channel, add_video + + +def test_insert_and_get(conn, channel): + row = add_video(conn, channel["id"], "vid00000001") + assert row["state"] == videos.LISTED + assert videos.exists(conn, "vid00000001") + assert not videos.exists(conn, "nosuchvideo") + + +def test_insert_is_idempotent(conn, channel): + add_video(conn, channel["id"], "vid00000001", title="First") + add_video(conn, channel["id"], "vid00000001", title="Second") + assert conn.execute("SELECT COUNT(*) FROM video").fetchone()[0] == 1 + assert videos.get(conn, "vid00000001")["title"] == "First" + + +def test_aged_out_is_terminal_and_never_revived(): + assert videos.AGED_OUT in videos.TERMINAL + assert videos.AGED_OUT in videos.NEVER_REVIVE + # skipped_old is terminal but IS revivable by rescan; conflating the two + # would resurrect months of deleted episodes as new. + assert videos.SKIPPED_OLD in videos.TERMINAL + assert videos.SKIPPED_OLD not in videos.NEVER_REVIVE + + +def test_no_download_era_states_survive(): + """Materialising a text file cannot fail, so there is no retry ladder.""" + for gone in ("pending", "downloading", "downloaded", "failed", "deferred"): + assert gone not in ( + videos.LISTED, videos.MATERIALISED, videos.SKIPPED_SHORT, + videos.SKIPPED_LIVE, videos.SKIPPED_OLD, videos.AGED_OUT, + ) + + +def test_mark_aged_out_clears_rel_path_but_keeps_the_row(conn, channel): + add_video(conn, channel["id"], "vid00000001", state=videos.MATERIALISED) + with conn: + conn.execute("UPDATE video SET rel_path = 'a/b.strm' WHERE video_id = ?", + ("vid00000001",)) + + videos.mark_aged_out(conn, "vid00000001") + + row = videos.get(conn, "vid00000001") + assert row is not None + assert row["state"] == videos.AGED_OUT + assert row["rel_path"] is None + assert row["deleted_at"] + + +def test_mark_materialised_records_everything(conn, channel): + add_video(conn, channel["id"], "vid00000001") + videos.mark_materialised(conn, "vid00000001", rel_path="a/b.strm", season=2026, + episode=8120, upload_date="2026-08-12", duration=99, + title="T") + row = videos.get(conn, "vid00000001") + assert (row["state"], row["season"], row["episode"], row["duration"]) == ( + videos.MATERIALISED, 2026, 8120, 99 + ) + + +def test_set_duration_alone(conn, channel): + add_video(conn, channel["id"], "vid00000001", duration=None) + videos.set_duration(conn, "vid00000001", 1234) + assert videos.get(conn, "vid00000001")["duration"] == 1234 + + +# ------------------------------------------------------------ episode numbers + + +def test_first_episode_of_a_day(conn, channel): + add_video(conn, channel["id"], "vid00000001") + assert videos.next_episode( + conn, channel["id"], date(2026, 8, 12), "vid00000001" + ) == (2026, 8120) + + +def test_ordinal_counts_existing_rows_for_that_day(conn, channel): + add_video(conn, channel["id"], "vid00000001") + with conn: + conn.execute("UPDATE video SET season = 2026, episode = 8120 " + "WHERE video_id = ?", ("vid00000001",)) + add_video(conn, channel["id"], "vid00000002") + + assert videos.next_episode( + conn, channel["id"], date(2026, 8, 12), "vid00000002" + ) == (2026, 8121) + + +def test_ordinal_ignores_other_days(conn, channel): + add_video(conn, channel["id"], "vid00000001") + with conn: + conn.execute("UPDATE video SET season = 2026, episode = 8110 " + "WHERE video_id = ?", ("vid00000001",)) + add_video(conn, channel["id"], "vid00000002") + + assert videos.next_episode( + conn, channel["id"], date(2026, 8, 12), "vid00000002" + ) == (2026, 8120) + + +def test_ordinal_ignores_other_channels(conn, channel): + other = add_channel(conn, "UC" + "z" * 22, "Other", "Other") + add_video(conn, other["id"], "vid00000001") + with conn: + conn.execute("UPDATE video SET season = 2026, episode = 8120 " + "WHERE video_id = ?", ("vid00000001",)) + add_video(conn, channel["id"], "vid00000002") + + assert videos.next_episode( + conn, channel["id"], date(2026, 8, 12), "vid00000002" + ) == (2026, 8120) + + +def test_ordinal_clamps_at_ten_uploads_a_day(conn, channel): + for index in range(10): + video_id = f"vid{index:08d}" + add_video(conn, channel["id"], video_id) + with conn: + conn.execute("UPDATE video SET season = 2026, episode = ? " + "WHERE video_id = ?", (8120 + index, video_id)) + add_video(conn, channel["id"], "vid00000099") + + season, episode = videos.next_episode( + conn, channel["id"], date(2026, 8, 12), "vid00000099" + ) + + assert (season, episode) == (2026, 8129) + + +def test_a_videos_own_row_does_not_bump_its_ordinal(conn, channel): + """Re-materialising must produce the same number, not the next one.""" + add_video(conn, channel["id"], "vid00000001") + with conn: + conn.execute("UPDATE video SET season = 2026, episode = 8120 " + "WHERE video_id = ?", ("vid00000001",)) + + assert videos.next_episode( + conn, channel["id"], date(2026, 8, 12), "vid00000001" + ) == (2026, 8120) + + +# --------------------------------------------------------------------- queues + + +def test_claim_listed_returns_only_listed_rows(conn, channel): + add_video(conn, channel["id"], "listed00001", state=videos.LISTED) + add_video(conn, channel["id"], "done0000001", state=videos.MATERIALISED) + add_video(conn, channel["id"], "short000001", state=videos.SKIPPED_SHORT) + add_video(conn, channel["id"], "aged0000001", state=videos.AGED_OUT) + + rows = videos.claim_listed(conn) + + assert [row["video_id"] for row in rows] == ["listed00001"] + + +def test_claim_listed_is_oldest_upload_first(conn, channel): + add_video(conn, channel["id"], "newer000001", upload_date="2026-08-10") + add_video(conn, channel["id"], "older000001", upload_date="2026-01-01") + + rows = videos.claim_listed(conn) + + assert [row["video_id"] for row in rows] == ["older000001", "newer000001"] + + +def test_claim_listed_honours_the_limit(conn, channel): + for index in range(5): + add_video(conn, channel["id"], f"vid{index:08d}") + assert len(videos.claim_listed(conn, limit=2)) == 2 + + +def test_materialised_for_channel_is_newest_first(conn, channel): + """The retention sweep counts down from the newest to honour min_keep_videos.""" + add_video(conn, channel["id"], "older000001", upload_date="2026-01-01", + state=videos.MATERIALISED) + add_video(conn, channel["id"], "newer000001", upload_date="2026-08-10", + state=videos.MATERIALISED) + + rows = videos.materialised_for_channel(conn, channel["id"]) + + assert [row["video_id"] for row in rows] == ["newer000001", "older000001"] + + +def test_queue_depth_and_counts(conn, channel): + add_video(conn, channel["id"], "listed00001", state=videos.LISTED) + add_video(conn, channel["id"], "listed00002", state=videos.LISTED) + add_video(conn, channel["id"], "done0000001", state=videos.MATERIALISED) + + assert videos.queue_depth(conn) == 2 + counts = videos.counts_by_state(conn) + assert counts[videos.LISTED] == 2 + assert counts[videos.MATERIALISED] == 1 + + +def test_deleting_a_channel_cascades_to_its_videos(conn, channel): + add_video(conn, channel["id"], "vid00000001") + with conn: + conn.execute("DELETE FROM channel WHERE id = ?", (channel["id"],)) + assert conn.execute("SELECT COUNT(*) FROM video").fetchone()[0] == 0 diff --git a/tests/test_web.py b/tests/test_web.py new file mode 100644 index 0000000..2bdb307 --- /dev/null +++ b/tests/test_web.py @@ -0,0 +1,146 @@ +"""The admin UI: rendering, auth gating and CSRF. + +The templates are f-strings over a dict, so a renamed key is a KeyError at render +time rather than a type error at import time — which is exactly the kind of +breakage that only shows up when somebody opens the page. These tests render every +page for real. +""" + +from __future__ import annotations + +import pytest + +from ytstream import videos +from ytstream.web import auth, templates + +from conftest import add_channel, add_video + + +@pytest.fixture() +def channel_row(conn, settings, channel): + """The dict shape server.py builds for the channel table.""" + add_video(conn, channel["id"], "vid00000001", state=videos.MATERIALISED) + return { + "id": channel["id"], + "title": channel["title"], + "handle": channel["handle"], + "channel_id": channel["channel_id"], + "retention_days": None, + "global_retention": 30, + "last_polled_at": "2026-08-12T14:00:00+00:00", + "last_poll_ok": 1, + "consecutive_poll_failures": 0, + "episodes": 1, + "source": "youtube", + "missing_syncs": 0, + "missing_threshold": 3, + "latest": "2026-08-12", + } + + +def test_login_page_renders(settings): + html = templates.login_page().decode() + assert "ytstream" in html + assert "youtube-automate" not in html + assert 'type="password"' in html + + +def test_login_page_shows_an_error(): + assert "Wrong" in templates.login_page("Wrong password").decode() + + +def test_index_page_renders(channel_row): + html = templates.index_page( + channels=[channel_row], settings_values={"retention_days": "30"}, + settings_errors={}, csrf="tok", add_error=None, queue_depth=0, + ).decode() + + assert "clabretro" in html + assert "ytstream" in html + # The provenance column replaced the meaningless size column. + assert "youtube" in html + assert "Size" not in html + + +def test_index_page_renders_with_no_channels(): + html = templates.index_page( + channels=[], settings_values={}, settings_errors={}, csrf="tok", + add_error=None, queue_depth=0, + ).decode() + assert "0 channel(s)" in html + + +def test_absence_badge_appears_before_the_channel_disappears(channel_row): + """A channel counting towards removal must be visible while it still exists.""" + channel_row["missing_syncs"] = 2 + html = templates.index_page( + channels=[channel_row], settings_values={}, settings_errors={}, csrf="tok", + add_error=None, queue_depth=0, + ).decode() + assert "absent 2/3 syncs" in html + + +def test_no_absence_badge_when_healthy(channel_row): + html = templates.index_page( + channels=[channel_row], settings_values={}, settings_errors={}, csrf="tok", + add_error=None, queue_depth=0, + ).decode() + assert "absent" not in html + + +def test_hostile_channel_title_is_escaped(channel_row): + channel_row["title"] = '' + html = templates.index_page( + channels=[channel_row], settings_values={}, settings_errors={}, csrf="tok", + add_error=None, queue_depth=0, + ).decode() + assert "