siftlode/backend/tests/test_discover.py

"""E24 S1 — the Home surface's read-model.

Covers the four things that can silently rot here: the time-budget bands (pure), the cross-source
resume rail's merge order, the hero's tier ladder, and the comfort shelf (its schema-level one-of
invariant, its cascade, and the resolved `next` that makes one click enough).

DB lane (`db` fixture) except for the budget block — skipped without Postgres.
"""

from datetime import datetime, timedelta, timezone

import pytest
from sqlalchemy import delete, func, select, text
from sqlalchemy.exc import IntegrityError

from app.discover import hero, pins, queries, rails
from app.discover.budget import BANDS, fits, remaining_seconds
from app.models import (
    Channel,
    ComfortPin,
    PlexItem,
    PlexLibrary,
    PlexShow,
    PlexState,
    Playlist,
    PlaylistItem,
    Subscription,
    User,
    Video,
    VideoState,
)

NOW = datetime(2026, 6, 1, tzinfo=timezone.utc)


# ── budget (pure) ─────────────────────────────────────────────────────────────────────────────────


def test_bands_cover_every_runtime_and_overlap_deliberately():
    """Every runtime has a home in `any`, and the bands overlap so "30 perc" is not answered with a
    two-minute Short just because it fits under the cap."""
    assert fits(3, "any") and fits(10_000, "any")
    assert fits(2 * 60, "short") and not fits(2 * 60, "medium")  # a Short is not a 30-minute answer
    assert fits(30 * 60, "medium") and not fits(30 * 60, "short")
    assert fits(110 * 60, "long") and not fits(110 * 60, "medium")


def test_unknown_duration_only_fits_the_unbounded_band():
    """We cannot promise an unknown-length item finishes inside ten minutes, so it is not offered as
    an answer to a bounded question — the one control on the page has to mean what it says."""
    for budget in BANDS:
        assert fits(None, budget) is (budget == "any")


def test_budget_is_matched_against_what_is_LEFT():
    """A 90-minute film with 8 minutes to go is a ten-minute answer."""
    assert remaining_seconds(90 * 60, 82 * 60) == 8 * 60
    assert fits(remaining_seconds(90 * 60, 82 * 60), "short")
    assert not fits(remaining_seconds(90 * 60, 0), "short")
    assert remaining_seconds(None, 30) is None
    assert remaining_seconds(600, 900) == 0  # a position past the end never goes negative


# ── fixtures ──────────────────────────────────────────────────────────────────────────────────────


def _seed_yt(db) -> tuple[User, Channel]:
    """A subscribed channel with three videos of distinct lengths, so a budget can pick between them."""
    user = User(email="discover-e24@example.com")
    ch = Channel(id="UCe24", title="Comfort Channel", handle="@comfort")
    db.add_all([user, ch])
    db.flush()
    db.add(Subscription(user_id=user.id, channel_id=ch.id))
    db.add_all(
        [
            Video(
                id="vshort",
                channel_id=ch.id,
                title="Ten minute video",
                duration_seconds=9 * 60,
                published_at=NOW - timedelta(days=1),
            ),
            Video(
                id="vmed",
                channel_id=ch.id,
                title="Half hour video",
                duration_seconds=30 * 60,
                published_at=NOW - timedelta(days=2),
            ),
            Video(
                id="vlong",
                channel_id=ch.id,
                title="Feature length video",
                duration_seconds=95 * 60,
                published_at=NOW - timedelta(days=3),
            ),
        ]
    )
    db.commit()
    return user, ch


def _seed_plex(db, user: User) -> tuple[PlexItem, PlexShow, PlexItem]:
    lib_m = PlexLibrary(plex_key="90", title="Movies", kind="movie", enabled=True)
    lib_s = PlexLibrary(plex_key="91", title="Shows", kind="show", enabled=True)
    db.add_all([lib_m, lib_s])
    db.flush()
    movie = PlexItem(
        rating_key="pm1",
        library_id=lib_m.id,
        kind="movie",
        title="A Movie",
        duration_s=100 * 60,
        playable="direct",
    )
    show = PlexShow(rating_key="ps1", library_id=lib_s.id, title="A Show", child_count=1)
    db.add_all([movie, show])
    db.flush()
    ep1 = PlexItem(
        rating_key="ps1e1",
        library_id=lib_s.id,
        kind="episode",
        title="Ep 1",
        show_id=show.id,
        season_number=1,
        episode_number=1,
        duration_s=45 * 60,
    )
    ep2 = PlexItem(
        rating_key="ps1e2",
        library_id=lib_s.id,
        kind="episode",
        title="Ep 2",
        show_id=show.id,
        season_number=1,
        episode_number=2,
        duration_s=45 * 60,
    )
    db.add_all([ep1, ep2])
    db.commit()
    return movie, show, ep1


