Roman/ohm_streaming
1
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

feat: add multiple video player support for Frieren S2 downloads

Browse Source 
- 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)
This commit is contained in:
root
2026-02-25 16:29:53 +00:00
parent 8b7a419b4c
commit 3cf2f8eca5
9 changed files with 1370 additions and 184 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_anime_sama_fallback.py
+273
View File
@@ -0,0 +1,273 @@
"""
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