peter
533571f572
PlexLibrary, PlexShow, PlexSeason, PlexItem, PlexPerson, PlexCollection, PlexState, PlexLink, PlexPlaylist and PlexPlaylistItem leave app/models.py for app/plex/models.py; app.tables registers them. Every import is re-pointed; no schema change.
279 lines
9.9 KiB
Python
279 lines
9.9 KiB
Python
"""Every A→Z list the server orders is in the reader's alphabet. The database image is Alpine,
|
|
whose `en_US.utf8` is BYTE order: "Életrajzi" and "Ötödik" sorted after "Zorró" in the feed, the channel
|
|
list, the Plex grid, the genre list. Hungarian puts "Ö" after "O" and "Cs" after "Cu". Needs the DB."""
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
import app.state as state
|
|
from app.auth import current_user
|
|
from app.collation import sorted_words
|
|
from app.db import get_db
|
|
from app.main import app
|
|
from app.models import User
|
|
from app.plex.models import PlexCollection, PlexItem, PlexLibrary
|
|
from app.catalog.models import Channel, Subscription, Video
|
|
from app.reader.models import Playlist, PlaylistItem
|
|
from datetime import datetime, timezone
|
|
|
|
WORDS = ["Zorró", "Életrajzi", "Ötödik", "oroszlán", "Csillag", "Cukor"]
|
|
HU = ["Cukor", "Csillag", "Életrajzi", "oroszlán", "Ötödik", "Zorró"]
|
|
EN = ["Csillag", "Cukor", "Életrajzi", "oroszlán", "Ötödik", "Zorró"]
|
|
|
|
|
|
def _user(db, lang: str) -> User:
|
|
user = User(email=f"collate-{lang}@example.com", preferences={"language": lang})
|
|
db.add(user)
|
|
db.flush()
|
|
return user
|
|
|
|
|
|
def _client(db, user, monkeypatch, override_dep) -> TestClient:
|
|
monkeypatch.setattr(state, "_configured_cache", True)
|
|
override_dep(get_db, db.info["request"])
|
|
override_dep(current_user, lambda: user)
|
|
return TestClient(app)
|
|
|
|
|
|
def test_words_sort_in_the_readers_alphabet(db):
|
|
assert sorted_words(db, WORDS, "hu") == HU
|
|
assert sorted_words(db, WORDS, "en") == EN
|
|
assert sorted_words(db, WORDS, "xx") == EN # an unknown language sorts as English
|
|
assert sorted_words(db, [], "hu") == []
|
|
|
|
|
|
def _seed_feed(db, user):
|
|
db.add(Channel(id="UCc", title="Channel"))
|
|
db.flush()
|
|
db.add(Subscription(user_id=user.id, channel_id="UCc", priority=0, hidden=False))
|
|
db.add_all(
|
|
Video(
|
|
id=f"v{i}",
|
|
channel_id="UCc",
|
|
published_at=datetime(2024, 1, i + 1, tzinfo=timezone.utc),
|
|
title=w,
|
|
is_short=False,
|
|
live_status="none",
|
|
via_search=False,
|
|
via_explore=False,
|
|
)
|
|
for i, w in enumerate(WORDS)
|
|
)
|
|
db.commit()
|
|
|
|
|
|
def test_the_feed_title_sort_pages_in_the_readers_alphabet(db, monkeypatch, override_dep):
|
|
user = _user(db, "hu")
|
|
_seed_feed(db, user)
|
|
client = _client(db, user, monkeypatch, override_dep)
|
|
|
|
full = [it["title"] for it in client.get("/api/feed?sort=title&limit=50").json()["items"]]
|
|
assert full == HU
|
|
# The cursor compares with the same collated expression, so paging walks the same order.
|
|
walked: list[str] = []
|
|
cursor = None
|
|
for _ in range(len(WORDS)):
|
|
url = "/api/feed?sort=title&limit=2" + (f"&cursor={cursor}" if cursor else "")
|
|
body = client.get(url).json()
|
|
walked += [it["title"] for it in body["items"]]
|
|
cursor = body["next_cursor"]
|
|
if not body["has_more"]:
|
|
break
|
|
assert walked == HU
|
|
desc = client.get("/api/feed?sort=title_desc&limit=50").json()["items"]
|
|
assert [it["title"] for it in desc] == HU[::-1]
|
|
|
|
|
|
def test_the_channel_list_is_in_the_readers_alphabet(db, monkeypatch, override_dep):
|
|
user = _user(db, "hu")
|
|
db.add_all(Channel(id=f"UC{i}", title=w) for i, w in enumerate(WORDS))
|
|
db.flush()
|
|
db.add_all(
|
|
Subscription(user_id=user.id, channel_id=f"UC{i}", priority=0, hidden=False)
|
|
for i in range(len(WORDS))
|
|
)
|
|
db.commit()
|
|
r = _client(db, user, monkeypatch, override_dep).get("/api/channels")
|
|
assert [c["title"] for c in r.json()] == HU
|
|
|
|
|
|
def _seed_plex(db):
|
|
lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True)
|
|
db.add(lib)
|
|
db.flush()
|
|
col = PlexCollection(rating_key="col1", library_id=lib.id, title="Set", child_count=len(WORDS))
|
|
db.add(col)
|
|
db.add_all(
|
|
PlexItem(
|
|
rating_key=f"m{i}",
|
|
library_id=lib.id,
|
|
kind="movie",
|
|
title=f"EN {i}",
|
|
title_hu=w,
|
|
year=2000,
|
|
genres_hu=[w],
|
|
collection_keys=["col1"],
|
|
)
|
|
for i, w in enumerate(WORDS)
|
|
)
|
|
db.commit()
|
|
|
|
|
|
def _plex(db, monkeypatch, override_dep) -> TestClient:
|
|
user = _user(db, "hu")
|
|
_seed_plex(db)
|
|
return _client(db, user, monkeypatch, override_dep)
|
|
|
|
|
|
def test_the_plex_grid_sorts_by_the_hungarian_title(db, monkeypatch, override_dep):
|
|
client = _plex(db, monkeypatch, override_dep)
|
|
grid = client.get("/api/plex/library?scope=movie&sort=title&sort_dir=asc&lang=hu").json()
|
|
assert [it["title"] for it in grid["items"]] == HU
|
|
|
|
|
|
def test_the_plex_genre_list_is_in_the_readers_alphabet(db, monkeypatch, override_dep):
|
|
client = _plex(db, monkeypatch, override_dep)
|
|
facets = client.get("/api/plex/facets?scope=movie&lang=hu").json()
|
|
assert [g["value"] for g in facets["genres"]] == HU
|
|
|
|
|
|
def test_a_collection_sorts_by_the_hungarian_title(db, monkeypatch, override_dep):
|
|
client = _plex(db, monkeypatch, override_dep)
|
|
members = client.get("/api/plex/collections/col1?sort=title&sort_dir=asc&lang=hu").json()
|
|
assert [it["title"] for it in members["items"]] == HU
|
|
# year ties everywhere, so the default (year) order falls through to the title tiebreak
|
|
default = client.get("/api/plex/collections/col1?lang=hu").json()
|
|
assert [it["title"] for it in default["items"]] == HU
|
|
|
|
|
|
def _seed_discovery(db, user, synced: bool):
|
|
"""Channels the user's playlist holds one video from each, not subscribed: discovery lists them,
|
|
count-tied, so the title decides. `synced=False` leaves them stubs, which sends the endpoint down
|
|
its enrich path — the one that re-sorts in Python and has to agree with the ORDER BY."""
|
|
pl = Playlist(user_id=user.id, name="Mix")
|
|
db.add(pl)
|
|
db.flush()
|
|
stamp = datetime(2024, 1, 1, tzinfo=timezone.utc) if synced else None
|
|
for i, w in enumerate(WORDS):
|
|
db.add(Channel(id=f"UCd{i}", title=w, details_synced_at=stamp))
|
|
db.flush()
|
|
db.add(
|
|
Video(
|
|
id=f"d{i}",
|
|
channel_id=f"UCd{i}",
|
|
published_at=datetime(2024, 1, 1, tzinfo=timezone.utc),
|
|
title="x",
|
|
is_short=False,
|
|
live_status="none",
|
|
via_search=False,
|
|
via_explore=False,
|
|
)
|
|
)
|
|
db.flush()
|
|
db.add(PlaylistItem(playlist_id=pl.id, video_id=f"d{i}", position=i))
|
|
db.commit()
|
|
|
|
|
|
def test_channel_discovery_is_in_the_readers_alphabet(db, monkeypatch, override_dep):
|
|
user = _user(db, "hu")
|
|
_seed_discovery(db, user, synced=True)
|
|
r = _client(db, user, monkeypatch, override_dep).get("/api/channels/discovery")
|
|
assert [c["title"] for c in r.json()] == HU
|
|
|
|
|
|
def test_channel_discovery_keeps_that_order_after_enriching(db, monkeypatch, override_dep):
|
|
import app.routes.channels as channels
|
|
|
|
class FakeYouTube:
|
|
def __init__(self, *a, **k): ...
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *a):
|
|
return False
|
|
|
|
def get_channels(self, ids):
|
|
return []
|
|
|
|
monkeypatch.setattr(channels, "YouTubeClient", FakeYouTube)
|
|
monkeypatch.setattr(channels, "apply_channel_details", lambda *a, **k: None)
|
|
user = _user(db, "hu")
|
|
_seed_discovery(db, user, synced=False)
|
|
r = _client(db, user, monkeypatch, override_dep).get("/api/channels/discovery")
|
|
assert [c["title"] for c in r.json()] == HU
|
|
|
|
|
|
# --- The other lists the server puts in the reader's alphabet: each call site, once. ---
|
|
|
|
|
|
def _others(db, n_offset: int = 0) -> list[User]:
|
|
users = [
|
|
User(email=f"{w.lower()}@example.com", display_name=w) for w in WORDS
|
|
] # emails and names in WORDS' order
|
|
db.add_all(users)
|
|
db.flush()
|
|
return users
|
|
|
|
|
|
def test_the_share_recipients_are_in_the_readers_alphabet(db, monkeypatch, override_dep):
|
|
user = _user(db, "hu")
|
|
_others(db)
|
|
db.commit()
|
|
client = _client(db, user, monkeypatch, override_dep)
|
|
emails = [r["email"] for r in client.get("/api/downloads/recipients").json()]
|
|
assert emails == [f"{w.lower()}@example.com" for w in HU]
|
|
|
|
|
|
def test_the_blocked_channels_are_in_the_readers_alphabet(db, monkeypatch, override_dep):
|
|
from app.catalog.models import BlockedChannel
|
|
|
|
user = _user(db, "hu")
|
|
db.add_all(Channel(id=f"UC{i}", title=w) for i, w in enumerate(WORDS))
|
|
db.flush()
|
|
db.add_all(BlockedChannel(user_id=user.id, channel_id=f"UC{i}") for i in range(len(WORDS)))
|
|
db.commit()
|
|
r = _client(db, user, monkeypatch, override_dep).get("/api/channels/blocked/list")
|
|
assert [c["title"] for c in r.json()] == HU
|
|
|
|
|
|
def test_the_plex_libraries_and_collections_are_in_the_readers_alphabet(
|
|
db, monkeypatch, override_dep
|
|
):
|
|
user = _user(db, "hu")
|
|
libs = [
|
|
PlexLibrary(plex_key=str(i), title=w, kind="movie", enabled=True)
|
|
for i, w in enumerate(WORDS)
|
|
]
|
|
db.add_all(libs)
|
|
db.flush()
|
|
db.add_all(
|
|
PlexCollection(rating_key=f"c{i}", library_id=libs[0].id, title=w, child_count=1)
|
|
for i, w in enumerate(WORDS)
|
|
)
|
|
db.commit()
|
|
client = _client(db, user, monkeypatch, override_dep)
|
|
assert [lib["title"] for lib in client.get("/api/plex/libraries").json()["libraries"]] == HU
|
|
cols = client.get("/api/plex/collections").json()["collections"]
|
|
assert [c["title"] for c in cols] == HU
|
|
|
|
|
|
def test_a_persons_titles_of_one_year_are_in_the_readers_alphabet(db, monkeypatch, override_dep):
|
|
user = _user(db, "hu")
|
|
lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True)
|
|
db.add(lib)
|
|
db.flush()
|
|
db.add_all(
|
|
PlexItem(
|
|
rating_key=f"m{i}",
|
|
library_id=lib.id,
|
|
kind="movie",
|
|
title=f"EN {i}",
|
|
title_hu=w,
|
|
year=2000,
|
|
cast_names=["Anna"],
|
|
)
|
|
for i, w in enumerate(WORDS)
|
|
)
|
|
db.commit()
|
|
body = _client(db, user, monkeypatch, override_dep).get("/api/plex/person/Anna?lang=hu").json()
|
|
assert [m["title"] for m in body["movies"]] == HU
|