fix(providers): rotate, probe and fall back on 404
Iterate providers actually advertised on the episode page (ordered by SUPPORTED_PROVIDERS preference) instead of always re-resolving VOE. Each candidate is HEAD-probed before yt-dlp runs, so dead links are skipped immediately; direct video URLs use a streaming fast path that bypasses yt-dlp; total failure now logs the exhausted provider list. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
@@ -1,9 +1,11 @@
|
||||
"""Unit tests for aniworld_provider.py - Anime catalog scraping, episode listing, streaming link extraction."""
|
||||
|
||||
import json
|
||||
import os
|
||||
from unittest.mock import MagicMock, Mock, patch
|
||||
|
||||
import pytest
|
||||
import requests
|
||||
|
||||
from src.core.providers.aniworld_provider import AniworldLoader
|
||||
|
||||
@@ -472,3 +474,250 @@ class TestAniworldEvents:
|
||||
# Fire event - handler should NOT be called
|
||||
loader.events.download_progress({"status": "downloading"})
|
||||
handler.assert_not_called()
|
||||
|
||||
|
||||
class TestAniworldHealthCheck:
|
||||
"""Tests for the _check_url_alive HEAD probe."""
|
||||
|
||||
def test_returns_true_on_200(self, loader):
|
||||
loader.session.head.return_value = MagicMock(status_code=200)
|
||||
assert loader._check_url_alive("https://provider/x") is True
|
||||
|
||||
def test_returns_false_on_404(self, loader):
|
||||
loader.session.head.return_value = MagicMock(status_code=404)
|
||||
assert loader._check_url_alive("https://provider/x") is False
|
||||
|
||||
def test_returns_false_on_403(self, loader):
|
||||
loader.session.head.return_value = MagicMock(status_code=403)
|
||||
assert loader._check_url_alive("https://provider/x") is False
|
||||
|
||||
def test_falls_back_to_get_when_head_disallowed(self, loader):
|
||||
loader.session.head.return_value = MagicMock(status_code=405)
|
||||
get_resp = MagicMock(status_code=200)
|
||||
get_resp.close = MagicMock()
|
||||
loader.session.get.return_value = get_resp
|
||||
assert loader._check_url_alive("https://provider/x") is True
|
||||
loader.session.get.assert_called_once()
|
||||
|
||||
def test_returns_false_on_connection_error(self, loader):
|
||||
loader.session.head.side_effect = requests.ConnectionError("boom")
|
||||
assert loader._check_url_alive("https://provider/x") is False
|
||||
|
||||
|
||||
class TestAniworldDirectStream:
|
||||
"""Tests for the _try_direct_stream fast-path."""
|
||||
|
||||
def _build_response(self, status, content_type, body=b""):
|
||||
resp = MagicMock()
|
||||
resp.ok = status < 400
|
||||
resp.status_code = status
|
||||
resp.headers = {"Content-Type": content_type}
|
||||
resp.iter_content = MagicMock(return_value=[body])
|
||||
resp.__enter__ = MagicMock(return_value=resp)
|
||||
resp.__exit__ = MagicMock(return_value=False)
|
||||
return resp
|
||||
|
||||
def test_skips_non_video_content(self, loader, tmp_path):
|
||||
target = tmp_path / "out.mp4"
|
||||
loader.session.get.return_value = self._build_response(
|
||||
200, "text/html"
|
||||
)
|
||||
assert loader._try_direct_stream(
|
||||
"https://x", str(target), None, 10
|
||||
) is False
|
||||
assert not target.exists()
|
||||
|
||||
def test_writes_video_content(self, loader, tmp_path):
|
||||
target = tmp_path / "out.mp4"
|
||||
loader.session.get.return_value = self._build_response(
|
||||
200, "video/mp4", body=b"abc123"
|
||||
)
|
||||
assert loader._try_direct_stream(
|
||||
"https://x", str(target), None, 10
|
||||
) is True
|
||||
assert target.read_bytes() == b"abc123"
|
||||
|
||||
def test_returns_false_on_http_error(self, loader, tmp_path):
|
||||
target = tmp_path / "out.mp4"
|
||||
loader.session.get.return_value = self._build_response(
|
||||
404, "video/mp4"
|
||||
)
|
||||
assert loader._try_direct_stream(
|
||||
"https://x", str(target), None, 10
|
||||
) is False
|
||||
|
||||
def test_returns_false_on_request_exception(self, loader, tmp_path):
|
||||
loader.session.get.side_effect = requests.RequestException("nope")
|
||||
assert loader._try_direct_stream(
|
||||
"https://x", str(tmp_path / "out.mp4"), None, 10
|
||||
) is False
|
||||
|
||||
|
||||
class TestAniworldProviderSelection:
|
||||
"""Tests for _select_providers_for_episode ordering and filtering."""
|
||||
|
||||
def test_orders_by_supported_preference(self, loader):
|
||||
loader.is_language = MagicMock(return_value=True)
|
||||
loader._get_provider_from_html = MagicMock(return_value={
|
||||
"Vidoza": {1: "https://aniworld.to/redirect/2"},
|
||||
"VOE": {1: "https://aniworld.to/redirect/1"},
|
||||
})
|
||||
result = loader._select_providers_for_episode(1, 1, "k", "German Dub")
|
||||
assert [name for name, _ in result] == ["VOE", "Vidoza"]
|
||||
|
||||
def test_filters_by_language(self, loader):
|
||||
loader.is_language = MagicMock(return_value=True)
|
||||
loader._get_provider_from_html = MagicMock(return_value={
|
||||
"VOE": {2: "https://aniworld.to/redirect/1"}, # English only
|
||||
})
|
||||
result = loader._select_providers_for_episode(1, 1, "k", "German Dub")
|
||||
assert result == []
|
||||
|
||||
def test_returns_empty_when_language_unavailable(self, loader):
|
||||
loader.is_language = MagicMock(return_value=False)
|
||||
loader._get_provider_from_html = MagicMock()
|
||||
result = loader._select_providers_for_episode(1, 1, "k", "German Dub")
|
||||
assert result == []
|
||||
loader._get_provider_from_html.assert_not_called()
|
||||
|
||||
|
||||
class TestAniworldDownloadFailover:
|
||||
"""Tests for the failover rotation in download()."""
|
||||
|
||||
@pytest.fixture
|
||||
def patched_loader(self, loader, tmp_path):
|
||||
"""Loader with side-effect heavy methods stubbed."""
|
||||
loader.get_title = MagicMock(return_value="Anime")
|
||||
loader._select_providers_for_episode = MagicMock(return_value=[
|
||||
("VOE", "https://aniworld.to/redirect/1"),
|
||||
("Doodstream", "https://aniworld.to/redirect/2"),
|
||||
])
|
||||
loader._check_url_alive = MagicMock(return_value=True)
|
||||
loader._try_direct_stream = MagicMock(return_value=False)
|
||||
loader.clear_cache = MagicMock()
|
||||
loader._resolve_direct_link = MagicMock(
|
||||
return_value=("https://cdn/video.m3u8", {"Referer": "https://x"})
|
||||
)
|
||||
return loader
|
||||
|
||||
def test_skips_provider_when_url_dead(self, patched_loader, tmp_path):
|
||||
# First provider URL fails health check, second succeeds and downloads
|
||||
patched_loader._check_url_alive.side_effect = [False, True]
|
||||
|
||||
def fake_ytdl(opts):
|
||||
outpath = opts["outtmpl"]
|
||||
os.makedirs(os.path.dirname(outpath), exist_ok=True)
|
||||
with open(outpath, "wb") as fh:
|
||||
fh.write(b"data")
|
||||
ydl = MagicMock()
|
||||
ydl.__enter__ = MagicMock(return_value=ydl)
|
||||
ydl.__exit__ = MagicMock(return_value=False)
|
||||
ydl.extract_info = MagicMock(return_value={"title": "t"})
|
||||
return ydl
|
||||
|
||||
with patch(
|
||||
"src.core.providers.aniworld_provider.YoutubeDL",
|
||||
side_effect=fake_ytdl,
|
||||
):
|
||||
result = patched_loader.download(
|
||||
str(tmp_path), "Anime", 1, 1, "k", "German Dub"
|
||||
)
|
||||
assert result is True
|
||||
assert patched_loader._check_url_alive.call_count == 2
|
||||
# Only second provider (Doodstream) attempted resolve
|
||||
patched_loader._resolve_direct_link.assert_called_once_with(
|
||||
"https://aniworld.to/redirect/2", "Doodstream"
|
||||
)
|
||||
|
||||
def test_falls_back_to_next_provider_on_ytdl_error(
|
||||
self, patched_loader, tmp_path
|
||||
):
|
||||
calls = {"n": 0}
|
||||
|
||||
def fake_ytdl(opts):
|
||||
calls["n"] += 1
|
||||
if calls["n"] == 1:
|
||||
raise Exception("HTTP 404 from VOE")
|
||||
outpath = opts["outtmpl"]
|
||||
os.makedirs(os.path.dirname(outpath), exist_ok=True)
|
||||
with open(outpath, "wb") as fh:
|
||||
fh.write(b"ok")
|
||||
ydl = MagicMock()
|
||||
ydl.__enter__ = MagicMock(return_value=ydl)
|
||||
ydl.__exit__ = MagicMock(return_value=False)
|
||||
ydl.extract_info = MagicMock(return_value={"title": "t"})
|
||||
return ydl
|
||||
|
||||
with patch(
|
||||
"src.core.providers.aniworld_provider.YoutubeDL",
|
||||
side_effect=fake_ytdl,
|
||||
):
|
||||
result = patched_loader.download(
|
||||
str(tmp_path), "Anime", 1, 1, "k", "German Dub"
|
||||
)
|
||||
assert result is True
|
||||
assert calls["n"] == 2
|
||||
|
||||
def test_uses_direct_stream_when_available(
|
||||
self, patched_loader, tmp_path
|
||||
):
|
||||
def write_direct(link, output, headers, timeout):
|
||||
os.makedirs(os.path.dirname(output), exist_ok=True)
|
||||
with open(output, "wb") as fh:
|
||||
fh.write(b"vid")
|
||||
return True
|
||||
|
||||
patched_loader._try_direct_stream.side_effect = write_direct
|
||||
|
||||
with patch(
|
||||
"src.core.providers.aniworld_provider.YoutubeDL"
|
||||
) as mock_ydl:
|
||||
result = patched_loader.download(
|
||||
str(tmp_path), "Anime", 1, 1, "k", "German Dub"
|
||||
)
|
||||
assert result is True
|
||||
mock_ydl.assert_not_called()
|
||||
|
||||
def test_returns_false_when_all_providers_fail(
|
||||
self, patched_loader, tmp_path, caplog
|
||||
):
|
||||
with patch(
|
||||
"src.core.providers.aniworld_provider.YoutubeDL",
|
||||
side_effect=Exception("HTTP 404"),
|
||||
):
|
||||
result = patched_loader.download(
|
||||
str(tmp_path), "Anime", 1, 1, "k", "German Dub"
|
||||
)
|
||||
assert result is False
|
||||
assert "All download providers failed" in caplog.text
|
||||
# Both providers attempted
|
||||
assert patched_loader._resolve_direct_link.call_count == 2
|
||||
|
||||
def test_returns_false_when_no_providers_advertised(
|
||||
self, patched_loader, tmp_path, caplog
|
||||
):
|
||||
patched_loader._select_providers_for_episode.return_value = []
|
||||
result = patched_loader.download(
|
||||
str(tmp_path), "Anime", 1, 1, "k", "German Dub"
|
||||
)
|
||||
assert result is False
|
||||
assert "No providers advertised" in caplog.text
|
||||
|
||||
|
||||
class TestAniworldHeaderParsing:
|
||||
"""_parse_provider_headers normalizes legacy strings to dict."""
|
||||
|
||||
def test_parses_referer(self):
|
||||
result = AniworldLoader._parse_provider_headers(
|
||||
['Referer: "https://vidmoly.to"']
|
||||
)
|
||||
assert result == {"Referer": "https://vidmoly.to"}
|
||||
|
||||
def test_handles_none(self):
|
||||
assert AniworldLoader._parse_provider_headers(None) == {}
|
||||
|
||||
def test_skips_malformed_entries(self):
|
||||
result = AniworldLoader._parse_provider_headers(
|
||||
["not-a-header", "Key: value"]
|
||||
)
|
||||
assert result == {"Key": "value"}
|
||||
|
||||
Reference in New Issue
Block a user