root
29c7040b20
- Integrated SQLModel with SQLite for robust data persistence - Refactored UserManager and WatchlistManager to use SQL queries - Migrated models to SQLModel with relationships and primary keys - Updated test suite with in-memory database isolation - Removed deprecated JSON storage files
302 lines
8.2 KiB
Python
302 lines
8.2 KiB
Python
"""
|
|
Pytest configuration and fixtures for Ohm Stream Downloader tests
|
|
"""
|
|
import pytest
|
|
import asyncio
|
|
import tempfile
|
|
import shutil
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
from unittest.mock import Mock, AsyncMock, patch
|
|
import sys
|
|
import os
|
|
|
|
import sys
|
|
import os
|
|
|
|
# Ensure the project root is in the Python path
|
|
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
|
|
|
|
# FORCE DATABASE_URL to in-memory for ALL tests before ANY app imports
|
|
os.environ["DATABASE_URL"] = "sqlite://"
|
|
|
|
from app.models import DownloadTask, DownloadStatus, DownloadRequest, HostType
|
|
from app.favorites import FavoritesManager
|
|
from app.download_manager import DownloadManager
|
|
from sqlmodel import SQLModel, create_engine, Session
|
|
|
|
|
|
@pytest.fixture(scope="session", autouse=True)
|
|
def init_db():
|
|
"""Initialize the in-memory database once for the test session"""
|
|
from app.database import engine
|
|
SQLModel.metadata.create_all(engine)
|
|
return engine
|
|
|
|
|
|
@pytest.fixture(name="engine")
|
|
def engine_fixture():
|
|
"""Returns the global test engine"""
|
|
from app.database import engine
|
|
return engine
|
|
|
|
|
|
@pytest.fixture(name="session")
|
|
def session_fixture(engine):
|
|
"""Create a temporary database session for testing"""
|
|
# Clear and recreate tables for each test to ensure isolation
|
|
SQLModel.metadata.drop_all(engine)
|
|
SQLModel.metadata.create_all(engine)
|
|
with Session(engine) as session:
|
|
yield session
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def mock_db(engine):
|
|
"""Ensure each test starts with fresh tables"""
|
|
SQLModel.metadata.drop_all(engine)
|
|
SQLModel.metadata.create_all(engine)
|
|
yield engine
|
|
|
|
|
|
@pytest.fixture
|
|
def user_manager():
|
|
"""Create a UserManager instance"""
|
|
from app.auth import UserManager
|
|
return UserManager()
|
|
|
|
|
|
@pytest.fixture
|
|
def watchlist_manager():
|
|
"""Create a WatchlistManager instance"""
|
|
from app.watchlist import WatchlistManager
|
|
return WatchlistManager()
|
|
|
|
|
|
def pytest_configure(config):
|
|
"""Configure pytest with custom markers"""
|
|
config.addinivalue_line(
|
|
"markers", "slow: marks tests as slow (deselect with '-m \"not slow\"')"
|
|
)
|
|
config.addinivalue_line(
|
|
"markers", "integration: marks tests as integration tests"
|
|
)
|
|
config.addinivalue_line(
|
|
"markers", "unit: marks tests as unit tests"
|
|
)
|
|
config.addinivalue_line(
|
|
"markers", "network: marks tests that require network access"
|
|
)
|
|
|
|
|
|
def pytest_collection_modifyitems(config, items):
|
|
"""Modify test collection to add markers automatically"""
|
|
for item in items:
|
|
# Mark async tests
|
|
if asyncio.iscoroutinefunction(item.obj):
|
|
item.add_marker("asyncio")
|
|
|
|
# Mark tests in test_api.py as integration tests
|
|
if "test_api.py" in str(item.fspath):
|
|
item.add_marker("integration")
|
|
|
|
# Mark other tests as unit tests
|
|
else:
|
|
item.add_marker("unit")
|
|
|
|
|
|
def pytest_report_header(config):
|
|
"""Add custom header to pytest report"""
|
|
return [
|
|
"Ohm Stream Downloader - Test Suite",
|
|
f"Python: {sys.version}",
|
|
]
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def event_loop():
|
|
"""Create an instance of the default event loop for the test session."""
|
|
loop = asyncio.get_event_loop_policy().new_event_loop()
|
|
yield loop
|
|
loop.close()
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_dir():
|
|
"""Create a temporary directory for test files"""
|
|
temp_path = Path(tempfile.mkdtemp())
|
|
yield temp_path
|
|
# Cleanup after test
|
|
shutil.rmtree(temp_path, ignore_errors=True)
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_download_dir(temp_dir):
|
|
"""Create a temporary download directory"""
|
|
download_dir = temp_dir / "downloads"
|
|
download_dir.mkdir(exist_ok=True)
|
|
return download_dir
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_download_task():
|
|
"""Create a sample download task"""
|
|
return DownloadTask(
|
|
id="test-task-123",
|
|
url="https://example.com/file.mp4",
|
|
filename="test_video.mp4",
|
|
host=HostType.OTHER,
|
|
status=DownloadStatus.PENDING,
|
|
progress=0.0,
|
|
downloaded_bytes=0,
|
|
total_bytes=None,
|
|
speed=0.0,
|
|
created_at=datetime.now()
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_download_request():
|
|
"""Create a sample download request"""
|
|
return DownloadRequest(
|
|
url="https://example.com/file.mp4",
|
|
filename="test_video.mp4"
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def download_manager(temp_download_dir):
|
|
"""Create a DownloadManager instance with temporary directory"""
|
|
manager = DownloadManager(download_dir=str(temp_download_dir), max_parallel=2)
|
|
yield manager
|
|
# Cleanup
|
|
for task_id in list(manager.tasks.keys()):
|
|
if task_id in manager.active_downloads:
|
|
manager.active_downloads[task_id].cancel()
|
|
|
|
|
|
@pytest.fixture
|
|
async def favorites_manager(temp_dir):
|
|
"""Create a FavoritesManager instance with temporary storage"""
|
|
storage_path = temp_dir / "test_favorites.json"
|
|
manager = FavoritesManager(storage_path=str(storage_path))
|
|
# Initialize asynchronously
|
|
await manager._load()
|
|
yield manager
|
|
# Cleanup
|
|
if storage_path.exists():
|
|
storage_path.unlink()
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_httpx_client():
|
|
"""Mock httpx.AsyncClient for network requests"""
|
|
mock_client = AsyncMock()
|
|
mock_response = AsyncMock()
|
|
mock_response.status_code = 200
|
|
mock_response.headers = {"content-length": "1000000"}
|
|
mock_response.raise_for_status = Mock()
|
|
mock_client.get.return_value = mock_response
|
|
mock_client.stream.return_value.__aenter__.return_value = mock_response
|
|
return mock_client
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_anime_metadata():
|
|
"""Sample anime metadata for testing"""
|
|
return {
|
|
"synopsis": "Test anime synopsis",
|
|
"genres": ["Action", "Adventure"],
|
|
"rating": "8.5/10",
|
|
"release_year": 2023,
|
|
"studio": "Test Studio",
|
|
"poster_image": "https://example.com/poster.jpg",
|
|
"total_episodes": 12,
|
|
"status": "Completed"
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_favorite_data():
|
|
"""Sample favorite anime data"""
|
|
return {
|
|
"anime_id": "anime-sama-test-anime",
|
|
"title": "Test Anime",
|
|
"url": "https://anime-sama.si/test/",
|
|
"provider": "anime-sama",
|
|
"metadata": {
|
|
"synopsis": "Test synopsis",
|
|
"genres": ["Action", "Adventure"],
|
|
"rating": "8.5/10",
|
|
"release_year": 2023
|
|
},
|
|
"poster_url": "https://example.com/poster.jpg"
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_async_context_manager():
|
|
"""Mock async context manager for streaming responses"""
|
|
class MockAsyncContextManager:
|
|
def __init__(self, mock_response):
|
|
self.mock_response = mock_response
|
|
|
|
async def __aenter__(self):
|
|
return self.mock_response
|
|
|
|
async def __aexit__(self, *args):
|
|
pass
|
|
|
|
return MockAsyncContextManager
|
|
|
|
|
|
# Mock data for API testing
|
|
@pytest.fixture
|
|
def mock_anime_search_results():
|
|
"""Mock anime search results"""
|
|
return [
|
|
{
|
|
"title": "Naruto Shippuden",
|
|
"url": "https://anime-sama.si/catalogue/naruto/saison1/vostfr/",
|
|
"cover_image": "https://example.com/naruto.jpg",
|
|
"type": "search_result"
|
|
},
|
|
{
|
|
"title": "One Piece",
|
|
"url": "https://anime-sama.si/catalogue/one-piece/saison1/vostfr/",
|
|
"cover_image": "https://example.com/onepiece.jpg",
|
|
"type": "search_result"
|
|
}
|
|
]
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_episode_list():
|
|
"""Mock episode list"""
|
|
return [
|
|
{"episode": 1, "url": "https://anime-sama.si/catalogue/test/ep1/vostfr/"},
|
|
{"episode": 2, "url": "https://anime-sama.si/catalogue/test/ep2/vostfr/"},
|
|
{"episode": 3, "url": "https://anime-sama.si/catalogue/test/ep3/vostfr/"}
|
|
]
|
|
|
|
|
|
# Patch fixtures
|
|
@pytest.fixture
|
|
def patch_httpx_client():
|
|
"""Patch httpx.AsyncClient to avoid real network calls"""
|
|
with patch('httpx.AsyncClient') as mock:
|
|
yield mock
|
|
|
|
|
|
@pytest.fixture
|
|
def patch_aiofiles():
|
|
"""Patch aiofiles for file operations"""
|
|
with patch('aiofiles.open', create=True) as mock:
|
|
yield mock
|
|
|
|
|
|
# Test data generators
|
|
def generate_chunk_data(size: int = 1024 * 1024) -> bytes:
|
|
"""Generate mock chunk data for downloads"""
|
|
return b"x" * size
|