root
18c3c4d27b
- tests/test_user_journey.py: 23 pytest tests covering auth, search, settings, and download flows using TestClient with mocked providers (no real network calls) - tests/e2e/user_journey.spec.ts: 6 fixme Playwright test placeholders for full UI journey (register, login, browse, search, settings, logout)
588 lines
20 KiB
Python
588 lines
20 KiB
Python
"""
|
|
End-to-end user journey tests for Ohm Stream Downloader.
|
|
|
|
These tests verify complete user workflows including authentication,
|
|
search, settings management, and download operations.
|
|
|
|
All tests use mocked providers and in-memory SQLite — no real network calls.
|
|
"""
|
|
|
|
import os
|
|
import time
|
|
import uuid
|
|
|
|
# FORCE DATABASE_URL to in-memory for ALL tests before ANY app imports
|
|
os.environ["DATABASE_URL"] = "sqlite://"
|
|
|
|
import pytest
|
|
from typing import Dict, List, Optional, Tuple
|
|
from unittest.mock import AsyncMock, Mock, patch
|
|
from fastapi.testclient import TestClient
|
|
|
|
from main import app
|
|
|
|
|
|
# =============================================================================
|
|
# MOCK DATA CONSTANTS
|
|
# =============================================================================
|
|
|
|
MOCK_ANIME_SEARCH_RESULTS: Dict[str, List[Dict]] = {
|
|
"anime-sama": [
|
|
{
|
|
"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",
|
|
},
|
|
],
|
|
"anime-ultime": [
|
|
{
|
|
"title": "Naruto Shippuden",
|
|
"url": "https://www.anime-ultime.net/naruto-shippuden",
|
|
"cover_image": "https://example.com/naruto-au.jpg",
|
|
"type": "search_result",
|
|
},
|
|
],
|
|
}
|
|
|
|
MOCK_SERIES_SEARCH_RESULTS: Dict[str, Dict] = {
|
|
"Breaking Bad": {
|
|
"base_name": "Breaking Bad",
|
|
"cover": "https://example.com/bb.jpg",
|
|
"synopsis": "A chemistry teacher turns to manufacturing methamphetamine...",
|
|
"seasons": {
|
|
1: [
|
|
{
|
|
"id": "fs7",
|
|
"url": "https://fs7.fr/breaking-bad/saison-1",
|
|
"provider_id": "fs7",
|
|
}
|
|
],
|
|
2: [
|
|
{
|
|
"id": "fs7",
|
|
"url": "https://fs7.fr/breaking-bad/saison-2",
|
|
"provider_id": "fs7",
|
|
}
|
|
],
|
|
},
|
|
}
|
|
}
|
|
|
|
MOCK_EPISODE_LIST: List[Dict] = [
|
|
{
|
|
"episode": 1,
|
|
"url": "https://example.com/video1.mp4|https://anime-sama.si/catalogue/naruto/s1/ep1|Naruto - Episode 1",
|
|
},
|
|
{
|
|
"episode": 2,
|
|
"url": "https://example.com/video2.mp4|https://anime-sama.si/catalogue/naruto/s1/ep2|Naruto - Episode 2",
|
|
},
|
|
{
|
|
"episode": 3,
|
|
"url": "https://example.com/video3.mp4|https://anime-sama.si/catalogue/naruto/s1/ep3|Naruto - Episode 3",
|
|
},
|
|
{
|
|
"episode": 4,
|
|
"url": "https://example.com/video4.mp4|https://anime-sama.si/catalogue/naruto/s1/ep4|Naruto - Episode 4",
|
|
},
|
|
{
|
|
"episode": 5,
|
|
"url": "https://example.com/video5.mp4|https://anime-sama.si/catalogue/naruto/s1/ep5|Naruto - Episode 5",
|
|
},
|
|
]
|
|
|
|
MOCK_DOWNLOAD_LINK: Tuple[str, str] = (
|
|
"https://example.com/video1.mp4",
|
|
"Naruto_Episode_1.mp4",
|
|
)
|
|
|
|
|
|
# =============================================================================
|
|
# FIXTURES
|
|
# =============================================================================
|
|
|
|
|
|
@pytest.fixture
|
|
def journey_client():
|
|
"""
|
|
Create a TestClient with a registered and logged-in user.
|
|
|
|
Uses 'with' context manager to ensure startup events fire
|
|
and database tables are properly initialized.
|
|
"""
|
|
unique_id = f"{int(time.time())}_{uuid.uuid4().hex[:8]}"
|
|
test_username = f"journey_user_{unique_id}"
|
|
test_password = "TestPassword123!"
|
|
test_email = f"{test_username}@test.example.com"
|
|
|
|
with TestClient(app) as client:
|
|
register_response = client.post(
|
|
"/api/auth/register",
|
|
json={
|
|
"username": test_username,
|
|
"password": test_password,
|
|
"email": test_email,
|
|
},
|
|
)
|
|
assert register_response.status_code == 200, (
|
|
f"Registration failed: {register_response.json()}"
|
|
)
|
|
|
|
login_response = client.post(
|
|
"/api/auth/login",
|
|
json={
|
|
"username": test_username,
|
|
"password": test_password,
|
|
},
|
|
)
|
|
assert login_response.status_code == 200, (
|
|
f"Login failed: {login_response.json()}"
|
|
)
|
|
|
|
access_token = login_response.json()["access_token"]
|
|
client.headers["Authorization"] = f"Bearer {access_token}"
|
|
|
|
yield client
|
|
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
@pytest.fixture
|
|
def unauth_client():
|
|
"""Create a plain TestClient without authentication."""
|
|
with TestClient(app) as client:
|
|
yield client
|
|
|
|
|
|
# =============================================================================
|
|
# TEST CLASSES
|
|
# =============================================================================
|
|
|
|
|
|
class TestAuthJourney:
|
|
"""Tests for the authentication user journey — registration, login, tokens."""
|
|
|
|
def test_register_new_user(self):
|
|
unique_id = f"{int(time.time())}_{uuid.uuid4().hex[:8]}"
|
|
username = f"auth_test_{unique_id}"
|
|
with TestClient(app) as client:
|
|
response = client.post(
|
|
"/api/auth/register",
|
|
json={"username": username, "password": "SecurePass123!"},
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["status"] == "success"
|
|
assert data["user"]["username"] == username
|
|
assert "id" in data["user"]
|
|
|
|
def test_register_duplicate_user(self):
|
|
username = f"dup_{uuid.uuid4().hex[:8]}"
|
|
with TestClient(app) as client:
|
|
first = client.post(
|
|
"/api/auth/register",
|
|
json={"username": username, "password": "SecurePass123!"},
|
|
)
|
|
assert first.status_code == 200
|
|
|
|
second = client.post(
|
|
"/api/auth/register",
|
|
json={"username": username, "password": "SecurePass123!"},
|
|
)
|
|
assert second.status_code in (400, 500)
|
|
|
|
def test_login_success(self):
|
|
unique_id = f"{int(time.time())}_{uuid.uuid4().hex[:8]}"
|
|
username = f"login_test_{unique_id}"
|
|
password = "SecurePass123!"
|
|
|
|
with TestClient(app) as client:
|
|
client.post(
|
|
"/api/auth/register",
|
|
json={"username": username, "password": password},
|
|
)
|
|
response = client.post(
|
|
"/api/auth/login",
|
|
json={"username": username, "password": password},
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "access_token" in data
|
|
assert data["token_type"] == "bearer"
|
|
assert data["user"]["username"] == username
|
|
|
|
def test_login_wrong_password(self):
|
|
unique_id = f"{int(time.time())}_{uuid.uuid4().hex[:8]}"
|
|
username = f"wrong_pw_{unique_id}"
|
|
|
|
with TestClient(app) as client:
|
|
client.post(
|
|
"/api/auth/register",
|
|
json={"username": username, "password": "CorrectPass123!"},
|
|
)
|
|
|
|
with TestClient(app) as client:
|
|
response = client.post(
|
|
"/api/auth/login",
|
|
json={"username": username, "password": "WrongPass999!"},
|
|
)
|
|
assert response.status_code == 401
|
|
|
|
def test_get_current_user(self, journey_client):
|
|
response = journey_client.get("/api/auth/me")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "user" in data
|
|
assert "username" in data["user"]
|
|
assert "id" in data["user"]
|
|
|
|
def test_unauthorized_access(self):
|
|
with TestClient(app) as client:
|
|
response = client.get("/api/auth/me")
|
|
assert response.status_code == 403
|
|
|
|
|
|
class TestSearchJourney:
|
|
"""Tests for the search user journey — anime/series search, episodes, metadata."""
|
|
|
|
@patch("app.routers.router_anime.providers_manager")
|
|
@patch("app.routers.router_anime.get_metadata_enricher")
|
|
def test_search_anime(self, mock_enricher, mock_pm, journey_client):
|
|
mock_provider = Mock()
|
|
mock_provider.id = "anime-sama"
|
|
mock_provider.search = AsyncMock(
|
|
return_value=[
|
|
Mock(
|
|
model_dump=Mock(
|
|
return_value={
|
|
"title": "Naruto Shippuden",
|
|
"url": "https://anime-sama.si/catalogue/naruto/s1/vostfr/",
|
|
"cover_image": "https://example.com/naruto.jpg",
|
|
"type": "search_result",
|
|
}
|
|
)
|
|
),
|
|
]
|
|
)
|
|
mock_pm.get_active_providers.return_value = [mock_provider]
|
|
|
|
mock_meta = AsyncMock()
|
|
mock_meta.enrich_metadata = AsyncMock(return_value=None)
|
|
mock_enricher.return_value = mock_meta
|
|
|
|
response = journey_client.get("/api/anime/search?q=naruto")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["query"] == "naruto"
|
|
assert "results" in data
|
|
|
|
@patch("app.routers.router_anime.providers_manager")
|
|
@patch("app.routers.router_anime.get_metadata_enricher")
|
|
def test_search_anime_results_have_providers(
|
|
self, mock_enricher, mock_pm, journey_client
|
|
):
|
|
mock_provider = Mock()
|
|
mock_provider.id = "anime-sama"
|
|
mock_provider.search = AsyncMock(
|
|
return_value=[
|
|
Mock(
|
|
model_dump=Mock(
|
|
return_value={
|
|
"title": "Naruto Shippuden",
|
|
"url": "https://anime-sama.si/catalogue/naruto/s1/vostfr/",
|
|
"cover_image": "https://example.com/naruto.jpg",
|
|
"type": "search_result",
|
|
}
|
|
)
|
|
),
|
|
]
|
|
)
|
|
mock_pm.get_active_providers.return_value = [mock_provider]
|
|
|
|
mock_meta = AsyncMock()
|
|
mock_meta.enrich_metadata = AsyncMock(return_value=None)
|
|
mock_enricher.return_value = mock_meta
|
|
|
|
response = journey_client.get("/api/anime/search?q=naruto")
|
|
data = response.json()
|
|
assert "anime-sama" in data["results"]
|
|
|
|
@patch("app.routers.router_anime.get_series_providers")
|
|
@patch("app.routers.router_anime.get_metadata_enricher")
|
|
def test_search_series(self, mock_enricher, mock_series_providers, journey_client):
|
|
mock_series_providers.return_value = {"fs7": {"name": "FS7"}}
|
|
|
|
mock_meta = AsyncMock()
|
|
mock_meta.enrich_metadata = AsyncMock(return_value=None)
|
|
mock_enricher.return_value = mock_meta
|
|
|
|
mock_fs7_instance = Mock()
|
|
mock_fs7_instance.search_anime = AsyncMock(
|
|
return_value=[
|
|
{
|
|
"title": "Breaking Bad",
|
|
"url": "https://fs7.fr/breaking-bad/saison-1",
|
|
"cover_image": "https://example.com/bb.jpg",
|
|
}
|
|
]
|
|
)
|
|
|
|
with patch.dict(
|
|
"sys.modules",
|
|
{
|
|
"app.downloaders.series_sites.fs7": Mock(
|
|
FS7Downloader=Mock(return_value=mock_fs7_instance)
|
|
)
|
|
},
|
|
):
|
|
response = journey_client.get("/api/series/search?q=breaking+bad")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["query"] == "breaking bad"
|
|
assert "results" in data
|
|
|
|
@patch("app.routers.router_anime.providers_manager")
|
|
@patch("app.routers.router_anime.get_metadata_enricher")
|
|
def test_search_anime_no_results(self, mock_enricher, mock_pm, journey_client):
|
|
mock_provider = Mock()
|
|
mock_provider.id = "anime-sama"
|
|
mock_provider.search = AsyncMock(return_value=[])
|
|
mock_pm.get_active_providers.return_value = [mock_provider]
|
|
|
|
mock_meta = AsyncMock()
|
|
mock_meta.enrich_metadata = AsyncMock(return_value=None)
|
|
mock_enricher.return_value = mock_meta
|
|
|
|
response = journey_client.get("/api/anime/search?q=test_no_results_xyz")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "results" in data
|
|
assert "anime-sama" not in data["results"]
|
|
|
|
@patch("app.routers.router_anime.get_downloader")
|
|
def test_get_episodes(self, mock_get_dl):
|
|
mock_dl = Mock()
|
|
mock_dl.get_episodes = AsyncMock(return_value=MOCK_EPISODE_LIST)
|
|
mock_get_dl.return_value = mock_dl
|
|
|
|
with TestClient(app) as client:
|
|
response = client.get(
|
|
"/api/anime/episodes?url=https://anime-sama.si/test&lang=vostfr"
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["lang"] == "vostfr"
|
|
assert len(data["episodes"]) == 5
|
|
assert data["episodes"][0]["episode"] == 1
|
|
|
|
@patch("app.routers.router_anime.get_downloader")
|
|
def test_get_anime_metadata(self, mock_get_dl):
|
|
mock_dl = Mock()
|
|
mock_dl.get_anime_metadata = AsyncMock(
|
|
return_value={
|
|
"synopsis": "A ninja story",
|
|
"genres": ["Action", "Adventure"],
|
|
"rating": "8.5/10",
|
|
}
|
|
)
|
|
mock_dl.hasattr = Mock(return_value=True)
|
|
mock_get_dl.return_value = mock_dl
|
|
|
|
with TestClient(app) as client:
|
|
response = client.get(
|
|
"/api/anime/metadata?url=https://anime-sama.si/naruto"
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["url"] == "https://anime-sama.si/naruto"
|
|
assert data["metadata"]["synopsis"] == "A ninja story"
|
|
|
|
|
|
class TestSettingsJourney:
|
|
"""Tests for the settings management user journey — preferences, providers, theme."""
|
|
|
|
def test_get_default_settings(self, journey_client):
|
|
response = journey_client.get("/api/settings")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["default_lang"] == "vostfr"
|
|
assert data["theme"] == "dark"
|
|
|
|
def test_update_lang(self, journey_client):
|
|
response = journey_client.patch("/api/settings", json={"default_lang": "vf"})
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["default_lang"] == "vf"
|
|
|
|
verify = journey_client.get("/api/settings")
|
|
assert verify.json()["default_lang"] == "vf"
|
|
|
|
def test_update_theme(self, journey_client):
|
|
response = journey_client.patch("/api/settings", json={"theme": "oled"})
|
|
assert response.status_code == 200
|
|
assert response.json()["theme"] == "oled"
|
|
|
|
def test_settings_persist_across_requests(self, journey_client):
|
|
journey_client.patch(
|
|
"/api/settings",
|
|
json={
|
|
"default_lang": "vf",
|
|
"theme": "oled",
|
|
},
|
|
)
|
|
|
|
first = journey_client.get("/api/settings").json()
|
|
second = journey_client.get("/api/settings").json()
|
|
|
|
assert first["default_lang"] == "vf"
|
|
assert first["theme"] == "oled"
|
|
assert second["default_lang"] == first["default_lang"]
|
|
assert second["theme"] == first["theme"]
|
|
|
|
@patch("app.routers.router_settings.providers_manager")
|
|
def test_get_providers_availability(self, mock_pm, journey_client):
|
|
mock_pm.get_all_status.return_value = {
|
|
"anime-sama": {"status": "up"},
|
|
"fs7": {"status": "up"},
|
|
}
|
|
|
|
response = journey_client.get("/api/settings/providers/availability")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
provider_ids = {p["id"] for p in data}
|
|
assert len(provider_ids) > 0
|
|
|
|
@patch("app.routers.router_settings.providers_manager")
|
|
def test_toggle_provider(self, mock_pm, journey_client):
|
|
mock_pm.get_all_status.return_value = {}
|
|
|
|
toggle = journey_client.post("/api/settings/providers/anime-sama/toggle")
|
|
assert toggle.status_code == 200
|
|
data = toggle.json()
|
|
assert data["id"] == "anime-sama"
|
|
assert data["enabled"] is False
|
|
|
|
toggle_back = journey_client.post("/api/settings/providers/anime-sama/toggle")
|
|
assert toggle_back.json()["enabled"] is True
|
|
|
|
settings = journey_client.get("/api/settings").json()
|
|
assert "anime-sama" not in settings["disabled_providers"]
|
|
|
|
|
|
class TestDownloadJourney:
|
|
"""Tests for the download management user journey — create, list, status, cancel."""
|
|
|
|
@patch("app.routers.router_anime.get_download_manager")
|
|
def test_create_single_download(self, mock_get_dm, journey_client):
|
|
import tempfile
|
|
from pathlib import Path
|
|
from app.download_manager import DownloadManager
|
|
|
|
tmp_dir = Path(tempfile.mkdtemp()) / "downloads"
|
|
tmp_dir.mkdir(exist_ok=True)
|
|
manager = DownloadManager(download_dir=str(tmp_dir), max_parallel=1)
|
|
mock_get_dm.return_value = manager
|
|
|
|
response = journey_client.post(
|
|
"/api/anime/download?url=https://example.com/video.mp4"
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "task_id" in data
|
|
|
|
@patch("app.routers.router_downloads.get_download_manager")
|
|
def test_list_downloads(self, mock_get_dm, journey_client):
|
|
import tempfile
|
|
from pathlib import Path
|
|
from app.download_manager import DownloadManager
|
|
|
|
tmp_dir = Path(tempfile.mkdtemp()) / "downloads"
|
|
tmp_dir.mkdir(exist_ok=True)
|
|
manager = DownloadManager(download_dir=str(tmp_dir), max_parallel=1)
|
|
mock_get_dm.return_value = manager
|
|
|
|
journey_client.post(
|
|
"/api/downloads",
|
|
json={"url": "https://example.com/video1.mp4"},
|
|
)
|
|
response = journey_client.get("/api/downloads")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "downloads" in data
|
|
assert len(data["downloads"]) >= 1
|
|
|
|
@patch("app.routers.router_downloads.get_download_manager")
|
|
def test_download_task_status(self, mock_get_dm, journey_client):
|
|
import tempfile
|
|
from pathlib import Path
|
|
from app.download_manager import DownloadManager
|
|
|
|
tmp_dir = Path(tempfile.mkdtemp()) / "downloads"
|
|
tmp_dir.mkdir(exist_ok=True)
|
|
manager = DownloadManager(download_dir=str(tmp_dir), max_parallel=1)
|
|
mock_get_dm.return_value = manager
|
|
|
|
task_resp = journey_client.post(
|
|
"/api/downloads",
|
|
json={"url": "https://example.com/status_test.mp4"},
|
|
)
|
|
task_id = task_resp.json()["id"]
|
|
|
|
status_resp = journey_client.get(f"/api/downloads/{task_id}")
|
|
assert status_resp.status_code == 200
|
|
assert status_resp.json()["id"] == task_id
|
|
|
|
@patch("app.routers.router_downloads.get_download_manager")
|
|
def test_cancel_download(self, mock_get_dm, journey_client):
|
|
import tempfile
|
|
from pathlib import Path
|
|
from app.download_manager import DownloadManager
|
|
|
|
tmp_dir = Path(tempfile.mkdtemp()) / "downloads"
|
|
tmp_dir.mkdir(exist_ok=True)
|
|
manager = DownloadManager(download_dir=str(tmp_dir), max_parallel=1)
|
|
mock_get_dm.return_value = manager
|
|
|
|
task_resp = journey_client.post(
|
|
"/api/downloads",
|
|
json={"url": "https://example.com/cancel_test.mp4"},
|
|
)
|
|
assert task_resp.status_code == 200
|
|
task_id = task_resp.json()["id"]
|
|
|
|
cancel_resp = journey_client.delete(f"/api/downloads/{task_id}")
|
|
assert cancel_resp.status_code == 200
|
|
|
|
@patch("app.routers.router_anime.get_download_manager")
|
|
@patch("app.routers.router_anime.get_downloader")
|
|
def test_download_season(self, mock_get_dl, mock_get_dm, journey_client):
|
|
import tempfile
|
|
from pathlib import Path
|
|
from app.download_manager import DownloadManager
|
|
|
|
mock_dl = Mock()
|
|
mock_dl.get_episodes = AsyncMock(return_value=MOCK_EPISODE_LIST)
|
|
mock_get_dl.return_value = mock_dl
|
|
|
|
tmp_dir = Path(tempfile.mkdtemp()) / "downloads"
|
|
tmp_dir.mkdir(exist_ok=True)
|
|
manager = DownloadManager(download_dir=str(tmp_dir), max_parallel=1)
|
|
mock_get_dm.return_value = manager
|
|
mock_get_dm.return_value = manager
|
|
|
|
response = journey_client.post(
|
|
"/api/anime/download-season?url=https://anime-sama.si/test&lang=vostfr"
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["total_episodes"] == 5
|
|
assert len(data["task_ids"]) == 5
|