From 155f05773da1d9bcb2010c119dbee9d9556bf0a8 Mon Sep 17 00:00:00 2001 From: Tom Flux Date: Wed, 12 Aug 2026 16:35:23 +0100 Subject: [PATCH] Build ytstream: catalogue, retention, subscription mirror, proxy MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phases 1-4 of plan.md §13. Forked from youtube-automate as planned rather than written from scratch: naming, NFO, auth, the admin UI, settings and the DB layer came across largely unchanged, download.py is gone, and the pieces that only make sense for a streaming library are new. api.py YouTube Data API v3 client. The whole metadata path. strm.py Materialising: a .strm, an .nfo and a thumbnail. Replaces the 330-line download.py, because the job is writing a URL to a file. subsync.py The subscription mirror, most of which is refusals. reap.py Retention, rewritten around the 30-day window and min_keep_videos. discovery.py RSS polling plus an API-backed, resumable, bounded backfill. proxy/ The verified PoC, moved in with a systemd unit. 330 tests, all passing, no network and no yt-dlp in any of them. The suite leans towards the failure paths, because that is where this design can actually hurt someone: a 403 that looks like an unsubscribe, a video that ages out and comes back, a title that never arrives. tests/test_proxy.py replaces the two standalone scripts under proxy/ and now drives the real make_handler(mgr, ...) rather than the PoC's make_handler(path, done), so routing and video-id validation are covered too. Ran it end to end against the live API and it found three real bugs. The first was mine and the tests caught it: strm.remove pruned empty directories up to the media root, so a channel directory whose tvshow.nfo happened to be missing would be deleted along with the season. It only looked safe because tvshow.nfo normally stops the walk. The prune boundary is now the channel directory explicitly. The other two only showed up against real data, and they compounded. The backfill inserted rows with no title and left the RSS poll to fill them in — but RSS returns 15 entries, which for Pitch Side spans 23 days against a 30-day window, so five of twenty episodes were named after their video ids. Worse, strm.materialise wrote that fallback back to the database as the title, which made the row look titled and permanently disabled the repair path. Both fixed: playlistItems.list now requests snippet as well as contentDetails, which costs the same single quota unit and carries the title alongside the exact publish date, and the fallback is used for the filename without being persisted. A title that does arrive late now also removes the badly-named files and re-queues, so the episode is rewritten rather than keeping its video-id name forever. Verified against the live API: all twenty Pitch Side episodes now carry real titles. Measured on the real account: 119 subscriptions queued for approval and none added on the first sync, then a two-channel run backfilled and materialised 26 episodes in under seven seconds. Two deliberate departures from plan.md, both recorded there: min_keep_videos defaults to 5 rather than being left as an open question. Without it 52 of 117 measured channels are empty Jellyfin series that flicker in and out as their single video crosses the retention line, and the plan already recommended it. The Jellyfin refresh is a bare /Library/Refresh with a comment explaining why it must stay that way. A normal scan makes zero media probes; FullRefresh does probe, and at 400 episodes that is 400 cold starts. Not yet done: no systemd units are installed (needs root — deploy/deploy.sh), the admin UI has no routes for sources or the approval queue yet, and nothing has been pointed at the real media root. Co-Authored-By: Claude Opus 5 --- deploy/bootstrap.sh | 46 +++ deploy/crontab.fragment | 15 + deploy/deploy.sh | 89 ++++ deploy/update-ytdlp.sh | 23 ++ deploy/ytstream-admin.service | 34 ++ deploy/ytstream-proxy.service | 46 +++ proxy/ytstream_proxy.py | 755 ++++++++++++++++++++++++++++++++++ pyproject.toml | 33 ++ tests/conftest.py | 202 +++++++++ tests/fixtures/uulf_feed.xml | 37 ++ tests/test_api.py | 329 +++++++++++++++ tests/test_auth.py | 203 +++++++++ tests/test_channels.py | 151 +++++++ tests/test_discovery.py | 391 ++++++++++++++++++ tests/test_naming.py | 128 ++++++ tests/test_nfo.py | 94 +++++ tests/test_proxy.py | 354 ++++++++++++++++ tests/test_reap.py | 201 +++++++++ tests/test_runner.py | 232 +++++++++++ tests/test_settings.py | 125 ++++++ tests/test_strm.py | 260 ++++++++++++ tests/test_subsync.py | 366 ++++++++++++++++ tests/test_videos.py | 209 ++++++++++ tests/test_web.py | 146 +++++++ ytstream/__init__.py | 7 + ytstream/__main__.py | 4 + ytstream/api.py | 327 +++++++++++++++ ytstream/channels.py | 301 ++++++++++++++ ytstream/cli.py | 556 +++++++++++++++++++++++++ ytstream/config.py | 36 ++ ytstream/db.py | 149 +++++++ ytstream/discovery.py | 422 +++++++++++++++++++ ytstream/doctor.py | 211 ++++++++++ ytstream/jellyfin.py | 145 +++++++ ytstream/naming.py | 104 +++++ ytstream/nfo.py | 86 ++++ ytstream/reap.py | 89 ++++ ytstream/runner.py | 149 +++++++ ytstream/settings.py | 160 +++++++ ytstream/strm.py | 207 ++++++++++ ytstream/subsync.py | 305 ++++++++++++++ ytstream/util.py | 70 ++++ ytstream/videos.py | 214 ++++++++++ ytstream/web/__init__.py | 1 + ytstream/web/auth.py | 191 +++++++++ ytstream/web/server.py | 399 ++++++++++++++++++ ytstream/web/templates.py | 253 ++++++++++++ ytstream/ytdlp.py | 109 +++++ 48 files changed, 8964 insertions(+) create mode 100755 deploy/bootstrap.sh create mode 100644 deploy/crontab.fragment create mode 100755 deploy/deploy.sh create mode 100755 deploy/update-ytdlp.sh create mode 100644 deploy/ytstream-admin.service create mode 100644 deploy/ytstream-proxy.service create mode 100644 proxy/ytstream_proxy.py create mode 100644 pyproject.toml create mode 100644 tests/conftest.py create mode 100644 tests/fixtures/uulf_feed.xml create mode 100644 tests/test_api.py create mode 100644 tests/test_auth.py create mode 100644 tests/test_channels.py create mode 100644 tests/test_discovery.py create mode 100644 tests/test_naming.py create mode 100644 tests/test_nfo.py create mode 100644 tests/test_proxy.py create mode 100644 tests/test_reap.py create mode 100644 tests/test_runner.py create mode 100644 tests/test_settings.py create mode 100644 tests/test_strm.py create mode 100644 tests/test_subsync.py create mode 100644 tests/test_videos.py create mode 100644 tests/test_web.py create mode 100644 ytstream/__init__.py create mode 100644 ytstream/__main__.py create mode 100644 ytstream/api.py create mode 100644 ytstream/channels.py create mode 100644 ytstream/cli.py create mode 100644 ytstream/config.py create mode 100644 ytstream/db.py create mode 100644 ytstream/discovery.py create mode 100644 ytstream/doctor.py create mode 100644 ytstream/jellyfin.py create mode 100644 ytstream/naming.py create mode 100644 ytstream/nfo.py create mode 100644 ytstream/reap.py create mode 100644 ytstream/runner.py create mode 100644 ytstream/settings.py create mode 100644 ytstream/strm.py create mode 100644 ytstream/subsync.py create mode 100644 ytstream/util.py create mode 100644 ytstream/videos.py create mode 100644 ytstream/web/__init__.py create mode 100644 ytstream/web/auth.py create mode 100644 ytstream/web/server.py create mode 100644 ytstream/web/templates.py create mode 100644 ytstream/ytdlp.py 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 "