94 lines
3.7 KiB
Python
94 lines
3.7 KiB
Python
"""Exercise user-scoped movie queries without loading local credentials."""
|
|
|
|
import importlib.util
|
|
import sys
|
|
import types
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import httpx
|
|
|
|
|
|
class JellyfinQueryTests(unittest.IsolatedAsyncioTestCase):
|
|
async def asyncSetUp(self):
|
|
config = types.ModuleType("app.config")
|
|
config.settings = types.SimpleNamespace(
|
|
jellyfin_api_key="fixture-token",
|
|
jellyfin_url="https://jellyfin.invalid",
|
|
jellyfin_external_url="https://jellyfin.invalid",
|
|
)
|
|
source = Path(__file__).resolve().parents[1] / "app/services/jellyfin.py"
|
|
spec = importlib.util.spec_from_file_location("jellyfin_fixture", source)
|
|
self.service = importlib.util.module_from_spec(spec)
|
|
with patch.dict(sys.modules, {"app.config": config}):
|
|
spec.loader.exec_module(self.service)
|
|
|
|
async def asyncTearDown(self):
|
|
if self.service._client is not None:
|
|
await self.service._client.aclose()
|
|
|
|
async def query(self, user_id, *, played_only):
|
|
requests = []
|
|
|
|
def respond(request):
|
|
requests.append(request)
|
|
self.assertEqual(request.url.path, "/Items")
|
|
self.assertEqual(request.url.params["UserId"], user_id)
|
|
self.assertEqual(request.url.params["IncludeItemTypes"], "Movie")
|
|
self.assertEqual(request.url.params["Recursive"], "true")
|
|
self.assertEqual(request.url.params["Limit"], "200")
|
|
self.assertIn('Token="fixture-token"', request.headers["Authorization"])
|
|
if played_only:
|
|
self.assertEqual(request.url.params["IsPlayed"], "true")
|
|
else:
|
|
self.assertNotIn("IsPlayed", request.url.params)
|
|
self.assertIn("RunTimeTicks", request.url.params["Fields"])
|
|
start = int(request.url.params["StartIndex"])
|
|
self.assertIn(start, (0, 200))
|
|
items = [
|
|
{
|
|
"Id": f"{user_id}-{index}",
|
|
"Name": f"Movie {index}",
|
|
"RunTimeTicks": 60_000_000_000,
|
|
"UserData": {"Played": played_only or index % 2 == 0},
|
|
}
|
|
for index in range(start, min(start + 200, 201))
|
|
]
|
|
return httpx.Response(200, json={"Items": items, "TotalRecordCount": 201})
|
|
|
|
self.service._client = httpx.AsyncClient(
|
|
base_url="https://jellyfin.invalid",
|
|
headers=self.service._headers,
|
|
transport=httpx.MockTransport(respond),
|
|
)
|
|
if played_only:
|
|
actual = await self.service.get_played_movie_ids(user_id)
|
|
self.assertEqual(actual, {f"{user_id}-{i}" for i in range(201)})
|
|
else:
|
|
actual = await self.service.get_all_movies(user_id)
|
|
self.assertEqual(
|
|
[movie["jellyfin_id"] for movie in actual],
|
|
[f"{user_id}-{i}" for i in range(201)],
|
|
)
|
|
self.assertEqual(actual[0]["runtime_minutes"], 100)
|
|
self.assertTrue(actual[0]["is_played"])
|
|
self.assertFalse(actual[1]["is_played"])
|
|
self.assertEqual([r.url.params["StartIndex"] for r in requests], ["0", "200"])
|
|
|
|
async def test_first_user_movie_library(self):
|
|
await self.query("first-user", played_only=False)
|
|
|
|
async def test_second_user_movie_library(self):
|
|
await self.query("second-user", played_only=False)
|
|
|
|
async def test_first_user_watch_history(self):
|
|
await self.query("first-user", played_only=True)
|
|
|
|
async def test_second_user_watch_history(self):
|
|
await self.query("second-user", played_only=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|