import asyncio import base64 import copy import unittest from datetime import timedelta from types import SimpleNamespace from unittest.mock import AsyncMock, patch from urllib.parse import parse_qs, urlsplit import httpx from fastapi import FastAPI, HTTPException from fastapi.testclient import TestClient from backend.app.routers import insights as router from backend.app.services import insights, insights_artwork as artwork from backend.tests.test_insights import NOW, LIBRARIES, USER, play ITEM = "a" * 32 OTHER = "b" * 32 PNG = base64.b64decode("iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+jK1sAAAAASUVORK5CYII=") def transcode(activity, minutes, video_direct=False, audio_direct=False, hardware="nvenc", **kwargs): return play(activity, NowPlayingItemId=ITEM, PlaybackDuration=minutes * 60, PlayMethod="Transcode", TranscodingInfo={"IsVideoDirect": video_direct, "IsAudioDirect": audio_direct, "HardwareAccelerationType": hardware, "VideoCodec": "h264", "AudioCodec": "aac"}, **kwargs) class TranscodingSummaryTests(unittest.TestCase): def test_gpu_audio_software_and_unknown_time_are_separate_and_deduplicated(self): gpu = transcode("gpu", 10) rows = [gpu, dict(gpu), transcode("audio-only", 5, video_direct=True), transcode("software", 3, audio_direct=True, hardware="none"), transcode("remux", 2, video_direct=True, audio_direct=True), {**transcode("stale", 1), "PlayMethod": "DirectPlay"}, transcode("unknown-hardware", 2, audio_direct=True, hardware=None), {**transcode("unknown-streams", 1), "TranscodingInfo": None}, transcode("old", 100, ActivityDateInserted=(NOW - timedelta(days=31)).isoformat())] result = insights.summarize(rows, LIBRARIES, NOW - timedelta(days=30), NOW) stats = result["transcoding"] self.assertEqual(stats["hardware_video_minutes"], 10) self.assertEqual(stats["audio_minutes"], 15) self.assertEqual(stats["video_minutes"], 15) self.assertEqual(stats["software_video_minutes"], 3) self.assertEqual(stats["unknown_hardware_minutes"], 2) self.assertEqual(stats["unknown_video_minutes"], 1) self.assertEqual(stats["unknown_audio_minutes"], 1) self.assertEqual(stats["hardware"], [{"name": "NVIDIA NVENC", "minutes": 10}]) self.assertIsNone(stats["gpu_busy_minutes"]) def test_audio_media_cannot_accumulate_video_gpu_time(self): row = transcode("music", 4, ParentId="music") row["TranscodingInfo"]["VideoCodec"] = None result = insights.summarize([row], LIBRARIES, NOW - timedelta(days=7), NOW)["transcoding"] self.assertEqual(result["audio_minutes"], 4) self.assertEqual(result["video_minutes"], 0) self.assertEqual(result["hardware_video_minutes"], 0) def test_direct_stream_counts_audio_but_does_not_claim_video_encoding(self): row = {**transcode("stream", 5), "PlayMethod": "DirectStream"} result = insights.summarize([row], LIBRARIES, NOW - timedelta(days=7), NOW)["transcoding"] self.assertEqual(result["audio_minutes"], 5) self.assertEqual(result["hardware_video_minutes"], 0) def test_numeric_hardware_enum_and_legacy_json_are_supported(self): import json row = transcode("enum", 5, hardware=3) row["TranscodingInfo"] = json.dumps(row["TranscodingInfo"]) result = insights.summarize([row], LIBRARIES, NOW - timedelta(days=7), NOW)["transcoding"] self.assertEqual(result["hardware_video_minutes"], 5) row["TranscodingInfo"] = "invalid JSON" result = insights.summarize([row], LIBRARIES, NOW - timedelta(days=7), NOW)["transcoding"] self.assertEqual(result["unknown_video_minutes"], 5) self.assertEqual(result["hardware_video_minutes"], 0) def test_episode_artwork_uses_series_id_and_invalid_ids_are_ignored(self): rows = [play("episode", EpisodeId=OTHER, NowPlayingItemId=ITEM), play("invalid", NowPlayingItemId="../../secret")] result = insights.summarize(rows, LIBRARIES, NOW - timedelta(days=7), NOW) indexed = {row["id"]: row for row in result["recent"]} self.assertEqual(indexed["episode"]["artwork_item_id"], ITEM) self.assertIsNone(indexed["invalid"]["artwork_item_id"]) class ArtworkTests(unittest.IsolatedAsyncioTestCase): def setUp(self): self.runtime = SimpleNamespace(jellyfin_base_url="http://jellyfin", jellyfin_api_key="PRIVATE-API-KEY") self.secret = patch.object(artwork.settings, "jwt_secret", "test-artwork-signing-secret") self.secret.start() self.addCleanup(self.secret.stop) artwork._cache.clear() artwork._downloads = asyncio.Semaphore(6) def url_and_token(self, user=USER): data = {"recent": [{"id": "play-1", "artwork_item_id": ITEM}]} before = copy.deepcopy(data) result = artwork.with_artwork(data, user, self.runtime) self.assertEqual(data, before) row = result["recent"][0] self.assertNotIn("artwork_item_id", row) self.assertNotIn("PRIVATE-API-KEY", row["artwork_url"]) return row["artwork_url"], parse_qs(urlsplit(row["artwork_url"]).query)["token"][0] async def test_ticket_is_bound_to_user_item_server_credentials_and_time(self): with patch.object(artwork.time, "time", return_value=1000): _, token = self.url_and_token() artwork.verify_artwork_token(USER, self.runtime, ITEM, token) for user, runtime, media_id in [({**USER, "username": "different"}, self.runtime, ITEM), (USER, self.runtime, OTHER), (USER, SimpleNamespace(jellyfin_base_url="http://other", jellyfin_api_key="PRIVATE-API-KEY"), ITEM), (USER, SimpleNamespace(jellyfin_base_url="http://jellyfin", jellyfin_api_key="changed"), ITEM)]: with self.assertRaises(HTTPException) as raised: artwork.verify_artwork_token(user, runtime, media_id, token) self.assertEqual(raised.exception.status_code, 403) with patch.object(artwork.time, "time", return_value=5000), self.assertRaises(HTTPException): artwork.verify_artwork_token(USER, self.runtime, ITEM, token) for invalid in ["invalid", "1.", "1." + "\u2603" * 64]: with self.assertRaises(HTTPException): artwork.verify_artwork_token(USER, self.runtime, ITEM, invalid) async def test_private_proxy_returns_image_and_validates_before_cache_access(self): calls = [] def handler(request): calls.append(request) self.assertEqual(request.url.path, f"/Items/{ITEM}/Images/Primary") self.assertNotIn("PRIVATE", str(request.url)) self.assertEqual(request.headers["X-Emby-Token"], "PRIVATE-API-KEY") return httpx.Response(200, content=PNG, headers={"Content-Type": "image/png"}) real = httpx.AsyncClient _, token = self.url_and_token() with patch.object(artwork.httpx, "AsyncClient", side_effect=lambda **kwargs: real(transport=httpx.MockTransport(handler), **kwargs)): self.assertEqual(await artwork.get_artwork(USER, self.runtime, ITEM, token), (PNG, "image/png")) self.assertEqual(await artwork.get_artwork(USER, self.runtime, ITEM, token), (PNG, "image/png")) with self.assertRaises(HTTPException): await artwork.get_artwork({**USER, "username": "someone-else"}, self.runtime, ITEM, token) self.assertEqual(len(calls), 1) async def test_non_images_missing_images_and_large_images_fail_closed(self): _, token = self.url_and_token() real = httpx.AsyncClient for status, body, mime in [(404, b"PRIVATE", "text/plain"), (200, b"PRIVATE", "image/svg+xml"), (200, b"x" * (artwork.MAX_IMAGE_BYTES + 1), "image/png")]: transport = httpx.MockTransport(lambda request: httpx.Response(status, content=body, headers={"Content-Type": mime})) with patch.object(artwork.httpx, "AsyncClient", side_effect=lambda **kwargs: real(transport=transport, **kwargs)): with self.assertRaises(HTTPException) as raised: await artwork.get_artwork(USER, self.runtime, ITEM, token) self.assertEqual(raised.exception.status_code, 404) self.assertNotIn("PRIVATE", raised.exception.detail) self.assertEqual(len(artwork._cache), 0) class ArtworkRouteTests(unittest.TestCase): def test_authentication_and_private_response_headers(self): app = FastAPI() app.include_router(router.router) client = TestClient(app) self.assertEqual(client.get(f"/insights/artwork/{ITEM}?token=invalid").status_code, 401) app.dependency_overrides[router.get_current_user] = lambda: {**USER, "features": {"stats": True}} with patch.object(router, "get_runtime_settings", return_value=None), \ patch.object(router, "get_artwork", new_callable=AsyncMock, return_value=(PNG, "image/png")): result = client.get(f"/insights/artwork/{ITEM}?token=fixture") self.assertEqual(result.status_code, 200) self.assertEqual(result.content, PNG) self.assertEqual(result.headers["cache-control"], "private, max-age=600") self.assertEqual(result.headers["vary"], "Cookie, Authorization") self.assertEqual(result.headers["x-content-type-options"], "nosniff")