"""Tests découverte : parsing latest(), service for_you et endpoint /api/discover.""" import asyncio import pytest from httpx import ASGITransport, AsyncClient from app.db import db from app.main import app from app.scrapers.base import ScrapeError, SearchResult from app.services.discover import DiscoverService, category_slug, discover # --------------------------------------------------------------- parsing latest VOSTFREE_LATEST_HTML = """

VOSTFR

""" FRENCH_MANGA_LATEST_HTML = """

11 / 12

Draw This, Then Die! - Saison 1 (2024)

""" class _FakeSource: """Source factice injectée dans le registre du service discover.""" name = "fake" label = "Fake" def __init__(self, results: list[SearchResult] | None = None, error: bool = False): self._results = results or [] self._error = error async def latest(self) -> list[SearchResult]: if self._error: raise ScrapeError("source indisponible") return self._results @pytest.fixture async def client() -> AsyncClient: transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as c: yield c @pytest.fixture async def admin_cookies(client: AsyncClient): r = await client.post("/auth/register", data={"username": "admin", "password": "secret123"}) assert r.status_code == 303 return r.cookies def test_category_slug(): assert category_slug("Action") == "action" assert category_slug("Slice of Life") == "slice-of-life" assert category_slug("Comédie") == "comedie" # accents retirés assert category_slug(" Super Power! ") == "super-power" async def test_latest_vostfree_parsing(monkeypatch): async def fake_fetch_soup(url, **kwargs): from bs4 import BeautifulSoup return BeautifulSoup(VOSTFREE_LATEST_HTML, "lxml") monkeypatch.setattr("app.scrapers.sources.vostfree.fetch_soup", fake_fetch_soup) results = await asyncio.shield(_latest_of("vostfree")) assert len(results) == 1 # les blocs sans lien/slug valide sont ignorés item = results[0] assert item.source == "vostfree" assert item.source_id == "1404-helck-vostfr-ddl-streaming-1fichier-uptobox.html".removesuffix( ".html" ) assert item.title == "Helck VOSTFR" # depuis l'attribut alt assert item.image_url == "https://vostfree.ws/uploads/posts/helck.jpg" async def _latest_of(name: str) -> list[SearchResult]: from app.scrapers.base import get_source return await get_source(name).latest() async def test_latest_french_manga_parsing(monkeypatch): async def fake_fetch_soup(url, **kwargs): from bs4 import BeautifulSoup return BeautifulSoup(FRENCH_MANGA_LATEST_HTML, "lxml") monkeypatch.setattr("app.scrapers.sources.french_manga.fetch_soup", fake_fetch_soup) results = await _latest_of("french_manga") assert len(results) == 1 item = results[0] assert item.source == "french_manga" assert item.source_id == "1498905" assert item.title == "Draw This, Then Die! - Saison 1" # année en queue retirée assert item.image_url == "https://image.tmdb.org/t/p/w500/dZp.jpg" assert item.url.endswith("newsid=1498905") async def test_latest_merges_and_sorts_by_release_date(monkeypatch): """Sources fusionnées, doublons retirés, tri du plus récent au plus ancien.""" import app.services.discover as discover_module src_a = _FakeSource( [ SearchResult(source="a", source_id="1", title="Frieren VOSTFR", url="https://x/1"), SearchResult(source="a", source_id="2", title="Helck VOSTFR", url="https://x/2"), ] ) src_b = _FakeSource( [ SearchResult(source="b", source_id="9", title="Frieren - Saison 1", url="https://y/9"), SearchResult(source="b", source_id="8", title="Old Anime", url="https://y/8"), ] ) monkeypatch.setattr(discover_module, "all_sources", lambda: [src_a, src_b]) async def enabled(name: str) -> bool: return True monkeypatch.setattr(discover_module, "is_source_enabled", enabled) async def fake_match(title: str): return None # aucun match → items sans date, triés en fin de liste service = DiscoverService() monkeypatch.setattr(service, "_kitsu_match_for_title", fake_match) result = await service.latest(limit=10) titles = [item["title"] for item in result] assert titles.count("Frieren VOSTFR") + titles.count("Frieren - Saison 1") == 1 # dédoublonné assert "Helck VOSTFR" in titles and "Old Anime" in titles assert all(item["start_date"] is None for item in result) async def test_latest_orders_by_kitsu_start_date(monkeypatch): """Les dates de sortie Kitsu pilotent l'ordre (plus récent d'abord).""" import app.services.discover as discover_module src = _FakeSource( [ SearchResult(source="a", source_id=str(i), title=f"Anime {i}", url=f"https://x/{i}") for i in range(3) ] ) monkeypatch.setattr(discover_module, "all_sources", lambda: [src]) async def enabled(name: str) -> bool: return True monkeypatch.setattr(discover_module, "is_source_enabled", enabled) async def fake_match(title: str): dates = {"Anime 0": "2021-01-01", "Anime 1": "2026-01-01", "Anime 2": None} start = dates[title] return {"id": "1", "attributes": {"startDate": start}, "genres": []} service = DiscoverService() monkeypatch.setattr(service, "_kitsu_match_for_title", fake_match) result = await service.latest(limit=10) assert [item["title"] for item in result] == ["Anime 1", "Anime 0", "Anime 2"] assert result[0]["rating"] is None async def test_latest_serie_preference_keeps_undated_titles(monkeypatch): """Régression : en mode séries, les titres sans date Kitsu ne doivent pas être évincés du rail par la troncature (les séries n'existent pas chez Kitsu).""" import app.services.discover as discover_module animes = [ SearchResult(source="vostfree", source_id=str(n), title=f"Animé {n}", url=f"https://a/{n}") for n in range(1, 31) ] series = [ SearchResult( source="french_stream", source_id=str(n), title=f"Série {n}", url=f"https://s/{n}", media_type="serie", ) for n in range(1, 10) ] class AnimeSource(_FakeSource): name, label, base_url, media_types = "vostfree", "Vostfree", "https://a", ("anime",) class SerieSource(_FakeSource): name, label, base_url, media_types = "french_stream", "French-Stream", "https://s", ("serie", "film") monkeypatch.setattr(discover_module, "all_sources", lambda: [AnimeSource(animes), SerieSource(series)]) async def enabled(name: str) -> bool: return True monkeypatch.setattr(discover_module, "is_source_enabled", enabled) async def fake_match(title: str): return None # rien chez Kitsu → aucune date de sortie service = DiscoverService() monkeypatch.setattr(service, "_kitsu_match_for_title", fake_match) result = await service.latest(limit=24, allowed={"serie"}) assert {item["media_type"] for item in result} == {"serie"} assert len(result) == 9 # les 9 séries, malgré 30 animés sans filtre interne async def test_latest_skips_broken_source_and_uses_cache(monkeypatch): """Une source en échec disparaît sans erreur, et le TTL évite les re-scrapes.""" import app.services.discover as discover_module calls = {"n": 0} def make_source(): async def latest(): calls["n"] += 1 return [] return type("S", (), {"name": "s", "label": "S", "latest": staticmethod(latest)})() monkeypatch.setattr( discover_module, "all_sources", lambda: [_FakeSource(error=True), make_source()] ) async def enabled(name: str) -> bool: return True monkeypatch.setattr(discover_module, "is_source_enabled", enabled) service = DiscoverService() assert await service.latest() == [] assert await service.latest() == [] assert calls["n"] == 1 # deuxième appel servi depuis le cache TTL # --------------------------------------------------------------- pour toi @pytest.fixture async def history(monkeypatch): """Téléchargements + favoris en base, Kitsu mocké.""" cursor = await db.execute( "INSERT INTO users (username, password_hash) VALUES (?, ?)", ("tester", "x" * 64), ) user_id = cursor.lastrowid await db.execute( "INSERT INTO downloads (source_key, video_url, title, status) VALUES (?,?,?,?)", ("k1", "https://v/1", "Frieren S1 - E1", "done"), ) await db.execute( "INSERT INTO downloads (source_key, video_url, title, status) VALUES (?,?,?,?)", ("k2", "https://v/2", "Helck VOSTFR", "done"), ) await db.execute( "INSERT INTO favorites (user_id, source, source_id, title, image_url, payload) " "VALUES (?,?,?,?,?,?)", (user_id, "vostfree", "x", "Favori", None, '{"genres": ["Comedy", "Action"]}'), ) async def fake_search_anime(title: str): lowered = title.lower() if "frieren" in lowered: return {"id": "46474", "attributes": {}, "genres": []} # genres → via catégories if "helck" in lowered: return {"id": "999", "attributes": {}, "genres": ["Action"]} return None async def fake_categories(anime_id: object) -> list[str]: return ["Fantasy", "Adventure"] if anime_id == "46474" else [] async def fake_kitsu_anime(params: dict) -> list[dict]: slugs = frozenset(params["filter[categories]"].split(",")) assert params["sort"] == "-userCount" catalog = { frozenset(["fantasy", "adventure", "action", "comedy"]): [ {"title": "Helck", "kitsu_id": "999"}, {"title": "Sousou no Frieren", "kitsu_id": "46474"}, {"title": "Konosuba", "kitsu_id": "1"}, ], } return catalog.get(slugs, []) service = DiscoverService() service.test_user_id = user_id # pour les assertions du test monkeypatch.setattr(service._kitsu, "search_anime", fake_search_anime) monkeypatch.setattr(service, "_kitsu_categories", fake_categories) monkeypatch.setattr(service, "_kitsu_anime", fake_kitsu_anime) return service async def test_for_you_aggregates_genres_and_excludes_owned(history): result = await history.for_you(user_id=history.test_user_id, limit=10) assert set(result["based_on"]) == {"Action", "Fantasy", "Adventure", "Comedy"} assert result["based_on"][0] == "Action" # 2 occurrences (téléchargement + favori) titles = [item["title"] for item in result["items"]] assert "Helck" not in titles # déjà possédé (« Helck VOSTFR » → « helck ») → exclu assert "Sousou no Frieren" in titles assert "Konosuba" in titles async def test_for_you_empty_without_history(monkeypatch): service = DiscoverService() result = await service.for_you(user_id=42) assert result == {"based_on": [], "items": []} class _FakeSonarr: """Profil Sonarr factice : titres possédés + genres.""" def __init__(self, owned: set[str], genres: dict[str, int]) -> None: self._profile = (owned, genres) async def profile(self) -> tuple[set[str], dict[str, int]]: return self._profile async def test_for_you_merges_sonarr_genres_and_owned(monkeypatch, history): monkeypatch.setattr( "app.services.discover.sonarr", _FakeSonarr( owned={"mob psycho 100"}, # déjà possédé sur Sonarr → exclu genres={"Fantasy": 5, "Action": 5}, # Action cumule avec l'historique local ), ) async def fake_kitsu_anime(params: dict) -> list[dict]: return [ {"title": "Konosuba", "kitsu_id": "1"}, {"title": "Mob Psycho 100", "kitsu_id": "2"}, ] monkeypatch.setattr(history, "_kitsu_anime", fake_kitsu_anime) result = await history.for_you(user_id=history.test_user_id, limit=10) # genres locaux (Action×2, Fantasy, Adventure, Comedy) + Sonarr (Fantasy+5, Action+5) counts = {"Action": 7, "Fantasy": 6, "Adventure": 1, "Comedy": 1} assert sorted(result["based_on"]) == sorted( sorted(counts, key=counts.get, reverse=True)[:4] ) titles = [item["title"] for item in result["items"]] assert "Mob Psycho 100" not in titles # possédé sur Sonarr → exclu assert "Konosuba" in titles # --------------------------------------------------------------- endpoint async def test_api_discover_requires_auth(client): r = await client.get("/api/discover", follow_redirects=False) assert r.status_code == 303 async def test_api_discover_sections(client, admin_cookies, monkeypatch): async def fake_latest(limit: int = 24, allowed: set[str] | None = None): return [ {"source": "vostfree", "label": "Vostfree", "source_id": "a", "title": "T", "start_date": "2026-01-01", "status": "current", "rating": 8.1}, ] async def fake_must_watch(limit: int = 20): return [{"kitsu_id": "1", "title": "Attack on Titan", "rating": 8.5}] async def fake_for_you(user_id: int, limit: int = 20): return {"based_on": ["Action"], "items": [{"kitsu_id": "2", "title": "X"}]} monkeypatch.setattr(discover, "latest", fake_latest) monkeypatch.setattr(discover, "must_watch", fake_must_watch) monkeypatch.setattr(discover, "for_you", fake_for_you) r = await client.get("/api/discover", cookies=admin_cookies) assert r.status_code == 200 data = r.json() assert data["latest"][0]["label"] == "Vostfree" assert data["latest"][0]["start_date"] == "2026-01-01" assert data["must_watch"][0]["title"] == "Attack on Titan" assert data["for_you"]["based_on"] == ["Action"] async def test_discover_page_renders(client, admin_cookies): r = await client.get("/discover", cookies=admin_cookies) assert r.status_code == 200 assert "Découvrir" in r.text