siftlode/backend/tests/test_collation.py

"""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 Channel, PlexCollection, PlexItem, PlexLibrary, Subscription, User, 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.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