root
3cf2f8eca5
- Add Lpayer API decryption using AES (key: kiemtienmua911ca) - Add yt-dlp extraction for bypassing player blocking - Add HTTP 206 support for video validation (Range header) - Add VidMoly .biz domain support (alternative to .to) - Add SendVid extraction (working - downloaded S1 and S2 E1) - Add player fallback system with caching per anime URL - Add video URL validation before returning to downloader - Update HTTP clients with realistic browser headers - Add pycryptodome to requirements.txt - Add test file for fallback system Downloads working: SendVid (primary), Lpayer (403 issue), VidMoly (testing)
274 lines
12 KiB
Python
274 lines
12 KiB
Python
"""
|
|
Unit tests for Anime-Sama fallback mechanism
|
|
Tests player priority, caching, and URL validation
|
|
"""
|
|
import pytest
|
|
from unittest.mock import Mock, AsyncMock, patch, MagicMock
|
|
from httpx import TimeoutException, ConnectError
|
|
|
|
from app.downloaders.anime_sites.animesama import AnimeSamaDownloader
|
|
|
|
|
|
class TestAnimeSamaFallback:
|
|
"""Tests for Anime-Sama fallback mechanism"""
|
|
|
|
@pytest.fixture
|
|
def downloader(self):
|
|
"""Create AnimeSamaDownloader instance"""
|
|
return AnimeSamaDownloader()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fallback_tries_players_in_priority_order(self, downloader):
|
|
"""
|
|
Test that fallback tries players in priority order:
|
|
VidMoly -> SendVid -> Sibnet -> Lpayer
|
|
"""
|
|
# Mock each player extraction method
|
|
with patch.object(downloader, '_extract_from_vidmoly') as mock_vidmoly, \
|
|
patch.object(downloader, '_extract_from_sendvid') as mock_sendvid, \
|
|
patch.object(downloader, '_extract_from_sibnet') as mock_sibnet, \
|
|
patch.object(downloader, '_extract_from_lpayer') as mock_lpayer, \
|
|
patch.object(downloader, '_test_video_url', new_callable=AsyncMock) as mock_test_url:
|
|
|
|
# Make vidmoly and sendvid fail, sibnet succeed
|
|
mock_vidmoly.side_effect = Exception("VidMoly failed")
|
|
mock_sendvid.side_effect = Exception("SendVid failed")
|
|
mock_sibnet.return_value = ("http://sibnet.com/video.mp4", "video.mp4")
|
|
mock_lpayer.return_value = ("http://lpayer.com/video.mp4", "video.mp4")
|
|
|
|
# Make validation pass for sibnet
|
|
mock_test_url.return_value = True
|
|
|
|
result = await downloader.get_download_link_with_fallback(
|
|
"http://vidmoly.to/test",
|
|
anime_page_url="https://anime-sama.si/catalogue/test/vostfr/"
|
|
)
|
|
|
|
# Verify player order was correct
|
|
assert mock_vidmoly.called, "VidMoly should be tried first"
|
|
assert mock_sendvid.called, "SendVid should be tried second"
|
|
assert mock_sibnet.called, "Sibnet should be tried third"
|
|
assert not mock_lpayer.called, "Lpayer should not be called since Sibnet succeeded"
|
|
|
|
assert result == ("http://sibnet.com/video.mp4", "video.mp4")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_caching_mechanism_stores_working_player(self, downloader):
|
|
"""
|
|
Test that caching mechanism stores working player for same anime URL.
|
|
After first successful player, subsequent requests should use cached player first.
|
|
"""
|
|
# Setup: First request - vidmoly fails, sendvid succeeds
|
|
with patch.object(downloader, '_extract_from_vidmoly') as mock_vidmoly, \
|
|
patch.object(downloader, '_extract_from_sendvid') as mock_sendvid, \
|
|
patch.object(downloader, '_extract_from_sibnet') as mock_sibnet, \
|
|
patch.object(downloader, '_extract_from_lpayer') as mock_lpayer, \
|
|
patch.object(downloader, '_test_video_url', new_callable=AsyncMock) as mock_test_url:
|
|
|
|
# First request: vidmoly fails, sendvid succeeds
|
|
mock_vidmoly.side_effect = Exception("VidMoly failed")
|
|
mock_sendvid.return_value = ("http://sendvid.com/video.mp4", "video.mp4")
|
|
mock_test_url.return_value = True
|
|
|
|
anime_url = "https://anime-sama.si/catalogue/test/vostfr/"
|
|
|
|
result1 = await downloader.get_download_link_with_fallback(
|
|
"http://vidmoly.to/test",
|
|
anime_page_url=anime_url
|
|
)
|
|
|
|
# Verify caching worked
|
|
assert anime_url in downloader._working_players
|
|
assert downloader._working_players[anime_url] == "sendvid"
|
|
|
|
# Reset mocks for second request
|
|
mock_vidmoly.reset_mock()
|
|
mock_sendvid.reset_mock()
|
|
mock_sibnet.reset_mock()
|
|
mock_lpayer.reset_mock()
|
|
|
|
# Second request: Should try sendvid first (cached)
|
|
mock_sendvid.return_value = ("http://sendvid.com/video2.mp4", "video2.mp4")
|
|
mock_test_url.return_value = True
|
|
|
|
result2 = await downloader.get_download_link_with_fallback(
|
|
"http://vidmoly.to/test",
|
|
anime_page_url=anime_url
|
|
)
|
|
|
|
# Verify sendvid was tried first (due to cache)
|
|
assert mock_sendvid.call_count == 1, "Cached player (sendvid) should be tried first"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_all_players_failing_raises_exception(self, downloader):
|
|
"""
|
|
Test that when all players fail, an exception is raised with proper error message.
|
|
"""
|
|
with patch.object(downloader, '_extract_from_vidmoly') as mock_vidmoly, \
|
|
patch.object(downloader, '_extract_from_sendvid') as mock_sendvid, \
|
|
patch.object(downloader, '_extract_from_sibnet') as mock_sibnet, \
|
|
patch.object(downloader, '_extract_from_lpayer') as mock_lpayer:
|
|
|
|
# All players fail
|
|
mock_vidmoly.side_effect = Exception("VidMoly error")
|
|
mock_sendvid.side_effect = Exception("SendVid error")
|
|
mock_sibnet.side_effect = Exception("Sibnet error")
|
|
mock_lpayer.side_effect = Exception("Lpayer error")
|
|
|
|
anime_url = "https://anime-sama.si/catalogue/test/vostfr/"
|
|
|
|
with pytest.raises(Exception) as exc_info:
|
|
await downloader.get_download_link_with_fallback(
|
|
"http://vidmoly.to/test",
|
|
anime_page_url=anime_url
|
|
)
|
|
|
|
# Verify error message mentions all players failed
|
|
assert "All players failed" in str(exc_info.value)
|
|
|
|
# Verify all players were tried
|
|
assert mock_vidmoly.called
|
|
assert mock_sendvid.called
|
|
assert mock_sibnet.called
|
|
assert mock_lpayer.called
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_test_video_url_returns_true_for_valid_url(self, downloader):
|
|
"""
|
|
Test that _test_video_url returns True for valid video URL (HTTP 200 with content).
|
|
"""
|
|
# Mock the client to return valid response
|
|
mock_response = Mock()
|
|
mock_response.status_code = 200
|
|
mock_response.content = b"video content data"
|
|
|
|
with patch.object(downloader.client, 'get', new_callable=AsyncMock) as mock_get:
|
|
mock_get.return_value = mock_response
|
|
|
|
result = await downloader._test_video_url("http://example.com/video.mp4")
|
|
|
|
assert result is True
|
|
mock_get.assert_called_once()
|
|
# Verify Range header was included
|
|
call_args = mock_get.call_args
|
|
assert "Range" in call_args.kwargs.get("headers", {})
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_test_video_url_returns_false_for_invalid_url(self, downloader):
|
|
"""
|
|
Test that _test_video_url returns False for invalid/non-working URL.
|
|
"""
|
|
# Test case 1: HTTP error status
|
|
mock_response = Mock()
|
|
mock_response.status_code = 404
|
|
|
|
with patch.object(downloader.client, 'get', new_callable=AsyncMock) as mock_get:
|
|
mock_get.return_value = mock_response
|
|
|
|
result = await downloader._test_video_url("http://example.com/notfound.mp4")
|
|
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_test_video_url_returns_false_for_empty_response(self, downloader):
|
|
"""
|
|
Test that _test_video_url returns False for empty response content.
|
|
"""
|
|
mock_response = Mock()
|
|
mock_response.status_code = 200
|
|
mock_response.content = b"" # Empty content
|
|
|
|
with patch.object(downloader.client, 'get', new_callable=AsyncMock) as mock_get:
|
|
mock_get.return_value = mock_response
|
|
|
|
result = await downloader._test_video_url("http://example.com/empty.mp4")
|
|
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_test_video_url_returns_false_for_timeout(self, downloader):
|
|
"""
|
|
Test that _test_video_url returns False for timeout.
|
|
"""
|
|
with patch.object(downloader.client, 'get', new_callable=AsyncMock) as mock_get:
|
|
mock_get.side_effect = TimeoutException("Request timeout")
|
|
|
|
result = await downloader._test_video_url("http://example.com/slow.mp4")
|
|
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_test_video_url_returns_false_for_connection_error(self, downloader):
|
|
"""
|
|
Test that _test_video_url returns False for connection error.
|
|
"""
|
|
with patch.object(downloader.client, 'get', new_callable=AsyncMock) as mock_get:
|
|
mock_get.side_effect = ConnectError("Connection failed")
|
|
|
|
result = await downloader._test_video_url("http://example.com/badhost.mp4")
|
|
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fallback_skips_invalid_player_url(self, downloader):
|
|
"""
|
|
Test that fallback skips players that return invalid URLs (validation fails).
|
|
"""
|
|
with patch.object(downloader, '_extract_from_vidmoly') as mock_vidmoly, \
|
|
patch.object(downloader, '_extract_from_sendvid') as mock_sendvid, \
|
|
patch.object(downloader, '_extract_from_sibnet') as mock_sibnet, \
|
|
patch.object(downloader, '_test_video_url', new_callable=AsyncMock) as mock_test_url:
|
|
|
|
# Vidmoly returns URL but validation fails
|
|
mock_vidmoly.return_value = ("http://vidmoly.com/video.mp4", "video.mp4")
|
|
|
|
# SendVid returns URL and validation passes
|
|
mock_sendvid.return_value = ("http://sendvid.com/video.mp4", "video.mp4")
|
|
mock_sibnet.return_value = ("http://sibnet.com/video.mp4", "video.mp4")
|
|
|
|
# First call (vidmoly): validation fails
|
|
# Second call (sendvid): validation passes
|
|
# Third call (sibnet): not called because sendvid succeeded
|
|
mock_test_url.side_effect = [False, True]
|
|
|
|
result = await downloader.get_download_link_with_fallback(
|
|
"http://vidmoly.to/test",
|
|
anime_page_url="https://anime-sama.si/catalogue/test/vostfr/"
|
|
)
|
|
|
|
# Verify validation was called for vidmoly
|
|
assert mock_test_url.call_count >= 1
|
|
# Verify sendvid was also tried after vidmoly failed validation
|
|
assert mock_sendvid.called
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cache_not_used_without_anime_page_url(self, downloader):
|
|
"""
|
|
Test that caching is not used when anime_page_url is not provided.
|
|
"""
|
|
with patch.object(downloader, '_extract_from_vidmoly') as mock_vidmoly, \
|
|
patch.object(downloader, '_extract_from_sendvid') as mock_sendvid, \
|
|
patch.object(downloader, '_test_video_url', new_callable=AsyncMock) as mock_test_url:
|
|
|
|
# First request: no anime_page_url, vidmoly succeeds
|
|
mock_vidmoly.return_value = ("http://vidmoly.com/video.mp4", "video.mp4")
|
|
mock_test_url.return_value = True
|
|
|
|
result1 = await downloader.get_download_link_with_fallback(
|
|
"http://vidmoly.to/test"
|
|
)
|
|
|
|
# Cache should be empty (no anime_page_url provided)
|
|
assert len(downloader._working_players) == 0
|
|
|
|
# Second request: still no anime_page_url, should not use cache
|
|
mock_vidmoly.reset_mock()
|
|
|
|
result2 = await downloader.get_download_link_with_fallback(
|
|
"http://vidmoly.to/test"
|
|
)
|
|
|
|
# Vidmoly should still be called (no cache used)
|
|
assert mock_vidmoly.call_count == 1
|