From 3ea4732c5d813a2902c8b6bda57e2f3f86d28e7f Mon Sep 17 00:00:00 2001 From: Alex Shnitman Date: Sun, 12 Jul 2026 08:08:14 +0300 Subject: [PATCH] fix: harden download lifecycle, subscriptions, and UI robustness Addresses a full-project review. Backend correctness and availability: - ytdl: cancel() only SIGKILLs the child's process group when the child actually became its own group leader, so a race (or failed setpgrp) can no longer kill the whole server; kill the group on cancel and on shutdown to avoid orphaned ffmpeg children - ytdl: dedicated ThreadPoolExecutor for download supervision so active downloads can't starve extract_info / live probes on the default pool - ytdl/main/subscriptions: route fire-and-forget tasks through a bg_tasks helper that keeps a strong ref and logs failures - subscriptions: run flat-playlist extraction in an executor and check feeds with bounded concurrency so one slow feed can't block the loop; set last_checked on failure so broken feeds aren't retried every 60s - main: validate ids on /start & /delete and numeric env vars at startup; return 400 (not 500) on bad subscriptions/update input; serve /history from memory; move get_custom_dirs off the event loop; restrict t= stripping to YouTube hosts; drop double percent-decode in state guard - dl_formats/ytdl: enforce requested caption format via FFmpegSubtitlesConvertor and strip VTT header metadata only in the pre-cue region so real dialogue is preserved - ytdl: throttle progress events, dedup adds against pending, clear filename/size on error and reject out-of-dir trashcan deletes, pin fork start-method on Linux only Frontend: - retry deletes the done record only after a successful re-add - surface HTTP errors for delete/start and reset the deleting flag - ignore late 'updated' events for rows no longer in the queue - track table rows by map key; FileSizePipe uses base-1024 Also: HTTPS-aware Docker healthcheck, dead-code removal, and shared helpers for path-containment and yt-dlp option merging. Adds/updates unit tests throughout (250 backend tests passing). Co-Authored-By: Claude Opus 4.8 --- Dockerfile | 9 +- app/bg_tasks.py | 26 ++ app/dl_formats.py | 34 +- app/main.py | 66 +++- app/subscriptions.py | 121 +++++-- app/tests/test_api.py | 87 ++++- app/tests/test_bg_tasks.py | 55 ++++ app/tests/test_config.py | 29 ++ app/tests/test_dl_formats.py | 45 ++- app/tests/test_download_queue.py | 164 +++++++++- app/tests/test_main_helpers.py | 43 ++- app/tests/test_subscriptions.py | 233 +++++++++++++- app/tests/test_ytdl_utils.py | 173 ++++++++++ app/ytdl.py | 298 +++++++++++++----- ui/src/app/app.html | 4 +- ui/src/app/app.spec.ts | 6 + ui/src/app/app.ts | 52 ++- ui/src/app/pipes/file-size.pipe.spec.ts | 13 +- ui/src/app/pipes/file-size.pipe.ts | 5 +- ui/src/app/services/downloads.service.spec.ts | 64 ++++ ui/src/app/services/downloads.service.ts | 31 +- 21 files changed, 1392 insertions(+), 166 deletions(-) create mode 100644 app/bg_tasks.py create mode 100644 app/tests/test_bg_tasks.py diff --git a/Dockerfile b/Dockerfile index f9896bc..4a72f0e 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,4 +1,8 @@ -FROM node:lts-alpine AS builder +# Pinned to a major version rather than the lts-alpine floating tag: that tag +# has lagged behind and resolved to a Node patch older than the Angular CLI's +# minimum supported version, breaking the build. node:22-alpine currently +# satisfies @angular/cli's >=22.22.3 requirement. +FROM node:22-alpine AS builder WORKDIR /metube COPY ui ./ @@ -66,7 +70,8 @@ ENV TEMP_DIR=/downloads ENV PORT=8081 VOLUME /downloads EXPOSE 8081 -HEALTHCHECK --interval=30s --timeout=5s --start-period=20s --retries=3 CMD curl -fsS "http://localhost:${PORT}/" || exit 1 +HEALTHCHECK --interval=30s --timeout=5s --start-period=20s --retries=3 \ + CMD case "$HTTPS" in true|True|on|1) curl -fsSk "https://localhost:${PORT}/";; *) curl -fsS "http://localhost:${PORT}/";; esac || exit 1 # Add build-time argument for version ARG VERSION=dev diff --git a/app/bg_tasks.py b/app/bg_tasks.py new file mode 100644 index 0000000..2a1da6c --- /dev/null +++ b/app/bg_tasks.py @@ -0,0 +1,26 @@ +import asyncio +import logging + +log = logging.getLogger("bg_tasks") +_TASKS: set[asyncio.Task] = set() + + +def create_task(coro, *, name: str | None = None) -> asyncio.Task: + """create_task that keeps a strong reference and logs unexpected failures. + + A bare ``asyncio.create_task(...)`` is only weakly referenced by the event + loop; if nothing else holds the returned Task, it can be garbage collected + mid-flight. Keeping a module-level strong reference (removed once the task + finishes) avoids that, and the done-callback surfaces otherwise-silent + failures. + """ + task = asyncio.get_running_loop().create_task(coro, name=name) + _TASKS.add(task) + + def _done(t: asyncio.Task) -> None: + _TASKS.discard(t) + if not t.cancelled() and t.exception() is not None: + log.error("Background task %s failed", t.get_name(), exc_info=t.exception()) + + task.add_done_callback(_done) + return task diff --git a/app/dl_formats.py b/app/dl_formats.py index 7933451..9fcd779 100644 --- a/app/dl_formats.py +++ b/app/dl_formats.py @@ -3,6 +3,21 @@ import copy AUDIO_FORMATS = ("m4a", "mp3", "opus", "wav", "flac") CAPTION_MODES = ("auto_only", "manual_only", "prefer_manual", "prefer_auto") + +def merge_ytdl_option_layers(presets, overrides, presets_config) -> dict: + """Overlay named presets (in order) then per-item overrides onto a fresh dict. + + Does NOT include any base ``YTDL_OPTIONS`` — callers layer this on top of + their own base (a per-download build adds the global base; a subscription + scan relies on ``**config.YTDL_OPTIONS`` already being present in its + params). ``presets_config`` maps a preset name to its options dict. + """ + merged: dict = {} + for name in presets or []: + merged.update(presets_config.get(name, {})) + merged.update(overrides or {}) + return merged + CODEC_FILTER_MAP = { 'h264': "[vcodec~='^(h264|avc)']", 'h265': "[vcodec~='^(h265|hevc)']", @@ -43,6 +58,10 @@ def get_format(download_type: str, codec: str, format: str, quality: str) -> str quality = (quality or "best").strip().lower() if format.startswith("custom:"): + # Unreachable via the HTTP API (format is validated against a fixed + # set in main.py), but legacy persisted downloads may carry a + # custom: format from before that validation existed; removing this + # would crash PersistentQueue.load() for those records. return format[7:] if download_type == "thumbnail": @@ -137,7 +156,20 @@ def get_opts( requested_subtitle_format = (format or "srt").lower() if requested_subtitle_format == "txt": requested_subtitle_format = "srt" - opts["subtitlesformat"] = requested_subtitle_format + opts["subtitlesformat"] = f"{requested_subtitle_format}/best" + if requested_subtitle_format in ("srt", "vtt"): + # subtitlesformat above is only a preference: if the extractor + # doesn't natively offer this ext (e.g. YouTube has no native srt), + # yt-dlp silently falls back to whatever it has. ffmpeg can only + # convert to srt/vtt/ass/lrc, so only guarantee the requested + # container for those; other formats stay best-effort. + postprocessors.append( + { + "key": "FFmpegSubtitlesConvertor", + "format": requested_subtitle_format, + "when": "before_dl", + } + ) if mode == "manual_only": opts["writesubtitles"] = True opts["writeautomaticsub"] = False diff --git a/app/main.py b/app/main.py index 03a222c..fa575a9 100644 --- a/app/main.py +++ b/app/main.py @@ -16,9 +16,11 @@ import logging import json import pathlib import re -from urllib.parse import parse_qs, unquote, urlencode, urlparse, urlunparse +import time +from urllib.parse import parse_qs, urlencode, urlparse, urlunparse from watchfiles import DefaultFilter, Change, awatch +import bg_tasks from ytdl import DownloadQueueNotifier, DownloadQueue, Download from subscriptions import SubscriptionManager, SubscriptionNotifier, SubscriptionInfo, coerce_optional_bool from yt_dlp.version import __version__ as yt_dlp_version @@ -140,6 +142,10 @@ class Config: self._validate_int('MAX_CONCURRENT_DOWNLOADS', minimum=1) self._validate_int('PORT', minimum=1, maximum=65535) self._validate_int('CLEAR_COMPLETED_AFTER', minimum=0) + self._validate_int('DEFAULT_OPTION_PLAYLIST_ITEM_LIMIT', minimum=0) + self._validate_int('SUBSCRIPTION_DEFAULT_CHECK_INTERVAL', minimum=1) + self._validate_int('SUBSCRIPTION_SCAN_PLAYLIST_END', minimum=1) + self._validate_int('SUBSCRIPTION_MAX_SEEN_IDS', minimum=1) self._runtime_overrides = {} @@ -301,7 +307,10 @@ async def state_dir_guard(request, handler): (config.URL_PREFIX + 'audio_download/', config.AUDIO_DOWNLOAD_DIR), ): if request.path.startswith(prefix): - rel = unquote(request.path[len(prefix):]) + # request.path is already percent-decoded by aiohttp; decoding it + # again would mangle a download whose filename contains a literal + # '%' (e.g. "%" turning into a truncated escape) into a false 404. + rel = request.path[len(prefix):] target = os.path.realpath(os.path.join(base, rel)) if _is_within_state_dir(target): raise web.HTTPNotFound() @@ -425,12 +434,20 @@ def _clip_field_provided_in_post(raw) -> bool: def _extract_t_query_from_url(url: str) -> tuple[str, float | None]: - """If ``t=`` is present and parseable, return URL without ``t`` and start seconds.""" + """If ``t=`` is present and parseable, return URL without ``t`` and start seconds. + + Restricted to YouTube hosts: ``t`` is a generic query parameter name that + other sites may use for unrelated purposes, so rewriting it there would + silently mutate the URL and inject a bogus clip start. + """ try: parsed = urlparse(url) params = parse_qs(parsed.query) except Exception: return url, None + host = (parsed.hostname or '').lower() + if not (host in ('youtu.be', 'youtube.com') or host.endswith('.youtube.com')): + return url, None t_values = params.get('t') if not t_values: return url, None @@ -552,6 +569,7 @@ async def _download_queue_startup(app): async def _shutdown_download_manager(app): + dqueue.close() Download.shutdown_manager() @@ -607,7 +625,7 @@ async def _schedule_nightly_update() -> None: async def _start_nightly_update_schedule(app): - asyncio.create_task(_schedule_nightly_update()) + bg_tasks.create_task(_schedule_nightly_update(), name="nightly_update_schedule") app.on_startup.append(_start_nightly_update_schedule) @@ -656,7 +674,7 @@ async def watch_files(): await sio.emit('ytdl_options_changed', serializer.encode(result)) log.info(f'Starting Watch File: {config.YTDL_OPTIONS_FILE}') - asyncio.create_task(_watch_files()) + bg_tasks.create_task(_watch_files(), name="watch_ytdl_options_file") async def _watch_files_startup(app): await watch_files() @@ -970,13 +988,20 @@ async def subscriptions_check(request): result = await submgr.check_now([str(i) for i in ids] if ids else None) return web.Response(text=serializer.encode(result)) +def _require_id_list(post: dict) -> list: + ids = post.get('ids') + if not isinstance(ids, list) or not ids or not all(isinstance(i, str) for i in ids): + raise web.HTTPBadRequest(reason="'ids' must be a non-empty list of strings") + return ids + + @routes.post(config.URL_PREFIX + 'delete') async def delete(request): post = await _read_json_request(request) - ids = post.get('ids') + ids = _require_id_list(post) where = post.get('where') - if not ids or where not in ['queue', 'done']: - log.error("Bad request: missing 'ids' or incorrect 'where' value") + if where not in ['queue', 'done']: + log.error("Bad request: incorrect 'where' value") raise web.HTTPBadRequest() status = await (dqueue.cancel(ids) if where == 'queue' else dqueue.clear(ids)) log.info(f"Download delete request processed for ids: {ids}, where: {where}") @@ -985,7 +1010,7 @@ async def delete(request): @routes.post(config.URL_PREFIX + 'start') async def start(request): post = await _read_json_request(request) - ids = post.get('ids') + ids = _require_id_list(post) log.info(f"Received request to start pending downloads for ids: {ids}") status = await dqueue.start_pending(ids) return web.Response(text=serializer.encode(status)) @@ -1065,12 +1090,15 @@ async def cookie_status(request): async def history(request): history = { 'done': [], 'queue': [], 'pending': []} - for _, v in dqueue.queue.saved_items(): - history['queue'].append(v) - for _, v in dqueue.done.saved_items(): - history['done'].append(v) - for _, v in dqueue.pending.saved_items(): - history['pending'].append(v) + # Served from the in-memory queues (like the socket 'all' event) rather + # than saved_items(), which reloads and re-compacts the on-disk state on + # every call. + for _, v in dqueue.queue.items(): + history['queue'].append(v.info) + for _, v in dqueue.done.items(): + history['done'].append(v.info) + for _, v in dqueue.pending.items(): + history['pending'].append(v.info) log.info("Sending download history") return web.Response(text=serializer.encode(history)) @@ -1082,13 +1110,17 @@ async def connect(sid, environ): await sio.emit('subscriptions_all', serializer.encode([s.to_public_dict() for s in submgr.list_all()]), to=sid) await sio.emit('configuration', serializer.encode(config.frontend_safe()), to=sid) if config.CUSTOM_DIRS: - await sio.emit('custom_dirs', serializer.encode(get_custom_dirs()), to=sid) + # get_custom_dirs() can walk the whole download tree on a cache miss; + # keep that off the event loop so a large library doesn't stall every + # client's connect handshake. + dirs = await asyncio.get_running_loop().run_in_executor(None, get_custom_dirs) + await sio.emit('custom_dirs', serializer.encode(dirs), to=sid) if config.YTDL_OPTIONS_FILE: await sio.emit('ytdl_options_changed', serializer.encode(get_options_update_time()), to=sid) def get_custom_dirs(): cache_ttl_seconds = 5 - now = asyncio.get_running_loop().time() + now = time.monotonic() cache_key = ( config.DOWNLOAD_DIR, config.AUDIO_DOWNLOAD_DIR, diff --git a/app/subscriptions.py b/app/subscriptions.py index b2662d5..81e9513 100644 --- a/app/subscriptions.py +++ b/app/subscriptions.py @@ -11,14 +11,22 @@ import time import types import uuid from dataclasses import dataclass, field, fields +from functools import partial from typing import Any, Optional import yt_dlp import yt_dlp.networking.impersonate +import bg_tasks +from dl_formats import merge_ytdl_option_layers from state_store import AtomicJsonStore, read_legacy_shelf log = logging.getLogger("subscriptions") +# How many subscription feeds to scan at once. Bounded so one slow/hung feed +# doesn't serialize the rest, without bursting a large subscription list at the +# extractor (which risks rate-limiting / bot detection). +_MAX_CONCURRENT_CHECKS = 4 + VIDEO_ONLY_MSG = ( "This URL points to a single video, not a channel or playlist. Use Download instead." ) @@ -42,7 +50,9 @@ def _impersonate_opt(ytdl_options: dict) -> dict: return opts -def _build_ydl_params(config, *, playlistend: Optional[int] = None) -> dict: +def _build_ydl_params( + config, *, playlistend: Optional[int] = None, extra_opts: Optional[dict[str, Any]] = None +) -> dict: params: dict[str, Any] = { "quiet": not logging.getLogger().isEnabledFor(logging.DEBUG), "verbose": logging.getLogger().isEnabledFor(logging.DEBUG), @@ -52,6 +62,7 @@ def _build_ydl_params(config, *, playlistend: Optional[int] = None) -> dict: "lazy_playlist": True, "paths": {"home": config.DOWNLOAD_DIR, "temp": config.TEMP_DIR}, **config.YTDL_OPTIONS, + **(extra_opts or {}), } params = _impersonate_opt(params) if playlistend is not None and playlistend > 0: @@ -76,9 +87,11 @@ def _is_media_entry(entry: Any) -> bool: return True -def extract_flat_playlist(config, url: str, playlistend: int, *, _depth: int = 0): +def extract_flat_playlist( + config, url: str, playlistend: int, *, extra_opts: Optional[dict[str, Any]] = None, _depth: int = 0 +): """Return (info_dict, entries_list) for playlist/channel URLs.""" - params = _build_ydl_params(config, playlistend=playlistend) + params = _build_ydl_params(config, playlistend=playlistend, extra_opts=extra_opts) with yt_dlp.YoutubeDL(params=params) as ydl: info = ydl.extract_info(url, download=False) if not info: @@ -104,6 +117,7 @@ def extract_flat_playlist(config, url: str, playlistend: int, *, _depth: int = 0 config, nested_url, playlistend, + extra_opts=extra_opts, _depth=_depth + 1, ) if nested_entries: @@ -370,6 +384,23 @@ class SubscriptionManager: def _save_locked(self) -> None: self._store.save({"items": [_subscription_to_record(sub) for sub in self._subs.values()]}) + def _scan_extra_opts( + self, + ytdl_options_presets: Optional[list[str]], + ytdl_options_overrides: Optional[dict[str, Any]], + ) -> dict[str, Any]: + """Merge configured presets (in order) with per-subscription overrides. + + Applied on top of the global YTDL_OPTIONS when scanning a + subscription's feed, so cookies/impersonation/etc. configured via a + preset or override also take effect during the flat-playlist scan, + not just the eventual per-video download. (The global YTDL_OPTIONS base + is already spread into the scan params by ``_build_ydl_params``.) + """ + return merge_ytdl_option_layers( + ytdl_options_presets, ytdl_options_overrides, self.config.YTDL_OPTIONS_PRESETS + ) + async def _queue_subscription_entries( self, entries: list[dict], @@ -436,12 +467,8 @@ class SubscriptionManager: def start_background_loop(self) -> None: if self._loop_task is not None and not self._loop_task.done(): return - self._loop_task = asyncio.create_task(self._periodic_loop()) - self._loop_task.add_done_callback( - lambda t: log.error("Subscription loop failed: %s", t.exception()) - if not t.cancelled() and t.exception() - else None - ) + # bg_tasks.create_task already logs unexpected task failures with the name. + self._loop_task = bg_tasks.create_task(self._periodic_loop(), name="subscription_loop") async def _periodic_loop(self) -> None: while True: @@ -465,8 +492,30 @@ class SubscriptionManager: if now - sub.last_checked < interval_sec: continue due.append(sub) - for sub in due: - await self._check_one_unlocked(sub) + await self._check_many(due) + + async def _check_many(self, subs: list[SubscriptionInfo]) -> None: + """Check subscriptions with bounded concurrency so one slow feed does + not serialize the rest. Failures are isolated per subscription.""" + if not subs: + return + sem = asyncio.Semaphore(_MAX_CONCURRENT_CHECKS) + + async def _guarded(sub: SubscriptionInfo) -> None: + async with sem: + await self._check_one_unlocked(sub) + + results = await asyncio.gather( + *(_guarded(sub) for sub in subs), return_exceptions=True + ) + for sub, result in zip(subs, results): + if isinstance(result, Exception): + log.error( + "Subscription check crashed for %s: %s", + sub.name, + result, + exc_info=result, + ) async def add_subscription( self, @@ -513,8 +562,12 @@ class SubscriptionManager: try: scan_first = max(int(getattr(self.config, "SUBSCRIPTION_SCAN_PLAYLIST_END", 50)), 1) + scan_extra_opts = self._scan_extra_opts(ytdl_options_presets, ytdl_options_overrides) try: - info, entries = extract_flat_playlist(self.config, url, scan_first) + info, entries = await asyncio.get_running_loop().run_in_executor( + None, + partial(extract_flat_playlist, self.config, url, scan_first, extra_opts=scan_extra_opts), + ) except yt_dlp.utils.YoutubeDLError as exc: return {"status": "error", "msg": str(exc)} @@ -629,6 +682,24 @@ class SubscriptionManager: except ValueError as exc: return {"status": "error", "msg": str(exc)} + enabled_set = False + validated_enabled = False + if "enabled" in changes: + try: + validated_enabled = _coerce_bool(changes["enabled"]) + enabled_set = True + except ValueError as exc: + return {"status": "error", "msg": str(exc)} + + interval_set = False + validated_interval = 0 + if "check_interval_minutes" in changes: + try: + validated_interval = max(1, int(changes["check_interval_minutes"])) + except (TypeError, ValueError): + return {"status": "error", "msg": "check_interval_minutes must be an integer"} + interval_set = True + async with self._lock: sub = self._subs.get(sub_id) if not sub: @@ -636,10 +707,10 @@ class SubscriptionManager: previous = copy.deepcopy(sub) old_enabled = sub.enabled - if "enabled" in changes: - sub.enabled = _coerce_bool(changes["enabled"]) - if "check_interval_minutes" in changes: - sub.check_interval_minutes = max(1, int(changes["check_interval_minutes"])) + if enabled_set: + sub.enabled = validated_enabled + if interval_set: + sub.check_interval_minutes = validated_interval if "name" in changes and changes["name"]: sub.name = str(changes["name"]) if validated_tr is not None: @@ -673,8 +744,7 @@ class SubscriptionManager: "Manual subscription check requested for %d subscription(s)", len(targets), ) - for sub in targets: - await self._check_one_unlocked(sub) + await self._check_many(targets) return {"status": "ok"} async def _check_one_unlocked(self, sub: SubscriptionInfo) -> None: @@ -696,15 +766,23 @@ class SubscriptionManager: async def _check_one_inner(self, sub: SubscriptionInfo) -> None: sid = sub.id scan = int(getattr(self.config, "SUBSCRIPTION_SCAN_PLAYLIST_END", 50)) + # ytdl_options_presets/overrides are set at subscription creation and + # never mutated afterwards (update_subscription doesn't allow it), so + # reading them off `sub` here without holding the lock is safe. + scan_extra_opts = self._scan_extra_opts(sub.ytdl_options_presets, sub.ytdl_options_overrides) log.info("Checking subscription: %s", sub.name) try: - info, entries = extract_flat_playlist(self.config, sub.url, scan) + info, entries = await asyncio.get_running_loop().run_in_executor( + None, + partial(extract_flat_playlist, self.config, sub.url, scan, extra_opts=scan_extra_opts), + ) except yt_dlp.utils.YoutubeDLError as exc: async with self._lock: cur = self._subs.get(sid) if cur: previous = copy.deepcopy(cur) cur.error = str(exc) + cur.last_checked = time.time() try: self._save_locked() except Exception: @@ -723,6 +801,7 @@ class SubscriptionManager: if cur: previous = copy.deepcopy(cur) cur.error = VIDEO_ONLY_MSG + cur.last_checked = time.time() try: self._save_locked() except Exception: @@ -777,6 +856,10 @@ class SubscriptionManager: eid = _entry_id(ent) if not eid: continue + # Seen entries that are currently live are deliberately re-queued: + # a stream first seen as 'upcoming' must still be captured once it + # goes live. The download queue dedups by URL while a capture is + # in flight, so this can't double-queue an active capture. if eid in seen and ent.get("live_status") != "is_live": continue new_entries.append(ent) diff --git a/app/tests/test_api.py b/app/tests/test_api.py index 6481520..8ac86b3 100644 --- a/app/tests/test_api.py +++ b/app/tests/test_api.py @@ -6,6 +6,7 @@ import json import os from pathlib import Path from unittest.mock import AsyncMock, MagicMock +from urllib.parse import quote import pytest from aiohttp import web @@ -20,6 +21,7 @@ def mock_dqueue(monkeypatch): d.initialize = AsyncMock(return_value=None) d.add = AsyncMock(return_value={"status": "ok"}) d.cancel = AsyncMock(return_value={"status": "ok"}) + d.clear = AsyncMock(return_value={"status": "ok"}) d.start_pending = AsyncMock(return_value={"status": "ok"}) d.cancel_add = MagicMock() d.queue = MagicMock() @@ -28,6 +30,9 @@ def mock_dqueue(monkeypatch): d.queue.saved_items = MagicMock(return_value=[]) d.done.saved_items = MagicMock(return_value=[]) d.pending.saved_items = MagicMock(return_value=[]) + d.queue.items = MagicMock(return_value=[]) + d.done.items = MagicMock(return_value=[]) + d.pending.items = MagicMock(return_value=[]) d.get = MagicMock(return_value=([], [])) monkeypatch.setattr(main, "dqueue", d) return d @@ -215,11 +220,35 @@ async def test_start_pending(mock_dqueue): mock_dqueue.start_pending.assert_awaited_once_with(["a"]) +@pytest.mark.asyncio +@pytest.mark.parametrize("body", [{}, {"ids": "abc"}, {"ids": []}, {"ids": [1, 2]}]) +async def test_start_rejects_malformed_ids(mock_dqueue, body): + req = _json_request(body) + with pytest.raises(web.HTTPBadRequest): + await main.start(req) + mock_dqueue.start_pending.assert_not_awaited() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "body", + [ + {"where": "queue"}, + {"where": "queue", "ids": "abc"}, + {"where": "queue", "ids": []}, + {"where": "queue", "ids": [1, 2]}, + ], +) +async def test_delete_rejects_malformed_ids(mock_dqueue, body): + req = _json_request(body) + with pytest.raises(web.HTTPBadRequest): + await main.delete(req) + mock_dqueue.cancel.assert_not_awaited() + mock_dqueue.clear.assert_not_awaited() + + @pytest.mark.asyncio async def test_history_shape(mock_dqueue): - mock_dqueue.queue.saved_items.return_value = [] - mock_dqueue.done.saved_items.return_value = [] - mock_dqueue.pending.saved_items.return_value = [] req = MagicMock(spec=web.Request) resp = await main.history(req) assert resp.status == 200 @@ -227,6 +256,30 @@ async def test_history_shape(mock_dqueue): assert set(data.keys()) == {"done", "queue", "pending"} +@pytest.mark.asyncio +async def test_history_reads_in_memory_queues_not_disk_state(mock_dqueue): + fake_queue_dl = MagicMock() + fake_queue_dl.info = {"id": "q1", "title": "Queued"} + fake_done_dl = MagicMock() + fake_done_dl.info = {"id": "d1", "title": "Done"} + fake_pending_dl = MagicMock() + fake_pending_dl.info = {"id": "p1", "title": "Pending"} + mock_dqueue.queue.items.return_value = [("q1", fake_queue_dl)] + mock_dqueue.done.items.return_value = [("d1", fake_done_dl)] + mock_dqueue.pending.items.return_value = [("p1", fake_pending_dl)] + + req = MagicMock(spec=web.Request) + resp = await main.history(req) + assert resp.status == 200 + data = json.loads(resp.text) + assert [item["id"] for item in data["queue"]] == ["q1"] + assert [item["id"] for item in data["done"]] == ["d1"] + assert [item["id"] for item in data["pending"]] == ["p1"] + mock_dqueue.queue.saved_items.assert_not_called() + mock_dqueue.done.saved_items.assert_not_called() + mock_dqueue.pending.saved_items.assert_not_called() + + @pytest.mark.asyncio async def test_version_json(mock_dqueue): req = MagicMock(spec=web.Request) @@ -311,6 +364,24 @@ async def test_subscribe_rejects_clip_options(mock_dqueue, monkeypatch): main.submgr.add_subscription.assert_not_awaited() +@pytest.mark.asyncio +async def test_subscriptions_update_invalid_enabled_returns_error_not_500(mock_dqueue): + req = _json_request({"id": "nonexistent", "enabled": "maybe"}) + resp = await main.subscriptions_update(req) + assert resp.status == 200 + body = json.loads(resp.text) + assert body["status"] == "error" + + +@pytest.mark.asyncio +async def test_subscriptions_update_invalid_interval_returns_error_not_500(mock_dqueue): + req = _json_request({"id": "nonexistent", "check_interval_minutes": "abc"}) + resp = await main.subscriptions_update(req) + assert resp.status == 200 + body = json.loads(resp.text) + assert body["status"] == "error" + + def test_is_within_state_dir_blocks_state_subtree(): state_dir = main._STATE_DIR_REAL assert main._is_within_state_dir(state_dir) @@ -331,6 +402,11 @@ async def test_download_blocks_state_dir_files(monkeypatch): state_dir.mkdir(parents=True, exist_ok=True) (state_dir / "cookies.txt").write_text("# Netscape HTTP Cookie File\n", encoding="utf-8") (download_dir / "video.mp4").write_bytes(b"video") + # request.path is already percent-decoded by aiohttp; state_dir_guard must + # not decode it a second time, or a filename containing a literal '%' + # gets mangled into a false 404. + percent_filename = "100% done.mp4" + (download_dir / percent_filename).write_bytes(b"percent video") monkeypatch.setattr(main.config, "STATE_DIR", str(state_dir)) monkeypatch.setattr(main, "_STATE_DIR_REAL", os.path.realpath(str(state_dir))) @@ -343,7 +419,12 @@ async def test_download_blocks_state_dir_files(monkeypatch): allowed = await client.get("/download/video.mp4") assert allowed.status == 200 assert await allowed.read() == b"video" + + percent_resp = await client.get("/download/" + quote(percent_filename)) + assert percent_resp.status == 200 + assert await percent_resp.read() == b"percent video" finally: (state_dir / "cookies.txt").unlink(missing_ok=True) (download_dir / "video.mp4").unlink(missing_ok=True) + (download_dir / percent_filename).unlink(missing_ok=True) state_dir.rmdir() diff --git a/app/tests/test_bg_tasks.py b/app/tests/test_bg_tasks.py new file mode 100644 index 0000000..a9c1237 --- /dev/null +++ b/app/tests/test_bg_tasks.py @@ -0,0 +1,55 @@ +"""Tests for the ``bg_tasks.create_task`` strong-reference/logging helper.""" + +from __future__ import annotations + +import asyncio +import logging + +import pytest + +import bg_tasks + + +@pytest.mark.asyncio +async def test_create_task_removes_itself_from_registry_on_success(): + async def _ok(): + return 42 + + task = bg_tasks.create_task(_ok(), name="ok_task") + assert task in bg_tasks._TASKS + result = await task + assert result == 42 + assert task not in bg_tasks._TASKS + + +@pytest.mark.asyncio +async def test_create_task_logs_unhandled_exception(caplog): + async def _boom(): + raise ValueError("kaboom") + + with caplog.at_level(logging.ERROR, logger="bg_tasks"): + task = bg_tasks.create_task(_boom(), name="boom_task") + with pytest.raises(ValueError): + await task + # Let the done-callback (scheduled via call_soon) run. + await asyncio.sleep(0) + + assert task not in bg_tasks._TASKS + assert any("boom_task" in record.message for record in caplog.records) + + +@pytest.mark.asyncio +async def test_create_task_does_not_log_on_cancellation(caplog): + async def _sleep_forever(): + await asyncio.sleep(10) + + with caplog.at_level(logging.ERROR, logger="bg_tasks"): + task = bg_tasks.create_task(_sleep_forever(), name="cancel_task") + await asyncio.sleep(0) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + await asyncio.sleep(0) + + assert task not in bg_tasks._TASKS + assert not any("cancel_task" in record.message for record in caplog.records) diff --git a/app/tests/test_config.py b/app/tests/test_config.py index b918a93..5fed6fd 100644 --- a/app/tests/test_config.py +++ b/app/tests/test_config.py @@ -159,6 +159,35 @@ class ConfigTests(unittest.TestCase): c = Config() self.assertEqual(c.CLEAR_COMPLETED_AFTER, "0") + def test_invalid_default_option_playlist_item_limit_exits(self): + for bad in ("-1", "many"): + with patch.dict(os.environ, _base_env(DEFAULT_OPTION_PLAYLIST_ITEM_LIMIT=bad), clear=False): + with self.assertRaises(SystemExit): + Config() + + def test_default_option_playlist_item_limit_zero_allowed(self): + with patch.dict(os.environ, _base_env(DEFAULT_OPTION_PLAYLIST_ITEM_LIMIT="0"), clear=False): + c = Config() + self.assertEqual(c.DEFAULT_OPTION_PLAYLIST_ITEM_LIMIT, "0") + + def test_invalid_subscription_default_check_interval_exits(self): + for bad in ("0", "-1", "often"): + with patch.dict(os.environ, _base_env(SUBSCRIPTION_DEFAULT_CHECK_INTERVAL=bad), clear=False): + with self.assertRaises(SystemExit): + Config() + + def test_invalid_subscription_scan_playlist_end_exits(self): + for bad in ("0", "-1", "all"): + with patch.dict(os.environ, _base_env(SUBSCRIPTION_SCAN_PLAYLIST_END=bad), clear=False): + with self.assertRaises(SystemExit): + Config() + + def test_invalid_subscription_max_seen_ids_exits(self): + for bad in ("0", "-1", "unlimited"): + with patch.dict(os.environ, _base_env(SUBSCRIPTION_MAX_SEEN_IDS=bad), clear=False): + with self.assertRaises(SystemExit): + Config() + def test_runtime_override_roundtrip(self): with patch.dict(os.environ, _base_env(), clear=False): c = Config() diff --git a/app/tests/test_dl_formats.py b/app/tests/test_dl_formats.py index f1fbf48..07e26dd 100644 --- a/app/tests/test_dl_formats.py +++ b/app/tests/test_dl_formats.py @@ -10,6 +10,7 @@ from app.dl_formats import ( _normalize_subtitle_language, get_format, get_opts, + merge_ytdl_option_layers, ) @@ -118,7 +119,31 @@ class DlFormatsTests(unittest.TestCase): def test_get_opts_captions_txt_maps_to_srt_format(self): opts = get_opts("captions", "auto", "txt", "best", {}) - self.assertEqual(opts["subtitlesformat"], "srt") + self.assertEqual(opts["subtitlesformat"], "srt/best") + keys = [p["key"] for p in opts["postprocessors"]] + self.assertIn("FFmpegSubtitlesConvertor", keys) + convertor = next(p for p in opts["postprocessors"] if p["key"] == "FFmpegSubtitlesConvertor") + self.assertEqual(convertor["format"], "srt") + + def test_get_opts_captions_srt_guarantees_convertor(self): + opts = get_opts("captions", "auto", "srt", "best", {}) + self.assertEqual(opts["subtitlesformat"], "srt/best") + keys = [p["key"] for p in opts["postprocessors"]] + self.assertIn("FFmpegSubtitlesConvertor", keys) + + def test_get_opts_captions_vtt_guarantees_convertor(self): + opts = get_opts("captions", "auto", "vtt", "best", {}) + self.assertEqual(opts["subtitlesformat"], "vtt/best") + keys = [p["key"] for p in opts["postprocessors"]] + self.assertIn("FFmpegSubtitlesConvertor", keys) + convertor = next(p for p in opts["postprocessors"] if p["key"] == "FFmpegSubtitlesConvertor") + self.assertEqual(convertor["format"], "vtt") + + def test_get_opts_captions_ttml_has_no_convertor(self): + opts = get_opts("captions", "auto", "ttml", "best", {}) + self.assertEqual(opts["subtitlesformat"], "ttml/best") + keys = [p["key"] for p in opts["postprocessors"]] + self.assertNotIn("FFmpegSubtitlesConvertor", keys) def test_get_opts_merges_existing_postprocessors(self): opts = get_opts("audio", "auto", "opus", "best", {"postprocessors": [{"key": "SponsorBlock"}]}) @@ -135,5 +160,23 @@ class DlFormatsTests(unittest.TestCase): self.assertEqual(_normalize_subtitle_language(" "), "en") +class MergeYtdlOptionLayersTests(unittest.TestCase): + def test_presets_applied_in_order_then_overrides(self): + presets_config = { + "a": {"x": 1, "y": 1}, + "b": {"y": 2, "z": 2}, + } + merged = merge_ytdl_option_layers(["a", "b"], {"z": 3, "w": 4}, presets_config) + # b overrides a's y; explicit overrides win over presets. + self.assertEqual(merged, {"x": 1, "y": 2, "z": 3, "w": 4}) + + def test_no_base_options_included(self): + # The helper only produces the preset/override layer, never base opts. + self.assertEqual(merge_ytdl_option_layers(None, None, {}), {}) + + def test_unknown_preset_names_ignored(self): + self.assertEqual(merge_ytdl_option_layers(["missing"], {"a": 1}, {}), {"a": 1}) + + if __name__ == "__main__": unittest.main() diff --git a/app/tests/test_download_queue.py b/app/tests/test_download_queue.py index 2c0dceb..4b375f6 100644 --- a/app/tests/test_download_queue.py +++ b/app/tests/test_download_queue.py @@ -3,13 +3,14 @@ from __future__ import annotations import os +import re import tempfile from unittest.mock import AsyncMock, MagicMock, patch import pytest import time -from ytdl import DownloadInfo, DownloadQueue +from ytdl import Download, DownloadInfo, DownloadQueue @pytest.fixture @@ -46,6 +47,37 @@ def test_cancel_add_increments_generation(dq_env): assert dq._add_generation == before + 1 +def test_download_queue_has_dedicated_executor_sized_from_config(dq_env): + notifier = MagicMock() + dq = DownloadQueue(dq_env, notifier) + assert dq._download_executor is not None + assert dq._download_executor._max_workers == 2 * int(dq_env.MAX_CONCURRENT_DOWNLOADS) + 2 + dq.close() + + +def test_close_cancels_running_downloads_before_shutdown(dq_env): + notifier = MagicMock() + dq = DownloadQueue(dq_env, notifier) + + running = MagicMock() + running.started.return_value = True + running.running.return_value = True + idle = MagicMock() + idle.started.return_value = False + idle.running.return_value = False + + dq.queue.dict["u-running"] = running + dq.queue.dict["u-idle"] = idle + + dq.close() + + # The active download's subprocess group is killed; the not-started one is + # left alone. Executor is shut down afterwards. + running.cancel.assert_called_once() + idle.cancel.assert_not_called() + assert dq._download_executor._shutdown + + def test_get_returns_tuple_of_lists(dq_env): notifier = MagicMock() dq = DownloadQueue(dq_env, notifier) @@ -222,6 +254,58 @@ async def test_add_entry_queues_single_video_without_reextracting(dq_env): assert dq.pending.exists("https://example.com/watch?v=1") +@pytest.mark.asyncio +async def test_add_entry_duplicate_while_pending_is_skipped_not_clobbered(dq_env): + notifier = AsyncMock() + dq = DownloadQueue(dq_env, notifier) + entry = { + "_type": "video", + "id": "vid1", + "title": "Original Title", + "url": "https://example.com/watch?v=1", + "webpage_url": "https://example.com/watch?v=1", + } + + with patch.object(DownloadQueue, "_DownloadQueue__extract_info", side_effect=AssertionError("should not re-extract")): + first = await dq.add_entry(entry, "video", "auto", "any", "best", "", "", 0, auto_start=False) + assert first["status"] == "ok" + assert "msg" not in first + + dupe_entry = {**entry, "title": "Different Title"} + second = await dq.add_entry(dupe_entry, "audio", "auto", "mp3", "best", "", "", 0, auto_start=False) + + assert second["status"] == "ok" + assert "Already in queue" in second["msg"] + # The original pending download's options must survive untouched. + pending_dl = dq.pending.get("https://example.com/watch?v=1") + assert pending_dl.info.download_type == "video" + assert pending_dl.info.title == "Original Title" + + +@pytest.mark.asyncio +async def test_add_entry_duplicate_while_queued_is_skipped(dq_env): + notifier = AsyncMock() + dq = DownloadQueue(dq_env, notifier) + entry = { + "_type": "video", + "id": "vid1", + "title": "Test Video", + "url": "https://example.com/watch?v=1", + "webpage_url": "https://example.com/watch?v=1", + } + + with patch.object(DownloadQueue, "_DownloadQueue__extract_info", side_effect=AssertionError("should not re-extract")), \ + patch.object(DownloadQueue, "_DownloadQueue__start_download", new=AsyncMock()): + first = await dq.add_entry(entry, "video", "auto", "any", "best", "", "", 0, auto_start=True) + assert first["status"] == "ok" + assert dq.queue.exists("https://example.com/watch?v=1") + + second = await dq.add_entry(entry, "video", "auto", "any", "best", "", "", 0, auto_start=True) + + assert second["status"] == "ok" + assert "Already in queue" in second["msg"] + + @pytest.mark.asyncio async def test_channel_download_uses_output_template_when_channel_template_empty(dq_env): """Channel tabs reported as playlists must honor OUTPUT_TEMPLATE when OUTPUT_TEMPLATE_CHANNEL is empty.""" @@ -526,6 +610,9 @@ async def test_add_upcoming_stream_scheduled_without_starting(dq_env): assert download.info.live_release_timestamp is not None start_mock.assert_not_called() assert url in dq._scheduled_probe_at + # The "scheduled to start at ..." message must include a UTC offset + # (a naive datetime's %z would render as an empty string here). + assert re.search(r"[+-]\d{4}$", download.info.error) @pytest.mark.asyncio @@ -784,3 +871,78 @@ def test_download_info_to_public_dict_excludes_server_only_fields(): assert public["url"] == "https://example.com/watch?v=1" assert public["title"] == "Test Video" assert public["status"] == "pending" + + +def _make_download(dq_env, *, download_type="video", status="downloading", filename=None): + info = DownloadInfo( + id="id1", + title="t", + url="http://example.com/v", + quality="best", + download_type=download_type, + codec="auto", + format="any", + folder="", + custom_name_prefix="", + error=None, + entry=None, + playlist_item_limit=0, + split_by_chapters=False, + chapter_template="", + ) + info.status = status + info.filename = filename + info.size = 123 if filename else None + return Download( + dq_env.DOWNLOAD_DIR, dq_env.TEMP_DIR, "%(title)s.%(ext)s", "%(title)s.%(ext)s", "best", "any", {}, info + ) + + +@pytest.mark.asyncio +async def test_post_download_cleanup_clears_filename_on_error(dq_env): + notifier = AsyncMock() + dq = DownloadQueue(dq_env, notifier) + download = _make_download(dq_env, status="downloading", filename="../tmp/partial.mp4") + dq.queue.put(download) + + dq._post_download_cleanup(download) + + assert download.info.status == "error" + assert download.info.filename is None + assert download.info.size is None + + +@pytest.mark.asyncio +async def test_post_download_cleanup_keeps_captured_subtitles_on_error(dq_env): + notifier = AsyncMock() + dq = DownloadQueue(dq_env, notifier) + download = _make_download(dq_env, download_type="captions", status="downloading", filename="en.srt") + download.info.subtitle_files = [{"filename": "en.srt", "size": 42}] + dq.queue.put(download) + + dq._post_download_cleanup(download) + + assert download.info.status == "error" + assert download.info.filename == "en.srt" + + +@pytest.mark.asyncio +async def test_clear_skips_deletion_outside_download_directory(dq_env): + notifier = AsyncMock() + dq_env.DELETE_FILE_ON_TRASHCAN = True + dq = DownloadQueue(dq_env, notifier) + + outside_dir = tempfile.mkdtemp() + outside_file = os.path.join(outside_dir, "outside.txt") + with open(outside_file, "w") as f: + f.write("do not delete me") + + # A crafted/legacy relative filename that escapes DOWNLOAD_DIR via '..'. + escaping_filename = os.path.relpath(outside_file, dq_env.DOWNLOAD_DIR) + download = _make_download(dq_env, status="finished", filename=escaping_filename) + dq.done.put(download) + + await dq.clear([download.info.url]) + + assert os.path.exists(outside_file) + assert not dq.done.exists(download.info.url) diff --git a/app/tests/test_main_helpers.py b/app/tests/test_main_helpers.py index d39e54b..282e940 100644 --- a/app/tests/test_main_helpers.py +++ b/app/tests/test_main_helpers.py @@ -220,42 +220,61 @@ class ParseDownloadOptionsTests(unittest.TestCase): def test_clip_url_t_param_strips_query_and_sets_start(self): parsed = main.parse_download_options({ - "url": "https://example.com/watch?v=1&t=855s", + "url": "https://www.youtube.com/watch?v=1&t=855s", "download_type": "video", "codec": "auto", "format": "any", "quality": "best", }) - self.assertEqual(parsed["url"], "https://example.com/watch?v=1") + self.assertEqual(parsed["url"], "https://www.youtube.com/watch?v=1") self.assertEqual(parsed["clip_start"], 855.0) self.assertIsNone(parsed["clip_end"]) def test_clip_explicit_start_wins_over_url_t(self): parsed = main.parse_download_options({ - "url": "https://example.com/watch?v=1&t=100", + "url": "https://www.youtube.com/watch?v=1&t=100", "download_type": "video", "codec": "auto", "format": "any", "quality": "best", "clip_start": "50", }) - self.assertEqual(parsed["url"], "https://example.com/watch?v=1") + self.assertEqual(parsed["url"], "https://www.youtube.com/watch?v=1") self.assertEqual(parsed["clip_start"], 50.0) self.assertIsNone(parsed["clip_end"]) def test_clip_end_only_sets_start_zero_and_strips_url_t(self): parsed = main.parse_download_options({ - "url": "https://example.com/watch?v=1&t=999", + "url": "https://www.youtube.com/watch?v=1&t=999", "download_type": "video", "codec": "auto", "format": "any", "quality": "best", "clip_end": "60", }) - self.assertEqual(parsed["url"], "https://example.com/watch?v=1") + self.assertEqual(parsed["url"], "https://www.youtube.com/watch?v=1") self.assertEqual(parsed["clip_start"], 0.0) self.assertEqual(parsed["clip_end"], 60.0) + def test_clip_url_t_param_ignored_on_non_youtube_host(self): + # 't' is a generic query param name; only rewrite it on YouTube hosts + # so an unrelated site's URL isn't silently mutated with a bogus clip. + parsed = main.parse_download_options({ + "url": "https://example.com/watch?v=1&t=855s", + "download_type": "video", + "codec": "auto", + "format": "any", + "quality": "best", + }) + self.assertEqual(parsed["url"], "https://example.com/watch?v=1&t=855s") + self.assertIsNone(parsed["clip_start"]) + self.assertIsNone(parsed["clip_end"]) + + def test_extract_t_query_youtu_be_short_host(self): + cleaned, start = main._extract_t_query_from_url("https://youtu.be/abc123?t=90") + self.assertEqual(cleaned, "https://youtu.be/abc123") + self.assertEqual(start, 90.0) + def test_clip_rejects_end_before_start(self): with self.assertRaises(main.web.HTTPBadRequest): main.parse_download_options({ @@ -280,5 +299,17 @@ class ParseDownloadOptionsTests(unittest.TestCase): }) +class GetCustomDirsTests(unittest.TestCase): + def test_works_without_a_running_event_loop(self): + # get_custom_dirs() used to time its cache via + # asyncio.get_running_loop().time(), which raises RuntimeError outside + # a running loop (e.g. when called from a plain executor thread). It + # must work from a synchronous context too. + result = main.get_custom_dirs() + self.assertIn("download_dir", result) + self.assertIn("audio_download_dir", result) + self.assertIn("", result["download_dir"]) + + if __name__ == "__main__": unittest.main() diff --git a/app/tests/test_subscriptions.py b/app/tests/test_subscriptions.py index 525db11..0f97af1 100644 --- a/app/tests/test_subscriptions.py +++ b/app/tests/test_subscriptions.py @@ -1,10 +1,12 @@ from __future__ import annotations +import asyncio import json import os import shelve import sys import tempfile +import time import types import unittest from unittest.mock import patch @@ -29,6 +31,7 @@ sys.modules.setdefault("yt_dlp.networking", fake_networking) sys.modules.setdefault("yt_dlp.networking.impersonate", fake_impersonate) from subscriptions import ( + SubscriptionInfo, SubscriptionManager, _is_subscriber_only_entry, coerce_optional_bool, @@ -44,6 +47,7 @@ class _Config: self.DOWNLOAD_DIR = state_dir self.TEMP_DIR = state_dir self.YTDL_OPTIONS = {} + self.YTDL_OPTIONS_PRESETS = {} class _Queue: @@ -571,8 +575,16 @@ class SubscriptionPersistenceTests(unittest.IsolatedAsyncioTestCase): ) sub_id = result["subscription"]["id"] - with self.assertRaises(ValueError): - await mgr.update_subscription(sub_id, {"enabled": "maybe"}) + update_result = await mgr.update_subscription(sub_id, {"enabled": "maybe"}) + self.assertEqual(update_result["status"], "error") + stored = mgr.get(sub_id) + self.assertTrue(stored.enabled) + + update_result = await mgr.update_subscription( + sub_id, {"check_interval_minutes": "abc"} + ) + self.assertEqual(update_result["status"], "error") + self.assertEqual(mgr.get(sub_id).check_interval_minutes, 60) async def test_add_subscription_rejects_invalid_title_regex(self): with tempfile.TemporaryDirectory() as tmp: @@ -1012,6 +1024,223 @@ class ExtractFlatPlaylistTests(unittest.TestCase): self.assertEqual(info.get("_type"), "playlist") self.assertEqual([entry["webpage_url"] for entry in entries], ["https://example.com/v1"]) + def test_extra_opts_applied_on_top_of_config_options(self): + captured: dict = {} + + class _FakeYDL: + def __init__(self, params): + captured.update(params) + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return False + + def extract_info(self, url, download=False): + return {"_type": "video"} + + cfg = _Config(tempfile.mkdtemp()) + with patch("subscriptions.yt_dlp.YoutubeDL", _FakeYDL, create=True): + extract_flat_playlist(cfg, "https://example.com/v1", 50, extra_opts={"cookiefile": "x"}) + + self.assertEqual(captured.get("cookiefile"), "x") + + +def _make_scan_capturing_fake_ydl(captured_params: list, entries: list[dict]): + class _FakeYDL: + def __init__(self, params): + captured_params.append(params) + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return False + + def extract_info(self, url, download=False): + return {"_type": "channel", "title": "Channel", "entries": entries} + + return _FakeYDL + + +class SubscriptionScanExtraOptsTests(unittest.IsolatedAsyncioTestCase): + async def test_add_subscription_scan_applies_presets_and_overrides(self): + captured_params: list = [] + fake_ydl = _make_scan_capturing_fake_ydl( + captured_params, + [{"id": "v1", "title": "One", "webpage_url": "https://example.com/v1"}], + ) + + with tempfile.TemporaryDirectory() as tmp: + cfg = _Config(tmp) + cfg.YTDL_OPTIONS_PRESETS = {"mypreset": {"cookiefile": "preset.txt"}} + mgr = SubscriptionManager(cfg, _Queue(), _Notifier()) + + with patch("subscriptions.yt_dlp.YoutubeDL", fake_ydl, create=True): + await mgr.add_subscription( + "https://example.com/channel", + check_interval_minutes=60, + download_type="video", + codec="auto", + format="any", + quality="best", + folder="", + custom_name_prefix="", + auto_start=True, + playlist_item_limit=0, + split_by_chapters=False, + chapter_template="", + subtitle_language="en", + subtitle_mode="prefer_manual", + ytdl_options_presets=["mypreset"], + ytdl_options_overrides={"extra": "override"}, + ) + + self.assertTrue(captured_params) + self.assertEqual(captured_params[0].get("cookiefile"), "preset.txt") + self.assertEqual(captured_params[0].get("extra"), "override") + + async def test_check_now_scan_applies_stored_subscription_presets(self): + entries = [{"id": "v1", "title": "One", "webpage_url": "https://example.com/v1"}] + + with tempfile.TemporaryDirectory() as tmp: + cfg = _Config(tmp) + cfg.YTDL_OPTIONS_PRESETS = {"mypreset": {"cookiefile": "preset.txt"}} + mgr = SubscriptionManager(cfg, _Queue(), _Notifier()) + + add_captured: list = [] + with patch( + "subscriptions.yt_dlp.YoutubeDL", + _make_scan_capturing_fake_ydl(add_captured, entries), + create=True, + ): + result = await mgr.add_subscription( + "https://example.com/channel", + check_interval_minutes=60, + download_type="video", + codec="auto", + format="any", + quality="best", + folder="", + custom_name_prefix="", + auto_start=True, + playlist_item_limit=0, + split_by_chapters=False, + chapter_template="", + subtitle_language="en", + subtitle_mode="prefer_manual", + ytdl_options_presets=["mypreset"], + ) + sub_id = result["subscription"]["id"] + + check_captured: list = [] + with patch( + "subscriptions.yt_dlp.YoutubeDL", + _make_scan_capturing_fake_ydl(check_captured, entries), + create=True, + ): + await mgr.check_now([sub_id]) + + self.assertTrue(check_captured) + self.assertEqual(check_captured[0].get("cookiefile"), "preset.txt") + + +class SubscriptionEventLoopTests(unittest.IsolatedAsyncioTestCase): + async def test_check_now_does_not_block_event_loop(self): + with tempfile.TemporaryDirectory() as tmp: + queue = _Queue() + mgr = SubscriptionManager(_Config(tmp), queue, _Notifier()) + + with patch( + "subscriptions.extract_flat_playlist", + return_value=( + {"_type": "channel", "title": "Channel"}, + [{"id": "v1", "title": "One", "webpage_url": "https://example.com/v1"}], + ), + ): + result = await mgr.add_subscription( + "https://example.com/channel", + check_interval_minutes=60, + download_type="video", + codec="auto", + format="any", + quality="best", + folder="", + custom_name_prefix="", + auto_start=True, + playlist_item_limit=0, + split_by_chapters=False, + chapter_template="", + subtitle_language="en", + subtitle_mode="prefer_manual", + ) + sub_id = result["subscription"]["id"] + + def _slow_extract(config, url, playlistend, **kwargs): + time.sleep(0.3) + return ( + {"_type": "channel", "title": "Channel"}, + [{"id": "v1", "title": "One", "webpage_url": "https://example.com/v1"}], + ) + + with patch("subscriptions.extract_flat_playlist", side_effect=_slow_extract): + check_task = asyncio.ensure_future(mgr.check_now([sub_id])) + # If check_now() blocked the event loop, this would not complete + # until after the slow extraction finishes. + await asyncio.wait_for(asyncio.sleep(0.05), timeout=0.2) + self.assertFalse(check_task.done()) + await check_task + + async def test_check_many_isolates_a_crashing_subscription(self): + with tempfile.TemporaryDirectory() as tmp: + mgr = SubscriptionManager(_Config(tmp), _Queue(), _Notifier()) + + good = SubscriptionInfo(id="good", name="Good", url="https://example.com/good") + bad = SubscriptionInfo(id="bad", name="Bad", url="https://example.com/bad") + other = SubscriptionInfo(id="other", name="Other", url="https://example.com/other") + + checked: list[str] = [] + + async def fake_check(sub): + if sub.id == "bad": + raise RuntimeError("boom") + checked.append(sub.id) + + with patch.object(mgr, "_check_one_unlocked", side_effect=fake_check): + # The crashing subscription must not prevent the others running. + await mgr._check_many([good, bad, other]) + + self.assertIn("good", checked) + self.assertIn("other", checked) + + async def test_check_many_bounded_concurrency(self): + with tempfile.TemporaryDirectory() as tmp: + mgr = SubscriptionManager(_Config(tmp), _Queue(), _Notifier()) + subs = [ + SubscriptionInfo(id=str(i), name=str(i), url=f"https://example.com/{i}") + for i in range(10) + ] + + import subscriptions as subs_mod + + concurrent = 0 + peak = 0 + + async def fake_check(sub): + nonlocal concurrent, peak + concurrent += 1 + peak = max(peak, concurrent) + await asyncio.sleep(0.02) + concurrent -= 1 + + with patch.object(mgr, "_check_one_unlocked", side_effect=fake_check): + await mgr._check_many(subs) + + # Never exceed the configured bound, but do run more than one at once. + self.assertLessEqual(peak, subs_mod._MAX_CONCURRENT_CHECKS) + self.assertGreater(peak, 1) + if __name__ == "__main__": unittest.main() diff --git a/app/tests/test_ytdl_utils.py b/app/tests/test_ytdl_utils.py index 0f2e66d..36c2451 100644 --- a/app/tests/test_ytdl_utils.py +++ b/app/tests/test_ytdl_utils.py @@ -3,12 +3,14 @@ from __future__ import annotations import pickle +import signal import sys import tempfile import threading import types import unittest from pathlib import Path +from unittest.mock import MagicMock, patch fake_yt_dlp = types.ModuleType("yt_dlp") fake_networking = types.ModuleType("yt_dlp.networking") @@ -37,6 +39,7 @@ sys.modules.setdefault("yt_dlp.networking.impersonate", fake_impersonate) sys.modules.setdefault("yt_dlp.utils", fake_utils) from ytdl import ( + Download, DownloadInfo, _compact_persisted_entry, _convert_srt_to_txt_file, @@ -160,6 +163,105 @@ class SanitizeEntryForPickleTests(unittest.TestCase): self.assertEqual(out, {"z": 1, "a": 2}) +def _make_test_download() -> Download: + info = DownloadInfo( + id="id1", + title="t", + url="http://example.com/v", + quality="best", + download_type="video", + codec="auto", + format="any", + folder="", + custom_name_prefix="", + error=None, + entry=None, + playlist_item_limit=0, + split_by_chapters=False, + chapter_template="", + ) + return Download("/tmp", "/tmp", "%(title)s.%(ext)s", "%(title)s.%(ext)s", "best", "any", {}, info) + + +class ProgressThrottleTests(unittest.TestCase): + def test_downloading_ticks_are_throttled(self): + dl = _make_test_download() + forwarded = [] + dl.status_queue = types.SimpleNamespace(put=forwarded.append) + hook = dl._make_progress_hook() + + with patch("ytdl.time.monotonic", side_effect=[100.0, 100.1, 100.6, 100.7]): + hook({"status": "downloading", "downloaded_bytes": 1}) + hook({"status": "downloading", "downloaded_bytes": 2}) + hook({"status": "downloading", "downloaded_bytes": 3}) + hook({"status": "downloading", "downloaded_bytes": 4}) + + # Only the 1st and 3rd ticks are >= 0.5s apart from the last forwarded one. + self.assertEqual(len(forwarded), 2) + + def test_finished_and_error_statuses_always_forwarded(self): + dl = _make_test_download() + forwarded = [] + dl.status_queue = types.SimpleNamespace(put=forwarded.append) + hook = dl._make_progress_hook() + + with patch("ytdl.time.monotonic", side_effect=[200.0, 200.1]): + hook({"status": "downloading"}) + hook({"status": "finished"}) + hook({"status": "downloading"}) + hook({"status": "error", "msg": "boom"}) + + statuses = [item.get("status") for item in forwarded] + self.assertIn("finished", statuses) + self.assertIn("error", statuses) + + +class CancelProcessGroupTests(unittest.TestCase): + def test_cancel_kills_group_when_child_is_group_leader(self): + # Child successfully ran os.setpgrp(): its pgid equals its own pid. + dl = _make_test_download() + dl.proc = types.SimpleNamespace(pid=4321) + dl.status_queue = types.SimpleNamespace(put=lambda _item: None) + + with patch.object(Download, "running", return_value=True), \ + patch("ytdl.os.getpgid", return_value=4321) as mock_getpgid, \ + patch("ytdl.os.killpg") as mock_killpg: + dl.cancel() + + mock_getpgid.assert_called_once_with(4321) + mock_killpg.assert_called_once_with(4321, signal.SIGKILL) + self.assertTrue(dl.canceled) + + def test_cancel_does_not_killpg_parent_group_kills_child_only(self): + # Child has NOT become its own group leader yet (pgid != pid, e.g. it is + # still in the server's process group). killpg must NOT be called — that + # would SIGKILL the whole server — and we fall back to proc.kill(). + dl = _make_test_download() + dl.proc = types.SimpleNamespace(pid=4321, kill=MagicMock()) + dl.status_queue = types.SimpleNamespace(put=lambda _item: None) + + with patch.object(Download, "running", return_value=True), \ + patch("ytdl.os.getpgid", return_value=999), \ + patch("ytdl.os.killpg") as mock_killpg: + dl.cancel() + + mock_killpg.assert_not_called() + dl.proc.kill.assert_called_once() + self.assertTrue(dl.canceled) + + def test_cancel_falls_back_to_proc_kill_when_getpgid_unavailable(self): + dl = _make_test_download() + dl.proc = types.SimpleNamespace(pid=4321, kill=MagicMock()) + dl.status_queue = types.SimpleNamespace(put=lambda _item: None) + + with patch.object(Download, "running", return_value=True), \ + patch("ytdl.os.getpgid", side_effect=OSError("no such process")): + dl.cancel() + + dl.proc.kill.assert_called_once() + self.assertTrue(dl.canceled) + + class ConvertSrtToTxtTests(unittest.TestCase): def test_basic_conversion(self): srt = """1 @@ -180,6 +282,77 @@ Second line self.assertIn("Hello world", content) self.assertIn("Second line", content) + def test_vtt_input_strips_header_and_metadata(self): + # yt-dlp can fall back to VTT even when srt/txt was requested (the + # extractor may not offer a native srt track); the converter must not + # leak VTT-only header/metadata lines into the plain-text output. + vtt = """WEBVTT +Kind: captions +Language: en + +NOTE +This is a note block + +1 +00:00:01.000 --> 00:00:02.000 +Hello world + +2 +00:00:03.000 --> 00:00:04.000 +Second line +""" + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "sub.vtt" + path.write_text(vtt, encoding="utf-8") + txt_path = _convert_srt_to_txt_file(str(path)) + self.assertIsNotNone(txt_path) + content = Path(txt_path).read_text(encoding="utf-8") + self.assertIn("Hello world", content) + self.assertIn("Second line", content) + self.assertNotIn("WEBVTT", content) + self.assertNotIn("Kind:", content) + self.assertNotIn("Language:", content) + self.assertNotIn("This is a note block", content) + + def test_vtt_standalone_header_block_is_stripped(self): + # Some VTT files put a blank line after WEBVTT, so Kind:/Language: form + # their own block. That header block (before the first timed cue) must + # still be stripped. + vtt = """WEBVTT + +Kind: captions +Language: en + +1 +00:00:01.000 --> 00:00:02.000 +Hello world +""" + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "sub.vtt" + path.write_text(vtt, encoding="utf-8") + content = Path(_convert_srt_to_txt_file(str(path))).read_text(encoding="utf-8") + self.assertIn("Hello world", content) + self.assertNotIn("Kind:", content) + self.assertNotIn("Language:", content) + + def test_cue_text_starting_with_metadata_keyword_is_kept(self): + # A real caption line beginning with "Kind:"/"Language:" must NOT be + # dropped as if it were VTT header metadata. + srt = """1 +00:00:01,000 --> 00:00:02,000 +Kind: regards, everyone + +2 +00:00:03,000 --> 00:00:04,000 +Language: they spoke French +""" + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "sub.srt" + path.write_text(srt, encoding="utf-8") + content = Path(_convert_srt_to_txt_file(str(path))).read_text(encoding="utf-8") + self.assertIn("Kind: regards, everyone", content) + self.assertIn("Language: they spoke French", content) + class DownloadInfoSetstateTests(unittest.TestCase): def _base_state(self, **kwargs): diff --git a/app/ytdl.py b/app/ytdl.py index 207d35e..43a7772 100644 --- a/app/ytdl.py +++ b/app/ytdl.py @@ -9,21 +9,42 @@ from collections import OrderedDict import time import asyncio import multiprocessing +from concurrent.futures import ThreadPoolExecutor from functools import partial import logging import re +import signal +import sys import types from typing import Any, Optional import yt_dlp.networking.impersonate from yt_dlp.utils import STR_FORMAT_RE_TMPL, STR_FORMAT_TYPES -from dl_formats import get_format, get_opts, AUDIO_FORMATS +import bg_tasks +from dl_formats import get_format, get_opts, AUDIO_FORMATS, merge_ytdl_option_layers from datetime import datetime from state_store import AtomicJsonStore, from_json_compatible, read_legacy_shelf, to_json_compatible from subscriptions import _entry_id log = logging.getLogger('ytdl') +# Python 3.14 switches the default multiprocessing start method on Linux +# (this app's only supported deployment target, per the Dockerfile) from fork +# to forkserver. Download._download relies on inheriting process state the +# way fork provides, so pin it back on Linux specifically. +# +# This must NOT be widened to "prefer fork wherever available": on macOS the +# platform default has long been spawn (never fork) precisely because forking +# a multi-threaded parent is hazardous — inherited locks held by threads that +# vanish in the child can deadlock it silently before it does any work. This +# app creates background threads (executors, notifier callbacks) well before +# any download starts, so forcing fork there reproduces exactly that hazard. +_MP_CTX = ( + multiprocessing.get_context("fork") + if sys.platform.startswith("linux") and "fork" in multiprocessing.get_all_start_methods() + else multiprocessing.get_context() +) + _LIVE_CHECK_INTERVAL = 60 _LIVE_MAX_CHECK_INTERVAL = 3600 # Consecutive probe failures (network blips, rate limits, transient extractor @@ -31,6 +52,22 @@ _LIVE_MAX_CHECK_INTERVAL = 3600 _LIVE_PROBE_MAX_FAILURES = 5 +def _is_within_directory(real_base: str, real_target: str) -> bool: + """True if ``real_target`` is inside (or equal to) ``real_base``. + + Both arguments must already be resolved with ``os.path.realpath``. Uses + ``commonpath`` rather than ``startswith`` so a sibling directory sharing a + name prefix (e.g. base ``/downloads`` vs ``/downloads-secret``) cannot be + reached via ``../downloads-secret``. + """ + try: + return os.path.commonpath([real_base, real_target]) == real_base + except ValueError: + # Raised when paths are on different drives (Windows) or mix + # absolute/relative; treat as outside the base directory. + return False + + # Characters that are invalid in Windows/NTFS path components. These are pre- # sanitised when substituting playlist/channel titles into output templates so # that downloads do not fail on NTFS-mounted volumes or Windows Docker hosts. @@ -133,10 +170,28 @@ def _convert_srt_to_txt_file(subtitle_path: str): # Normalize newlines so cue splitting is consistent across platforms. content = content.replace("\r\n", "\n").replace("\r", "\n") cues = [] + # The caption format may resolve to a VTT file even when srt/txt was + # requested (yt-dlp falls back when the extractor doesn't offer the + # requested ext). VTT header metadata (WEBVTT/NOTE/STYLE and + # "Kind:"/"Language:" fields) only ever appears BEFORE the first timed + # cue, so only strip it while still in that header region — otherwise a + # real caption line like "Kind: regards" would be dropped as metadata. + seen_cue = False for block in re.split(r"\n{2,}", content): lines = [line.strip() for line in block.split("\n") if line.strip()] if not lines: continue + has_timing = any("-->" in line for line in lines) + if not seen_cue and not has_timing: + # Still in the header region: drop metadata-only blocks + # (WEBVTT/NOTE/STYLE, or blocks made entirely of "Key: value" + # header fields such as a standalone "Kind:/Language:" block). + if re.match(r"^(WEBVTT|NOTE|STYLE)\b", lines[0]) or all( + re.match(r"^[A-Za-z][\w-]*:\s", line) for line in lines + ): + continue + if has_timing: + seen_cue = True if re.fullmatch(r"\d+", lines[0]): lines = lines[1:] if lines and "-->" in lines[0]: @@ -442,23 +497,50 @@ class Download: self.proc = None self.loop = None self.notifier = None + self._executor = None + + # Minimum interval between forwarded 'downloading' progress ticks. yt-dlp + # emits these many times per second; without throttling, each active + # download broadcasts hundreds of socket.io events/sec to every client. + _PROGRESS_THROTTLE_SECONDS = 0.5 + + def _make_progress_hook(self): + last_forward = 0.0 + + def put_status(st): + nonlocal last_forward + if st.get('status') == 'downloading': + now = time.monotonic() + if now - last_forward < self._PROGRESS_THROTTLE_SECONDS: + return + last_forward = now + self.status_queue.put({k: v for k, v in st.items() if k in ( + 'tmpfilename', + 'filename', + 'status', + 'msg', + 'total_bytes', + 'total_bytes_estimate', + 'downloaded_bytes', + 'speed', + 'eta', + )}) + + return put_status def _download(self): + # Run in our own process group so cancel() can SIGKILL the whole + # group (yt-dlp + any ffmpeg children it spawned for merge/postproc), + # instead of orphaning ffmpeg when only the yt-dlp process is killed. + if hasattr(os, 'setpgrp'): + try: + os.setpgrp() + except OSError: + pass log.info(f"Starting download for: {self.info.title} ({self.info.url})") try: debug_logging = logging.getLogger().isEnabledFor(logging.DEBUG) - def put_status(st): - self.status_queue.put({k: v for k, v in st.items() if k in ( - 'tmpfilename', - 'filename', - 'status', - 'msg', - 'total_bytes', - 'total_bytes_estimate', - 'downloaded_bytes', - 'speed', - 'eta', - )}) + put_status = self._make_progress_hook() def put_status_postprocessor(d): if d['postprocessor'] == 'MoveFiles' and d['status'] == 'finished': @@ -530,19 +612,20 @@ class Download: log.error(f"Download error for {self.info.title}: {str(exc)}") self.status_queue.put({'status': 'error', 'msg': str(exc)}) - async def start(self, notifier): + async def start(self, notifier, executor=None): log.info(f"Preparing download for: {self.info.title}") if Download.manager is None: - Download.manager = multiprocessing.Manager() + Download.manager = _MP_CTX.Manager() self.status_queue = Download.manager.Queue() - self.proc = multiprocessing.Process(target=self._download) + self.proc = _MP_CTX.Process(target=self._download) self.proc.start() self.loop = asyncio.get_running_loop() self.notifier = notifier + self._executor = executor self.info.status = 'preparing' await self.notifier.updated(self.info) self.status_task = asyncio.create_task(self.update_status()) - await self.loop.run_in_executor(None, self.proc.join) + await self.loop.run_in_executor(self._executor, self.proc.join) # Signal update_status to stop and wait for it to finish # so that all status updates (including MoveFiles with correct # file size) are processed before _post_download_cleanup runs. @@ -553,10 +636,26 @@ class Download: def cancel(self): log.info(f"Cancelling download: {self.info.title}") if self.running(): + killed_group = False try: - self.proc.kill() - except Exception as e: - log.error(f"Error killing process for {self.info.title}: {e}") + pgid = os.getpgid(self.proc.pid) + # Only kill the whole group (yt-dlp + any ffmpeg children) when + # the child actually became its own group leader via + # os.setpgrp() in _download() — that sets its pgid equal to its + # own pid. If it hasn't run setpgrp() yet, or setpgrp() failed, + # its pgid is still the SERVER's group and killpg would SIGKILL + # the entire MeTube process (PID 1 in Docker). Fall back to + # killing just the child process by pid in that case. + if pgid == self.proc.pid: + os.killpg(pgid, signal.SIGKILL) + killed_group = True + except (OSError, AttributeError): + pass + if not killed_group: + try: + self.proc.kill() + except Exception as e: + log.error(f"Error killing process for {self.info.title}: {e}") self.canceled = True if self.status_queue is not None: self.status_queue.put(None) @@ -577,7 +676,13 @@ class Download: async def update_status(self): while True: - status = await self.loop.run_in_executor(None, self.status_queue.get) + try: + status = await self.loop.run_in_executor(self._executor, self.status_queue.get) + except RuntimeError: + # The download executor was shut down (server shutting down); + # stop polling instead of raising a noisy background-task error. + log.info(f"Status polling stopped (executor shut down) for: {self.info.title}") + return if status is None: log.info(f"Status update finished for: {self.info.title}") return @@ -771,10 +876,6 @@ class PersistentQueue: self.dict[key] = old raise - def next(self): - k, v = next(iter(self.dict.items())) - return k, v - def empty(self): return not bool(self.dict) @@ -787,6 +888,13 @@ class DownloadQueue: self.pending = PersistentQueue("pending", self.config.STATE_DIR + '/pending') self.active_downloads = set() self.semaphore = asyncio.Semaphore(int(self.config.MAX_CONCURRENT_DOWNLOADS)) + # Each active download parks two threads for its whole duration + # (proc.join + status_queue.get). A dedicated pool keeps those from + # starving the default executor, which extract_info/live-probes also use. + self._download_executor = ThreadPoolExecutor( + max_workers=2 * int(self.config.MAX_CONCURRENT_DOWNLOADS) + 2, + thread_name_prefix="dl", + ) self.done.load() self._add_generation = 0 self._canceled_urls = set() # URLs canceled during current playlist add @@ -820,18 +928,14 @@ class DownloadQueue: async def initialize(self): log.info("Initializing DownloadQueue") self._start_live_monitor() - asyncio.create_task(self.__import_queue()) - asyncio.create_task(self.__import_pending()) + bg_tasks.create_task(self.__import_queue(), name="import_queue") + bg_tasks.create_task(self.__import_pending(), name="import_pending") def _start_live_monitor(self) -> None: if self._live_monitor_task is not None and not self._live_monitor_task.done(): return - self._live_monitor_task = asyncio.create_task(self._live_monitor_loop()) - self._live_monitor_task.add_done_callback( - lambda t: log.error("Live monitor loop failed: %s", t.exception()) - if not t.cancelled() and t.exception() - else None - ) + # bg_tasks.create_task already logs unexpected task failures with the name. + self._live_monitor_task = bg_tasks.create_task(self._live_monitor_loop(), name="live_monitor") def _register_scheduled(self, download: Download) -> None: self._scheduled_probe_at[download.info.url] = 0 @@ -964,7 +1068,7 @@ class DownloadQueue: info.error = None info.msg = None await self.notifier.updated(info) - asyncio.create_task(self.__start_download(download)) + bg_tasks.create_task(self.__start_download(download), name="start_download") def _schedule_upcoming_download(self, download: Download) -> None: download.info.status = 'scheduled' @@ -976,7 +1080,7 @@ class DownloadQueue: download.info.status = 'pending' download.info.error = None download.info.msg = None - asyncio.create_task(self.__start_download(download)) + bg_tasks.create_task(self.__start_download(download), name="start_download") async def __start_download(self, download): if download.canceled: @@ -986,7 +1090,7 @@ class DownloadQueue: if download.canceled: log.info(f"Download {download.info.title} was canceled, skipping start.") return - await download.start(self.notifier) + await download.start(self.notifier, self._download_executor) self._post_download_cleanup(download) def _post_download_cleanup(self, download): @@ -997,22 +1101,35 @@ class DownloadQueue: except OSError: pass download.info.status = 'error' + # A progress tick may have set filename to a temp-directory + # relative path before the error occurred; clear it so the UI + # doesn't render a broken link (or, worse, so a later trashcan + # delete doesn't act on a path outside the download directory). + # Captions downloads may still have captured valid subtitle + # files even when the overall status is 'error' — keep those. + has_captured_subtitles = bool(getattr(download.info, 'subtitle_files', None)) + if not (download.info.download_type == 'captions' and has_captured_subtitles): + download.info.filename = None + download.info.size = None download.close() if self.queue.exists(download.info.url): self.queue.delete(download.info.url) if download.canceled: - asyncio.create_task(self.notifier.canceled(download.info.url)) + bg_tasks.create_task(self.notifier.canceled(download.info.url), name="notify_canceled") else: self.done.put(download) - asyncio.create_task(self.notifier.completed(download.info)) + bg_tasks.create_task(self.notifier.completed(download.info), name="notify_completed") try: clear_after = int(self.config.CLEAR_COMPLETED_AFTER) except ValueError: log.error(f'CLEAR_COMPLETED_AFTER is set to an invalid value "{self.config.CLEAR_COMPLETED_AFTER}", expected an integer number of seconds') clear_after = 0 if clear_after > 0: - task = asyncio.create_task(self.__auto_clear_after_delay(download.info.url, clear_after)) - task.add_done_callback(lambda t: log.error(f'Auto-clear task failed: {t.exception()}') if not t.cancelled() and t.exception() else None) + # bg_tasks.create_task already logs unexpected task failures. + bg_tasks.create_task( + self.__auto_clear_after_delay(download.info.url, clear_after), + name="auto_clear", + ) async def __auto_clear_after_delay(self, url, delay_seconds): await asyncio.sleep(delay_seconds) @@ -1023,9 +1140,9 @@ class DownloadQueue: def _build_ytdl_options(self, ytdl_options_presets=None, ytdl_options_overrides=None): """Merge global options, presets (in order), and per-download overrides.""" opts = dict(self.config.YTDL_OPTIONS) - for preset_name in ytdl_options_presets or []: - opts.update(self.config.YTDL_OPTIONS_PRESETS.get(preset_name, {})) - opts.update(ytdl_options_overrides or {}) + opts.update(merge_ytdl_option_layers( + ytdl_options_presets, ytdl_options_overrides, self.config.YTDL_OPTIONS_PRESETS + )) return opts def __extract_info(self, url, ytdl_options_presets=None, ytdl_options_overrides=None): @@ -1053,16 +1170,7 @@ class DownloadQueue: return None, {'status': 'error', 'msg': 'A folder for the download was specified but CUSTOM_DIRS is not true in the configuration.'} dldirectory = os.path.realpath(os.path.join(base_directory, folder)) real_base_directory = os.path.realpath(base_directory) - # Use commonpath rather than startswith so that a sibling directory - # sharing a name prefix (e.g. base "/downloads" vs "/downloads-secret") - # cannot be reached via "../downloads-secret". - try: - inside_base = os.path.commonpath([real_base_directory, dldirectory]) == real_base_directory - except ValueError: - # Raised when paths are on different drives (Windows) or mix - # absolute/relative; treat as outside the base directory. - inside_base = False - if not inside_base: + if not _is_within_directory(real_base_directory, dldirectory): return None, {'status': 'error', 'msg': f'Folder "{folder}" must resolve inside the base download directory "{real_base_directory}"'} if not os.path.isdir(dldirectory): if not self.config.CREATE_CUSTOM_DIRS: @@ -1107,7 +1215,7 @@ class DownloadQueue: self._schedule_upcoming_download(download) else: self.queue.put(download) - asyncio.create_task(self.__start_download(download)) + bg_tasks.create_task(self.__start_download(download), name="start_download") else: self.pending.put(download) await self.notifier.added(dl) @@ -1139,7 +1247,9 @@ class DownloadQueue: error = None if "live_status" in entry and "release_timestamp" in entry and entry.get("live_status") == "is_upcoming": - dt_ts = datetime.fromtimestamp(entry.get("release_timestamp")).strftime('%Y-%m-%d %H:%M:%S %z') + # astimezone() makes this an aware datetime in the server's local + # zone; a naive datetime's %z renders as an empty string. + dt_ts = datetime.fromtimestamp(entry.get("release_timestamp")).astimezone().strftime('%Y-%m-%d %H:%M:%S %z') error = f"Live stream is scheduled to start at {dt_ts}" else: if "msg" in entry: @@ -1237,38 +1347,43 @@ class DownloadQueue: if any(res['status'] == 'error' for res in results): return {'status': 'error', 'msg': ', '.join(res['msg'] for res in results if res['status'] == 'error' and 'msg' in res)} return {'status': 'ok'} - elif etype == 'video' or (etype.startswith('url') and 'id' in entry and 'title' in entry): + elif etype == 'video': log.debug('Processing as a video') key = entry.get('webpage_url') or entry['url'] if key in self._canceled_urls: log.info(f'Skipping canceled URL: {entry.get("title") or key}') return {'status': 'ok'} - if not self.queue.exists(key): - dl = DownloadInfo( - id=entry['id'], - title=entry.get('title') or entry['id'], - url=key, - quality=quality, - download_type=download_type, - codec=codec, - format=format, - folder=folder, - custom_name_prefix=custom_name_prefix, - error=error, - entry=entry, - playlist_item_limit=playlist_item_limit, - split_by_chapters=split_by_chapters, - chapter_template=chapter_template, - subtitle_language=subtitle_language, - subtitle_mode=subtitle_mode, - ytdl_options_presets=ytdl_options_presets, - ytdl_options_overrides=ytdl_options_overrides, - clip_start=clip_start, - clip_end=clip_end, - live_status=entry.get('live_status'), - live_release_timestamp=entry.get('release_timestamp'), - ) - await self.__add_download(dl, auto_start) + if self.queue.exists(key) or self.pending.exists(key): + # Surface the skip instead of silently no-op'ing, and avoid + # clobbering an existing pending entry's options with a + # fresh DownloadInfo built from possibly-different args. + title = entry.get('title') or key + return {'status': 'ok', 'msg': f'Already in queue: {title}'} + dl = DownloadInfo( + id=entry['id'], + title=entry.get('title') or entry['id'], + url=key, + quality=quality, + download_type=download_type, + codec=codec, + format=format, + folder=folder, + custom_name_prefix=custom_name_prefix, + error=error, + entry=entry, + playlist_item_limit=playlist_item_limit, + split_by_chapters=split_by_chapters, + chapter_template=chapter_template, + subtitle_language=subtitle_language, + subtitle_mode=subtitle_mode, + ytdl_options_presets=ytdl_options_presets, + ytdl_options_overrides=ytdl_options_overrides, + clip_start=clip_start, + clip_end=clip_end, + live_status=entry.get('live_status'), + live_release_timestamp=entry.get('release_timestamp'), + ) + await self.__add_download(dl, auto_start) return {'status': 'ok'} return {'status': 'error', 'msg': f'Unsupported resource "{etype}"'} @@ -1394,7 +1509,7 @@ class DownloadQueue: self._schedule_upcoming_download(dl) else: self.queue.put(dl) - asyncio.create_task(self.__start_download(dl)) + bg_tasks.create_task(self.__start_download(dl), name="start_download") continue if self.queue.exists(id): dl = self.queue.get(id) @@ -1448,9 +1563,14 @@ class DownloadQueue: for extra in (getattr(dl.info, 'subtitle_files', None) or []): if isinstance(extra, dict) and extra.get('filename'): rel_names.append(extra['filename']) + real_base_directory = os.path.realpath(dldirectory) for rel_name in rel_names: + full_path = os.path.realpath(os.path.join(dldirectory, rel_name)) + if not _is_within_directory(real_base_directory, full_path): + log.warning(f'skipping deletion of "{rel_name}" for download {id}: resolves outside the download directory') + continue try: - os.remove(os.path.join(dldirectory, rel_name)) + os.remove(full_path) except FileNotFoundError: pass except OSError as e: @@ -1463,3 +1583,13 @@ class DownloadQueue: return (list((k, v.info) for k, v in self.queue.items()) + list((k, v.info) for k, v in self.pending.items()), list((k, v.info) for k, v in self.done.items())) + + def close(self): + # Kill any still-running download subprocesses (and their ffmpeg + # children) before tearing down the executor, so they aren't orphaned + # when the server exits. Their queue entries stay persisted and are + # re-imported/restarted on next startup. + for _key, download in list(self.queue.items()): + if download.started() and download.running(): + download.cancel() + self._download_executor.shutdown(wait=False, cancel_futures=True) diff --git a/ui/src/app/app.html b/ui/src/app/app.html index 1299b05..5e875f3 100644 --- a/ui/src/app/app.html +++ b/ui/src/app/app.html @@ -699,7 +699,7 @@ - @for (download of downloads.queue | keyvalue: asIsOrder; track download.value.id) { + @for (download of downloads.queue | keyvalue: asIsOrder; track download.key) { @@ -769,7 +769,7 @@ - @for (entry of cachedSortedDone; track entry[1].id) { + @for (entry of cachedSortedDone; track entry[0]) { diff --git a/ui/src/app/app.spec.ts b/ui/src/app/app.spec.ts index be19b65..c1f61a5 100644 --- a/ui/src/app/app.spec.ts +++ b/ui/src/app/app.spec.ts @@ -139,6 +139,12 @@ describe('App', () => { expect(app).toBeTruthy(); }); + it('asIsOrder returns a stable comparator value (insertion order preserved)', () => { + const fixture = TestBed.createComponent(App); + const app = fixture.componentInstance; + expect(app.asIsOrder()).toBe(0); + }); + it('hides manual override input when disabled', () => { const fixture = TestBed.createComponent(App); fixture.componentInstance.isAdvancedOpen = true; diff --git a/ui/src/app/app.ts b/ui/src/app/app.ts index e48161e..67bd702 100644 --- a/ui/src/app/app.ts +++ b/ui/src/app/app.ts @@ -351,13 +351,10 @@ export class App implements AfterViewInit, OnInit, OnDestroy { this.colorSchemeMediaQuery.removeEventListener('change', this.onColorSchemeChanged); } - // workaround to allow fetching of Map values in the order they were inserted - // https://github.com/angular/angular/issues/31420 - - - + // keyvalue comparator that preserves insertion order (Angular's keyvalue + // pipe sorts by key by default): https://github.com/angular/angular/issues/31420 asIsOrder() { - return 1; + return 0; } qualityChanged() { @@ -430,8 +427,8 @@ export class App implements AfterViewInit, OnInit, OnDestroy { this.downloads.configurationChanged.pipe(takeUntilDestroyed(this.destroyRef)).subscribe({ // eslint-disable-next-line @typescript-eslint/no-explicit-any next: (config: any) => { - const playlistItemLimit = config['DEFAULT_OPTION_PLAYLIST_ITEM_LIMIT']; - if (playlistItemLimit !== '0') { + const playlistItemLimit = parseInt(String(config['DEFAULT_OPTION_PLAYLIST_ITEM_LIMIT'] ?? '0'), 10); + if (!Number.isNaN(playlistItemLimit) && playlistItemLimit > 0) { this.playlistItemLimit = playlistItemLimit; } // Set chapter template from backend config if not already set by cookie @@ -526,6 +523,14 @@ export class App implements AfterViewInit, OnInit, OnDestroy { return status?.status === 'error' ? status.msg || null : null; } + private handleActionResult(res: unknown, fallbackMsg: string) { + const error = this.getStatusError(res); + if (error) { + this.toasts.error(error || fallbackMsg); + } + this.cdr.markForCheck(); + } + private refreshSubscriptionsWithAlert() { this.subscriptionsSvc.refreshList().pipe(takeUntilDestroyed(this.destroyRef)).subscribe((refreshRes) => { const error = this.getStatusError(refreshRes); @@ -1085,6 +1090,10 @@ export class App implements AfterViewInit, OnInit, OnDestroy { if (status.status === 'error' && !this.cancelRequested) { this.toasts.error(`Error adding URL: ${status.msg}`); } else if (status.status !== 'error') { + // e.g. "Already in queue: ..." when the backend skipped a duplicate. + if (status.msg) { + this.toasts.info(status.msg); + } this.addUrl = ''; } this.resetAddState(); @@ -1113,7 +1122,7 @@ export class App implements AfterViewInit, OnInit, OnDestroy { } downloadItemByKey(id: string) { - this.downloads.startById([id]).subscribe(); + this.downloads.startById([id]).subscribe((res) => this.handleActionResult(res, 'Start download failed')); } liveCountdownSeconds(download: Download): number | null { @@ -1137,7 +1146,7 @@ export class App implements AfterViewInit, OnInit, OnDestroy { } retryDownload(key: string, download: Download) { - this.addDownload({ + const payload = this.buildAddPayload({ url: download.url, downloadType: download.download_type, codec: download.codec, @@ -1158,27 +1167,38 @@ export class App implements AfterViewInit, OnInit, OnDestroy { clipStart: download.clip_start != null ? String(download.clip_start) : '', clipEnd: download.clip_end != null ? String(download.clip_end) : '', }); - this.downloads.delById('done', [key]).subscribe(); + // Only remove the done-list record once the retry is confirmed queued — + // deleting it eagerly would silently lose history if the re-add fails. + this.downloads.add(payload) + .pipe(takeUntilDestroyed(this.destroyRef)) + .subscribe((status: Status) => { + if (status.status === 'error') { + this.toasts.error(`Error retrying ${download.title}: ${status.msg}`); + this.cdr.markForCheck(); + return; + } + this.downloads.delById('done', [key]).subscribe(); + }); } delDownload(where: State, id: string) { - this.downloads.delById(where, [id]).subscribe(); + this.downloads.delById(where, [id]).subscribe((res) => this.handleActionResult(res, 'Delete failed')); } startSelectedDownloads(where: State){ - this.downloads.startByFilter(where, dl => !!dl.checked).subscribe(); + this.downloads.startByFilter(where, dl => !!dl.checked).subscribe((res) => this.handleActionResult(res, 'Start download failed')); } delSelectedDownloads(where: State) { - this.downloads.delByFilter(where, dl => !!dl.checked).subscribe(); + this.downloads.delByFilter(where, dl => !!dl.checked).subscribe((res) => this.handleActionResult(res, 'Delete failed')); } clearCompletedDownloads() { - this.downloads.delByFilter('done', dl => dl.status === 'finished').subscribe(); + this.downloads.delByFilter('done', dl => dl.status === 'finished').subscribe((res) => this.handleActionResult(res, 'Clear completed failed')); } clearFailedDownloads() { - this.downloads.delByFilter('done', dl => dl.status === 'error').subscribe(); + this.downloads.delByFilter('done', dl => dl.status === 'error').subscribe((res) => this.handleActionResult(res, 'Clear failed downloads failed')); } retryFailedDownloads() { diff --git a/ui/src/app/pipes/file-size.pipe.spec.ts b/ui/src/app/pipes/file-size.pipe.spec.ts index 06b649a..15d20b0 100644 --- a/ui/src/app/pipes/file-size.pipe.spec.ts +++ b/ui/src/app/pipes/file-size.pipe.spec.ts @@ -10,15 +10,16 @@ describe('FileSizePipe', () => { it('formats bytes and larger units', () => { const pipe = new FileSizePipe(); expect(pipe.transform(500)).toContain('Bytes'); - expect(pipe.transform(1000)).toContain('KB'); - expect(pipe.transform(1000 * 1000)).toContain('MB'); - expect(pipe.transform(1000 ** 3)).toContain('GB'); + expect(pipe.transform(1000)).toContain('Bytes'); + expect(pipe.transform(1024)).toContain('KB'); + expect(pipe.transform(1024 ** 2)).toContain('MB'); + expect(pipe.transform(1024 ** 3)).toContain('GB'); }); it('handles boundaries between units', () => { const pipe = new FileSizePipe(); - expect(pipe.transform(999)).toContain('Bytes'); - expect(pipe.transform(1000)).toContain('KB'); - expect(pipe.transform(1001)).toContain('KB'); + expect(pipe.transform(1023)).toContain('Bytes'); + expect(pipe.transform(1024)).toContain('KB'); + expect(pipe.transform(1025)).toContain('KB'); }); }); diff --git a/ui/src/app/pipes/file-size.pipe.ts b/ui/src/app/pipes/file-size.pipe.ts index 7e8e993..e87b23e 100644 --- a/ui/src/app/pipes/file-size.pipe.ts +++ b/ui/src/app/pipes/file-size.pipe.ts @@ -8,9 +8,10 @@ export class FileSizePipe implements PipeTransform { if (isNaN(value) || value === 0) return '0 Bytes'; const units = ['Bytes', 'KB', 'MB', 'GB', 'TB', 'PB', 'EB', 'ZB', 'YB']; - const unitIndex = Math.floor(Math.log(value) / Math.log(1000)); // Use 1000 for common units + const k = 1024; // Matches SpeedPipe's base so file sizes and transfer speeds agree. + const unitIndex = Math.floor(Math.log(value) / Math.log(k)); - const unitValue = value / Math.pow(1000, unitIndex); + const unitValue = value / Math.pow(k, unitIndex); return `${unitValue.toFixed(2)} ${units[unitIndex]}`; } } \ No newline at end of file diff --git a/ui/src/app/services/downloads.service.spec.ts b/ui/src/app/services/downloads.service.spec.ts index 2fbc4be..897ce51 100644 --- a/ui/src/app/services/downloads.service.spec.ts +++ b/ui/src/app/services/downloads.service.spec.ts @@ -159,6 +159,61 @@ describe('DownloadsService', () => { req.flush({}); }); + it('delById resets deleting flag and emits error status on HTTP failure', () => { + const dl: Download = { + id: '1', + title: 't', + url: 'u1', + download_type: 'video', + quality: 'best', + format: 'any', + folder: '', + custom_name_prefix: '', + playlist_item_limit: 0, + status: 'finished', + msg: '', + percent: 0, + speed: 0, + eta: 0, + filename: '', + checked: false, + deleting: false, + }; + service.queue.set('u1', dl); + let queueChangedCount = 0; + service.queueChanged.subscribe(() => queueChangedCount++); + let result: unknown; + let threw = false; + + service.delById('queue', ['u1']).subscribe({ + next: (res) => { result = res; }, + error: () => { threw = true; }, + }); + expect(dl.deleting).toBe(true); + const req = httpMock.expectOne('delete'); + req.flush({ msg: 'boom' }, { status: 500, statusText: 'Server Error' }); + + expect(threw).toBe(false); + expect(dl.deleting).toBe(false); + expect(queueChangedCount).toBeGreaterThan(0); + expect((result as { status: string }).status).toBe('error'); + }); + + it('startById surfaces HTTP errors as a status object instead of throwing', () => { + let result: unknown; + let threw = false; + + service.startById(['a']).subscribe({ + next: (res) => { result = res; }, + error: () => { threw = true; }, + }); + const req = httpMock.expectOne('start'); + req.flush({ msg: 'nope' }, { status: 500, statusText: 'Server Error' }); + + expect(threw).toBe(false); + expect((result as { status: string }).status).toBe('error'); + }); + it('handleHTTPError extracts msg from object body', async () => { const err = new HttpErrorResponse({ error: { msg: 'bad' }, @@ -226,6 +281,15 @@ describe('DownloadsService', () => { expect(updated?.deleting).toBe(true); }); + it('socket updated ignores events for urls not already in the queue', () => { + expect(service.queue.has('unknown-url')).toBe(false); + socket.emit( + 'updated', + JSON.stringify({ url: 'unknown-url', title: 't', status: 'downloading' }), + ); + expect(service.queue.has('unknown-url')).toBe(false); + }); + it('socket completed moves entry to done', () => { service.queue.set('u1', { id: '1', diff --git a/ui/src/app/services/downloads.service.ts b/ui/src/app/services/downloads.service.ts index 60c3df6..8579cc8 100644 --- a/ui/src/app/services/downloads.service.ts +++ b/ui/src/app/services/downloads.service.ts @@ -69,8 +69,14 @@ export class DownloadsService { .subscribe((strdata: string) => { const data: Download = JSON.parse(strdata); const dl: Download | undefined = this.queue.get(data.url); - data.checked = !!dl?.checked; - data.deleting = !!dl?.deleting; + // An 'added' event always precedes legitimate updates. If the row is + // gone (canceled/completed already processed), this update is stale — + // applying it would resurrect a ghost row until the next full refresh. + if (!dl) { + return; + } + data.checked = !!dl.checked; + data.deleting = !!dl.deleting; this.queue.set(data.url, data); this.updated.next(); }); @@ -164,7 +170,9 @@ export class DownloadsService { } public startById(ids: string[]) { - return this.http.post('start', {ids: ids}); + return this.http.post('start', {ids: ids}).pipe( + catchError(this.handleHTTPError) + ); } public delById(where: State, ids: string[]) { @@ -177,7 +185,22 @@ export class DownloadsService { } } } - return this.http.post('delete', {where: where, ids: ids}); + return this.http.post('delete', {where: where, ids: ids}).pipe( + catchError((err: HttpErrorResponse) => { + // Request failed — the rows would otherwise stay disabled forever + // with no way to retry, since nothing ever clears `deleting`. + if (map) { + for (const id of ids) { + const obj = map.get(id); + if (obj) { + obj.deleting = false; + } + } + } + (where === 'queue' ? this.queueChanged : this.doneChanged).next(); + return this.handleHTTPError(err); + }) + ); } public startByFilter(where: State, filter: (dl: Download) => boolean) {