Files
Roman fed61adc56 Pour toi : recommandations diverses et déjà-possédé réellement exclu
Deux défauts enchaînés : la requête Kitsu cumulait les genres en ET
(quatre catégories → 1 seul titre, précisément le téléchargement à
l'origine des genres), et l'exclusion échouait car « Titre - Saison 1 -
E3 » ne normalisait jamais vers « Titre » canonique.

- Une requête Kitsu par genre, fusion entrelacée (chaque genre contribue)
- normalize_title retire saison/épisode en boucle avec frontières de mots
  (« - Saison 1 - E3 », « S1 E1 » → titre canonique ; « Série 9 » intact)
- test de non-régession : le possédé avec marqueurs n'est plus recommandé
2026-09-25 15:23:29 +00:00

526 lines
20 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""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 = """
<html><body>
<div class="movie-poster">
<div class="play"><a class="fa fa-play link"
href="https://ipv4.vostfree.ws/1404-helck-vostfr-ddl-streaming-1fichier-uptobox.html"
alt="Helck VOSTFR"><span>Helck VOSTFR</span></a></div>
<div class="quality">VOSTFR</div>
<span class="image"><img src="https://vostfree.ws/uploads/posts/helck.jpg" alt="Helck VOSTFR"/></span>
</div>
<div class="movie-poster">
<div class="play"></div>
</div>
<div class="movie-poster">
<div class="play"><a class="fa fa-play link" href="/sans-slug-" alt="Bizarre"></a></div>
</div>
</body></html>
"""
FRENCH_MANGA_LATEST_HTML = """
<html><body>
<div class="short"><div class="short-in nl">
<a class="short-poster img-box with-mask"
href="https://w16.french-manga.net/index.php?newsid=1498905" alt="Draw This, Then Die! - Saison 1">
<img src="https://image.tmdb.org/t/p/w500/dZp.jpg" width="160" height="240" alt="affiche"/>
</a>
<span class="mli-eps">11 / 12</span>
<div class="short-title">Draw This, Then Die! - Saison 1 (2024)</div>
</div></div>
<div class="short"><div class="short-in nl">
<a class="short-poster img-box with-mask" href="/relative-no-id.html"></a>
</div></div>
</body></html>
"""
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_by_type(limit=10))["anime"]
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_by_type(limit=10))["anime"]
assert [item["title"] for item in result] == ["Anime 1", "Anime 0", "Anime 2"]
assert result[0]["rating"] is None
async def test_latest_by_type_keeps_every_type(monkeypatch):
"""Chaque type garde son rail complet : les animés nombreux ne vicient pas
le rail séries (les séries n'existent pas chez Kitsu, donc sans date)."""
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)
rails = await service.latest_by_type(limit=24)
assert len(rails["anime"]) == 24 # tronqué à la limite
assert len(rails["serie"]) == 9 # les 9 séries intactes, malgré 30 animés
assert all(item["media_type"] == "serie" for item in rails["serie"])
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_by_type() == {"anime": [], "serie": [], "film": []}
assert await service.latest_by_type() == {"anime": [], "serie": [], "film": []}
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]:
assert params["sort"] == "-userCount"
# Une catégorie par requête (le cumul Kitsu est un ET trop restrictif)
assert "," not in params["filter[categories]"]
return [
{"title": "Helck", "kitsu_id": "999"},
{"title": "Sousou no Frieren", "kitsu_id": "46474"},
{"title": "Konosuba", "kitsu_id": "1"},
]
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_excludes_owned_with_episode_markers(monkeypatch):
"""Régression : « Titre - Saison 1 - E3 » possédé doit exclure le « Titre »
canonique recommandé (la normalisation retire saison ET épisode, en boucle)."""
cursor = await db.execute(
"INSERT INTO users (username, password_hash) VALUES (?, ?)",
("reg-user", "x" * 64),
)
user_id = cursor.lastrowid
await db.execute(
"INSERT INTO downloads (source_key, video_url, title, status) VALUES (?,?,?,?)",
("kr", "https://v/r", "The Eminence in Shadow - Saison 1 - E3", "done"),
)
await db.execute(
"INSERT INTO favorites (user_id, source, source_id, title, image_url, payload) "
"VALUES (?,?,?,?,?,?)",
(user_id, "vostfree", "r", "Favori R", None, '{"genres": ["Fantasy"]}'),
)
async def fake_match(title: str):
return None # genres apportés par le favori
async def fake_kitsu_anime(params: dict) -> list[dict]:
return [
{"title": "The Eminence in Shadow", "kitsu_id": "1"},
{"title": "Autre Anime", "kitsu_id": "2"},
]
service = DiscoverService()
monkeypatch.setattr(service, "_kitsu_match_for_title", fake_match)
monkeypatch.setattr(service, "_kitsu_anime", fake_kitsu_anime)
result = await service.for_you(user_id=user_id, limit=10)
titles = [item["title"] for item in result["items"]]
assert "The Eminence in Shadow" not in titles # possédé (« - Saison 1 - E3 »)
assert "Autre Anime" in titles
async def test_for_you_cold_start_without_history(monkeypatch):
"""Aucun téléchargement/favori/Sonarr : amorçage signalé (non caché)."""
service = DiscoverService()
result = await service.for_you(user_id=42)
assert result == {"based_on": [], "items": [], "cold_start": True}
assert service._cache.get("for_you:42:20") is None # pas de cache : bon marché, change au 1er téléchargement
async def test_for_you_uses_series_page_genres_without_sonarr(monkeypatch):
"""Sans Sonarr, les séries téléchargées donnent leurs genres via leur fiche
source (libellés FR convertis en catégories Kitsu)."""
cursor = await db.execute(
"INSERT INTO users (username, password_hash) VALUES (?, ?)",
("serie-user", "x" * 64),
)
user_id = cursor.lastrowid
await db.execute(
"INSERT INTO downloads (source_key, video_url, page_url, title, status) VALUES (?,?,?,?,?)",
("k9", "https://v/9", "https://french-stream.lat/15138132-stat-saison-5.html",
"STAT - Saison 5", "done"),
)
async def fake_no_match(title: str):
return None # série réelle : aucun match Kitsu
async def fake_genres_of_page(self, page_url: str):
assert "15138132" in page_url
return ["Drame", "Médical"] # Médical : sans équivalent Kitsu → ignoré
service = DiscoverService()
monkeypatch.setattr(service._kitsu, "search_anime", fake_no_match)
monkeypatch.setattr(
"app.scrapers.sources.french_stream.FrenchStreamScraper.genres_of_page",
fake_genres_of_page,
)
result = await service.for_you(user_id=user_id, limit=10)
assert result["based_on"] == ["Drama"] # « Drame » converti, « Médical » écarté
assert result["items"] # des recommandations issues de la catégorie drama
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_by_type(limit: int = 24):
return {
"anime": [
{"source": "vostfree", "label": "Vostfree", "source_id": "a", "title": "T",
"media_type": "anime", "start_date": "2026-01-01", "status": "current", "rating": 8.1},
],
"serie": [
{"source": "french_stream", "label": "French-Stream", "source_id": "b",
"title": "S", "media_type": "serie"},
],
"film": [],
}
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_by_type", fake_latest_by_type)
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_anime"][0]["label"] == "Vostfree"
assert data["latest_anime"][0]["start_date"] == "2026-01-01"
assert data["latest_serie"][0]["label"] == "French-Stream"
assert data["must_watch"][0]["title"] == "Attack on Titan"
assert data["for_you"]["based_on"] == ["Action"]
async def test_api_browse(client, admin_cookies, monkeypatch):
"""Explorer : animés via Kitsu, séries via French-Stream, hors préférence → vide."""
async def fake_browse_anime(genre: str, limit: int = 20):
assert genre == "comedy"
return [{"kitsu_id": "1", "title": "Nichijou"}]
async def fake_browse_serie_film(media_type: str, genre: str, limit: int = 24):
assert (media_type, genre) == ("serie", "medical")
return [{"source": "french_stream", "source_id": "9", "title": "STAT - Saison 5",
"media_type": "serie"}]
monkeypatch.setattr(discover, "browse_anime", fake_browse_anime)
monkeypatch.setattr(discover, "browse_serie_film", fake_browse_serie_film)
r = await client.get(
"/api/discover/browse", params={"type": "anime", "genre": "comedy"}, cookies=admin_cookies
)
assert r.json()["items"][0]["title"] == "Nichijou"
r = await client.get(
"/api/discover/browse", params={"type": "serie", "genre": "medical"}, cookies=admin_cookies
)
assert r.json()["items"][0]["title"] == "STAT - Saison 5"
# Préférence animés : le parcours séries renvoie vide sans même scraper
await client.put("/auth/preferences", json={"content_preference": "anime"}, cookies=admin_cookies)
r = await client.get(
"/api/discover/browse", params={"type": "serie", "genre": "medical"}, cookies=admin_cookies
)
assert r.json()["items"] == []
async def test_api_genres_catalog(client, admin_cookies):
"""Catalogue de genres : animés + séries/films French-Stream en mode « les deux »."""
r = await client.get("/api/discover/genres", cookies=admin_cookies)
assert r.status_code == 200
catalog = r.json()
assert {"anime", "serie", "film"} <= set(catalog)
assert {"key": "thriller", "label": "Thriller"} in catalog["serie"]
assert any(g["key"] == "comedies" for g in catalog["film"])
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