def _enable_plex(db):
    from app import sysconfig

    sysconfig.set_value(db, "plex_enabled", True)
    db.commit()


# ── continue watching ─────────────────────────────────────────────────────────────────────────────


def test_continue_watching_merges_both_sources_by_recency(db):
    """One rail, not two lists side by side — and ordered by when the user last touched each thing,
    across sources. Two side-by-side rails would just move the decision up a level."""
    user, _ = _seed_yt(db)
    movie, _, _ = _seed_plex(db, user)
    _enable_plex(db)
    db.add_all(
        [
            VideoState(
                user_id=user.id,
                video_id="vmed",
                status="new",
                position_seconds=300,
                progress_updated_at=NOW - timedelta(hours=5),
            ),
            PlexState(
                user_id=user.id,
                item_id=movie.id,
                status="new",
                position_seconds=600,
                progress_updated_at=NOW - timedelta(hours=1),
            ),
        ]
    )
    db.commit()

    got = rails.continue_watching(db, user)
    assert [(c["source"], c["id"]) for c in got] == [("plex", "pm1"), ("youtube", "vmed")]
    assert got[0]["remaining_seconds"] == 100 * 60 - 600


def test_continue_watching_is_youtube_only_when_plex_is_off(db):
    """Plex is a deployment toggle; with it off the surface says nothing about Plex at all."""
    user, _ = _seed_yt(db)
    movie, _, _ = _seed_plex(db, user)
    db.add_all(
        [
            VideoState(
                user_id=user.id,
                video_id="vmed",
                status="new",
                position_seconds=300,
                progress_updated_at=NOW,
            ),
            PlexState(
                user_id=user.id,
                item_id=movie.id,
                status="new",
                position_seconds=600,
                progress_updated_at=NOW,
            ),
        ]
    )
    db.commit()
    assert {c["source"] for c in rails.continue_watching(db, user)} == {"youtube"}


def test_a_null_timestamp_sorts_last_instead_of_raising(db):
    """The `(ts is not None, ts)` key this rail's Plex sibling used compares `None < None` when two
    rows both lack a timestamp, which raises. A missing progress timestamp must sort last, not 500
    the landing page."""
    user, _ = _seed_yt(db)
    db.add_all(
        [
            VideoState(
                user_id=user.id, video_id="vmed", status="new", position_seconds=1
            ),  # no progress_updated_at
            VideoState(
                user_id=user.id, video_id="vlong", status="new", position_seconds=1
            ),  # nor here
            VideoState(
                user_id=user.id,
                video_id="vshort",
                status="new",
                position_seconds=1,
                progress_updated_at=NOW,
            ),
        ]
    )
    db.commit()
    got = rails.continue_watching(db, user)
    assert got[0]["id"] == "vshort"
    assert len(got) == 3


# ── hero ──────────────────────────────────────────────────────────────────────────────────────────


def test_hero_prefers_resume_and_labels_the_tier(db):
    user, _ = _seed_yt(db)
    db.add(
        VideoState(
            user_id=user.id,
            video_id="vmed",
            status="new",
            position_seconds=300,
            progress_updated_at=NOW,
        )
    )
    db.commit()
    got = hero.pick(db, user, "any", seed=1)
    assert got["tier"] == "resume"
    assert got["card"]["id"] == "vmed"


def test_hero_skips_a_resume_that_does_not_fit_the_budget(db):
    """ "I have ten minutes" must not be answered with the 80 minutes left on a film, even though
    that film is the most recent thing the user touched."""
    user, _ = _seed_yt(db)
    db.add(
        VideoState(
            user_id=user.id,
            video_id="vlong",
            status="new",
            position_seconds=60,
            progress_updated_at=NOW,
        )
    )
    db.commit()
    got = hero.pick(db, user, "short", seed=1)
    assert got["tier"] == "surprise"
    assert got["card"]["duration_seconds"] <= BANDS["short"][1]


