mirror of
https://github.com/alexta69/metube.git
synced 2026-09-21 21:45:04 +00:00
Added handling for unsupported URL
This commit is contained in:
@@ -115,6 +115,54 @@ async def test_add_single_video_goes_to_pending_when_auto_start_false(dq_env):
|
||||
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, ytdl_options_presets=None, ytdl_options_overrides=None):
|
||||
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_cancel_removes_from_pending(dq_env):
|
||||
notifier = AsyncMock()
|
||||
|
||||
@@ -60,6 +60,7 @@ from ytdl import (
|
||||
_resolve_outtmpl_fields,
|
||||
_sanitize_entry_for_pickle,
|
||||
_sanitize_path_component,
|
||||
_short_title_for_failed_url,
|
||||
)
|
||||
|
||||
# Detect whether the real yt-dlp is loaded (as opposed to the minimal fake
|
||||
@@ -747,5 +748,23 @@ class CompactPersistedEntryTests(unittest.TestCase):
|
||||
self.assertIsNone(_compact_persisted_entry({"id": "x", "title": "y"}))
|
||||
|
||||
|
||||
class ShortTitleForFailedUrlTests(unittest.TestCase):
|
||||
def test_uses_hostname_for_a_normal_url(self):
|
||||
self.assertEqual(
|
||||
_short_title_for_failed_url("https://example.com/watch?v=1"),
|
||||
"example.com",
|
||||
)
|
||||
|
||||
def test_falls_back_to_raw_value_when_there_is_no_hostname(self):
|
||||
# file:// URIs and bare search terms/video IDs have no netloc to extract.
|
||||
self.assertEqual(_short_title_for_failed_url("file:///etc/passwd"), "file:///etc/passwd")
|
||||
self.assertEqual(_short_title_for_failed_url("ytsearch:some query"), "ytsearch:some query")
|
||||
|
||||
def test_falls_back_to_raw_value_on_unparseable_input(self):
|
||||
# A malformed IPv6-looking host raises ValueError in urlsplit().hostname.
|
||||
malformed = "https://[::1/watch"
|
||||
self.assertEqual(_short_title_for_failed_url(malformed), malformed)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user