Lukas
9078a6f3dc
- Fixed test_download_progress_websocket: stop() -> stop_downloads() - Fixed test_download_service: start() -> initialize(), stop() -> stop_downloads() - Resolved 8 test errors and 3 test failures - Test status: 970 passing, 31 failing (down from 967 passing, 34 failing, 8 errors) - All 104 NFO-related tests still passing (100%)
495 lines
16 KiB
Python
495 lines
16 KiB
Python
"""Unit tests for image downloader."""
|
|
|
|
import io
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import aiohttp
|
|
import pytest
|
|
from PIL import Image
|
|
|
|
from src.core.utils.image_downloader import ImageDownloader, ImageDownloadError
|
|
|
|
|
|
@pytest.fixture
|
|
def image_downloader():
|
|
"""Create image downloader instance."""
|
|
# Use smaller min_file_size for tests since test images are small
|
|
return ImageDownloader(min_file_size=100)
|
|
|
|
|
|
@pytest.fixture
|
|
def valid_image_bytes():
|
|
"""Create valid test image bytes."""
|
|
img = Image.new('RGB', (100, 100), color='red')
|
|
buf = io.BytesIO()
|
|
img.save(buf, format='JPEG')
|
|
return buf.getvalue()
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_session():
|
|
"""Create mock aiohttp session."""
|
|
mock = AsyncMock()
|
|
# Make get() return an async context manager
|
|
mock.get = MagicMock()
|
|
mock.closed = False
|
|
return mock
|
|
|
|
|
|
class TestImageDownloaderInit:
|
|
"""Test ImageDownloader initialization."""
|
|
|
|
def test_init_default_values(self):
|
|
"""Test initialization with default values."""
|
|
downloader = ImageDownloader()
|
|
|
|
assert downloader.min_file_size == 1024
|
|
assert downloader.max_retries == 3
|
|
assert downloader.retry_delay == 1.0
|
|
assert downloader.timeout == 30
|
|
assert downloader.session is None
|
|
|
|
def test_init_custom_values(self):
|
|
"""Test initialization with custom values."""
|
|
downloader = ImageDownloader(
|
|
min_file_size=5000,
|
|
max_retries=5,
|
|
retry_delay=2.0,
|
|
timeout=60
|
|
)
|
|
|
|
assert downloader.min_file_size == 5000
|
|
assert downloader.max_retries == 5
|
|
assert downloader.retry_delay == 2.0
|
|
assert downloader.timeout == 60
|
|
|
|
|
|
class TestImageDownloaderContextManager:
|
|
"""Test ImageDownloader as context manager."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_async_context_manager(self, image_downloader):
|
|
"""Test async context manager creates session."""
|
|
async with image_downloader as d:
|
|
assert d.session is not None
|
|
|
|
assert image_downloader.session is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_close_closes_session(self, image_downloader):
|
|
"""Test close method closes session."""
|
|
await image_downloader.__aenter__()
|
|
|
|
assert image_downloader.session is not None
|
|
await image_downloader.close()
|
|
assert image_downloader.session is None
|
|
|
|
|
|
class TestImageDownloaderValidateImage:
|
|
"""Test validate_image method."""
|
|
|
|
def test_validate_valid_image(self, image_downloader, valid_image_bytes, tmp_path):
|
|
"""Test validation of valid image."""
|
|
image_path = tmp_path / "valid.jpg"
|
|
image_path.write_bytes(valid_image_bytes)
|
|
|
|
result = image_downloader.validate_image(image_path)
|
|
assert result is True
|
|
|
|
def test_validate_too_small(self, image_downloader, tmp_path):
|
|
"""Test validation rejects too-small file."""
|
|
tiny_data = b"tiny"
|
|
image_path = tmp_path / "tiny.jpg"
|
|
image_path.write_bytes(tiny_data)
|
|
|
|
result = image_downloader.validate_image(image_path)
|
|
assert result is False
|
|
|
|
def test_validate_invalid_image_data(self, image_downloader, tmp_path):
|
|
"""Test validation rejects invalid image data."""
|
|
invalid_data = b"x" * 2000 # Large enough but not an image
|
|
image_path = tmp_path / "invalid.jpg"
|
|
image_path.write_bytes(invalid_data)
|
|
|
|
result = image_downloader.validate_image(image_path)
|
|
assert result is False
|
|
|
|
def test_validate_corrupted_image(self, image_downloader, tmp_path):
|
|
"""Test validation rejects corrupted image."""
|
|
# Create a corrupted JPEG-like file
|
|
corrupted = b"\xFF\xD8\xFF\xE0" + b"corrupted_data" * 100
|
|
image_path = tmp_path / "corrupted.jpg"
|
|
image_path.write_bytes(corrupted)
|
|
|
|
result = image_downloader.validate_image(image_path)
|
|
assert result is False
|
|
|
|
|
|
class TestImageDownloaderDownloadImage:
|
|
"""Test download_image method."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_download_image_success(
|
|
self,
|
|
image_downloader,
|
|
valid_image_bytes,
|
|
tmp_path
|
|
):
|
|
"""Test successful image download."""
|
|
mock_session = AsyncMock()
|
|
mock_session.closed = False
|
|
mock_response = AsyncMock()
|
|
mock_response.status = 200
|
|
mock_response.read = AsyncMock(return_value=valid_image_bytes)
|
|
|
|
# Setup async context manager for session.get()
|
|
mock_cm = MagicMock()
|
|
mock_cm.__aenter__ = AsyncMock(return_value=mock_response)
|
|
mock_cm.__aexit__ = AsyncMock(return_value=None)
|
|
mock_session.get = MagicMock(return_value=mock_cm)
|
|
|
|
image_downloader.session = mock_session
|
|
|
|
output_path = tmp_path / "test.jpg"
|
|
await image_downloader.download_image(
|
|
"https://test.com/image.jpg", output_path
|
|
)
|
|
|
|
assert output_path.exists()
|
|
assert output_path.stat().st_size == len(valid_image_bytes)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_download_image_skip_existing(
|
|
self,
|
|
image_downloader,
|
|
tmp_path
|
|
):
|
|
"""Test skipping existing file."""
|
|
output_path = tmp_path / "existing.jpg"
|
|
# Write a file large enough to pass min_file_size check
|
|
output_path.write_bytes(b"x" * 200)
|
|
|
|
mock_session = AsyncMock()
|
|
mock_session.closed = False
|
|
image_downloader.session = mock_session
|
|
|
|
result = await image_downloader.download_image(
|
|
"https://test.com/image.jpg",
|
|
output_path,
|
|
skip_existing=True
|
|
)
|
|
|
|
assert result is True
|
|
assert output_path.read_bytes() == b"x" * 200 # Unchanged
|
|
assert not mock_session.get.called
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_download_image_overwrite_existing(
|
|
self,
|
|
image_downloader,
|
|
valid_image_bytes,
|
|
tmp_path
|
|
):
|
|
"""Test overwriting existing file."""
|
|
output_path = tmp_path / "existing.jpg"
|
|
output_path.write_bytes(b"old")
|
|
|
|
mock_session = AsyncMock()
|
|
mock_session.closed = False
|
|
mock_response = AsyncMock()
|
|
mock_response.status = 200
|
|
mock_response.read = AsyncMock(return_value=valid_image_bytes)
|
|
|
|
# Setup async context manager for session.get()
|
|
mock_cm = MagicMock()
|
|
mock_cm.__aenter__ = AsyncMock(return_value=mock_response)
|
|
mock_cm.__aexit__ = AsyncMock(return_value=None)
|
|
mock_session.get = MagicMock(return_value=mock_cm)
|
|
|
|
image_downloader.session = mock_session
|
|
|
|
await image_downloader.download_image(
|
|
"https://test.com/image.jpg",
|
|
output_path,
|
|
skip_existing=False
|
|
)
|
|
|
|
assert output_path.exists()
|
|
assert output_path.read_bytes() == valid_image_bytes
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_download_image_invalid_url(self, image_downloader, tmp_path):
|
|
"""Test download with invalid URL."""
|
|
mock_session = AsyncMock()
|
|
mock_session.closed = False
|
|
mock_response = AsyncMock()
|
|
mock_response.status = 404
|
|
|
|
# Setup async context manager for session.get()
|
|
mock_cm = MagicMock()
|
|
mock_cm.__aenter__ = AsyncMock(return_value=mock_response)
|
|
mock_cm.__aexit__ = AsyncMock(return_value=None)
|
|
mock_session.get = MagicMock(return_value=mock_cm)
|
|
|
|
image_downloader.session = mock_session
|
|
|
|
output_path = tmp_path / "test.jpg"
|
|
|
|
result = await image_downloader.download_image(
|
|
"https://test.com/missing.jpg",
|
|
output_path
|
|
)
|
|
assert result is False # 404 returns False, not exception
|
|
|
|
|
|
class TestImageDownloaderSpecificMethods:
|
|
"""Test type-specific download methods."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_download_poster(self, image_downloader, valid_image_bytes, tmp_path):
|
|
"""Test download_poster method."""
|
|
with patch.object(
|
|
image_downloader,
|
|
'download_image',
|
|
new_callable=AsyncMock
|
|
) as mock_download:
|
|
await image_downloader.download_poster(
|
|
"https://test.com/poster.jpg",
|
|
tmp_path
|
|
)
|
|
|
|
mock_download.assert_called_once()
|
|
call_args = mock_download.call_args
|
|
assert call_args[0][1] == tmp_path / "poster.jpg"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_download_logo(self, image_downloader, tmp_path):
|
|
"""Test download_logo method."""
|
|
with patch.object(
|
|
image_downloader,
|
|
'download_image',
|
|
new_callable=AsyncMock
|
|
) as mock_download:
|
|
await image_downloader.download_logo(
|
|
"https://test.com/logo.png",
|
|
tmp_path
|
|
)
|
|
|
|
mock_download.assert_called_once()
|
|
call_args = mock_download.call_args
|
|
assert call_args[0][1] == tmp_path / "logo.png"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_download_fanart(self, image_downloader, tmp_path):
|
|
"""Test download_fanart method."""
|
|
with patch.object(
|
|
image_downloader,
|
|
'download_image',
|
|
new_callable=AsyncMock
|
|
) as mock_download:
|
|
await image_downloader.download_fanart(
|
|
"https://test.com/fanart.jpg",
|
|
tmp_path
|
|
)
|
|
|
|
mock_download.assert_called_once()
|
|
call_args = mock_download.call_args
|
|
assert call_args[0][1] == tmp_path / "fanart.jpg"
|
|
|
|
|
|
class TestImageDownloaderDownloadAll:
|
|
"""Test download_all_media method."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_download_all_success(self, image_downloader, tmp_path):
|
|
"""Test downloading all media types."""
|
|
with patch.object(
|
|
image_downloader,
|
|
'download_poster',
|
|
new_callable=AsyncMock,
|
|
return_value=True
|
|
), patch.object(
|
|
image_downloader,
|
|
'download_logo',
|
|
new_callable=AsyncMock,
|
|
return_value=True
|
|
), patch.object(
|
|
image_downloader,
|
|
'download_fanart',
|
|
new_callable=AsyncMock,
|
|
return_value=True
|
|
):
|
|
results = await image_downloader.download_all_media(
|
|
tmp_path,
|
|
poster_url="https://test.com/poster.jpg",
|
|
logo_url="https://test.com/logo.png",
|
|
fanart_url="https://test.com/fanart.jpg"
|
|
)
|
|
|
|
assert results["poster"] is True
|
|
assert results["logo"] is True
|
|
assert results["fanart"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_download_all_partial(self, image_downloader, tmp_path):
|
|
"""Test downloading with some URLs missing."""
|
|
with patch.object(
|
|
image_downloader,
|
|
'download_poster',
|
|
new_callable=AsyncMock,
|
|
return_value=True
|
|
), patch.object(
|
|
image_downloader,
|
|
'download_logo',
|
|
new_callable=AsyncMock
|
|
) as mock_logo, patch.object(
|
|
image_downloader,
|
|
'download_fanart',
|
|
new_callable=AsyncMock
|
|
) as mock_fanart:
|
|
results = await image_downloader.download_all_media(
|
|
tmp_path,
|
|
poster_url="https://test.com/poster.jpg",
|
|
logo_url=None,
|
|
fanart_url=None
|
|
)
|
|
|
|
assert results["poster"] is True
|
|
assert results["logo"] is None
|
|
assert results["fanart"] is None
|
|
assert not mock_logo.called
|
|
assert not mock_fanart.called
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_download_all_with_failures(self, image_downloader, tmp_path):
|
|
"""Test downloading with some failures."""
|
|
with patch.object(
|
|
image_downloader,
|
|
'download_poster',
|
|
new_callable=AsyncMock,
|
|
return_value=True
|
|
), patch.object(
|
|
image_downloader,
|
|
'download_logo',
|
|
new_callable=AsyncMock,
|
|
side_effect=ImageDownloadError("Failed")
|
|
), patch.object(
|
|
image_downloader,
|
|
'download_fanart',
|
|
new_callable=AsyncMock,
|
|
return_value=True
|
|
):
|
|
results = await image_downloader.download_all_media(
|
|
tmp_path,
|
|
poster_url="https://test.com/poster.jpg",
|
|
logo_url="https://test.com/logo.png",
|
|
fanart_url="https://test.com/fanart.jpg"
|
|
)
|
|
|
|
assert results["poster"] is True
|
|
assert results["logo"] is False
|
|
assert results["fanart"] is True
|
|
|
|
|
|
class TestImageDownloaderRetryLogic:
|
|
"""Test retry logic."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_retry_on_failure(
|
|
self, image_downloader, valid_image_bytes, tmp_path
|
|
):
|
|
"""Test retry logic on temporary failure."""
|
|
mock_session = AsyncMock()
|
|
mock_session.closed = False
|
|
|
|
# First two calls fail, third succeeds
|
|
mock_response_fail = AsyncMock()
|
|
mock_response_fail.status = 500
|
|
|
|
# Create mock RequestInfo for ClientResponseError
|
|
mock_request_info = MagicMock()
|
|
mock_request_info.real_url = "https://test.com/image.jpg"
|
|
|
|
mock_response_fail.raise_for_status = MagicMock(
|
|
side_effect=aiohttp.ClientResponseError(
|
|
request_info=mock_request_info,
|
|
history=(),
|
|
status=500,
|
|
message="Server Error"
|
|
)
|
|
)
|
|
|
|
mock_response_success = AsyncMock()
|
|
mock_response_success.status = 200
|
|
mock_response_success.read = AsyncMock(return_value=valid_image_bytes)
|
|
|
|
# Setup context managers
|
|
mock_cm_fail = MagicMock()
|
|
mock_cm_fail.__aenter__ = AsyncMock(return_value=mock_response_fail)
|
|
mock_cm_fail.__aexit__ = AsyncMock(return_value=None)
|
|
|
|
mock_cm_success = MagicMock()
|
|
mock_cm_success.__aenter__ = AsyncMock(
|
|
return_value=mock_response_success
|
|
)
|
|
mock_cm_success.__aexit__ = AsyncMock(return_value=None)
|
|
|
|
mock_session.get = MagicMock(
|
|
side_effect=[mock_cm_fail, mock_cm_fail, mock_cm_success]
|
|
)
|
|
|
|
image_downloader.session = mock_session
|
|
image_downloader.retry_delay = 0.1 # Speed up test
|
|
|
|
output_path = tmp_path / "test.jpg"
|
|
await image_downloader.download_image(
|
|
"https://test.com/image.jpg", output_path
|
|
)
|
|
|
|
# Should have retried twice then succeeded
|
|
assert mock_session.get.call_count == 3
|
|
assert output_path.exists()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_max_retries_exceeded(self, image_downloader, tmp_path):
|
|
"""Test failure after max retries."""
|
|
mock_session = AsyncMock()
|
|
mock_session.closed = False
|
|
mock_response = AsyncMock()
|
|
mock_response.status = 500
|
|
|
|
# Create mock RequestInfo for ClientResponseError
|
|
mock_request_info = MagicMock()
|
|
mock_request_info.real_url = "https://test.com/image.jpg"
|
|
|
|
mock_response.raise_for_status = MagicMock(
|
|
side_effect=aiohttp.ClientResponseError(
|
|
request_info=mock_request_info,
|
|
history=(),
|
|
status=500,
|
|
message="Server Error"
|
|
)
|
|
)
|
|
|
|
# Setup context manager
|
|
mock_cm = MagicMock()
|
|
mock_cm.__aenter__ = AsyncMock(return_value=mock_response)
|
|
mock_cm.__aexit__ = AsyncMock(return_value=None)
|
|
mock_session.get = MagicMock(return_value=mock_cm)
|
|
|
|
image_downloader.session = mock_session
|
|
image_downloader.max_retries = 2
|
|
image_downloader.retry_delay = 0.1
|
|
|
|
output_path = tmp_path / "test.jpg"
|
|
|
|
with pytest.raises(ImageDownloadError):
|
|
await image_downloader.download_image(
|
|
"https://test.com/image.jpg",
|
|
output_path
|
|
)
|
|
|
|
# Should have tried 2 times (max_retries=2 means 2 total attempts)
|
|
assert mock_session.get.call_count == 2
|