def test_hero_falls_back_to_the_shelf_before_guessing(db):
    """The user's own declared fallback beats anything we could infer."""
    user, ch = _seed_yt(db)
    db.add(ComfortPin(user_id=user.id, video_id="vmed", position=0))
    db.commit()
    got = hero.pick(db, user, "medium", seed=1)
    assert got["tier"] == "comfort"
    assert got["card"]["id"] == "vmed"


def test_hero_resolves_a_show_to_the_episode_that_would_play(db):
    """A show has no Play button. The hero of a partially-watched show is its next episode."""
    user, _ = _seed_yt(db)
    _, show, ep1 = _seed_plex(db, user)
    _enable_plex(db)
    db.add(
        PlexState(
            user_id=user.id,
            item_id=ep1.id,
            status="watched",
            watched_at=NOW,
            progress_updated_at=NOW,
        )
    )
    db.commit()
    got = hero.pick(db, user, "any", seed=1)
    assert got["tier"] == "resume"
    assert got["card"]["kind"] == "episode"
    assert got["card"]["id"] == "ps1e2"  # ep1 is finished → ep2 is what plays


def test_hero_is_none_when_nothing_fits(db):
    """An empty answer is an honest answer: the page renders its own empty state rather than being
    handed a card that does not answer what was asked. Here the whole library is under 40 minutes,
    so "egy este" has nothing to offer."""
    user, ch = _seed_yt(db)
    db.execute(text("DELETE FROM videos WHERE id = 'vlong'"))
    db.commit()
    assert hero.pick(db, user, "long", seed=1) is None


def test_surprise_is_reproducible_for_a_seed(db):
    """The reroll button depends on this: one seed drives both sources, so the same seed must give
    the same answer and a different seed must be free to give another."""
    user, _ = _seed_yt(db)
    a = rails.surprise(db, user, "any", seed=7, limit=3)
    b = rails.surprise(db, user, "any", seed=7, limit=3)
    assert [c["id"] for c in a] == [c["id"] for c in b]
    seeds = {tuple(c["id"] for c in rails.surprise(db, user, "any", s, 3)) for s in range(12)}
    assert len(seeds) > 1  # the seed actually reaches the ordering


def test_surprise_never_offers_something_already_started(db):
    """The resume tier may have skipped an item because its REMAINING time did not fit; handing the
    same item back as "Meglepetés" — where the full runtime does fit — would contradict the question
    the user just answered."""
    user, _ = _seed_yt(db)
    db.add(
        VideoState(
            user_id=user.id,
            video_id="vlong",
            status="new",
            position_seconds=90 * 60,
            progress_updated_at=NOW,
        )
    )
    db.commit()
    assert "vlong" not in {c["id"] for c in rails.surprise(db, user, "long", seed=3, limit=5)}


def test_the_seed_decides_which_source_leads(db):
    """With a pool of one, a fixed source order would mean the Plex library could never answer "egy
    este" whenever a YouTube video also fit — the cross-source promise would quietly be a YouTube
    promise."""
    user, _ = _seed_yt(db)
    _seed_plex(db, user)
    _enable_plex(db)
    leaders = {rails.surprise(db, user, "any", s, limit=1)[0]["source"] for s in range(6)}
    assert leaders == {"youtube", "plex"}


def test_the_reroll_must_be_able_to_move_past_the_resume_tier(db):
    """A new seed alone only re-draws the SURPRISE tier — the resume tier deterministically returns
    the most recently progressed thing that fits, so without an exclusion the reroll handed the same
    card straight back and the button looked dead (found in the browser, 2026-09-06)."""
    user, _ = _seed_yt(db)
    db.add_all(
        [
            VideoState(
                user_id=user.id,
                video_id="vmed",
                status="new",
                position_seconds=300,
                progress_updated_at=NOW,
            ),
            VideoState(
                user_id=user.id,
                video_id="vshort",
                status="new",
                position_seconds=60,
                progress_updated_at=NOW - timedelta(hours=1),
            ),
        ]
    )
    db.commit()

    first = hero.pick(db, user, "any", seed=1)
    assert first["card"]["id"] == "vmed"
    # Same seed, same everything — only the exclusion differs.
    second = hero.pick(db, user, "any", seed=1, exclude={"youtube:vmed"})
    assert second["card"]["id"] == "vshort"


