# -*- coding: utf-8 -*- """ Tests fuer TMDBDiscoverClient — TMDB Discover API Client. Alle HTTP-Requests werden via unittest.mock simuliert. Testet Genre-Mapping, Normalisierung, Paginierung, Fehlerbehandlung und Standard-Sprach-Zuordnungen pro Dienst. """ from __future__ import annotations import asyncio from datetime import date, timedelta from unittest.mock import AsyncMock, MagicMock, patch import pytest from plugins.streaming.catalog.tmdb_discover import ( GENRE_MAP_MOVIE, GENRE_MAP_TV, PROVIDER_IDS, TMDBDiscoverClient, _DEFAULT_LANGUAGES, ) # --------------------------------------------------------------------------- # Hilfsfunktionen und Testdaten # --------------------------------------------------------------------------- def _tmdb_movie( title: str = "Inception", movie_id: int = 27205, genre_ids: list[int] | None = None, rating: float = 8.4, release_date: str = "2010-07-16", overview: str = "Ein Dieb stiehlt Geheimnisse aus dem Unterbewusstsein.", original_language: str = "en", ) -> dict: """Erzeugt ein TMDB-Film-Rohergebnis.""" return { "id": movie_id, "title": title, "genre_ids": genre_ids or [28, 878], "vote_average": rating, "release_date": release_date, "overview": overview, "original_language": original_language, } def _tmdb_tv( name: str = "Breaking Bad", tv_id: int = 1396, genre_ids: list[int] | None = None, rating: float = 9.5, first_air_date: str = "2008-01-20", overview: str = "Ein Chemielehrer wird zum Drogenboss.", original_language: str = "en", ) -> dict: """Erzeugt ein TMDB-TV-Rohergebnis.""" return { "id": tv_id, "name": name, "genre_ids": genre_ids or [18, 80], "vote_average": rating, "first_air_date": first_air_date, "overview": overview, "original_language": original_language, } def _discover_page(results: list[dict], page: int = 1, total_pages: int = 1) -> dict: """Erzeugt eine TMDB-Discover-API-Antwort (einzelne Seite).""" return { "page": page, "total_pages": total_pages, "total_results": len(results), "results": results, } @pytest.fixture() def client() -> TMDBDiscoverClient: """Liefert einen TMDBDiscoverClient mit Test-API-Key.""" return TMDBDiscoverClient(api_key="test-api-key-123", region="DE", language="de-DE") # =================================================================== # Konstruktor # =================================================================== class TestKonstruktor: """Tests fuer __init__.""" def test_konfiguration_gespeichert(self) -> None: """API-Key, Region und Sprache werden korrekt gespeichert.""" c = TMDBDiscoverClient(api_key="my-key", region="US", language="en-US") assert c._api_key == "my-key" assert c._region == "US" assert c._language == "en-US" # =================================================================== # Genre-Mapping # =================================================================== class TestMapGenres: """Tests fuer _map_genres().""" def test_film_genres_korrekt(self, client: TMDBDiscoverClient) -> None: """Film-Genre-IDs werden zu deutschen Namen aufgeloest.""" ergebnis = client._map_genres([28, 878, 53], "movie") assert ergebnis == ["Action", "Science Fiction", "Thriller"] def test_tv_genres_korrekt(self, client: TMDBDiscoverClient) -> None: """TV-Genre-IDs werden zu deutschen Namen aufgeloest.""" ergebnis = client._map_genres([10759, 18], "tv") assert ergebnis == ["Action & Abenteuer", "Drama"] def test_unbekannte_genre_id_uebersprungen(self, client: TMDBDiscoverClient) -> None: """Unbekannte Genre-IDs werden uebersprungen.""" ergebnis = client._map_genres([28, 99999, 53], "movie") assert ergebnis == ["Action", "Thriller"] assert len(ergebnis) == 2 # =================================================================== # Normalisierung Film # =================================================================== class TestNormalizeMovie: """Tests fuer _normalize_movie().""" def test_erstellt_korrektes_titel_dict(self, client: TMDBDiscoverClient) -> None: """Film wird korrekt in das Katalog-Format konvertiert.""" rohdaten = _tmdb_movie() ergebnis = client._normalize_movie(rohdaten, "netflix") assert ergebnis is not None assert ergebnis["t"] == "Inception" assert ergebnis["ty"] == "film" assert ergebnis["yr"] == 2010 assert ergebnis["rt"] == 8.4 assert ergebnis["orig_lang"] == "en" assert ergebnis["tmdb_id"] == 27205 assert ergebnis["src"] == "tmdb" assert "netflix" in ergebnis["on"] def test_standard_dub_sub_netflix(self, client: TMDBDiscoverClient) -> None: """Netflix bekommt de/en Dub und de Sub als Standard.""" rohdaten = _tmdb_movie() ergebnis = client._normalize_movie(rohdaten, "netflix") netflix_info = ergebnis["on"]["netflix"] assert "de" in netflix_info["dub"] assert "en" in netflix_info["dub"] assert "de" in netflix_info["sub"] def test_standard_dub_sub_crunchyroll(self, client: TMDBDiscoverClient) -> None: """Crunchyroll bekommt ja Dub und de Sub als Standard.""" rohdaten = _tmdb_movie() ergebnis = client._normalize_movie(rohdaten, "crunchyroll") cr_info = ergebnis["on"]["crunchyroll"] assert cr_info["dub"] == ["ja"] assert cr_info["sub"] == ["de"] def test_leerer_titel_gibt_none(self, client: TMDBDiscoverClient) -> None: """Film ohne Titel gibt None.""" rohdaten = _tmdb_movie(title="") ergebnis = client._normalize_movie(rohdaten, "netflix") assert ergebnis is None def test_fehlende_felder_graceful(self, client: TMDBDiscoverClient) -> None: """Fehlende optionale Felder werden sauber behandelt.""" rohdaten = {"title": "Testfilm", "id": 1} ergebnis = client._normalize_movie(rohdaten, "netflix") assert ergebnis is not None assert ergebnis["t"] == "Testfilm" assert ergebnis["yr"] is None assert ergebnis["g"] == [] # =================================================================== # Normalisierung TV # =================================================================== class TestNormalizeTv: """Tests fuer _normalize_tv().""" def test_erstellt_korrektes_titel_dict(self, client: TMDBDiscoverClient) -> None: """Serie wird korrekt in das Katalog-Format konvertiert.""" rohdaten = _tmdb_tv() ergebnis = client._normalize_tv(rohdaten, "netflix") assert ergebnis is not None assert ergebnis["t"] == "Breaking Bad" assert ergebnis["ty"] == "serie" assert ergebnis["yr"] == 2008 assert ergebnis["rt"] == 9.5 assert ergebnis["tmdb_id"] == 1396 def test_fehlende_felder_graceful(self, client: TMDBDiscoverClient) -> None: """Fehlende optionale Felder werden sauber behandelt.""" rohdaten = {"name": "Testserie", "id": 2} ergebnis = client._normalize_tv(rohdaten, "netflix") assert ergebnis is not None assert ergebnis["t"] == "Testserie" assert ergebnis["yr"] is None assert ergebnis["rt"] is None def test_leerer_name_gibt_none(self, client: TMDBDiscoverClient) -> None: """Serie ohne Name gibt None.""" rohdaten = _tmdb_tv(name="") ergebnis = client._normalize_tv(rohdaten, "netflix") assert ergebnis is None # =================================================================== # discover_service # =================================================================== class TestDiscoverService: """Tests fuer discover_service().""" async def test_kombiniert_filme_und_serien(self, client: TMDBDiscoverClient) -> None: """discover_service() kombiniert Filme und Serien.""" movie_page = _discover_page([_tmdb_movie()]) tv_page = _discover_page([_tmdb_tv()]) with patch.object(client, "_get", new_callable=AsyncMock) as mock_get: mock_get.side_effect = [movie_page, tv_page] ergebnis = await client.discover_service("netflix", max_titles=20) assert len(ergebnis) == 2 typen = {t["ty"] for t in ergebnis} assert "film" in typen assert "serie" in typen async def test_unbekannter_dienst_gibt_leere_liste( self, client: TMDBDiscoverClient, ) -> None: """Unbekannter Dienst gibt leere Liste ohne API-Aufruf.""" ergebnis = await client.discover_service("nichtexistent") assert ergebnis == [] # =================================================================== # Paginierung # =================================================================== class TestDiscoverMovies: """Tests fuer _discover_movies().""" async def test_paginierung_mehrere_seiten(self, client: TMDBDiscoverClient) -> None: """Paginierung liest mehrere Seiten bis max_pages.""" seite1 = _discover_page( [_tmdb_movie(title="Film A", movie_id=1)], page=1, total_pages=3, ) seite2 = _discover_page( [_tmdb_movie(title="Film B", movie_id=2)], page=2, total_pages=3, ) with patch.object(client, "_get", new_callable=AsyncMock) as mock_get: mock_get.side_effect = [seite1, seite2] ergebnis = await client._discover_movies(provider_id=8, max_pages=2) assert len(ergebnis) == 2 assert mock_get.call_count == 2 async def test_leere_ergebnisse_beenden_paginierung( self, client: TMDBDiscoverClient, ) -> None: """Leere Ergebnisseite beendet die Paginierung.""" leere_seite = _discover_page([], page=1, total_pages=5) with patch.object(client, "_get", new_callable=AsyncMock) as mock_get: mock_get.return_value = leere_seite ergebnis = await client._discover_movies(provider_id=8, max_pages=5) assert ergebnis == [] assert mock_get.call_count == 1 # =================================================================== # discover_recent # =================================================================== class TestDiscoverRecent: """Tests fuer discover_recent().""" async def test_nutzt_datumsfilter(self, client: TMDBDiscoverClient) -> None: """discover_recent() uebergibt Datums-Range an die API.""" movie_page = _discover_page([_tmdb_movie()]) tv_page = _discover_page([_tmdb_tv()]) with patch.object(client, "_get", new_callable=AsyncMock) as mock_get: mock_get.side_effect = [movie_page, tv_page] ergebnis = await client.discover_recent("netflix", days=7) assert len(ergebnis) == 2 # Pruefen, dass Datums-Parameter uebergeben wurden movie_call_params = mock_get.call_args_list[0][1].get("params") or mock_get.call_args_list[0][0][1] assert "primary_release_date.gte" in movie_call_params assert "primary_release_date.lte" in movie_call_params # =================================================================== # HTTP-Fehlerbehandlung (_get) # =================================================================== class TestGetFehlerbehandlung: """Tests fuer _get() — Fehlerszenarien.""" async def test_rate_limit_429_retry(self, client: TMDBDiscoverClient) -> None: """429 Rate-Limit fuehrt zu Retry.""" # Erstelle Mock-Response-Objekte response_429 = AsyncMock() response_429.status = 429 response_429.headers = {"Retry-After": "0.01"} response_200 = AsyncMock() response_200.status = 200 response_200.json = AsyncMock(return_value={"results": []}) # aiohttp.ClientSession Mocking mock_session = AsyncMock() mock_session.__aenter__ = AsyncMock(return_value=mock_session) mock_session.__aexit__ = AsyncMock(return_value=False) # get() gibt zuerst 429, dann 200 mock_resp_ctx_429 = AsyncMock() mock_resp_ctx_429.__aenter__ = AsyncMock(return_value=response_429) mock_resp_ctx_429.__aexit__ = AsyncMock(return_value=False) mock_resp_ctx_200 = AsyncMock() mock_resp_ctx_200.__aenter__ = AsyncMock(return_value=response_200) mock_resp_ctx_200.__aexit__ = AsyncMock(return_value=False) mock_session.get = MagicMock(side_effect=[mock_resp_ctx_429, mock_resp_ctx_200]) with patch("plugins.streaming.catalog.tmdb_discover.aiohttp") as mock_aiohttp: mock_aiohttp.ClientSession = MagicMock(return_value=mock_session) mock_aiohttp.ClientTimeout = MagicMock() mock_aiohttp.ClientError = Exception ergebnis = await client._get("/discover/movie") assert ergebnis == {"results": []} async def test_timeout_gibt_none(self, client: TMDBDiscoverClient) -> None: """Timeout fuehrt zu None.""" mock_session = AsyncMock() mock_session.__aenter__ = AsyncMock(return_value=mock_session) mock_session.__aexit__ = AsyncMock(return_value=False) mock_session.get = MagicMock(side_effect=asyncio.TimeoutError) with patch("plugins.streaming.catalog.tmdb_discover.aiohttp") as mock_aiohttp: mock_aiohttp.ClientSession = MagicMock(return_value=mock_session) mock_aiohttp.ClientTimeout = MagicMock() mock_aiohttp.ClientError = Exception ergebnis = await client._get("/discover/movie") assert ergebnis is None async def test_verbindungsfehler_gibt_none(self, client: TMDBDiscoverClient) -> None: """Verbindungsfehler fuehrt zu None.""" mock_session = AsyncMock() mock_session.__aenter__ = AsyncMock(return_value=mock_session) mock_session.__aexit__ = AsyncMock(return_value=False) # aiohttp.ClientError simulieren class MockClientError(Exception): pass mock_session.get = MagicMock(side_effect=MockClientError("Verbindung fehlgeschlagen")) with patch("plugins.streaming.catalog.tmdb_discover.aiohttp") as mock_aiohttp: mock_aiohttp.ClientSession = MagicMock(return_value=mock_session) mock_aiohttp.ClientTimeout = MagicMock() mock_aiohttp.ClientError = MockClientError ergebnis = await client._get("/discover/movie") assert ergebnis is None # =================================================================== # Konstanten und Konfiguration # =================================================================== class TestKonstanten: """Tests fuer Modul-Konstanten.""" def test_provider_ids_erwartete_dienste(self) -> None: """PROVIDER_IDS enthaelt die erwarteten Streaming-Dienste.""" erwartete = {"netflix", "amazon_prime", "crunchyroll", "disney_plus", "wow"} assert set(PROVIDER_IDS.keys()) == erwartete def test_genre_map_movie_hat_erwartete_genres(self) -> None: """GENRE_MAP_MOVIE enthaelt die wichtigsten Film-Genres.""" assert 28 in GENRE_MAP_MOVIE # Action assert 18 in GENRE_MAP_MOVIE # Drama assert 53 in GENRE_MAP_MOVIE # Thriller assert 878 in GENRE_MAP_MOVIE # Science Fiction def test_standard_sprachen_variieren_nach_dienst(self) -> None: """Verschiedene Dienste haben unterschiedliche Standard-Sprachen.""" cr = _DEFAULT_LANGUAGES["crunchyroll"] nf = _DEFAULT_LANGUAGES["netflix"] # Crunchyroll: Japanisch-Dub assert cr["dub"] == ["ja"] # Netflix: Deutsch/Englisch-Dub assert "de" in nf["dub"] assert "en" in nf["dub"]