mirror of
https://github.com/alexta69/metube.git
synced 2026-09-21 21:45:04 +00:00
Alex Shnitman
a6d81d4513
A download waiting for a MAX_CONCURRENT_DOWNLOADS slot sat at status
'pending' — the same status as an item added with auto-start off, which
is waiting for the user to press Start. The Downloading table draws its
Start button for exactly that status, so every queued row offered one.
Pressing it did nothing. start_pending() looks the id up in self.pending
first, and a queued download is not there; the fallback branch only acts
on 'scheduled' items, so the call fell through and still returned
{'status': 'ok'} — the UI reported success for a no-op.
One status name was covering two different states. A download in
self.queue waiting on the semaphore is now 'queued', leaving 'pending'
to mean only "waiting for you to press Start". The template condition is
unchanged and now excludes these rows by construction, and a 'Queued'
badge says why the row is idle instead of leaving a bare empty bar.
start_pending() notifies on promotion as well: with the slots saturated
the wait before Download.start() reports 'preparing' is unbounded, and
until something lands the client keeps showing the button it outgrew.
Persisted state needs no migration — __import_queue re-adds saved items
through __add_download, which sets the new status.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
1570 lines
52 KiB
Python
1570 lines
52 KiB
Python
"""Tests for ``DownloadQueue`` with mocked yt-dlp extraction."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import copy
|
|
import os
|
|
import re
|
|
import socket
|
|
import tempfile
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
import time
|
|
|
|
from ytdl import Download, DownloadInfo, DownloadQueue
|
|
|
|
|
|
@pytest.fixture
|
|
def dq_env():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
dl = os.path.join(tmp, "downloads")
|
|
st = os.path.join(tmp, "state")
|
|
os.makedirs(dl, exist_ok=True)
|
|
os.makedirs(st, exist_ok=True)
|
|
cfg = MagicMock()
|
|
cfg.STATE_DIR = st
|
|
cfg.DOWNLOAD_DIR = dl
|
|
cfg.AUDIO_DOWNLOAD_DIR = dl
|
|
cfg.TEMP_DIR = dl
|
|
cfg.MAX_CONCURRENT_DOWNLOADS = "3"
|
|
# Explicit: an unset attribute on a MagicMock is truthy, which would
|
|
# make validate_url bypass every SSRF check it is asked to run.
|
|
cfg.ALLOW_PRIVATE_ADDRESSES = False
|
|
cfg.YTDL_OPTIONS = {}
|
|
cfg.YTDL_OPTIONS_PRESETS = {}
|
|
cfg.CUSTOM_DIRS = True
|
|
cfg.CREATE_CUSTOM_DIRS = True
|
|
cfg.CLEAR_COMPLETED_AFTER = "0"
|
|
cfg.DELETE_FILE_ON_TRASHCAN = False
|
|
cfg.OUTPUT_TEMPLATE = "%(title)s.%(ext)s"
|
|
cfg.OUTPUT_TEMPLATE_CHAPTER = "%(title)s.%(ext)s"
|
|
cfg.OUTPUT_TEMPLATE_PLAYLIST = ""
|
|
cfg.OUTPUT_TEMPLATE_CHANNEL = ""
|
|
yield cfg
|
|
|
|
|
|
def test_cancel_add_increments_generation(dq_env):
|
|
notifier = MagicMock()
|
|
dq = DownloadQueue(dq_env, notifier)
|
|
before = dq._add_generation
|
|
dq.cancel_add()
|
|
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)
|
|
q, done = dq.get()
|
|
assert q == [] and done == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_single_video_goes_to_pending_when_auto_start_false(dq_env):
|
|
notifier = AsyncMock()
|
|
|
|
def fake_extract(self, url, *_args, **_kwargs):
|
|
return {
|
|
"_type": "video",
|
|
"id": "vid1",
|
|
"title": "Test Video",
|
|
"url": url,
|
|
"webpage_url": url,
|
|
}
|
|
|
|
dq = DownloadQueue(dq_env, notifier)
|
|
with patch.object(DownloadQueue, "_DownloadQueue__extract_info", fake_extract):
|
|
result = await dq.add(
|
|
"https://example.com/watch?v=1",
|
|
"video",
|
|
"auto",
|
|
"any",
|
|
"best",
|
|
"",
|
|
"",
|
|
0,
|
|
auto_start=False,
|
|
)
|
|
assert result["status"] == "ok"
|
|
assert dq.pending.exists("https://example.com/watch?v=1")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_unsupported_url_recorded_as_failed_entry(dq_env):
|
|
"""An unsupported/unextractable URL must show up as a red-cross entry in the
|
|
done list, not just a transient toast and a server log line."""
|
|
import ytdl
|
|
|
|
notifier = AsyncMock()
|
|
url = "https://example.com/not-a-video"
|
|
|
|
def boom(self, url, *_args, **_kwargs):
|
|
raise ytdl.yt_dlp.utils.YoutubeDLError(f'Unsupported URL: {url}')
|
|
|
|
dq = DownloadQueue(dq_env, notifier)
|
|
with patch.object(DownloadQueue, "_DownloadQueue__extract_info", boom):
|
|
result = await dq.add(
|
|
url, "video", "auto", "any", "best", "", "", 0, auto_start=True,
|
|
)
|
|
assert result["status"] == "error"
|
|
assert dq.done.exists(url)
|
|
failed = dq.done.get(url)
|
|
assert failed.info.status == "error"
|
|
assert failed.info.error == result["msg"]
|
|
assert failed.info.url == url
|
|
# The full URL stays in .url/.error for the detail panel; the display
|
|
# title is shortened to the hostname so the Completed row stays readable.
|
|
assert failed.info.title == "example.com"
|
|
notifier.completed.assert_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_ssrf_rejected_url_recorded_as_failed_entry(dq_env):
|
|
"""A URL rejected by the SSRF guard (before yt-dlp ever runs) must also
|
|
surface as a failed entry, not just an error status returned to the caller."""
|
|
notifier = AsyncMock()
|
|
url = "file:///etc/passwd"
|
|
|
|
dq = DownloadQueue(dq_env, notifier)
|
|
result = await dq.add(
|
|
url, "video", "auto", "any", "best", "", "", 0, auto_start=True,
|
|
)
|
|
assert result["status"] == "error"
|
|
assert dq.done.exists(url)
|
|
failed = dq.done.get(url)
|
|
assert failed.info.status == "error"
|
|
assert failed.info.error == result["msg"]
|
|
notifier.completed.assert_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_does_not_resolve_hostname_when_proxied(dq_env):
|
|
"""With a remote-DNS proxy configured, adding a URL must not look its host
|
|
up here: that both leaks the hostname to the local resolver and fails closed
|
|
when only the proxy can resolve it (issue #1079)."""
|
|
dq_env.YTDL_OPTIONS = {"proxy": "socks5h://tor:9050"}
|
|
notifier = AsyncMock()
|
|
|
|
def fake_extract(self, url, *_args, **_kwargs):
|
|
return {"_type": "video", "id": "vid1", "title": "t", "webpage_url": url}
|
|
|
|
dq = DownloadQueue(dq_env, notifier)
|
|
with patch("url_guard.socket.getaddrinfo", side_effect=AssertionError("resolved")), \
|
|
patch.object(DownloadQueue, "_DownloadQueue__extract_info", fake_extract):
|
|
result = await dq.add(
|
|
"https://only-the-proxy-can-resolve.invalid/x",
|
|
"video", "auto", "any", "best", "", "", 0, auto_start=False,
|
|
)
|
|
assert result["status"] == "ok"
|
|
dq.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_resolves_hostname_when_not_proxied(dq_env):
|
|
"""Without a proxy the address check still runs and still rejects."""
|
|
notifier = AsyncMock()
|
|
url = "https://internal.invalid/x"
|
|
|
|
dq = DownloadQueue(dq_env, notifier)
|
|
with patch("url_guard.socket.getaddrinfo",
|
|
return_value=[(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, "",
|
|
("169.254.169.254", 0))]):
|
|
result = await dq.add(url, "video", "auto", "any", "best", "", "", 0, auto_start=False)
|
|
assert result["status"] == "error"
|
|
dq.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancel_removes_from_pending(dq_env):
|
|
notifier = AsyncMock()
|
|
|
|
def fake_extract(self, url, *_args, **_kwargs):
|
|
return {
|
|
"_type": "video",
|
|
"id": "vid1",
|
|
"title": "Test Video",
|
|
"url": url,
|
|
"webpage_url": url,
|
|
}
|
|
|
|
dq = DownloadQueue(dq_env, notifier)
|
|
with patch.object(DownloadQueue, "_DownloadQueue__extract_info", fake_extract):
|
|
await dq.add(
|
|
"https://example.com/pending",
|
|
"video",
|
|
"auto",
|
|
"any",
|
|
"best",
|
|
"",
|
|
"",
|
|
0,
|
|
auto_start=False,
|
|
)
|
|
url = "https://example.com/pending"
|
|
await dq.cancel([url])
|
|
assert not dq.pending.exists(url)
|
|
notifier.canceled.assert_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancel_before_start_marks_download_canceled(dq_env):
|
|
"""Regression test for the race condition where cancel() arrives after the
|
|
download has been placed in the queue and ``__start_download`` has been
|
|
scheduled via ``asyncio.create_task`` but has not yet executed. Without the
|
|
fix, the pending task would run ``download.start()`` despite the user
|
|
cancelling, because its ``download.canceled`` guard was never flipped."""
|
|
notifier = AsyncMock()
|
|
|
|
def fake_extract(self, url, *_args, **_kwargs):
|
|
return {
|
|
"_type": "video",
|
|
"id": "vid1",
|
|
"title": "Test Video",
|
|
"url": url,
|
|
"webpage_url": url,
|
|
}
|
|
|
|
dq = DownloadQueue(dq_env, notifier)
|
|
url = "https://example.com/race"
|
|
start_mock = AsyncMock()
|
|
with patch.object(DownloadQueue, "_DownloadQueue__extract_info", fake_extract), \
|
|
patch.object(DownloadQueue, "_DownloadQueue__start_download", start_mock):
|
|
await dq.add(
|
|
url,
|
|
"video",
|
|
"auto",
|
|
"any",
|
|
"best",
|
|
"",
|
|
"",
|
|
0,
|
|
auto_start=True,
|
|
)
|
|
assert dq.queue.exists(url)
|
|
download = dq.queue.get(url)
|
|
assert download.canceled is False
|
|
await dq.cancel([url])
|
|
assert not dq.queue.exists(url)
|
|
assert download.canceled is True
|
|
notifier.canceled.assert_awaited_with(url)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_start_pending_moves_to_queue(dq_env):
|
|
notifier = AsyncMock()
|
|
|
|
def fake_extract(self, url, *_args, **_kwargs):
|
|
return {
|
|
"_type": "video",
|
|
"id": "vid1",
|
|
"title": "Test Video",
|
|
"url": url,
|
|
"webpage_url": url,
|
|
}
|
|
|
|
dq = DownloadQueue(dq_env, notifier)
|
|
with patch.object(DownloadQueue, "_DownloadQueue__extract_info", fake_extract):
|
|
await dq.add(
|
|
"https://example.com/startme",
|
|
"video",
|
|
"auto",
|
|
"any",
|
|
"best",
|
|
"",
|
|
"",
|
|
0,
|
|
auto_start=False,
|
|
)
|
|
url = "https://example.com/startme"
|
|
# Starting will spawn real download — cancel immediately before worker runs much
|
|
with patch.object(DownloadQueue, "_DownloadQueue__start_download", AsyncMock()):
|
|
await dq.start_pending([url])
|
|
assert not dq.pending.exists(url)
|
|
# It is in the queue now and starts on its own, so it must not keep
|
|
# advertising the Start button the UI draws for 'pending' (#1081), and the
|
|
# client has to be told before the concurrency slot frees up.
|
|
assert dq.queue.get(url).info.status == "queued"
|
|
assert notifier.updated.await_args[0][0].status == "queued"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_queued_download_is_not_offered_as_startable(dq_env):
|
|
"""A download waiting on a MAX_CONCURRENT_DOWNLOADS slot used to sit at
|
|
'pending', which is the status the UI draws a Start button for — but
|
|
start_pending has nothing to do for an item already in the queue, so the
|
|
button silently did nothing and still reported success (#1081).
|
|
"""
|
|
notifier = AsyncMock()
|
|
dq_env.MAX_CONCURRENT_DOWNLOADS = "1"
|
|
dq = DownloadQueue(dq_env, notifier)
|
|
released = asyncio.Event()
|
|
|
|
def fake_extract(self, url, *_args, **_kwargs):
|
|
return {"_type": "video", "id": url[-1], "title": f"Video {url[-1]}",
|
|
"url": url, "webpage_url": url}
|
|
|
|
async def blocking_start(self, notifier_, executor=None):
|
|
await released.wait()
|
|
|
|
first, second = "https://example.com/v1", "https://example.com/v2"
|
|
with patch.object(DownloadQueue, "_DownloadQueue__extract_info", fake_extract), \
|
|
patch("ytdl.Download.start", blocking_start), \
|
|
patch("ytdl.Download.close", lambda self: None):
|
|
for url in (first, second):
|
|
await dq.add(url, "video", "auto", "any", "best", "", "", 0, auto_start=True)
|
|
await asyncio.sleep(0)
|
|
|
|
# The first holds the only slot; the second is waiting behind it.
|
|
assert dq.queue.get(second).info.status == "queued"
|
|
assert not dq.pending.exists(second)
|
|
|
|
released.set()
|
|
|
|
await asyncio.sleep(0)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_entry_queues_single_video_without_reextracting(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",
|
|
"playlist_index": "01",
|
|
"playlist_title": "Playlist",
|
|
}
|
|
|
|
with patch.object(DownloadQueue, "_DownloadQueue__extract_info", side_effect=AssertionError("should not re-extract")):
|
|
result = await dq.add_entry(
|
|
entry,
|
|
"video",
|
|
"auto",
|
|
"any",
|
|
"best",
|
|
"",
|
|
"",
|
|
0,
|
|
auto_start=False,
|
|
)
|
|
|
|
assert result["status"] == "ok"
|
|
assert dq.pending.exists("https://example.com/watch?v=1")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_retry_restores_playlist_output_context(dq_env):
|
|
notifier = AsyncMock()
|
|
dq_env.OUTPUT_TEMPLATE_PLAYLIST = "%(playlist_title)s/%(title)s.%(ext)s"
|
|
dq = DownloadQueue(dq_env, notifier)
|
|
url = "https://example.com/watch?v=1"
|
|
failed_info = DownloadInfo(
|
|
id="vid1",
|
|
title="Test Video",
|
|
url=url,
|
|
quality="best",
|
|
download_type="video",
|
|
codec="auto",
|
|
format="any",
|
|
folder="",
|
|
custom_name_prefix="",
|
|
error="temporary failure",
|
|
entry={
|
|
"playlist_index": "01",
|
|
"playlist_title": "My Playlist",
|
|
"playlist_count": 10,
|
|
},
|
|
playlist_item_limit=0,
|
|
split_by_chapters=False,
|
|
chapter_template="",
|
|
)
|
|
failed_info.status = "error"
|
|
await dq.done.put(Download(None, None, None, None, "best", "any", {}, failed_info))
|
|
|
|
def fake_extract(self, extracted_url, *_args, **_kwargs):
|
|
return {
|
|
"_type": "video",
|
|
"id": "vid1",
|
|
"title": "Test Video",
|
|
"url": extracted_url,
|
|
"webpage_url": extracted_url,
|
|
}
|
|
|
|
with patch.object(DownloadQueue, "_DownloadQueue__extract_info", fake_extract), \
|
|
patch.object(DownloadQueue, "_DownloadQueue__start_download", new=AsyncMock()):
|
|
result = await dq.retry(url)
|
|
|
|
assert result["status"] == "ok"
|
|
queued = dq.queue.get(url)
|
|
assert queued.output_template == "My Playlist/%(title)s.%(ext)s"
|
|
assert queued.info.entry["playlist_index"] == "01"
|
|
assert queued.info.entry["playlist_title"] == "My Playlist"
|
|
|
|
|
|
def _failed_playlist_item(url, **overrides):
|
|
"""A done-list entry for a playlist item that failed mid-download."""
|
|
info = DownloadInfo(
|
|
id="vid1",
|
|
title="Test Video",
|
|
url=url,
|
|
quality="best",
|
|
download_type="video",
|
|
codec="auto",
|
|
format="any",
|
|
folder="",
|
|
custom_name_prefix="",
|
|
error="temporary failure",
|
|
entry={
|
|
"playlist_index": "01",
|
|
"playlist_title": "My Playlist",
|
|
"playlist_count": 10,
|
|
},
|
|
playlist_item_limit=0,
|
|
split_by_chapters=False,
|
|
chapter_template="",
|
|
**overrides,
|
|
)
|
|
info.status = "error"
|
|
return info
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_retry_keeps_playlist_context_through_url_indirection(dq_env):
|
|
# extract_flat=True makes yt-dlp hand back url/url_transparent results
|
|
# unprocessed, so __add_entry recurses into add() a second time. The retry
|
|
# context has to survive that hop or the item lands in the root directory.
|
|
notifier = AsyncMock()
|
|
dq_env.OUTPUT_TEMPLATE_PLAYLIST = "%(playlist_title)s/%(title)s.%(ext)s"
|
|
dq = DownloadQueue(dq_env, notifier)
|
|
url = "https://example.com/watch?v=1"
|
|
resolved = "https://example.com/resolved?v=1"
|
|
await dq.done.put(Download(None, None, None, None, "best", "any", {}, _failed_playlist_item(url)))
|
|
|
|
def fake_extract(self, extracted_url, *_args, **_kwargs):
|
|
if extracted_url == url:
|
|
return {"_type": "url", "url": resolved, "id": "vid1"}
|
|
return {
|
|
"_type": "video",
|
|
"id": "vid1",
|
|
"title": "Test Video",
|
|
"url": extracted_url,
|
|
"webpage_url": extracted_url,
|
|
}
|
|
|
|
with patch.object(DownloadQueue, "_DownloadQueue__extract_info", fake_extract), \
|
|
patch.object(DownloadQueue, "_DownloadQueue__start_download", new=AsyncMock()):
|
|
result = await dq.retry(url)
|
|
|
|
assert result["status"] == "ok"
|
|
queued = dq.queue.get(resolved)
|
|
assert queued.output_template == "My Playlist/%(title)s.%(ext)s"
|
|
assert queued.info.entry["playlist_title"] == "My Playlist"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_retry_reapplies_current_options_gates(dq_env):
|
|
# The stored options passed parse_download_options when first submitted, but
|
|
# the configuration can have changed since; retry must not resurrect
|
|
# overrides or presets the current configuration no longer allows.
|
|
notifier = AsyncMock()
|
|
dq_env.ALLOW_YTDL_OPTIONS_OVERRIDES = False
|
|
dq_env.YTDL_OPTIONS_PRESETS = {"Still There": {"writesubtitles": True}}
|
|
dq = DownloadQueue(dq_env, notifier)
|
|
url = "https://example.com/watch?v=1"
|
|
info = _failed_playlist_item(
|
|
url,
|
|
ytdl_options_presets=["Still There", "Removed Preset"],
|
|
ytdl_options_overrides={"paths": {"home": "/etc"}},
|
|
)
|
|
await dq.done.put(Download(None, None, None, None, "best", "any", {}, info))
|
|
|
|
def fake_extract(self, extracted_url, *_args, **_kwargs):
|
|
return {
|
|
"_type": "video",
|
|
"id": "vid1",
|
|
"title": "Test Video",
|
|
"url": extracted_url,
|
|
"webpage_url": extracted_url,
|
|
}
|
|
|
|
with patch.object(DownloadQueue, "_DownloadQueue__extract_info", fake_extract), \
|
|
patch.object(DownloadQueue, "_DownloadQueue__start_download", new=AsyncMock()):
|
|
result = await dq.retry(url)
|
|
|
|
assert result["status"] == "ok"
|
|
queued = dq.queue.get(url)
|
|
assert queued.info.ytdl_options_overrides == {}
|
|
assert queued.info.ytdl_options_presets == ["Still There"]
|
|
assert queued.ytdl_opts.get("paths", {}).get("home") != "/etc"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_retry_keeps_overrides_while_still_allowed(dq_env):
|
|
notifier = AsyncMock()
|
|
dq_env.ALLOW_YTDL_OPTIONS_OVERRIDES = True
|
|
dq_env.YTDL_OPTIONS_PRESETS = {}
|
|
dq = DownloadQueue(dq_env, notifier)
|
|
url = "https://example.com/watch?v=1"
|
|
info = _failed_playlist_item(url, ytdl_options_overrides={"writesubtitles": True})
|
|
await dq.done.put(Download(None, None, None, None, "best", "any", {}, info))
|
|
|
|
def fake_extract(self, extracted_url, *_args, **_kwargs):
|
|
return {
|
|
"_type": "video",
|
|
"id": "vid1",
|
|
"title": "Test Video",
|
|
"url": extracted_url,
|
|
"webpage_url": extracted_url,
|
|
}
|
|
|
|
with patch.object(DownloadQueue, "_DownloadQueue__extract_info", fake_extract), \
|
|
patch.object(DownloadQueue, "_DownloadQueue__start_download", new=AsyncMock()):
|
|
result = await dq.retry(url)
|
|
|
|
assert result["status"] == "ok"
|
|
assert dq.queue.get(url).info.ytdl_options_overrides == {"writesubtitles": True}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_retry_carries_the_sponsorblock_flag(dq_env):
|
|
notifier = AsyncMock()
|
|
dq = DownloadQueue(dq_env, notifier)
|
|
url = "https://example.com/watch?v=1"
|
|
await dq.done.put(
|
|
Download(None, None, None, None, "best", "any", {}, _failed_playlist_item(url, sponsorblock=True))
|
|
)
|
|
|
|
def fake_extract(self, extracted_url, *_args, **_kwargs):
|
|
return {
|
|
"_type": "video",
|
|
"id": "vid1",
|
|
"title": "Test Video",
|
|
"url": extracted_url,
|
|
"webpage_url": extracted_url,
|
|
}
|
|
|
|
with patch.object(DownloadQueue, "_DownloadQueue__extract_info", fake_extract), \
|
|
patch.object(DownloadQueue, "_DownloadQueue__start_download", new=AsyncMock()):
|
|
result = await dq.retry(url)
|
|
|
|
assert result["status"] == "ok"
|
|
assert dq.queue.get(url).info.sponsorblock is True
|
|
|
|
|
|
@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."""
|
|
notifier = AsyncMock()
|
|
dq_env.OUTPUT_TEMPLATE = "%(channel)s [YT]/%(title)s.%(ext)s"
|
|
dq_env.OUTPUT_TEMPLATE_CHANNEL = ""
|
|
dq_env.OUTPUT_TEMPLATE_PLAYLIST = ""
|
|
|
|
channel_id = "UCabcd123"
|
|
|
|
def fake_extract(self, url, *_args, **_kwargs):
|
|
return {
|
|
"_type": "playlist",
|
|
"id": channel_id,
|
|
"channel_id": channel_id,
|
|
"channel": "Odin",
|
|
"title": "Odin - Videos",
|
|
"entries": [
|
|
{
|
|
"id": "vid1",
|
|
"title": "Salvia Plath - Pondering",
|
|
"url": "https://example.com/watch?v=1",
|
|
"webpage_url": "https://example.com/watch?v=1",
|
|
"channel": "Odin",
|
|
"upload_date": "20130804",
|
|
},
|
|
],
|
|
}
|
|
|
|
dq = DownloadQueue(dq_env, notifier)
|
|
with patch.object(DownloadQueue, "_DownloadQueue__extract_info", fake_extract):
|
|
result = await dq.add(
|
|
"https://www.youtube.com/@odin/videos",
|
|
"video",
|
|
"auto",
|
|
"any",
|
|
"best",
|
|
"",
|
|
"",
|
|
0,
|
|
auto_start=False,
|
|
)
|
|
|
|
assert result["status"] == "ok"
|
|
url = "https://example.com/watch?v=1"
|
|
assert dq.pending.exists(url)
|
|
download = dq.pending.get(url)
|
|
assert download.output_template.startswith("Odin [YT]/")
|
|
assert "Odin - Videos" not in download.output_template
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_playlist_download_not_treated_as_channel(dq_env):
|
|
"""Real playlists (id != channel_id) must not be promoted to channel downloads."""
|
|
notifier = AsyncMock()
|
|
dq_env.OUTPUT_TEMPLATE = "%(channel)s [YT]/%(title)s.%(ext)s"
|
|
dq_env.OUTPUT_TEMPLATE_CHANNEL = ""
|
|
dq_env.OUTPUT_TEMPLATE_PLAYLIST = "%(playlist_title)s/%(title)s.%(ext)s"
|
|
|
|
def fake_extract(self, url, *_args, **_kwargs):
|
|
return {
|
|
"_type": "playlist",
|
|
"id": "PLxyz789",
|
|
"channel_id": "UCabcd123",
|
|
"channel": "Odin",
|
|
"title": "My Playlist",
|
|
"entries": [
|
|
{
|
|
"id": "vid1",
|
|
"title": "Test Video",
|
|
"url": "https://example.com/watch?v=1",
|
|
"webpage_url": "https://example.com/watch?v=1",
|
|
},
|
|
],
|
|
}
|
|
|
|
dq = DownloadQueue(dq_env, notifier)
|
|
with patch.object(DownloadQueue, "_DownloadQueue__extract_info", fake_extract):
|
|
result = await dq.add(
|
|
"https://www.youtube.com/playlist?list=PLxyz789",
|
|
"video",
|
|
"auto",
|
|
"any",
|
|
"best",
|
|
"",
|
|
"",
|
|
0,
|
|
auto_start=False,
|
|
)
|
|
|
|
assert result["status"] == "ok"
|
|
url = "https://example.com/watch?v=1"
|
|
assert dq.pending.exists(url)
|
|
download = dq.pending.get(url)
|
|
assert download.output_template.startswith("My Playlist/")
|
|
|
|
|
|
def _channel_extraction(entry_id, **extra):
|
|
"""A channel yt-dlp reported as a playlist, addressed by *entry_id*."""
|
|
return {
|
|
"_type": "playlist",
|
|
"id": entry_id,
|
|
"channel_id": "UCabcd123",
|
|
"channel": "Odin",
|
|
"title": "Odin",
|
|
**extra,
|
|
"entries": [
|
|
{
|
|
"id": "vid1",
|
|
"title": "Salvia Plath - Pondering",
|
|
"url": "https://example.com/watch?v=1",
|
|
"webpage_url": "https://example.com/watch?v=1",
|
|
"channel": "Odin",
|
|
"upload_date": "20130804",
|
|
},
|
|
],
|
|
}
|
|
|
|
|
|
async def _add_and_get_template(dq_env, extraction, url):
|
|
dq_env.OUTPUT_TEMPLATE = "%(channel)s [YT]/%(title)s.%(ext)s"
|
|
dq_env.OUTPUT_TEMPLATE_CHANNEL = ""
|
|
dq_env.OUTPUT_TEMPLATE_PLAYLIST = "%(playlist_title)s/%(title)s.%(ext)s"
|
|
|
|
def fake_extract(self, _url, *_args, **_kwargs):
|
|
return extraction
|
|
|
|
dq = DownloadQueue(dq_env, AsyncMock())
|
|
with patch.object(DownloadQueue, "_DownloadQueue__extract_info", fake_extract):
|
|
result = await dq.add(url, "video", "auto", "any", "best", "", "", 0, auto_start=False)
|
|
assert result["status"] == "ok"
|
|
return dq.pending.get("https://example.com/watch?v=1").output_template
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bare_handle_channel_url_is_treated_as_a_channel(dq_env):
|
|
"""A channel addressed as /@handle reports its id as the handle, not the
|
|
channel id, and was falling through to OUTPUT_TEMPLATE_PLAYLIST."""
|
|
template = await _add_and_get_template(
|
|
dq_env,
|
|
_channel_extraction("@odin", uploader_id="@odin"),
|
|
"https://www.youtube.com/@odin",
|
|
)
|
|
|
|
assert template.startswith("Odin [YT]/")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_legacy_vanity_channel_url_is_treated_as_a_channel(dq_env):
|
|
"""A legacy /c/Name URL reports the vanity name as its id, while
|
|
uploader_id is still the handle."""
|
|
template = await _add_and_get_template(
|
|
dq_env,
|
|
_channel_extraction("Odin", uploader_id="@odin"),
|
|
"https://www.youtube.com/c/Odin",
|
|
)
|
|
|
|
assert template.startswith("Odin [YT]/")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_playlist_with_owner_uploader_id_is_still_a_playlist(dq_env):
|
|
"""A real playlist carries its owner's channel_id and uploader_id, but its
|
|
own id matches neither, so it must keep the playlist template."""
|
|
template = await _add_and_get_template(
|
|
dq_env,
|
|
_channel_extraction("PLxyz789", uploader_id="@odin", title="My Playlist"),
|
|
"https://www.youtube.com/playlist?list=PLxyz789",
|
|
)
|
|
|
|
assert template.startswith("My Playlist/")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_merges_global_preset_and_override_options(dq_env):
|
|
notifier = AsyncMock()
|
|
dq_env.YTDL_OPTIONS = {"writesubtitles": False, "cookiefile": "/tmp/global.txt"}
|
|
dq_env.YTDL_OPTIONS_PRESETS = {
|
|
"Preset A": {"writesubtitles": True, "proxy": "http://preset-a"},
|
|
"Preset B": {"writesubtitles": False, "ratelimit": 1000},
|
|
}
|
|
|
|
def fake_extract(self, url, *_args, **_kwargs):
|
|
return {
|
|
"_type": "video",
|
|
"id": "vid2",
|
|
"title": "Preset Video",
|
|
"url": url,
|
|
"webpage_url": url,
|
|
}
|
|
|
|
dq = DownloadQueue(dq_env, notifier)
|
|
with patch.object(DownloadQueue, "_DownloadQueue__extract_info", fake_extract):
|
|
result = await dq.add(
|
|
"https://example.com/preset",
|
|
"video",
|
|
"auto",
|
|
"any",
|
|
"best",
|
|
"",
|
|
"",
|
|
0,
|
|
auto_start=False,
|
|
ytdl_options_presets=["Preset A", "Preset B"],
|
|
ytdl_options_overrides={"proxy": "http://override", "embed_thumbnail": True},
|
|
)
|
|
|
|
assert result["status"] == "ok"
|
|
queued = dq.pending.get("https://example.com/preset")
|
|
assert queued.ytdl_opts["cookiefile"] == "/tmp/global.txt"
|
|
assert queued.ytdl_opts["writesubtitles"] is False
|
|
assert queued.ytdl_opts["ratelimit"] == 1000
|
|
assert queued.ytdl_opts["proxy"] == "http://override"
|
|
assert queued.ytdl_opts["embed_thumbnail"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_info_preset_null_download_archive_overrides_global(dq_env):
|
|
"""Preset download_archive:null must apply during extract_info (global archive otherwise wins first)."""
|
|
dq_env.YTDL_OPTIONS = {"download_archive": "/tmp/archive.txt"}
|
|
dq_env.YTDL_OPTIONS_PRESETS = {"NoArchive": {"download_archive": None}}
|
|
|
|
captured_params: list = []
|
|
|
|
class FakeYoutubeDL:
|
|
def __init__(self, params=None):
|
|
captured_params.append(params)
|
|
|
|
def extract_info(self, url, download=False):
|
|
return {
|
|
"_type": "video",
|
|
"id": "vid-archive",
|
|
"title": "Archive Test",
|
|
"url": url,
|
|
"webpage_url": url,
|
|
}
|
|
|
|
notifier = AsyncMock()
|
|
dq = DownloadQueue(dq_env, notifier)
|
|
with patch("ytdl.yt_dlp.YoutubeDL", FakeYoutubeDL):
|
|
result = await dq.add(
|
|
"https://example.com/archive-test",
|
|
"video",
|
|
"auto",
|
|
"any",
|
|
"best",
|
|
"",
|
|
"",
|
|
0,
|
|
auto_start=False,
|
|
ytdl_options_presets=["NoArchive"],
|
|
)
|
|
|
|
assert result["status"] == "ok"
|
|
assert len(captured_params) == 1
|
|
extract_params = captured_params[0]
|
|
assert extract_params.get("download_archive") is None
|
|
assert extract_params["extract_flat"] is True
|
|
assert extract_params["noplaylist"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_info_metube_extract_keys_win_over_preset(dq_env):
|
|
"""MeTube's flat-extract settings must not be overridden by presets."""
|
|
dq_env.YTDL_OPTIONS = {}
|
|
dq_env.YTDL_OPTIONS_PRESETS = {
|
|
"TryOverride": {"extract_flat": False, "noplaylist": False},
|
|
}
|
|
|
|
captured_params: list = []
|
|
|
|
class FakeYoutubeDL:
|
|
def __init__(self, params=None):
|
|
captured_params.append(params)
|
|
|
|
def extract_info(self, url, download=False):
|
|
return {
|
|
"_type": "video",
|
|
"id": "vid-flat",
|
|
"title": "Flat Test",
|
|
"url": url,
|
|
"webpage_url": url,
|
|
}
|
|
|
|
notifier = AsyncMock()
|
|
dq = DownloadQueue(dq_env, notifier)
|
|
with patch("ytdl.yt_dlp.YoutubeDL", FakeYoutubeDL):
|
|
result = await dq.add(
|
|
"https://example.com/flat-test",
|
|
"video",
|
|
"auto",
|
|
"any",
|
|
"best",
|
|
"",
|
|
"",
|
|
0,
|
|
auto_start=False,
|
|
ytdl_options_presets=["TryOverride"],
|
|
)
|
|
|
|
assert result["status"] == "ok"
|
|
assert captured_params[0]["extract_flat"] is True
|
|
assert captured_params[0]["noplaylist"] is True
|
|
|
|
|
|
def _feed_extract(feed):
|
|
"""Patch for __extract_info that returns a playlist/channel feed dict."""
|
|
|
|
def fake_extract(self, url, *_args, **_kwargs):
|
|
return copy.deepcopy(feed)
|
|
|
|
return fake_extract
|
|
|
|
|
|
_CHANNEL_FEED = {
|
|
"_type": "playlist",
|
|
"id": "UC123",
|
|
"title": "Vanessa - Videos",
|
|
"channel": "Vanessa",
|
|
"channel_id": "UC123",
|
|
"uploader": "Vanessa",
|
|
"extractor": "youtube:tab",
|
|
"extractor_key": "YoutubeTab",
|
|
"webpage_url": "https://example.com/@vanessa/videos",
|
|
"entries": [
|
|
{"id": "v1", "title": "One", "url": "https://example.com/v1",
|
|
"webpage_url": "https://example.com/v1", "_type": "url"},
|
|
],
|
|
}
|
|
|
|
_PLAYLIST_FEED = {
|
|
"_type": "playlist",
|
|
"id": "PL123",
|
|
"title": "My Playlist",
|
|
"extractor": "generic",
|
|
"extractor_key": "Generic",
|
|
"webpage_url": "https://example.com/playlist?list=PL123",
|
|
"entries": [
|
|
{"id": "v1", "title": "One", "url": "https://example.com/v1",
|
|
"webpage_url": "https://example.com/v1", "_type": "url"},
|
|
],
|
|
}
|
|
|
|
|
|
def _written_files(root):
|
|
found = []
|
|
for dirpath, _dirs, files in os.walk(root):
|
|
for f in files:
|
|
found.append(os.path.relpath(os.path.join(dirpath, f), root))
|
|
return sorted(found)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_channel_feed_metadata_lands_beside_its_items(dq_env):
|
|
"""Issues #660/#1040: the feed-level .info.json follows the same template
|
|
the items use, so it sits in the channel's own folder rather than in
|
|
DOWNLOAD_DIR under yt-dlp's pl_* default name."""
|
|
dq_env.YTDL_OPTIONS = {"writeinfojson": True}
|
|
dq_env.OUTPUT_TEMPLATE_CHANNEL = "%(channel)s/%(title)s.%(ext)s"
|
|
|
|
dq = DownloadQueue(dq_env, AsyncMock())
|
|
with patch.object(DownloadQueue, "_DownloadQueue__extract_info", _feed_extract(_CHANNEL_FEED)), \
|
|
patch.object(DownloadQueue, "_DownloadQueue__start_download", new=AsyncMock()):
|
|
result = await dq.add(
|
|
"https://example.com/@vanessa/videos", "video", "auto", "any", "best",
|
|
"", "", 0, auto_start=False,
|
|
)
|
|
|
|
assert result["status"] == "ok"
|
|
assert _written_files(dq_env.DOWNLOAD_DIR) == [
|
|
os.path.join("Vanessa", "Vanessa - Videos.info.json")
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_playlist_feed_metadata_uses_the_playlist_template(dq_env):
|
|
dq_env.YTDL_OPTIONS = {"writeinfojson": True}
|
|
dq_env.OUTPUT_TEMPLATE_PLAYLIST = "%(playlist_title)s/%(title)s.%(ext)s"
|
|
|
|
dq = DownloadQueue(dq_env, AsyncMock())
|
|
with patch.object(DownloadQueue, "_DownloadQueue__extract_info", _feed_extract(_PLAYLIST_FEED)), \
|
|
patch.object(DownloadQueue, "_DownloadQueue__start_download", new=AsyncMock()):
|
|
await dq.add(
|
|
"https://example.com/playlist?list=PL123", "video", "auto", "any", "best",
|
|
"", "", 0, auto_start=False,
|
|
)
|
|
|
|
assert _written_files(dq_env.DOWNLOAD_DIR) == [
|
|
os.path.join("My Playlist", "My Playlist.info.json")
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_feed_metadata_honours_custom_folder(dq_env):
|
|
dq_env.YTDL_OPTIONS = {"writeinfojson": True}
|
|
dq_env.OUTPUT_TEMPLATE_PLAYLIST = "%(playlist_title)s/%(title)s.%(ext)s"
|
|
|
|
dq = DownloadQueue(dq_env, AsyncMock())
|
|
with patch.object(DownloadQueue, "_DownloadQueue__extract_info", _feed_extract(_PLAYLIST_FEED)), \
|
|
patch.object(DownloadQueue, "_DownloadQueue__start_download", new=AsyncMock()):
|
|
await dq.add(
|
|
"https://example.com/playlist?list=PL123", "video", "auto", "any", "best",
|
|
"Music", "", 0, auto_start=False,
|
|
)
|
|
|
|
assert _written_files(dq_env.DOWNLOAD_DIR) == [
|
|
os.path.join("Music", "My Playlist", "My Playlist.info.json")
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_feed_metadata_without_writeinfojson(dq_env):
|
|
"""Nothing new appears for users who never asked for these files."""
|
|
dq_env.YTDL_OPTIONS = {}
|
|
|
|
dq = DownloadQueue(dq_env, AsyncMock())
|
|
with patch.object(DownloadQueue, "_DownloadQueue__extract_info", _feed_extract(_PLAYLIST_FEED)), \
|
|
patch.object(DownloadQueue, "_DownloadQueue__start_download", new=AsyncMock()):
|
|
await dq.add(
|
|
"https://example.com/playlist?list=PL123", "video", "auto", "any", "best",
|
|
"", "", 0, auto_start=False,
|
|
)
|
|
|
|
assert _written_files(dq_env.DOWNLOAD_DIR) == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_feed_metadata_can_be_turned_off_by_the_user(dq_env):
|
|
dq_env.YTDL_OPTIONS = {"writeinfojson": True, "allow_playlist_files": False}
|
|
|
|
dq = DownloadQueue(dq_env, AsyncMock())
|
|
with patch.object(DownloadQueue, "_DownloadQueue__extract_info", _feed_extract(_PLAYLIST_FEED)), \
|
|
patch.object(DownloadQueue, "_DownloadQueue__start_download", new=AsyncMock()):
|
|
await dq.add(
|
|
"https://example.com/playlist?list=PL123", "video", "auto", "any", "best",
|
|
"", "", 0, auto_start=False,
|
|
)
|
|
|
|
assert _written_files(dq_env.DOWNLOAD_DIR) == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_feed_metadata_failure_does_not_fail_the_add(dq_env):
|
|
dq_env.YTDL_OPTIONS = {"writeinfojson": True}
|
|
|
|
dq = DownloadQueue(dq_env, AsyncMock())
|
|
with patch.object(DownloadQueue, "_DownloadQueue__extract_info", _feed_extract(_PLAYLIST_FEED)), \
|
|
patch.object(DownloadQueue, "_DownloadQueue__start_download", new=AsyncMock()), \
|
|
patch.object(
|
|
DownloadQueue, "_DownloadQueue__write_feed_metadata_sync",
|
|
side_effect=OSError("read-only filesystem"),
|
|
):
|
|
result = await dq.add(
|
|
"https://example.com/playlist?list=PL123", "video", "auto", "any", "best",
|
|
"", "", 0, auto_start=False,
|
|
)
|
|
|
|
assert result["status"] == "ok"
|
|
assert dq.pending.exists("https://example.com/v1")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extraction_pass_never_writes_feed_metadata(dq_env):
|
|
"""The classification pass must not produce files: it runs before the add is
|
|
known to succeed, and yt-dlp writes playlist files regardless of `download`."""
|
|
dq_env.YTDL_OPTIONS = {"writeinfojson": True, "allow_playlist_files": True}
|
|
captured: list = []
|
|
|
|
class FakeYoutubeDL:
|
|
def __init__(self, params=None):
|
|
captured.append(params)
|
|
|
|
def extract_info(self, url, download=False):
|
|
return {"_type": "video", "id": "v", "title": "V", "url": url, "webpage_url": url}
|
|
|
|
dq = DownloadQueue(dq_env, AsyncMock())
|
|
with patch("ytdl.yt_dlp.YoutubeDL", FakeYoutubeDL):
|
|
await dq.add(
|
|
"https://example.com/watch?v=1", "video", "auto", "any", "best",
|
|
"", "", 0, auto_start=False,
|
|
)
|
|
|
|
assert captured[0]["allow_playlist_files"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_sets_clip_bounds_on_download_info(dq_env):
|
|
notifier = AsyncMock()
|
|
|
|
def fake_extract(self, url, *_args, **_kwargs):
|
|
return {
|
|
"_type": "video",
|
|
"id": "vid1",
|
|
"title": "Test Video",
|
|
"url": url,
|
|
"webpage_url": url,
|
|
}
|
|
|
|
dq = DownloadQueue(dq_env, notifier)
|
|
with patch.object(DownloadQueue, "_DownloadQueue__extract_info", fake_extract):
|
|
result = await dq.add(
|
|
"https://example.com/clip",
|
|
"video",
|
|
"auto",
|
|
"any",
|
|
"best",
|
|
"",
|
|
"",
|
|
0,
|
|
auto_start=False,
|
|
clip_start=10.0,
|
|
clip_end=99.5,
|
|
)
|
|
|
|
assert result["status"] == "ok"
|
|
download = dq.pending.get("https://example.com/clip")
|
|
assert download.info.clip_start == 10.0
|
|
assert download.info.clip_end == 99.5
|
|
|
|
|
|
def _upcoming_entry(url: str, *, release_timestamp: float | None = None) -> dict:
|
|
return {
|
|
"_type": "video",
|
|
"id": "live1",
|
|
"title": "Upcoming Stream",
|
|
"url": url,
|
|
"webpage_url": url,
|
|
"live_status": "is_upcoming",
|
|
"release_timestamp": release_timestamp if release_timestamp is not None else time.time() + 3600,
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_upcoming_stream_scheduled_without_starting(dq_env):
|
|
notifier = AsyncMock()
|
|
url = "https://example.com/live-upcoming"
|
|
start_mock = AsyncMock()
|
|
|
|
dq = DownloadQueue(dq_env, notifier)
|
|
with patch.object(DownloadQueue, "_DownloadQueue__start_download", start_mock):
|
|
result = await dq.add_entry(
|
|
_upcoming_entry(url),
|
|
"video",
|
|
"auto",
|
|
"any",
|
|
"best",
|
|
"",
|
|
"",
|
|
0,
|
|
auto_start=True,
|
|
)
|
|
|
|
assert result["status"] == "ok"
|
|
assert dq.queue.exists(url)
|
|
download = dq.queue.get(url)
|
|
assert download.info.status == "scheduled"
|
|
assert download.info.live_status == "is_upcoming"
|
|
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
|
|
async def test_probe_scheduled_starts_when_live(dq_env):
|
|
notifier = AsyncMock()
|
|
url = "https://example.com/live-upcoming"
|
|
start_mock = AsyncMock()
|
|
|
|
dq = DownloadQueue(dq_env, notifier)
|
|
with patch.object(DownloadQueue, "_DownloadQueue__start_download", start_mock):
|
|
await dq.add_entry(
|
|
_upcoming_entry(url),
|
|
"video",
|
|
"auto",
|
|
"any",
|
|
"best",
|
|
"",
|
|
"",
|
|
0,
|
|
auto_start=True,
|
|
)
|
|
|
|
download = dq.queue.get(url)
|
|
|
|
def fake_probe_extract(self, probe_url, ytdl_options_presets=None, ytdl_options_overrides=None):
|
|
assert probe_url == url
|
|
return {
|
|
"_type": "video",
|
|
"id": "live1",
|
|
"title": "Live Now",
|
|
"url": url,
|
|
"webpage_url": url,
|
|
"live_status": "is_live",
|
|
"formats": [{"format_id": "22"}],
|
|
}
|
|
|
|
with patch.object(DownloadQueue, "_DownloadQueue__extract_info", fake_probe_extract), \
|
|
patch.object(DownloadQueue, "_DownloadQueue__start_download", start_mock):
|
|
await dq._probe_scheduled_download(download)
|
|
|
|
assert url not in dq._scheduled_probe_at
|
|
assert download.info.live_status == "is_live"
|
|
assert download.info.status == "queued"
|
|
start_mock.assert_called_once_with(download)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_import_scheduled_re_registers_monitor(dq_env):
|
|
notifier = AsyncMock()
|
|
url = "https://example.com/live-restart"
|
|
release = time.time() + 7200
|
|
|
|
info = DownloadInfo(
|
|
id="live1",
|
|
title="Upcoming Stream",
|
|
url=url,
|
|
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="",
|
|
live_status="is_upcoming",
|
|
live_release_timestamp=release,
|
|
)
|
|
info.status = "scheduled"
|
|
|
|
dq = DownloadQueue(dq_env, notifier)
|
|
start_mock = AsyncMock()
|
|
with patch.object(DownloadQueue, "_DownloadQueue__start_download", start_mock):
|
|
await dq._DownloadQueue__add_download(info, True)
|
|
|
|
assert dq.queue.exists(url)
|
|
assert dq.queue.get(url).info.status == "scheduled"
|
|
assert url in dq._scheduled_probe_at
|
|
start_mock.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_probe_transient_error_retries_without_failing(dq_env):
|
|
"""A single probe failure must not abandon the scheduled stream."""
|
|
import ytdl
|
|
|
|
notifier = AsyncMock()
|
|
url = "https://example.com/live-transient"
|
|
start_mock = AsyncMock()
|
|
|
|
dq = DownloadQueue(dq_env, notifier)
|
|
with patch.object(DownloadQueue, "_DownloadQueue__start_download", start_mock):
|
|
await dq.add_entry(
|
|
_upcoming_entry(url),
|
|
"video", "auto", "any", "best", "", "", 0,
|
|
auto_start=True,
|
|
)
|
|
download = dq.queue.get(url)
|
|
|
|
def boom(self, *args, **kwargs):
|
|
raise ytdl.yt_dlp.utils.YoutubeDLError("temporary network glitch")
|
|
|
|
before = time.time()
|
|
with patch.object(DownloadQueue, "_DownloadQueue__extract_info", boom):
|
|
await dq._probe_scheduled_download(download)
|
|
|
|
# Still scheduled, still monitored, probe rescheduled into the future.
|
|
assert download.info.status == "scheduled"
|
|
assert url in dq._scheduled_probe_at
|
|
assert dq._scheduled_probe_at[url] >= before
|
|
assert dq._scheduled_probe_failures[url] == 1
|
|
notifier.completed.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_probe_gives_up_after_max_failures(dq_env):
|
|
import ytdl
|
|
|
|
notifier = AsyncMock()
|
|
url = "https://example.com/live-dead"
|
|
start_mock = AsyncMock()
|
|
|
|
dq = DownloadQueue(dq_env, notifier)
|
|
with patch.object(DownloadQueue, "_DownloadQueue__start_download", start_mock):
|
|
await dq.add_entry(
|
|
_upcoming_entry(url),
|
|
"video", "auto", "any", "best", "", "", 0,
|
|
auto_start=True,
|
|
)
|
|
download = dq.queue.get(url)
|
|
|
|
def boom(self, *args, **kwargs):
|
|
raise ytdl.yt_dlp.utils.YoutubeDLError("stream was deleted")
|
|
|
|
with patch.object(DownloadQueue, "_DownloadQueue__extract_info", boom):
|
|
for _ in range(ytdl._LIVE_PROBE_MAX_FAILURES):
|
|
await dq._probe_scheduled_download(download)
|
|
|
|
assert url not in dq._scheduled_probe_at
|
|
assert not dq.queue.exists(url)
|
|
assert dq.done.exists(url)
|
|
assert download.info.status == "error"
|
|
notifier.completed.assert_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_probe_recovers_after_transient_then_starts(dq_env):
|
|
"""A transient failure followed by a successful live probe should start the download."""
|
|
import ytdl
|
|
|
|
notifier = AsyncMock()
|
|
url = "https://example.com/live-recover"
|
|
start_mock = AsyncMock()
|
|
|
|
dq = DownloadQueue(dq_env, notifier)
|
|
with patch.object(DownloadQueue, "_DownloadQueue__start_download", start_mock):
|
|
await dq.add_entry(
|
|
_upcoming_entry(url),
|
|
"video", "auto", "any", "best", "", "", 0,
|
|
auto_start=True,
|
|
)
|
|
download = dq.queue.get(url)
|
|
# The scheduling placeholder error is set on add.
|
|
assert download.info.error
|
|
|
|
def boom(self, *args, **kwargs):
|
|
raise ytdl.yt_dlp.utils.YoutubeDLError("temporary glitch")
|
|
|
|
with patch.object(DownloadQueue, "_DownloadQueue__extract_info", boom):
|
|
await dq._probe_scheduled_download(download)
|
|
assert dq._scheduled_probe_failures[url] == 1
|
|
|
|
def live_now(self, *args, **kwargs):
|
|
return {
|
|
"_type": "video", "id": "live1", "title": "Live Now",
|
|
"url": url, "webpage_url": url, "live_status": "is_live",
|
|
"formats": [{"format_id": "22"}],
|
|
}
|
|
|
|
with patch.object(DownloadQueue, "_DownloadQueue__extract_info", live_now), \
|
|
patch.object(DownloadQueue, "_DownloadQueue__start_download", start_mock):
|
|
await dq._probe_scheduled_download(download)
|
|
|
|
assert url not in dq._scheduled_probe_at
|
|
assert url not in dq._scheduled_probe_failures
|
|
assert download.info.status == "queued"
|
|
# Placeholder error/msg cleared now that a real download is starting.
|
|
assert download.info.error is None
|
|
assert download.info.msg is None
|
|
start_mock.assert_called_once_with(download)
|
|
|
|
|
|
def test_seconds_until_next_probe_none_when_empty(dq_env):
|
|
notifier = AsyncMock()
|
|
dq = DownloadQueue(dq_env, notifier)
|
|
assert dq._seconds_until_next_probe() is None
|
|
|
|
|
|
def test_calc_download_path_allows_subfolder(dq_env):
|
|
notifier = AsyncMock()
|
|
dq = DownloadQueue(dq_env, notifier)
|
|
path, err = dq._DownloadQueue__calc_download_path("video", "sub/dir")
|
|
assert err is None
|
|
assert os.path.realpath(path) == os.path.join(os.path.realpath(dq_env.DOWNLOAD_DIR), "sub", "dir")
|
|
|
|
|
|
def test_calc_download_path_rejects_sibling_prefix_escape(dq_env):
|
|
"""A folder resolving to a sibling sharing a name prefix must be rejected.
|
|
|
|
Regression test: ``startswith`` would have accepted ``../downloads-secret``
|
|
when the base directory is ``.../downloads``.
|
|
"""
|
|
notifier = AsyncMock()
|
|
base = os.path.realpath(dq_env.DOWNLOAD_DIR)
|
|
sibling = base + "-secret"
|
|
os.makedirs(sibling, exist_ok=True)
|
|
dq = DownloadQueue(dq_env, notifier)
|
|
escape_folder = os.path.join("..", os.path.basename(sibling), "x")
|
|
path, err = dq._DownloadQueue__calc_download_path("video", escape_folder)
|
|
assert path is None
|
|
assert err is not None and err["status"] == "error"
|
|
|
|
|
|
def test_calc_download_path_rejects_parent_escape(dq_env):
|
|
notifier = AsyncMock()
|
|
dq = DownloadQueue(dq_env, notifier)
|
|
path, err = dq._DownloadQueue__calc_download_path("video", "../../etc")
|
|
assert path is None
|
|
assert err is not None and err["status"] == "error"
|
|
|
|
|
|
def test_download_info_to_public_dict_excludes_server_only_fields():
|
|
info = DownloadInfo(
|
|
id="vid1",
|
|
title="Test Video",
|
|
url="https://example.com/watch?v=1",
|
|
quality="best",
|
|
download_type="video",
|
|
codec="auto",
|
|
format="any",
|
|
folder="",
|
|
custom_name_prefix="",
|
|
error=None,
|
|
entry={"id": "vid1", "huge": "x" * 100000},
|
|
playlist_item_limit=0,
|
|
split_by_chapters=False,
|
|
chapter_template="",
|
|
)
|
|
info.subtitle_files = [{"filename": "a.srt", "size": 10}]
|
|
public = info.to_public_dict()
|
|
assert "entry" not in public
|
|
assert "subtitle_files" not in public
|
|
# Client-facing fields are still present.
|
|
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
|
|
)
|
|
|
|
|
|
def test_download_close_releases_status_queue(dq_env):
|
|
download = _make_download(dq_env)
|
|
status_queue = MagicMock()
|
|
proc = MagicMock()
|
|
download.status_queue = status_queue
|
|
download.proc = proc
|
|
|
|
download.close()
|
|
|
|
proc.close.assert_called_once()
|
|
assert download.status_queue is None
|
|
|
|
|
|
def test_download_close_releases_status_queue_without_process(dq_env):
|
|
download = _make_download(dq_env)
|
|
download.status_queue = MagicMock()
|
|
|
|
download.close()
|
|
|
|
assert download.status_queue is None
|
|
|
|
|
|
def test_download_close_releases_status_queue_when_process_close_fails(dq_env):
|
|
download = _make_download(dq_env)
|
|
download.status_queue = MagicMock()
|
|
download.proc = MagicMock()
|
|
download.proc.close.side_effect = RuntimeError('close failed')
|
|
|
|
with pytest.raises(RuntimeError, match='close failed'):
|
|
download.close()
|
|
|
|
assert download.status_queue is None
|
|
|
|
|
|
@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")
|
|
await dq.queue.put(download)
|
|
|
|
await 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}]
|
|
await dq.queue.put(download)
|
|
|
|
await 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)
|
|
await dq.done.put(download)
|
|
|
|
await dq.clear([download.info.url])
|
|
|
|
assert os.path.exists(outside_file)
|
|
assert not dq.done.exists(download.info.url)
|