def test_the_reroll_rotates_which_tier_answers(db):
    """Measured 2026-09-06: exclusion alone left the button walking down the resume rail (20 rows)
    and cycling nine cards, every one of them `resume` — the shelf and the surprise tier were
    unreachable by pressing it. `start` is the press count, so consecutive presses offer a different
    KIND of answer."""
    user, _ = _seed_yt(db)
    # One of each: something half-finished, something pinned, and a third video for the draw.
    db.add(
        VideoState(
            user_id=user.id,
            video_id="vmed",
            status="new",
            position_seconds=300,
            progress_updated_at=NOW,
        )
    )
    db.add(ComfortPin(user_id=user.id, video_id="vlong", position=0))
    db.commit()
    tiers = [hero.pick(db, user, "any", seed=1, start=n)["tier"] for n in range(4)]
    assert tiers == ["resume", "comfort", "surprise", "resume"]


def test_the_rotation_still_falls_through_an_empty_tier(db):
    """A press must not come back empty just because the tier it happened to lead with has nothing.
    Here there is no shelf at all, so the comfort-led press has to answer from somewhere else."""
    user, _ = _seed_yt(db)
    assert db.execute(select(func.count()).select_from(ComfortPin)).scalar_one() == 0
    got = hero.pick(db, user, "any", seed=1, start=1)
    assert got is not None and got["tier"] != "comfort"


def test_excluding_a_show_also_excludes_the_episode_it_resolves_to(db):
    """The user rerolled past the SHOW they were shown; handing back its next episode under a
    different title would be the same suggestion wearing a hat."""
    user, _ = _seed_yt(db)
    _, show, ep1 = _seed_plex(db, user)
    _enable_plex(db)
    db.add(
        PlexState(
            user_id=user.id,
            item_id=ep1.id,
            status="new",
            position_seconds=120,
            progress_updated_at=NOW,
        )
    )
    db.commit()

    shown = hero.pick(db, user, "any", seed=1)
    assert shown["card"]["id"] == "ps1e1"  # the show resolved to its in-progress episode
    # Excluding what the USER saw — the show — must move the tier on, not return the episode again.
    after = hero.pick(db, user, "any", seed=1, exclude={"plex:ps1"})
    assert after is None or after["card"]["id"] != "ps1e1"


# ── comfort shelf ─────────────────────────────────────────────────────────────────────────────────


def test_a_pin_must_reference_exactly_one_thing(db):
    """The union invariant lives in the schema, not in a comment at every insert site."""
    user, ch = _seed_yt(db)
    for kwargs in ({}, {"video_id": "vmed", "channel_id": ch.id}):
        db.add(ComfortPin(user_id=user.id, position=0, **kwargs))
        with pytest.raises(IntegrityError):
            db.flush()
        db.rollback()


def test_deleting_the_pinned_row_removes_the_pin(db):
    """A film dropped from the library must not leave a dead tile on the one surface whose whole job
    is to be reliable — which is why the shelf holds foreign keys, not strings."""
    user, _ = _seed_yt(db)
    db.add(ComfortPin(user_id=user.id, video_id="vmed", position=0))
    db.commit()
    db.execute(text("DELETE FROM videos WHERE id = 'vmed'"))
    db.commit()
    assert db.execute(select(func.count()).select_from(ComfortPin)).scalar() == 0


def test_shelf_resolves_a_channel_to_its_newest_unwatched_video(db):
    """ "Put that channel on" has to be one click, so the tile carries what would play."""
    user, ch = _seed_yt(db)
    db.add(ComfortPin(user_id=user.id, channel_id=ch.id, position=0))
    db.commit()
    (card,) = pins.shelf(db, user)
    assert card["kind"] == "channel"
    assert card["unwatched_count"] == 3
    assert card["next"]["id"] == "vshort"  # newest


