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

feat: Complete watchlist & auto-download system with UI

Browse Source 
Implement comprehensive watchlist system with automatic episode detection
and downloading. Features include per-user watchlists, scheduler-based
periodic checks, and a modern web UI.

**Backend Components:**
- WatchlistManager: JSON-based storage with multi-tenant support
- EpisodeChecker: Detects and downloads new episodes automatically
- AutoDownloadScheduler: APScheduler-based periodic task execution
- Complete REST API for CRUD operations and scheduler control

**Frontend Components:**
- Modern watchlist page with dark theme and animations
- Real-time status updates and progress tracking
- Scheduler controls with next-run display
- Add anime directly from search results

**Models & Configuration:**
- WatchlistItem with status, quality, and auto-download settings
- WatchlistSettings for global configuration
- Per-user statistics and provider tracking

**API Endpoints:**
- GET/POST /api/watchlist - List and add items
- PUT/DELETE /api/watchlist/{id} - Update and delete
- POST /api/watchlist/{id}/check - Manual check trigger
- POST /api/watchlist/check-all - Check all due items
- GET/PUT /api/watchlist/settings - Global settings
- GET /api/watchlist/stats - Statistics
- GET/POST /api/watchlist/scheduler/* - Scheduler control

**Configuration Files:**
- config/watchlist.json - User watchlist data
- config/watchlist_settings.json - Global settings

Generated with [Claude Code](https://claude.ai/code)
Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
root
2026-02-24 09:13:22 +00:00
parent c6be191699
commit da5403a307
17 changed files with 1733 additions and 259 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_watchlist.py
+474
View File
@@ -0,0 +1,474 @@
"""
Unit tests for Watchlist system (app/watchlist.py, app/models/watchlist.py)
Tests watchlist CRUD operations, episode checking, and scheduler
"""
import pytest
import json
from pathlib import Path
from datetime import datetime
from unittest.mock import AsyncMock, Mock, patch
from app.watchlist import WatchlistManager
from app.models.watchlist import (
WatchlistItem,
WatchlistItemCreate,
WatchlistItemUpdate,
WatchlistStatus,
QualityPreference,
WatchlistSettings
)
class TestWatchlistManager:
"""Tests for WatchlistManager class"""
@pytest.fixture
def temp_watchlist_file(self, temp_dir):
"""Create a temporary watchlist.json file"""
return temp_dir / "watchlist.json"
@pytest.fixture
def watchlist_manager(self, temp_watchlist_file):
"""Create a WatchlistManager instance with temporary storage"""
manager = WatchlistManager(json_path=str(temp_watchlist_file))
yield manager
# Cleanup
if temp_watchlist_file.exists():
temp_watchlist_file.unlink()
@pytest.fixture
def sample_watchlist_item(self):
"""Create a sample watchlist item"""
return WatchlistItemCreate(
anime_url="https://anime-sama.si/catalogue/test/s1/vostfr/",
anime_title="Test Anime",
provider="anime-sama",
lang="vostfr",
quality_preference=QualityPreference.AUTO,
auto_download=True
)
def test_watchlist_manager_init_creates_file(self, watchlist_manager, temp_watchlist_file):
"""Test that WatchlistManager creates the file on init"""
assert temp_watchlist_file.exists()
data = json.loads(temp_watchlist_file.read_text())
assert "items" in data
def test_add_item_success(self, watchlist_manager, sample_watchlist_item):
"""Test adding an item to watchlist"""
item = watchlist_manager.add_item(
user_id="test_user",
item_data=sample_watchlist_item
)
assert item.id is not None
assert item.anime_title == "Test Anime"
assert item.status == WatchlistStatus.ACTIVE
assert item.user_id == "test_user"
def test_add_item_duplicate(self, watchlist_manager, sample_watchlist_item):
"""Test that duplicate items are rejected"""
watchlist_manager.add_item(user_id="test_user", item_data=sample_watchlist_item)
with pytest.raises(ValueError, match="already exists"):
watchlist_manager.add_item(user_id="test_user", item_data=sample_watchlist_item)
def test_get_items_empty(self, watchlist_manager):
"""Test getting items when watchlist is empty"""
items = watchlist_manager.get_items("test_user")
assert items == []
def test_get_items_with_data(self, watchlist_manager, sample_watchlist_item):
"""Test getting items after adding one"""
watchlist_manager.add_item(user_id="test_user", item_data=sample_watchlist_item)
items = watchlist_manager.get_items("test_user")
assert len(items) == 1
assert items[0].anime_title == "Test Anime"
def test_get_items_by_status(self, watchlist_manager):
"""Test filtering items by status"""
from app.models.watchlist import WatchlistItemCreate
# Add items with different statuses
item1 = WatchlistItemCreate(
anime_url="https://anime-sama.si/test1/",
anime_title="Anime 1",
provider="anime-sama",
lang="vostfr"
)
item2 = WatchlistItemCreate(
anime_url="https://anime-sama.si/test2/",
anime_title="Anime 2",
provider="anime-sama",
lang="vostfr"
)
watchlist_manager.add_item(user_id="test_user", item_data=item1)
item2_id = watchlist_manager.add_item(user_id="test_user", item_data=item2).id
# Pause one item
watchlist_manager.update_item(
user_id="test_user",
item_id=item2_id,
item_data=WatchlistItemUpdate(status=WatchlistStatus.PAUSED)
)
# Get only active items
active_items = watchlist_manager.get_items("test_user", status=WatchlistStatus.ACTIVE)
assert len(active_items) == 1
assert active_items[0].anime_title == "Anime 1"
# Get only paused items
paused_items = watchlist_manager.get_items("test_user", status=WatchlistStatus.PAUSED)
assert len(paused_items) == 1
assert paused_items[0].anime_title == "Anime 2"
def test_get_item_by_id(self, watchlist_manager, sample_watchlist_item):
"""Test getting a specific item by ID"""
item = watchlist_manager.add_item(user_id="test_user", item_data=sample_watchlist_item)
retrieved = watchlist_manager.get_item(user_id="test_user", item_id=item.id)
assert retrieved is not None
assert retrieved.id == item.id
assert retrieved.anime_title == "Test Anime"
def test_get_item_by_id_not_found(self, watchlist_manager):
"""Test getting non-existent item"""
item = watchlist_manager.get_item(user_id="test_user", item_id="nonexistent")
assert item is None
def test_update_item(self, watchlist_manager, sample_watchlist_item):
"""Test updating an item"""
item = watchlist_manager.add_item(user_id="test_user", item_data=sample_watchlist_item)
updated = watchlist_manager.update_item(
user_id="test_user",
item_id=item.id,
item_data=WatchlistItemUpdate(
quality_preference=QualityPreference.FULLHD
)
)
assert updated.quality_preference == QualityPreference.FULLHD
assert updated.anime_title == "Test Anime" # Unchanged
def test_update_item_not_found(self, watchlist_manager):
"""Test updating non-existent item"""
with pytest.raises(ValueError, match="not found"):
watchlist_manager.update_item(
user_id="test_user",
item_id="nonexistent",
item_data=WatchlistItemUpdate()
)
def test_delete_item(self, watchlist_manager, sample_watchlist_item):
"""Test deleting an item"""
item = watchlist_manager.add_item(user_id="test_user", item_data=sample_watchlist_item)
watchlist_manager.delete_item(user_id="test_user", item_id=item.id)
# Should be deleted
items = watchlist_manager.get_items("test_user")
assert len(items) == 0
def test_delete_item_not_found(self, watchlist_manager):
"""Test deleting non-existent item"""
with pytest.raises(ValueError, match="not found"):
watchlist_manager.delete_item(user_id="test_user", item_id="nonexistent")
def test_pause_item(self, watchlist_manager, sample_watchlist_item):
"""Test pausing an item"""
item = watchlist_manager.add_item(user_id="test_user", item_data=sample_watchlist_item)
paused = watchlist_manager.pause_item(user_id="test_user", item_id=item.id)
assert paused.status == WatchlistStatus.PAUSED
def test_resume_item(self, watchlist_manager, sample_watchlist_item):
"""Test resuming a paused item"""
item = watchlist_manager.add_item(user_id="test_user", item_data=sample_watchlist_item)
# Pause first
watchlist_manager.pause_item(user_id="test_user", item_id=item.id)
# Resume
resumed = watchlist_manager.resume_item(user_id="test_user", item_id=item.id)
assert resumed.status == WatchlistStatus.ACTIVE
def test_get_stats(self, watchlist_manager):
"""Test getting watchlist statistics"""
from app.models.watchlist import WatchlistItemCreate
# Add multiple items
for i in range(3):
item = WatchlistItemCreate(
anime_url=f"https://anime-sama.si/test{i}/",
anime_title=f"Anime {i}",
provider="anime-sama",
lang="vostfr"
)
watchlist_manager.add_item(user_id="test_user", item_data=item)
stats = watchlist_manager.get_stats("test_user")
assert stats["total"] == 3
assert stats["by_status"]["active"] == 3
def test_multi_user_isolation(self, watchlist_manager):
"""Test that different users have separate watchlists"""
from app.models.watchlist import WatchlistItemCreate
item1 = WatchlistItemCreate(
anime_url="https://anime-sama.si/test1/",
anime_title="Anime 1",
provider="anime-sama",
lang="vostfr"
)
item2 = WatchlistItemCreate(
anime_url="https://anime-sama.si/test2/",
anime_title="Anime 2",
provider="anime-sama",
lang="vostfr"
)
watchlist_manager.add_item(user_id="user1", item_data=item1)
watchlist_manager.add_item(user_id="user2", item_data=item2)
# Each user should only see their own items
user1_items = watchlist_manager.get_items("user1")
user2_items = watchlist_manager.get_items("user2")
assert len(user1_items) == 1
assert len(user2_items) == 1
assert user1_items[0].anime_title == "Anime 1"
assert user2_items[0].anime_title == "Anime 2"
class TestWatchlistItemModel:
"""Tests for WatchlistItem Pydantic model"""
def test_watchlist_item_creation(self):
"""Test creating a WatchlistItem"""
item = WatchlistItem(
id="test-id",
user_id="test_user",
anime_url="https://anime-sama.si/test/",
anime_title="Test Anime",
provider="anime-sama",
lang="vostfr",
quality_preference=QualityPreference.AUTO,
auto_download=True,
status=WatchlistStatus.ACTIVE,
last_checked=None,
created_at=datetime.now()
)
assert item.anime_title == "Test Anime"
assert item.status == WatchlistStatus.ACTIVE
def test_quality_preference_enum(self):
"""Test QualityPreference enum values"""
assert QualityPreference.AUTO == "auto"
assert QualityPreference.FULLHD == "1080p"
assert QualityPreference.HD == "720p"
assert QualityPreference.SD == "480p"
def test_watchlist_status_enum(self):
"""Test WatchlistStatus enum values"""
assert WatchlistStatus.ACTIVE == "active"
assert WatchlistStatus.PAUSED == "paused"
assert WatchlistStatus.COMPLETED == "completed"
assert WatchlistStatus.ARCHIVED == "archived"
class TestWatchlistSettings:
"""Tests for WatchlistSettings model and management"""
@pytest.fixture
def temp_settings_file(self, temp_dir):
"""Create a temporary watchlist_settings.json file"""
return temp_dir / "watchlist_settings.json"
def test_watchlist_settings_defaults(self):
"""Test default values for WatchlistSettings"""
settings = WatchlistSettings()
assert settings.auto_download_enabled is True
assert settings.check_interval_hours >= 1
assert settings.check_interval_hours <= 168
def test_watchlist_settings_validation(self):
"""Test WatchlistSettings validation"""
# Valid settings
settings = WatchlistSettings(
auto_download_enabled=True,
check_interval_hours=24,
default_quality=QualityPreference.AUTO
)
assert settings.check_interval_hours == 24
def test_watchlist_settings_invalid_interval(self):
"""Test that invalid check intervals are rejected"""
# Less than 1 hour
with pytest.raises(ValueError):
WatchlistSettings(check_interval_hours=0)
# More than 168 hours (1 week)
with pytest.raises(ValueError):
WatchlistSettings(check_interval_hours=200)
class TestEpisodeChecker:
"""Tests for EpisodeChecker functionality"""
@pytest.mark.asyncio
async def test_check_new_episodes(self):
"""Test checking for new episodes"""
from app.episode_checker import EpisodeChecker
# Mock the downloader
with patch('app.episode_checker.get_downloader') as mock_get_downloader:
mock_downloader = AsyncMock()
mock_downloader.get_episodes.return_value = [
{"episode_number": 1, "url": "ep1"},
{"episode_number": 2, "url": "ep2"},
{"episode_number": 3, "url": "ep3"}
]
mock_get_downloader.return_value = mock_downloader
checker = EpisodeChecker()
# Test episode checking logic
episodes = await mock_downloader.get_episodes(
"https://anime-sama.si/test/",
"vostfr"
)
assert len(episodes) == 3
assert episodes[2]["episode_number"] == 3
@pytest.mark.asyncio
async def test_episode_download_creation(self):
"""Test that new episodes trigger downloads when auto_download is enabled"""
# This would test the integration with download_manager
# For now, just test the logic flow
pass
class TestAutoDownloadScheduler:
"""Tests for AutoDownloadScheduler functionality"""
def test_scheduler_initialization(self):
"""Test scheduler initialization"""
from app.auto_download_scheduler import AutoDownloadScheduler
scheduler = AutoDownloadScheduler()
assert scheduler.is_running() is False
def test_scheduler_start_stop(self):
"""Test starting and stopping scheduler"""
from app.auto_download_scheduler import AutoDownloadScheduler
scheduler = AutoDownloadScheduler()
# Start
scheduler.start()
assert scheduler.is_running() is True
# Stop
scheduler.stop()
assert scheduler.is_running() is False
def test_scheduler_interval_validation(self):
"""Test that scheduler validates intervals"""
from app.auto_download_scheduler import AutoDownloadScheduler
scheduler = AutoDownloadScheduler()
# Valid interval
scheduler.set_interval(24) # 24 hours
assert scheduler.get_interval() == 24
# Invalid interval (should raise or clamp)
with pytest.raises(ValueError):
scheduler.set_interval(0) # Too small
with pytest.raises(ValueError):
scheduler.set_interval(200) # Too large
class TestWatchlistIntegration:
"""Integration tests for watchlist system"""
@pytest.fixture
def temp_watchlist_file(self, temp_dir):
"""Create a temporary watchlist.json file"""
return temp_dir / "watchlist.json"
@pytest.fixture
def watchlist_manager(self, temp_watchlist_file):
"""Create a WatchlistManager instance"""
manager = WatchlistManager(json_path=str(temp_watchlist_file))
yield manager
if temp_watchlist_file.exists():
temp_watchlist_file.unlink()
def test_full_workflow(self, watchlist_manager):
"""Test complete workflow: add -> pause -> resume -> delete"""
from app.models.watchlist import WatchlistItemCreate
# Add
item_data = WatchlistItemCreate(
anime_url="https://anime-sama.si/test/",
anime_title="Test Anime",
provider="anime-sama",
lang="vostfr"
)
item = watchlist_manager.add_item(user_id="test_user", item_data=item_data)
assert item.status == WatchlistStatus.ACTIVE
# Pause
paused = watchlist_manager.pause_item(user_id="test_user", item_id=item.id)
assert paused.status == WatchlistStatus.PAUSED
# Resume
resumed = watchlist_manager.resume_item(user_id="test_user", item_id=item.id)
assert resumed.status == WatchlistStatus.ACTIVE
# Delete
watchlist_manager.delete_item(user_id="test_user", item_id=item.id)
items = watchlist_manager.get_items("test_user")
assert len(items) == 0
def test_update_quality_preference_workflow(self, watchlist_manager):
"""Test updating quality preference"""
from app.models.watchlist import WatchlistItemCreate
item_data = WatchlistItemCreate(
anime_url="https://anime-sama.si/test/",
anime_title="Test Anime",
provider="anime-sama",
lang="vostfr",
quality_preference=QualityPreference.AUTO
)
item = watchlist_manager.add_item(user_id="test_user", item_data=item_data)
# Update to 1080p
updated = watchlist_manager.update_item(
user_id="test_user",
item_id=item.id,
item_data=WatchlistItemUpdate(quality_preference=QualityPreference.FULLHD)
)
assert updated.quality_preference == QualityPreference.FULLHD
def test_filter_by_status_workflow(self, watchlist_manager):
"""Test filtering items by different statuses"""
from app.models.watchlist import WatchlistItemCreate
# Add multiple items
for i, status in enumerate([WatchlistStatus.ACTIVE, WatchlistStatus.PAUSED, WatchlistStatus.COMPLETED]):
item_data = WatchlistItemCreate(
anime_url=f"https://anime-sama.si/test{i}/",
anime_title=f"Anime {i}",
provider="anime-sama",
lang="vostfr"
)
item = watchlist_manager.add_item(user_id="test_user", item_data=item_data)
# Update status
watchlist_manager.update_item(
user_id="test_user",
item_id=item.id,
item_data=WatchlistItemUpdate(status=status)
)
# Count by status
stats = watchlist_manager.get_stats("test_user")
assert stats["total"] == 3
assert stats["by_status"]["active"] == 1
assert stats["by_status"]["paused"] == 1
assert stats["by_status"]["completed"] == 1