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.plex import rollup
from app.plex.cards import leaf_card
from app.models import (
    BlockedChannel,
    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_the_resume_rail_obeys_the_channel_blocklist(db):
    """Blocking a channel does not touch `video_states`, so a half-watched video from it survives
    with its resume position — and this rail builds its own SELECT instead of going through
    `filtered_query`. "Never show me this channel" has to outrank "you started it", or the landing
    page answers "what should I watch" with the thing the user has just banned."""
    user, ch = _seed_yt(db)
    db.add(
        VideoState(
            user_id=user.id,
            video_id="vmed",
            status="new",
            position_seconds=300,
            progress_updated_at=NOW,
        )
    )
    db.commit()
    assert [c["id"] for c in rails.continue_watching(db, user)] == ["vmed"]

    db.add(BlockedChannel(user_id=user.id, channel_id=ch.id))
    db.commit()
    assert rails.continue_watching(db, user) == []
    # And so the hero, whose resume tier draws from that rail, has nothing to offer either.
    assert hero.pick(db, user, "any", 0) is None


def test_a_null_timestamp_sorts_last(db):
    """A missing progress timestamp sorts last rather than 500ing the landing page.

    Narrower than it was written to be, and the epic-end review corrected it: the previous key,
    `(ts is not None, ts)`, does NOT raise on two nulls — tuple comparison finds the first UNEQUAL
    element via `==`, so `None < None` is never reached, and this test passes under that key too.
    What it does discriminate is the naive `key=lambda r: r[0]`, which really does raise. See
    `plex.rollup.NO_TS` for the corrected rationale."""
    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_the_same_thing_cannot_be_pinned_twice(db):
    """The per-kind unique constraints are what make the route's concurrent-tab path safe (it catches
    `IntegrityError` precisely because of them) — and the migration drift check filters constraint
    diffs out as noise, so nothing else would notice them going missing. The route's own dedupe
    returns early without ever reaching the DB, so only this test exercises them."""
    user, _ = _seed_yt(db)
    db.add(ComfortPin(user_id=user.id, video_id="vmed", position=0))
    db.commit()
    db.add(ComfortPin(user_id=user.id, video_id="vmed", position=1))
    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 _count_statements(db, fn) -> int:
    """How many SQL statements `fn` sends through the session's connection."""
    from sqlalchemy import event

    seen: list[str] = []
    engine = db.get_bind()
    listener = lambda *a, **k: seen.append(a[2])  # noqa: E731 — (conn, cursor, statement, …)
    event.listen(engine, "before_cursor_execute", listener)
    try:
        fn()
    finally:
        event.remove(engine, "before_cursor_execute", listener)
    return len(seen)


def test_the_shelf_resolves_each_channel_and_playlist_on_its_own_terms(db):
    """Batched, the per-channel count and newest video come from one windowed query: each channel
    must still get ITS count and ITS newest, not the shelf's overall first row."""
    user, ch = _seed_yt(db)
    other = Channel(id="UCother", title="Other")
    db.add(other)
    db.flush()
    db.add(Video(id="vother", channel_id=other.id, title="x", published_at=NOW - timedelta(days=9)))
    pl = Playlist(user_id=user.id, name="Két elem")
    empty = Playlist(user_id=user.id, name="Üres")
    db.add_all([pl, empty])
    db.flush()
    db.add(PlaylistItem(playlist_id=pl.id, video_id="vlong", position=0))
    for i, target in enumerate(
        [
            {"channel_id": ch.id},
            {"channel_id": other.id},
            {"playlist_id": pl.id},
            {"playlist_id": empty.id},
        ]
    ):
        db.add(ComfortPin(user_id=user.id, position=i, **target))
    db.commit()

    got = [
        (c.get("unwatched_count", c.get("item_count")), (c["next"] or {}).get("id"))
        for c in pins.shelf(db, user)
    ]
    assert got == [(3, "vshort"), (1, "vother"), (1, "vlong"), (0, None)]


def test_the_shelf_costs_the_same_queries_for_one_pin_or_many(db):
    """A channel pin used to cost three queries and a playlist pin two, each; the shelf now
    resolves every channel and every playlist together."""
    user, ch = _seed_yt(db)
    extra = [Channel(id=f"UCx{i}", title=f"x{i}") for i in range(3)]
    lists = [Playlist(user_id=user.id, name=f"p{i}") for i in range(4)]
    db.add_all(extra + lists)
    db.flush()
    db.add(ComfortPin(user_id=user.id, channel_id=ch.id, position=0))
    db.add(ComfortPin(user_id=user.id, playlist_id=lists[0].id, position=1))
    db.commit()
    few = _count_statements(db, lambda: pins.shelf(db, user))

    for i, c in enumerate(extra):
        db.add(ComfortPin(user_id=user.id, channel_id=c.id, position=10 + i))
    for i, p in enumerate(lists[1:]):
        db.add(ComfortPin(user_id=user.id, playlist_id=p.id, position=20 + i))
    db.commit()
    assert _count_statements(db, lambda: pins.shelf(db, user)) == few


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_the_plex_surprise_obeys_the_time_budget(db):
    """The one control on the page has to mean what it says: a 100-minute film is not an answer to
    "10 perc". The YouTube half of this rule has had its own test since S1; the Plex half's duration
    bounds could be deleted with the whole suite still green, because every test that reached
    `plex_surprise` passed budget="any"."""
    user, _ = _seed_yt(db)
    _seed_plex(db, user)
    _enable_plex(db)
    drawn = lambda budget: {  # noqa: E731
        (c["source"], c["id"]) for s in range(6) for c in rails.surprise(db, user, budget, s, 5)
    }
    assert ("plex", "pm1") not in drawn("short")
    # …and it IS offered where it fits, so this cannot pass by the rail being empty.
    assert ("plex", "pm1") in drawn("long")


def test_the_plex_surprise_never_offers_something_already_started(db):
    """A half-watched film is not a surprise, and the resume tier has already had its chance at it.
    The mirror of `test_surprise_never_offers_something_already_started` for the other source — the
    asymmetric half of one principle, which the house rules say to fix everywhere."""
    user, _ = _seed_yt(db)
    movie, _, _ = _seed_plex(db, user)
    _enable_plex(db)
    db.add(
        PlexState(
            user_id=user.id,
            item_id=movie.id,
            status="new",
            position_seconds=600,
            progress_updated_at=NOW,
        )
    )
    db.commit()
    drawn = {
        (c["source"], c["id"]) for s in range(6) for c in rails.surprise(db, user, "long", s, 5)
    }
    assert ("plex", "pm1") not in drawn


def test_a_new_pin_lands_at_the_end_even_after_a_hole(db):
    """Positions are never renumbered on removal — `remove_pin` deletes the row, and every FK cascade
    silently removes a pin when its target disappears — so the append position has to come from
    `max(position)`, not from the row count. With a count, the first hole drops the next pin into the
    MIDDLE of the shelf and the second makes two pins claim one slot."""
    user, ch = _seed_yt(db)
    db.add_all(
        [
            ComfortPin(user_id=user.id, video_id="vshort", position=0),
            ComfortPin(user_id=user.id, video_id="vmed", position=1),
            ComfortPin(user_id=user.id, video_id="vlong", position=2),
        ]
    )
    db.commit()
    db.execute(delete(ComfortPin).where(ComfortPin.video_id.in_(("vshort", "vmed"))))
    db.commit()

    db.add(ComfortPin(user_id=user.id, channel_id=ch.id, position=pins.next_position(db, user)))
    db.commit()
    assert [c["kind"] for c in pins.shelf(db, user)] == ["video", "channel"]
    positions = (
        db.execute(select(ComfortPin.position).order_by(ComfortPin.position)).scalars().all()
    )
    assert len(set(positions)) == len(positions)  # and no two tiles claim the same slot


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.

    The states are what make this test exist: without them every rail query inner-joins its way to
    zero rows, the loop body never runs, and the assertion is vacuous — which is exactly how the
    epic-end review found it."""
    user, _ = _seed_yt(db)
    movie, _, ep1 = _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,
            ),
            PlexState(
                user_id=user.id,
                item_id=movie.id,
                status="new",
                position_seconds=600,
                progress_updated_at=NOW,
            ),
            PlexState(
                user_id=user.id,
                item_id=ep1.id,
                status="new",
                position_seconds=120,
                progress_updated_at=NOW,
            ),
        ]
    )
    db.commit()

    rail = rails.continue_watching(db, user)
    # A video, a movie and a show — one row of each kind that can appear here, so the assertion below
    # actually exercises both sides of the equality.
    assert {c["kind"] for c in rail} == {"video", "movie", "show"}
    for card in rail:
        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_agrees_with_the_show_page_when_two_are_in_progress(db):
    """One show, one answer. `plex.rollup.rollup` — which drives the show page's Resume button —
    takes the LAST in-progress episode in broadcast order; taking the FIRST here gave the same show
    two different "continue" answers, with the rail ranking it by its most recent touch and then
    offering the episode abandoned weeks earlier. Two in-progress episodes are not exotic: a scrub to
    the end leaves a stale `status='new', position>0` row until the daily reconcile."""
    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="new",
                position_seconds=300,
                progress_updated_at=NOW - timedelta(days=9),
            ),
            PlexState(
                user_id=user.id,
                item_id=ep2.id,
                status="new",
                position_seconds=600,
                progress_updated_at=NOW,
            ),
        ]
    )
    db.commit()

    got = queries.next_episode(db, user.id, show.id)
    assert got["id"] == "ps1e2"
    # Asserted against the other implementation rather than against a literal, so the two cannot
    # drift apart again without this failing.
    ordered = [
        leaf_card(ep, db.execute(select(PlexState).where(PlexState.item_id == ep.id)).scalar_one())
        for ep in (ep1, ep2)
    ]
    assert rollup.rollup(ordered)["resume"]["id"] == got["id"]


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_a_plex_show_resolves_to_the_episode_that_would_play(db, as_user):
    """Two of the five pinnable kinds are Plex, and the rating_key → internal-id translation is
    theirs alone — a wrong column or a wrong model in either arm pins the wrong row, or 404s every
    Plex pin. Neither arm had a test before the epic-end review."""
    user, _ = _seed_yt(db)
    _, _, ep1 = _seed_plex(db, user)
    _enable_plex(db)
    db.add(PlexState(user_id=user.id, item_id=ep1.id, status="watched", watched_at=NOW))
    db.commit()
    c = as_user(user)
    assert c.post("/api/discover/pins", json={"kind": "plex_show", "ref": "ps1"}).status_code == 200
    (tile,) = c.get("/api/discover/pins").json()["items"]
    assert (tile["kind"], tile["id"]) == ("show", "ps1")
    assert tile["next"]["id"] == "ps1e2"  # ep1 is finished


def test_pinning_a_plex_movie_plays_itself_without_a_cycle(db, as_user):
    """The self-reference that once 500'd the whole shelf, on the other leaf kind: `next` has to be a
    distinct dict, not the tile aliased into itself. Covered for a video since S1, never for a movie."""
    user, _ = _seed_yt(db)
    _seed_plex(db, user)
    _enable_plex(db)
    c = as_user(user)
    assert c.post("/api/discover/pins", json={"kind": "plex_item", "ref": "pm1"}).status_code == 200
    (tile,) = c.get("/api/discover/pins").json()["items"]
    assert tile["id"] == "pm1" and tile["next"]["id"] == "pm1"


def test_the_reroll_endpoint_parses_its_whole_exclusion_list_and_its_press_count(db, as_user):
    """The reroll's behaviour is covered at the domain level; the ROUTE's query-string translation —
    the comma split and the `start` passthrough — was not covered at all, so a change there could
    hand back the same card on every press with the suite green. Excluding all three candidates is
    what proves the whole list parsed: a delimiter change leaves one unsplit string in the set, and a
    hero comes back.

    NOT the `MAX_EXCLUDE` clamp, which this sends three keys against a cap of 32 and cannot see. The
    docstring used to claim it did (round 2). The clamp is deliberately left unpinned: it guards
    nothing the route would otherwise do — see the constant's own comment — and a test that sent 33
    keys could only assert that the request still answers, which the comma split already proves."""
    user, _ = _seed_yt(db)
    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()
    c = as_user(user)

    # `start` rotates which tier answers first: 0 leads with resume, 1 with the shelf.
    assert c.get("/api/discover/pick?seed=1&start=0").json()["hero"]["tier"] == "resume"
    assert c.get("/api/discover/pick?seed=1&start=1").json()["hero"]["tier"] == "comfort"

    skip = ",".join(f"youtube:{v}" for v in ("vshort", "vmed", "vlong"))
    assert c.get(f"/api/discover/pick?seed=2&exclude={skip}").json()["hero"] is None
    assert c.get("/api/discover/pick?budget=whenever").status_code == 422


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_a_pin_whose_target_vanishes_mid_request_is_a_404_not_a_500(db, as_user, monkeypatch):
    """`add_pin`'s `except IntegrityError` arm was reached by no test at all, so the 500→404
    narrowing could be reverted with the whole suite green (round 2). The 404 above comes from the
    `resolve_ref` PRE-CHECK, a different branch that returns before the INSERT.

    The race the arm exists for — a Plex sync dropping the row between `resolve_ref` confirming it
    and the INSERT — needs two transactions, but the BRANCH does not: a `resolve_ref` pointing at an
    id no row carries lets the pre-check pass and trips the same FK on commit, with the re-query
    finding no winner. The sentinel has to FIT `comfort_pins.video_id` (varchar(16)) — a longer one
    raises `DataError` on the way in, which this arm does not catch and never meant to. Its sibling arm (a concurrent duplicate, where the re-query DOES find a winner
    and answers 200) genuinely needs a second connection and stays uncovered."""
    user, _ = _seed_yt(db)
    monkeypatch.setattr(pins, "resolve_ref", lambda *a, **k: ("video_id", "gone-mid-req"))
    r = as_user(user).post("/api/discover/pins", json={"kind": "video", "ref": "vshort"})
    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"] == []


def test_with_plex_off_the_home_shelf_hides_plex_pins_but_keeps_them(db, as_user):
    """The owner's call (2026-09-19): with the deployment's Plex toggle off, the Home shelf goes
    YouTube-only like the rails beside it — a Plex tile there offered a Play that landed on a gated
    module. The hero's comfort tier draws from the same filtered shelf, and the pin menus' own
    listing still sees every pin, so nothing the user curated is lost."""
    user, _ = _seed_yt(db)
    _seed_plex(db, user)
    db.add_all(
        [
            ComfortPin(
                user_id=user.id,
                plex_item_id=db.scalar(select(PlexItem.id).where(PlexItem.rating_key == "pm1")),
                position=0,
            ),
            ComfortPin(user_id=user.id, video_id="vlong", position=1),
        ]
    )
    db.commit()
    c = as_user(user)
    assert [t["id"] for t in c.get("/api/discover").json()["comfort"]] == ["vlong"]
    assert c.get("/api/discover/pick?start=1").json()["hero"]["card"]["id"] == "vlong"
    assert {t["id"] for t in c.get("/api/discover/pins").json()["items"]} == {"pm1", "vlong"}
    _enable_plex(db)
    assert [t["id"] for t in c.get("/api/discover").json()["comfort"]] == ["pm1", "vlong"]


def test_a_reorder_of_the_visible_pins_keeps_the_hidden_ones_in_place(db, as_user):
    """The client sends the order it SHOWS. With Plex off that leaves the Plex pins out, and
    numbering only the listed ids 0..n-1 put two pins on one position and moved the hidden one about
    when Plex came back. The listed pins trade the slots they hold; the rest stay put."""
    user, ch = _seed_yt(db)
    _seed_plex(db, user)
    c = as_user(user)
    _enable_plex(db)
    a = c.post("/api/discover/pins", json={"kind": "video", "ref": "vmed"}).json()["id"]
    hidden = c.post("/api/discover/pins", json={"kind": "plex_item", "ref": "pm1"}).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, b, a]}).status_code == 200
    assert [p["pin_id"] for p in c.get("/api/discover/pins").json()["items"]] == [b, hidden, a]
    positions = db.scalars(select(ComfortPin.position).where(ComfortPin.user_id == user.id)).all()
    assert sorted(positions) == [0, 1, 2]


def test_a_hero_from_a_show_pin_names_the_pin_it_came_from(db, as_user):
    """A show's next EPISODE is not itself pinned, so the hero's menu used to look the episode up,
    miss, and offer "Pin" under "From your shelf" — creating a second pin. The card now carries
    the pin it was resolved from, which the menu acts on (owner's call: "Unpin")."""
    user, _ = _seed_yt(db)
    _seed_plex(db, user)
    _enable_plex(db)
    c = as_user(user)
    assert c.post("/api/discover/pins", json={"kind": "plex_show", "ref": "ps1"}).status_code == 200
    got = c.get("/api/discover/pick?start=1").json()["hero"]
    assert got["tier"] == "comfort" and got["card"]["id"] == "ps1e1"
    assert got["card"]["pinned_from"] == {
        "source": "plex",
        "kind": "show",
        "id": "ps1",
        "title": "A Show",
    }


def test_two_overlapping_reorders_never_share_a_slot(db, as_user):
    """`reorder` writes only the rows that change. Unlocked, two tabs reordering from the same
    snapshot each skipped the rows that already matched in THEIR snapshot and interleaved row by
    row — two pins on one slot (deep review, debt burn). Locked, the second waits for the first and
    deals from its result: here a partial list (a tab with Plex off, one pin hidden), whose left-out
    pin must stay where the FIRST reorder put it — a lock taken in position order handed the waiter
    fresh rows in its stale order, and it dealt that pin back to its old slot (round 2)."""
    import threading

    from app.db import SessionLocal

    user, ch = _seed_yt(db)
    c = as_user(user)
    p1 = c.post("/api/discover/pins", json={"kind": "video", "ref": "vmed"}).json()["id"]
    p2 = c.post("/api/discover/pins", json={"kind": "channel", "ref": ch.id}).json()["id"]
    p3 = c.post("/api/discover/pins", json={"kind": "video", "ref": "vlong"}).json()["id"]
    uid = user.id

    with SessionLocal() as tab_a, SessionLocal() as tab_b, SessionLocal() as probe:
        pins.reorder(tab_a, tab_a.get(User, uid), [p2, p3, p1])  # written, not yet committed
        errors: list[BaseException] = []

        def second_tab():
            try:
                pins.reorder(tab_b, tab_b.get(User, uid), [p3, p2])
                tab_b.commit()
            except BaseException as exc:  # noqa: BLE001 — re-raised in the test thread below
                errors.append(exc)

        worker = threading.Thread(target=second_tab)
        worker.start()
        # Commit A only once B is blocked on a row lock, so B's read is the stale snapshot.
        waiting = 0
        for _ in range(200):
            waiting = probe.scalar(text("SELECT count(*) FROM pg_locks WHERE NOT granted"))
            probe.rollback()
            if waiting:
                break
            worker.join(0.05)
        assert waiting, "tab B never blocked — the race was not exercised"
        tab_a.commit()
        worker.join(10)
        assert not worker.is_alive() and errors == []

    db.expire_all()
    positions = db.scalars(select(ComfortPin.position).where(ComfortPin.user_id == uid)).all()
    assert sorted(positions) == [0, 1, 2]
    assert [p["pin_id"] for p in c.get("/api/discover/pins").json()["items"]] == [p3, p2, p1]