def test_shelf_resolves_a_playlist_to_its_first_unwatched_item(db):
    user, _ = _seed_yt(db)
    pl = Playlist(user_id=user.id, name="Elalvós")
    db.add(pl)
    db.flush()
    db.add_all(
        [
            PlaylistItem(playlist_id=pl.id, video_id="vshort", position=0),
            PlaylistItem(playlist_id=pl.id, video_id="vmed", position=1),
        ]
    )
    db.add(VideoState(user_id=user.id, video_id="vshort", status="watched", watched_at=NOW))
    db.add(ComfortPin(user_id=user.id, playlist_id=pl.id, position=0))
    db.commit()
    (card,) = pins.shelf(db, user)
    assert card["kind"] == "playlist"
    assert card["item_count"] == 2
    assert card["next"]["id"] == "vmed"


def test_a_pinned_video_survives_leaving_the_feed_scope(db):
    """The shelf is the fallback surface. An entry silently vanishing because the user unsubscribed
    from the channel is exactly the failure it must not have."""
    user, ch = _seed_yt(db)
    db.add(ComfortPin(user_id=user.id, video_id="vmed", position=0))
    db.commit()
    db.execute(delete(Subscription))
    db.commit()
    (card,) = pins.shelf(db, user)
    assert card["id"] == "vmed"
    assert card["next"]["id"] == "vmed"


def test_a_self_playing_pin_does_not_reference_itself(db):
    """`next` must be a distinct dict even when the pin plays itself. Aliasing the card into its own
    `next` is a cycle, and the serializer turns the whole shelf into a 500 over it."""
    user, _ = _seed_yt(db)
    db.add(ComfortPin(user_id=user.id, video_id="vmed", position=0))
    db.commit()
    (card,) = pins.shelf(db, user)
    assert card["next"] is not card
    assert "next" not in card["next"]


def test_the_resume_rail_says_what_would_actually_play(db):
    """A show tile that only opens the show page moves the decision instead of removing it — the
    click lands on six seasons to choose between, which is the paralysis this whole surface exists
    for. The shelf and the hero already resolved; this rail did not (reported 2026-09-06). The tile
    stays the show (its artwork is what the user recognises); `next` is what starts."""
    user, _ = _seed_yt(db)
    _, show, ep1 = _seed_plex(db, user)
    _enable_plex(db)
    db.add(
        PlexState(
            user_id=user.id,
            item_id=ep1.id,
            status="watched",
            watched_at=NOW,
            progress_updated_at=NOW,
        )
    )
    db.commit()
    rail = rails.continue_watching(db, user)
    tile = next(c for c in rail if c["kind"] == "show")
    assert tile["id"] == "ps1"  # still the show
    assert tile["next"]["kind"] == "episode"
    assert tile["next"]["id"] == "ps1e2"  # ep1 is finished → ep2 is what plays


def test_only_a_show_row_pays_for_a_resolution(db):
    """`next` is the answer to "a show has no Play button". Attaching it to a movie or a video would
    both cost a query for nothing and re-open the self-reference cycle the shelf already hit."""
    user, _ = _seed_yt(db)
    _seed_plex(db, user)
    _enable_plex(db)
    for card in rails.continue_watching(db, user):
        assert ("next" in card) == (card["kind"] == "show"), card["id"]


def test_next_episode_takes_an_episode_with_no_state_row_at_all(db):
    """The NULL trap in the SQL rewrite: `status NOT IN ('watched','hidden')` is UNKNOWN — not true —
    for an episode the user has never touched, so a bare filter drops exactly the common case and the
    rail would offer episode 1 of a show already half-watched."""
    user, _ = _seed_yt(db)
    _, show, ep1 = _seed_plex(db, user)
    db.add(PlexState(user_id=user.id, item_id=ep1.id, status="watched", watched_at=NOW))
    db.commit()  # ep2 has NO PlexState row
    assert queries.next_episode(db, user.id, show.id)["id"] == "ps1e2"


def test_a_finished_show_starts_again_from_the_top(db):
    """Nothing unwatched left is not "nothing to play" — it is "watch it again"."""
    user, _ = _seed_yt(db)
    _, show, ep1 = _seed_plex(db, user)
    ep2 = db.execute(select(PlexItem).where(PlexItem.rating_key == "ps1e2")).scalar_one()
    db.add_all(
        [
            PlexState(user_id=user.id, item_id=ep1.id, status="watched", watched_at=NOW),
            PlexState(user_id=user.id, item_id=ep2.id, status="watched", watched_at=NOW),
        ]
    )
    db.commit()
    assert queries.next_episode(db, user.id, show.id)["id"] == "ps1e1"


