myYouTube/tests/test_shorts_feed.py

"""Tests for the Shorts/long split on /api/feed: the `type` filter, the
`is_short` flag and the one-off `anchor` starting point (see
analytics/2026-09-27-shorts-split-and-feed.md)."""

from datetime import datetime, timedelta, timezone

import pytest
from fastapi.testclient import TestClient

from app.core.auth_dependency import require_session
from app.db import get_db
from app.main import app
from app.models.channel import Channel
from app.models.video import Video


@pytest.fixture
def client(db_session):
    def _get_db_override():
        yield db_session

    app.dependency_overrides[get_db] = _get_db_override
    app.dependency_overrides[require_session] = lambda: None
    # Deliberately not using `with TestClient(app)`: that runs the app's
    # lifespan, which would try to reach the real MeTube instance and DB.
    yield TestClient(app)
    del app.dependency_overrides[get_db]
    del app.dependency_overrides[require_session]


BASE = datetime(2026, 9, 10, tzinfo=timezone.utc)

# Insertion order gives increasing ids. published_at also increases with i,
# so the descending feed order is v4, v3, v2, v1, v0.
DURATIONS = {
    "v0": None,  # unknown -> long
    "v1": 179,  # short
    "v2": 180,  # short (boundary, inclusive)
    "v3": 181,  # long (just above the threshold)
    "v4": 100,  # short
}


def _seed(db_session):
    channel = Channel(youtube_channel_id="chanShorts", title="Channel Shorts", subscribed=True)
    db_session.add(channel)
    db_session.commit()

    videos = []
    for i, (youtube_video_id, duration) in enumerate(DURATIONS.items()):
        video = Video(
            youtube_video_id=youtube_video_id,
            channel_id=channel.id,
            title=f"Video {youtube_video_id}",
            published_at=BASE + timedelta(hours=i),
            duration_seconds=duration,
            youtube_url=f"https://www.youtube.com/watch?v={youtube_video_id}",
        )
        videos.append(video)
    db_session.add_all(videos)
    db_session.commit()
    return channel, videos


def _ids(resp):
    return [item["youtube_video_id"] for item in resp["items"]]


def test_feed_type_short_only_includes_at_or_below_threshold(client, db_session):
    _seed(db_session)

    assert _ids(client.get("/api/feed?type=short").json()) == ["v4", "v2", "v1"]


def test_feed_type_long_includes_null_and_above_threshold(client, db_session):
    _seed(db_session)

    assert _ids(client.get("/api/feed?type=long").json()) == ["v3", "v0"]


def test_feed_type_all_matches_default(client, db_session):
    _seed(db_session)

    explicit = client.get("/api/feed?type=all").json()
    default = client.get("/api/feed").json()
    assert _ids(explicit) == ["v4", "v3", "v2", "v1", "v0"]
    assert explicit["items"] == default["items"]


def test_feed_type_invalid_returns_422(client, db_session):
    _seed(db_session)

    assert client.get("/api/feed?type=weird").status_code == 422


def test_feed_type_threshold_is_configurable(client, db_session, monkeypatch):
    _seed(db_session)

    monkeypatch.setattr("app.config.settings.shorts_max_duration_seconds", 181)

    assert set(_ids(client.get("/api/feed?type=short").json())) == {"v4", "v3", "v2", "v1"}
    assert _ids(client.get("/api/feed?type=long").json()) == ["v0"]


def test_feed_item_is_short_flag(client, db_session):
    _seed(db_session)

    items = {item["youtube_video_id"]: item for item in client.get("/api/feed").json()["items"]}

    assert items["v1"]["is_short"] is True
    assert items["v2"]["is_short"] is True
    assert items["v4"]["is_short"] is True
    assert items["v3"]["is_short"] is False
    assert items["v0"]["is_short"] is False


def test_feed_type_short_pagination_in_filtered_set(client, db_session):
    _seed(db_session)

    page1 = client.get("/api/feed?type=short&limit=2").json()
    assert _ids(page1) == ["v4", "v2"]
    assert page1["next_cursor"] is not None

    page2 = client.get(f"/api/feed?type=short&limit=2&cursor={page1['next_cursor']}").json()
    assert _ids(page2) == ["v1"]
    assert page2["next_cursor"] is None


def test_feed_type_long_pagination_in_filtered_set(client, db_session):
    _seed(db_session)

    page1 = client.get("/api/feed?type=long&limit=1").json()
    assert _ids(page1) == ["v3"]
    assert page1["next_cursor"] is not None

    page2 = client.get(f"/api/feed?type=long&limit=1&cursor={page1['next_cursor']}").json()
    assert _ids(page2) == ["v0"]
    assert page2["next_cursor"] is None


def test_feed_anchor_starts_at_video_and_goes_older(client, db_session):
    _seed(db_session)

    assert _ids(client.get("/api/feed?anchor=v2").json()) == ["v2", "v1", "v0"]


def test_feed_anchor_unknown_falls_back_to_first_page(client, db_session):
    _seed(db_session)

    resp = client.get("/api/feed?anchor=does-not-exist").json()
    assert _ids(resp) == ["v4", "v3", "v2", "v1", "v0"]


def test_feed_anchor_ignored_when_cursor_present(client, db_session):
    _seed(db_session)

    page1 = client.get("/api/feed?anchor=v2&limit=2").json()
    assert _ids(page1) == ["v2", "v1"]
    assert page1["next_cursor"] is not None

    # A (different) anchor on a cursor page must not shift the selection:
    # only the first page uses the anchor.
    page2 = client.get(f"/api/feed?anchor=v4&cursor={page1['next_cursor']}").json()
    assert _ids(page2) == ["v0"]


def test_feed_anchor_combines_with_type_filter(client, db_session):
    _seed(db_session)

    assert client.get("/api/feed?type=short&anchor=v0").json()["items"] == []
    assert _ids(client.get("/api/feed?type=short&anchor=v3").json()) == ["v2", "v1"]