397 lines
17 KiB
Python
397 lines
17 KiB
Python
"""Tests for the User Favourites feature (Emby Collections / BoxSets).
|
|
|
|
Uses a FakeEmbyClient (no network, no app.py import) and plain ``asyncio.run`` so
|
|
no pytest-asyncio plugin is required. Run from the repo root: ``python -m pytest``.
|
|
"""
|
|
|
|
import asyncio
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
from services import emby_collections, emby_users, emby_watch_history, favorites
|
|
from services import recommendations as rec
|
|
|
|
|
|
def run(coro):
|
|
return asyncio.run(coro)
|
|
|
|
|
|
# ── Item builders ────────────────────────────────────────────────────────────
|
|
|
|
def movie(item_id, name, *, genres=None, year=2014, studio="Acme",
|
|
director="Jane Doe", actor="Bob Roe", ticks=72_000_000_000, rating=7.0):
|
|
return {
|
|
"Id": item_id, "Name": name, "Type": "Movie", "MediaType": "Video",
|
|
"ProductionYear": year, "RunTimeTicks": ticks, "CommunityRating": rating,
|
|
"Genres": genres or ["Action"],
|
|
"Studios": [{"Name": studio}] if studio else [],
|
|
"People": ([{"Name": director, "Type": "Director"}] if director else [])
|
|
+ ([{"Name": actor, "Type": "Actor"}] if actor else []),
|
|
}
|
|
|
|
|
|
# ── Fake Emby client ─────────────────────────────────────────────────────────
|
|
|
|
class FakeEmbyClient:
|
|
def __init__(self, users=None, collections=None, collection_items=None,
|
|
watched_by_user=None, watched_history=None, candidates=None):
|
|
self.users = users or []
|
|
self.collections = collections or []
|
|
self.collection_items = collection_items or {} # cid -> [raw items]
|
|
self.watched_by_user = watched_by_user or {} # uid -> set(item_id)
|
|
self.watched_history = watched_history or {} # uid -> [raw items]
|
|
self.candidates = candidates or {} # uid -> [raw items]
|
|
self.calls = [] # recorded writes
|
|
|
|
async def get(self, path, params=None):
|
|
params = params or {}
|
|
if path == "/Users":
|
|
return list(self.users)
|
|
if path.startswith("/Users/") and path.endswith("/Items"):
|
|
uid = path.split("/")[2]
|
|
if "ParentId" in params: # collection children
|
|
cid = params["ParentId"]
|
|
watched = self.watched_by_user.get(uid, set())
|
|
items = []
|
|
for raw in self.collection_items.get(cid, []):
|
|
copy = dict(raw)
|
|
copy["UserData"] = {"Played": raw["Id"] in watched}
|
|
items.append(copy)
|
|
return {"Items": items}
|
|
if "Ids" in params: # is_item_watched
|
|
wanted = params["Ids"].split(",")
|
|
watched = self.watched_by_user.get(uid, set())
|
|
return {"Items": [{"Id": i, "UserData": {"Played": i in watched}} for i in wanted]}
|
|
if params.get("Filters") == "IsUnplayed": # candidate pool
|
|
return {"Items": list(self.candidates.get(uid, []))}
|
|
raise AssertionError(f"unexpected get {path} {params}")
|
|
|
|
async def get_all(self, path, params=None):
|
|
params = params or {}
|
|
if path == "/Items" and params.get("IncludeItemTypes") == "BoxSet":
|
|
return list(self.collections)
|
|
if path.startswith("/Users/") and path.endswith("/Items"):
|
|
uid = path.split("/")[2]
|
|
if params.get("Filters") == "IsPlayed":
|
|
return list(self.watched_history.get(uid, []))
|
|
raise AssertionError(f"unexpected get_all {path} {params}")
|
|
|
|
async def post(self, path, params=None, **kwargs):
|
|
self.calls.append(("post", path, params or {}))
|
|
cid = path.split("/")[2]
|
|
for item_id in (params or {}).get("Ids", "").split(","):
|
|
if item_id:
|
|
self.collection_items.setdefault(cid, []).append(
|
|
{"Id": item_id, "Name": item_id, "Type": "Movie"})
|
|
return {"ok": True}
|
|
|
|
async def delete(self, path, params=None, **kwargs):
|
|
self.calls.append(("delete", path, params or {}))
|
|
cid = path.split("/")[2]
|
|
remove = set((params or {}).get("Ids", "").split(","))
|
|
self.collection_items[cid] = [
|
|
i for i in self.collection_items.get(cid, []) if i["Id"] not in remove
|
|
]
|
|
return {"ok": True}
|
|
|
|
@property
|
|
def writes(self):
|
|
return [c for c in self.calls if c[0] in ("post", "delete")]
|
|
|
|
|
|
def base_client():
|
|
"""Two users, each with a favourites collection; Matt and Dave have watched
|
|
different items to prove user-specific behaviour. Plus a themed collection
|
|
that is not a favourites collection."""
|
|
return FakeEmbyClient(
|
|
users=[{"Id": "u-matt", "Name": "Matt"}, {"Id": "u-dave", "Name": "Dave"}],
|
|
collections=[
|
|
{"Id": "c-matt", "Name": "Matt Favorites", "ChildCount": 3},
|
|
{"Id": "c-dave", "Name": "Dave Favorites", "ChildCount": 1},
|
|
{"Id": "c-misc", "Name": "Marvel Universe", "ChildCount": 5},
|
|
],
|
|
collection_items={
|
|
"c-matt": [movie("m1", "Alpha"), movie("m2", "Bravo"), movie("m3", "Charlie")],
|
|
},
|
|
# Matt watched m1; Dave watched m2 (same items, different user)
|
|
watched_by_user={"u-matt": {"m1"}, "u-dave": {"m2"}},
|
|
)
|
|
|
|
|
|
# ── 1. Detecting favourites collections by name ──────────────────────────────
|
|
|
|
@pytest.mark.parametrize("name,owner", [
|
|
("Matt Favorites", "Matt"),
|
|
("Dave Favorites", "Dave"),
|
|
("Anna Maria Favorites", "Anna Maria"),
|
|
("favorites", None),
|
|
("Favorites", None),
|
|
("Marvel Universe", None),
|
|
("Favorites of Matt", None),
|
|
("", None),
|
|
(None, None),
|
|
])
|
|
def test_parse_favorites_owner(name, owner):
|
|
assert emby_collections.parse_favorites_owner(name) == owner
|
|
|
|
|
|
def test_list_favorites_users_only_matches_real_users():
|
|
users = run(favorites.list_favorites_users(base_client()))
|
|
assert [u["user_name"] for u in users] == ["Dave", "Matt"] # "Marvel Universe" excluded
|
|
|
|
|
|
def test_find_all_collections_includes_non_favorites():
|
|
cols = run(emby_collections.find_all_collections(base_client()))
|
|
by_name = {c["collection_name"]: c for c in cols}
|
|
assert set(by_name) == {"Matt Favorites", "Dave Favorites", "Marvel Universe"}
|
|
assert by_name["Matt Favorites"]["is_favorites"] is True
|
|
assert by_name["Marvel Universe"]["is_favorites"] is False
|
|
assert by_name["Marvel Universe"]["owner_name"] is None
|
|
|
|
|
|
def test_list_collections_overview_maps_owner_user_id():
|
|
overview = run(favorites.list_collections_overview(base_client()))
|
|
by_name = {c["collection_name"]: c for c in overview["collections"]}
|
|
assert by_name["Matt Favorites"]["owner_user_id"] == "u-matt"
|
|
assert by_name["Marvel Universe"]["owner_user_id"] is None
|
|
assert {u["name"] for u in overview["users"]} == {"Matt", "Dave"}
|
|
|
|
|
|
# ── 2. Listing collection items, user-specific watched status ────────────────
|
|
|
|
def test_list_collection_items_for_user():
|
|
items = run(emby_collections.list_collection_items(base_client(), "c-matt", "u-matt"))
|
|
assert [i["title"] for i in items] == ["Alpha", "Bravo", "Charlie"]
|
|
assert items[0]["runtime_minutes"] == 120
|
|
|
|
|
|
def test_watched_status_is_user_specific():
|
|
client = base_client()
|
|
matt = run(emby_collections.list_collection_items(client, "c-matt", "u-matt"))
|
|
dave = run(emby_collections.list_collection_items(client, "c-matt", "u-dave"))
|
|
assert [i["watched"] for i in matt] == [True, False, False] # Matt watched m1
|
|
assert [i["watched"] for i in dave] == [False, True, False] # Dave watched m2
|
|
|
|
|
|
def test_is_item_watched_user_specific():
|
|
client = base_client()
|
|
assert run(emby_watch_history.is_item_watched(client, "u-matt", "m1")) is True
|
|
assert run(emby_watch_history.is_item_watched(client, "u-dave", "m1")) is False
|
|
|
|
|
|
# ── 3. Browse any collection; actions gated to favourites owner ──────────────
|
|
|
|
def test_collection_view_actions_enabled_for_any_collection():
|
|
view = run(favorites.get_collection_items_view(base_client(), "c-matt", "u-matt"))
|
|
assert view["actions_enabled"] is True
|
|
assert [i["watched"] for i in view["items"]] == [True, False, False]
|
|
|
|
|
|
def test_collection_view_watched_follows_selected_user():
|
|
view = run(favorites.get_collection_items_view(base_client(), "c-matt", "u-dave"))
|
|
assert view["actions_enabled"] is True # available regardless of owner
|
|
assert [i["watched"] for i in view["items"]] == [False, True, False]
|
|
|
|
|
|
def test_collection_view_actions_enabled_for_non_favorites_collection():
|
|
client = base_client()
|
|
client.collection_items["c-misc"] = [movie("x1", "Iron Man")]
|
|
view = run(favorites.get_collection_items_view(client, "c-misc", "u-matt"))
|
|
assert view["is_favorites"] is False
|
|
assert view["actions_enabled"] is True
|
|
|
|
|
|
def test_collection_view_missing_collection_raises_404():
|
|
with pytest.raises(favorites.FavoritesError) as exc:
|
|
run(favorites.get_collection_items_view(base_client(), "c-nope", "u-matt"))
|
|
assert exc.value.status == 404
|
|
|
|
|
|
# ── 4. Cleanup: dry-run vs apply ─────────────────────────────────────────────
|
|
|
|
def test_cleanup_dry_run_does_not_modify_emby():
|
|
client = base_client()
|
|
res = run(favorites.cleanup_watched(client, "c-matt", "u-matt", dry_run=True))
|
|
assert res["dry_run"] is True and res["applied"] is False
|
|
assert res["watched_found"] == 1
|
|
assert [r["item_id"] for r in res["removed"]] == ["m1"]
|
|
assert client.writes == []
|
|
assert len(client.collection_items["c-matt"]) == 3
|
|
|
|
|
|
def test_apply_cleanup_removes_only_watched_items():
|
|
client = base_client()
|
|
res = run(favorites.cleanup_watched(client, "c-matt", "u-matt", dry_run=False))
|
|
assert res["applied"] is True
|
|
assert res["summary"]["removed_count"] == 1
|
|
assert res["summary"]["final_count"] == 2
|
|
deletes = [c for c in client.writes if c[0] == "delete"]
|
|
assert len(deletes) == 1
|
|
assert deletes[0][1] == "/Collections/c-matt/Items"
|
|
assert deletes[0][2]["Ids"] == "m1"
|
|
assert {i["Id"] for i in client.collection_items["c-matt"]} == {"m2", "m3"}
|
|
assert res["log"][0]["item_id"] == "m1"
|
|
assert res["log"][0]["reason"] == "watched-by-user"
|
|
assert res["log"][0]["user"] == "Matt"
|
|
assert res["log"][0]["collection"] == "Matt Favorites"
|
|
|
|
|
|
def test_cleanup_does_not_cross_users():
|
|
"""Dave watched m2; cleaning Matt's collection must not remove m2."""
|
|
client = base_client()
|
|
res = run(favorites.cleanup_watched(client, "c-matt", "u-matt", dry_run=False))
|
|
assert "m2" not in {r["item_id"] for r in res["removed"]}
|
|
assert "m2" in {i["Id"] for i in client.collection_items["c-matt"]}
|
|
|
|
|
|
def test_cleanup_works_on_non_favorites_collection():
|
|
"""Cleanup is available for any collection, using the selected user's
|
|
watched status."""
|
|
client = base_client()
|
|
client.collection_items["c-misc"] = [movie("x1", "Iron Man"), movie("x2", "Iron Man 2")]
|
|
client.watched_by_user["u-matt"] = {"x1"}
|
|
res = run(favorites.cleanup_watched(client, "c-misc", "u-matt", dry_run=False))
|
|
assert res["applied"] is True
|
|
assert [r["item_id"] for r in res["removed"]] == ["x1"]
|
|
assert {i["Id"] for i in client.collection_items["c-misc"]} == {"x2"}
|
|
|
|
|
|
# ── 5. Recommendation scoring (pure) ─────────────────────────────────────────
|
|
|
|
def test_build_profile_and_score_candidate():
|
|
watched = [emby_collections.normalize_item(movie(
|
|
"w1", "Seed", genres=["Action", "Sci-Fi"], year=2014,
|
|
studio="Acme", director="Jane Doe", actor="Bob Roe"))]
|
|
profile = rec.build_profile(watched)
|
|
assert profile.genres == {"Action", "Sci-Fi"}
|
|
assert profile.decades == {2010}
|
|
|
|
candidate = emby_collections.normalize_item(movie(
|
|
"c1", "Match", genres=["Action", "Sci-Fi", "Drama"], year=2013,
|
|
studio="Acme", director="Jane Doe", actor="Bob Roe"))
|
|
# 2 genres*5 + director3 + actor2 + studio2 + decade1 + mediatype1 = 19
|
|
assert rec.score_candidate(candidate, profile) == 19
|
|
|
|
|
|
def test_score_series_franchise_bonus():
|
|
profile = rec.TasteProfile(series={"The Saga"}, media_types={"Video"})
|
|
cand = {"series_name": "The Saga", "media_type": "Video", "genres": [], "year": None}
|
|
assert rec.score_candidate(cand, profile) == rec.SCORE_SERIES + rec.SCORE_MEDIA_TYPE
|
|
|
|
|
|
def test_rank_orders_by_score_then_rating_then_title():
|
|
profile = rec.TasteProfile(genres={"Action"}, media_types={"Video"},
|
|
studios={"Acme"}, directors={"Jane Doe"})
|
|
low = emby_collections.normalize_item(movie("low", "Zeta", genres=["Action"],
|
|
studio=None, director=None, actor=None, rating=9.0))
|
|
high = emby_collections.normalize_item(movie("high", "Alpha", genres=["Action"],
|
|
studio="Acme", director="Jane Doe", actor=None, rating=1.0))
|
|
ranked = rec.rank_candidates([low, high], profile)
|
|
assert [r["id"] for r in ranked] == ["high", "low"]
|
|
|
|
|
|
def test_rank_drops_zero_score_candidates():
|
|
profile = rec.TasteProfile(genres={"Action"})
|
|
unrelated = {"id": "x", "genres": ["Cooking"], "media_type": "Audio", "year": 1980, "title": "X"}
|
|
assert rec.rank_candidates([unrelated], profile) == []
|
|
|
|
|
|
# ── 6. Regenerate: exclusions ────────────────────────────────────────────────
|
|
|
|
def regen_client():
|
|
client = base_client()
|
|
client.watched_history["u-matt"] = [
|
|
movie("w1", "Seed One", genres=["Action", "Sci-Fi"], year=2014),
|
|
movie("w2", "Seed Two", genres=["Action"], year=2016),
|
|
]
|
|
client.candidates["u-matt"] = [
|
|
movie("c1", "Good One", genres=["Action", "Sci-Fi"], year=2015),
|
|
movie("c2", "Good Two", genres=["Action"], year=2012),
|
|
movie("m2", "Already In Collection", genres=["Action"], year=2015), # in c-matt
|
|
movie("w1", "Already Watched", genres=["Action"], year=2014), # watched
|
|
]
|
|
return client
|
|
|
|
|
|
def test_regenerate_excludes_watched_and_existing_collection_items():
|
|
res = run(favorites.regenerate(regen_client(), "c-matt", "u-matt", dry_run=True, target_size=25))
|
|
ids = {r["item_id"] for r in res["recommended"]}
|
|
assert "m2" not in ids # already in collection
|
|
assert "w1" not in ids # already watched
|
|
assert {"c1", "c2"} <= ids
|
|
|
|
|
|
def test_regenerate_dry_run_makes_no_writes():
|
|
client = regen_client()
|
|
res = run(favorites.regenerate(client, "c-matt", "u-matt", dry_run=True, target_size=25))
|
|
assert res["applied"] is False
|
|
assert client.writes == []
|
|
|
|
|
|
def test_regenerate_apply_adds_up_to_target():
|
|
client = regen_client()
|
|
res = run(favorites.regenerate(client, "c-matt", "u-matt", dry_run=False, target_size=4))
|
|
# collection had 3, target 4 -> add exactly 1
|
|
assert res["summary"]["added_count"] == 1
|
|
assert res["summary"]["final_count"] == 4
|
|
posts = [c for c in client.writes if c[0] == "post"]
|
|
assert len(posts) == 1
|
|
assert posts[0][1] == "/Collections/c-matt/Items"
|
|
assert res["recommended"][0]["item_id"] == "c1" # highest score first
|
|
|
|
|
|
def test_regenerate_never_readds_watched():
|
|
res = run(favorites.regenerate(regen_client(), "c-matt", "u-matt", dry_run=False, target_size=25))
|
|
added_ids = {r["item_id"] for r in res["recommended"]}
|
|
assert "w1" not in added_ids and "w2" not in added_ids
|
|
|
|
|
|
# ── 7. Empty history ─────────────────────────────────────────────────────────
|
|
|
|
def test_regenerate_with_empty_history_recommends_nothing():
|
|
client = base_client()
|
|
client.watched_history["u-matt"] = []
|
|
res = run(favorites.regenerate(client, "c-matt", "u-matt", dry_run=True, target_size=25))
|
|
assert res["recommended"] == []
|
|
assert "No watch history" in res["message"]
|
|
assert client.writes == []
|
|
|
|
|
|
def test_empty_profile_build_candidates_returns_empty():
|
|
profile = rec.build_profile([])
|
|
assert profile.is_empty
|
|
assert run(rec.build_candidates(base_client(), "u-matt", profile, set())) == []
|
|
|
|
|
|
# ── 8. Missing users / collections / API failures ────────────────────────────
|
|
|
|
def test_missing_user_raises_404():
|
|
with pytest.raises(favorites.FavoritesError) as exc:
|
|
run(favorites.cleanup_watched(base_client(), "c-matt", "u-ghost"))
|
|
assert exc.value.status == 404
|
|
|
|
|
|
def test_missing_collection_raises_404():
|
|
with pytest.raises(favorites.FavoritesError) as exc:
|
|
run(favorites.cleanup_watched(base_client(), "c-nope", "u-matt"))
|
|
assert exc.value.status == 404
|
|
|
|
|
|
def test_emby_api_failure_propagates():
|
|
class BoomClient(FakeEmbyClient):
|
|
async def get_all(self, path, params=None):
|
|
raise RuntimeError("Emby exploded")
|
|
|
|
client = BoomClient(users=[{"Id": "u-matt", "Name": "Matt"}])
|
|
with pytest.raises(RuntimeError):
|
|
run(favorites.list_collections_overview(client))
|
|
|
|
|
|
def test_regenerate_rejects_negative_target():
|
|
with pytest.raises(favorites.FavoritesError):
|
|
run(favorites.regenerate(base_client(), "c-matt", "u-matt", dry_run=True, target_size=-1))
|