def test_next_episode_prefers_the_one_in_progress(db):
    """In progress beats "next unwatched": resuming is cheaper than starting."""
    user, _ = _seed_yt(db)
    _, show, ep1 = _seed_plex(db, user)
    ep2 = db.execute(select(PlexItem).where(PlexItem.rating_key == "ps1e2")).scalar_one()
    db.add_all(
        [
            PlexState(user_id=user.id, item_id=ep1.id, status="watched", watched_at=NOW),
            PlexState(user_id=user.id, item_id=ep2.id, status="new", position_seconds=120),
        ]
    )
    db.commit()
    assert queries.next_episode(db, user.id, show.id)["id"] == "ps1e2"


# ── HTTP surface ──────────────────────────────────────────────────────────────────────────────────


def test_discover_endpoint_answers_in_one_pass(db, as_user):
    user, _ = _seed_yt(db)
    db.add(
        VideoState(
            user_id=user.id,
            video_id="vmed",
            status="new",
            position_seconds=300,
            progress_updated_at=NOW,
        )
    )
    db.commit()
    r = as_user(user).get("/api/discover?budget=medium")
    assert r.status_code == 200
    body = r.json()
    assert body["budget"] == "medium"
    assert body["hero"]["tier"] == "resume"
    assert [c["id"] for c in body["continue_watching"]] == ["vmed"]
    assert body["comfort"] == []


def test_an_unknown_budget_is_refused_rather_than_silently_widened(db, as_user):
    user, _ = _seed_yt(db)
    assert as_user(user).get("/api/discover?budget=whenever").status_code == 422


def test_pinning_twice_is_the_same_shelf(db, as_user):
    """The gesture is "make sure this is on my shelf", so a re-pin is a no-op, not a 409."""
    user, _ = _seed_yt(db)
    c = as_user(user)
    first = c.post("/api/discover/pins", json={"kind": "video", "ref": "vmed"}).json()
    second = c.post("/api/discover/pins", json={"kind": "video", "ref": "vmed"}).json()
    assert first["created"] is True and second["created"] is False
    assert first["id"] == second["id"]
    assert len(c.get("/api/discover/pins").json()["items"]) == 1


def test_pinning_something_that_does_not_exist_is_a_404(db, as_user):
    user, _ = _seed_yt(db)
    r = as_user(user).post("/api/discover/pins", json={"kind": "video", "ref": "nope"})
    assert r.status_code == 404


def test_another_users_playlist_is_not_pinnable(db, as_user):
    """404, not 403 — a 403 would confirm the id exists to a stranger."""
    user, _ = _seed_yt(db)
    other = User(email="other-e24@example.com")
    db.add(other)
    db.flush()
    pl = Playlist(user_id=other.id, name="Theirs")
    db.add(pl)
    db.commit()
    r = as_user(user).post("/api/discover/pins", json={"kind": "playlist", "ref": str(pl.id)})
    assert r.status_code == 404


def test_reorder_ignores_ids_that_are_not_yours(db, as_user):
    """A stale tab replaying an order that includes a since-deleted pin should reorder what still
    exists rather than fail wholesale."""
    user, ch = _seed_yt(db)
    c = as_user(user)
    a = c.post("/api/discover/pins", json={"kind": "video", "ref": "vmed"}).json()["id"]
    b = c.post("/api/discover/pins", json={"kind": "channel", "ref": ch.id}).json()["id"]
    assert c.post("/api/discover/pins/reorder", json={"ids": [b, 9999, a]}).status_code == 200
    assert [p["pin_id"] for p in c.get("/api/discover/pins").json()["items"]] == [b, a]


def test_deleting_a_pin_reports_what_went(db, as_user):
    user, _ = _seed_yt(db)
    c = as_user(user)
    pin_id = c.post("/api/discover/pins", json={"kind": "video", "ref": "vmed"}).json()["id"]
    r = c.delete(f"/api/discover/pins/{pin_id}")
    assert r.status_code == 200 and r.json() == {"deleted": pin_id}
    assert c.get("/api/discover/pins").json()["items"] == []