mirror of
https://github.com/alexta69/metube.git
synced 2026-07-17 16:47:30 +00:00
3ea4732c5d
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 <noreply@anthropic.com>
231 lines
8.9 KiB
Python
231 lines
8.9 KiB
Python
"""Tests for ``Config`` (env parsing, yt-dlp options, frontend_safe)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from main import Config
|
|
|
|
|
|
def _base_env(**overrides: str) -> dict[str, str]:
|
|
env = {k: str(v) for k, v in Config._DEFAULTS.items()}
|
|
env.update(overrides)
|
|
return env
|
|
|
|
|
|
class ConfigTests(unittest.TestCase):
|
|
def test_url_prefix_gets_trailing_slash(self):
|
|
with patch.dict(os.environ, _base_env(URL_PREFIX="foo"), clear=False):
|
|
c = Config()
|
|
self.assertEqual(c.URL_PREFIX, "foo/")
|
|
|
|
def test_public_host_url_gets_trailing_slash(self):
|
|
with patch.dict(
|
|
os.environ,
|
|
_base_env(PUBLIC_HOST_URL="https://ytdl.example.com"),
|
|
clear=False,
|
|
):
|
|
c = Config()
|
|
self.assertEqual(c.PUBLIC_HOST_URL, "https://ytdl.example.com/")
|
|
|
|
def test_public_host_audio_url_gets_trailing_slash(self):
|
|
with patch.dict(
|
|
os.environ,
|
|
_base_env(PUBLIC_HOST_AUDIO_URL="https://audio.example.com"),
|
|
clear=False,
|
|
):
|
|
c = Config()
|
|
self.assertEqual(c.PUBLIC_HOST_AUDIO_URL, "https://audio.example.com/")
|
|
|
|
def test_public_host_url_empty_stays_empty(self):
|
|
with patch.dict(
|
|
os.environ,
|
|
_base_env(PUBLIC_HOST_URL="", PUBLIC_HOST_AUDIO_URL=""),
|
|
clear=False,
|
|
):
|
|
c = Config()
|
|
self.assertEqual(c.PUBLIC_HOST_URL, "")
|
|
self.assertEqual(c.PUBLIC_HOST_AUDIO_URL, "")
|
|
|
|
def test_blank_audio_host_falls_back_to_audio_download_route(self):
|
|
# Regression: a present-but-blank PUBLIC_HOST_AUDIO_URL must not stay empty
|
|
# (which produced root-relative, 404ing audio links). It falls back to the
|
|
# 'audio_download/' route that serves AUDIO_DOWNLOAD_DIR.
|
|
with patch.dict(
|
|
os.environ,
|
|
_base_env(PUBLIC_HOST_URL="https://ytdl.example.com", PUBLIC_HOST_AUDIO_URL=""),
|
|
clear=False,
|
|
):
|
|
c = Config()
|
|
self.assertEqual(c.PUBLIC_HOST_URL, "https://ytdl.example.com/")
|
|
self.assertEqual(c.PUBLIC_HOST_AUDIO_URL, "audio_download/")
|
|
|
|
def test_public_host_url_already_slashed_unchanged(self):
|
|
with patch.dict(
|
|
os.environ,
|
|
_base_env(
|
|
PUBLIC_HOST_URL="https://ytdl.example.com/",
|
|
PUBLIC_HOST_AUDIO_URL="https://audio.example.com/",
|
|
),
|
|
clear=False,
|
|
):
|
|
c = Config()
|
|
self.assertEqual(c.PUBLIC_HOST_URL, "https://ytdl.example.com/")
|
|
self.assertEqual(c.PUBLIC_HOST_AUDIO_URL, "https://audio.example.com/")
|
|
|
|
def test_ytdl_options_json_loaded(self):
|
|
opts = {"quiet": True, "no_warnings": True}
|
|
with patch.dict(
|
|
os.environ,
|
|
_base_env(YTDL_OPTIONS=json.dumps(opts)),
|
|
clear=False,
|
|
):
|
|
c = Config()
|
|
self.assertEqual(c.YTDL_OPTIONS["quiet"], True)
|
|
|
|
def test_ytdl_option_presets_json_loaded(self):
|
|
presets = {"Audio extras": {"embed_thumbnail": True}}
|
|
with patch.dict(
|
|
os.environ,
|
|
_base_env(YTDL_OPTIONS_PRESETS=json.dumps(presets)),
|
|
clear=False,
|
|
):
|
|
c = Config()
|
|
self.assertEqual(c.YTDL_OPTIONS_PRESETS["Audio extras"]["embed_thumbnail"], True)
|
|
|
|
def test_invalid_ytdl_options_exits(self):
|
|
with patch.dict(os.environ, _base_env(YTDL_OPTIONS="not-json"), clear=False):
|
|
with self.assertRaises(SystemExit):
|
|
Config()
|
|
|
|
def test_invalid_boolean_env_exits(self):
|
|
with patch.dict(os.environ, _base_env(CUSTOM_DIRS="maybe"), clear=False):
|
|
with self.assertRaises(SystemExit):
|
|
Config()
|
|
|
|
def test_frontend_safe_excludes_secrets(self):
|
|
with patch.dict(os.environ, _base_env(), clear=False):
|
|
c = Config()
|
|
safe = c.frontend_safe()
|
|
self.assertNotIn("YTDL_OPTIONS", safe)
|
|
self.assertNotIn("HOST", safe)
|
|
self.assertEqual(safe["ALLOW_YTDL_OPTIONS_OVERRIDES"], False)
|
|
|
|
def test_allow_ytdl_options_overrides_boolean_loaded(self):
|
|
with patch.dict(os.environ, _base_env(ALLOW_YTDL_OPTIONS_OVERRIDES="true"), clear=False):
|
|
c = Config()
|
|
self.assertTrue(c.ALLOW_YTDL_OPTIONS_OVERRIDES)
|
|
|
|
def test_ytdl_nightly_update_time_empty_default(self):
|
|
with patch.dict(os.environ, _base_env(YTDL_NIGHTLY_UPDATE_TIME=""), clear=False):
|
|
c = Config()
|
|
self.assertEqual(c.YTDL_NIGHTLY_UPDATE_TIME, "")
|
|
|
|
def test_ytdl_nightly_update_time_valid(self):
|
|
with patch.dict(os.environ, _base_env(YTDL_NIGHTLY_UPDATE_TIME="04:00"), clear=False):
|
|
c = Config()
|
|
self.assertEqual(c.YTDL_NIGHTLY_UPDATE_TIME, "04:00")
|
|
|
|
def test_ytdl_nightly_update_time_invalid_exits(self):
|
|
for bad in ("25:00", "4am", "12:60"):
|
|
with patch.dict(os.environ, _base_env(YTDL_NIGHTLY_UPDATE_TIME=bad), clear=False):
|
|
with self.assertRaises(SystemExit):
|
|
Config()
|
|
|
|
def test_invalid_max_concurrent_downloads_exits(self):
|
|
for bad in ("0", "-1", "abc"):
|
|
with patch.dict(os.environ, _base_env(MAX_CONCURRENT_DOWNLOADS=bad), clear=False):
|
|
with self.assertRaises(SystemExit):
|
|
Config()
|
|
|
|
def test_invalid_port_exits(self):
|
|
for bad in ("0", "70000", "notaport"):
|
|
with patch.dict(os.environ, _base_env(PORT=bad), clear=False):
|
|
with self.assertRaises(SystemExit):
|
|
Config()
|
|
|
|
def test_invalid_clear_completed_after_exits(self):
|
|
for bad in ("-5", "soon"):
|
|
with patch.dict(os.environ, _base_env(CLEAR_COMPLETED_AFTER=bad), clear=False):
|
|
with self.assertRaises(SystemExit):
|
|
Config()
|
|
|
|
def test_clear_completed_after_zero_allowed(self):
|
|
with patch.dict(os.environ, _base_env(CLEAR_COMPLETED_AFTER="0"), clear=False):
|
|
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()
|
|
c.set_runtime_override("cookiefile", "/tmp/c.txt")
|
|
self.assertEqual(c.YTDL_OPTIONS.get("cookiefile"), "/tmp/c.txt")
|
|
c.remove_runtime_override("cookiefile")
|
|
self.assertIsNone(c.YTDL_OPTIONS.get("cookiefile"))
|
|
|
|
def test_ytdl_options_file_merges(self):
|
|
with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as f:
|
|
json.dump({"extractor_args": {"youtube": {"player_client": ["web"]}}}, f)
|
|
path = f.name
|
|
try:
|
|
with patch.dict(
|
|
os.environ,
|
|
_base_env(YTDL_OPTIONS="{}", YTDL_OPTIONS_FILE=path),
|
|
clear=False,
|
|
):
|
|
c = Config()
|
|
self.assertIn("extractor_args", c.YTDL_OPTIONS)
|
|
finally:
|
|
os.unlink(path)
|
|
|
|
def test_ytdl_option_presets_file_merges(self):
|
|
with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as f:
|
|
json.dump({"With subtitles": {"writesubtitles": True}}, f)
|
|
path = f.name
|
|
try:
|
|
with patch.dict(
|
|
os.environ,
|
|
_base_env(YTDL_OPTIONS_PRESETS="{}", YTDL_OPTIONS_PRESETS_FILE=path),
|
|
clear=False,
|
|
):
|
|
c = Config()
|
|
self.assertIn("With subtitles", c.YTDL_OPTIONS_PRESETS)
|
|
finally:
|
|
os.unlink(path)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|