myYouTube/tests/test_feed.py

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.category import Category
from app.models.channel import Channel
from app.models.channel_category import channel_categories
from app.models.download_job import DownloadJob
from app.models.video import Video
from app.api.feed import get_feed


@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]


def _seed(db_session):
    channel_a = Channel(youtube_channel_id="chanA", title="Channel A", subscribed=True)
    channel_b = Channel(youtube_channel_id="chanB", title="Channel B", subscribed=True)
    db_session.add_all([channel_a, channel_b])
    db_session.commit()

    category = Category(name="Linux", slug="linux", sort_order=0)
    db_session.add(category)
    db_session.commit()

    db_session.execute(channel_categories.insert().values(channel_id=channel_a.id, category_id=category.id))
    db_session.commit()

    base = datetime(2026, 9, 10, tzinfo=timezone.utc)
    videos = []
    for i in range(5):
        video = Video(
            youtube_video_id=f"vid{i}",
            channel_id=channel_a.id if i % 2 == 0 else channel_b.id,
            title=f"Video {i}",
            published_at=base + timedelta(hours=i),
            youtube_url=f"https://www.youtube.com/watch?v=vid{i}",
        )
        videos.append(video)
    db_session.add_all(videos)
    db_session.commit()

    return channel_a, channel_b, category, videos


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

    resp = client.get("/api/feed").json()

    ids = [item["youtube_video_id"] for item in resp["items"]]
    assert ids == ["vid4", "vid3", "vid2", "vid1", "vid0"]
    assert resp["next_cursor"] is None


def test_feed_searches_video_and_channel_titles(db_session):
    _seed(db_session)

    by_video = get_feed(search="video 4", limit=30, db=db_session)
    assert [item["youtube_video_id"] for item in by_video["items"]] == ["vid4"]

    by_channel = get_feed(search="channel b", limit=30, db=db_session)
    assert [item["youtube_video_id"] for item in by_channel["items"]] == ["vid3", "vid1"]

    literal_wildcard = get_feed(search="%", limit=30, db=db_session)
    assert literal_wildcard["items"] == []


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

    page1 = client.get("/api/feed?limit=2").json()
    assert [i["youtube_video_id"] for i in page1["items"]] == ["vid4", "vid3"]
    assert page1["next_cursor"] is not None

    page2 = client.get(f"/api/feed?limit=2&cursor={page1['next_cursor']}").json()
    assert [i["youtube_video_id"] for i in page2["items"]] == ["vid2", "vid1"]
    assert page2["next_cursor"] is not None

    page3 = client.get(f"/api/feed?limit=2&cursor={page2['next_cursor']}").json()
    assert [i["youtube_video_id"] for i in page3["items"]] == ["vid0"]
    assert page3["next_cursor"] is None


def test_feed_filters_by_category(client, db_session):
    _, _, category, _ = _seed(db_session)

    resp = client.get(f"/api/feed?category_id={category.id}").json()

    ids = {i["youtube_video_id"] for i in resp["items"]}
    assert ids == {"vid0", "vid2", "vid4"}


def test_feed_filters_by_channel(client, db_session):
    channel_a, channel_b, _, _ = _seed(db_session)

    resp = client.get(f"/api/feed?channel_id={channel_b.id}").json()

    ids = {i["youtube_video_id"] for i in resp["items"]}
    assert ids == {"vid1", "vid3"}


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

    resp = client.get("/api/feed?uncategorized=true").json()

    ids = {i["youtube_video_id"] for i in resp["items"]}
    assert ids == {"vid1", "vid3"}


def _seed_new_only(db_session):
    """Two categories: channel A in a category (recent + old video),
    channel B uncategorized (recent + old video). Published dates are
    relative to now so the new_videos_window filter is exercised."""
    now = datetime.now(timezone.utc)
    channel_a = Channel(youtube_channel_id="chanNewA", title="Channel New A", subscribed=True)
    channel_b = Channel(youtube_channel_id="chanNewB", title="Channel New B", subscribed=True)
    db_session.add_all([channel_a, channel_b])
    db_session.commit()

    category = Category(name="Recent", slug="recent", sort_order=0)
    db_session.add(category)
    db_session.commit()

    db_session.execute(channel_categories.insert().values(channel_id=channel_a.id, category_id=category.id))
    db_session.commit()

    videos = [
        Video(
            youtube_video_id="vidRecentA",
            channel_id=channel_a.id,
            title="Recent A",
            published_at=now - timedelta(days=1),
            youtube_url="https://www.youtube.com/watch?v=vidRecentA",
        ),
        Video(
            youtube_video_id="vidOldA",
            channel_id=channel_a.id,
            title="Old A",
            published_at=now - timedelta(days=30),
            youtube_url="https://www.youtube.com/watch?v=vidOldA",
        ),
        Video(
            youtube_video_id="vidRecentB",
            channel_id=channel_b.id,
            title="Recent B",
            published_at=now - timedelta(hours=2),
            youtube_url="https://www.youtube.com/watch?v=vidRecentB",
        ),
        Video(
            youtube_video_id="vidOldB",
            channel_id=channel_b.id,
            title="Old B",
            published_at=now - timedelta(days=30),
            youtube_url="https://www.youtube.com/watch?v=vidOldB",
        ),
    ]
    db_session.add_all(videos)
    db_session.commit()

    return channel_a, channel_b, category, videos


def test_feed_new_only_includes_recent_and_excludes_old(client, db_session):
    _seed_new_only(db_session)

    resp = client.get("/api/feed?new_only=true").json()

    ids = {i["youtube_video_id"] for i in resp["items"]}
    assert ids == {"vidRecentA", "vidRecentB"}

    # Without new_only nothing changes: old videos appear as usual.
    all_ids = {i["youtube_video_id"] for i in client.get("/api/feed").json()["items"]}
    assert all_ids == {"vidRecentA", "vidOldA", "vidRecentB", "vidOldB"}


def test_feed_new_only_combines_with_category(client, db_session):
    _, _, category, _ = _seed_new_only(db_session)

    resp = client.get(f"/api/feed?category_id={category.id}&new_only=true").json()

    ids = {i["youtube_video_id"] for i in resp["items"]}
    assert ids == {"vidRecentA"}


def test_feed_new_only_excludes_unsubscribed_channels(client, db_session):
    now = datetime.now(timezone.utc)
    channel = Channel(youtube_channel_id="chanUnsub", title="Channel Unsub", subscribed=False)
    db_session.add(channel)
    db_session.commit()
    db_session.add(
        Video(
            youtube_video_id="vidFreshUnsub",
            channel_id=channel.id,
            title="Fresh Unsub",
            published_at=now - timedelta(hours=2),
            youtube_url="https://www.youtube.com/watch?v=vidFreshUnsub",
        )
    )
    db_session.commit()

    resp = client.get("/api/feed?new_only=true").json()
    assert resp["items"] == []

    # Regular feed is not filtered by subscription.
    all_ids = {i["youtube_video_id"] for i in client.get("/api/feed").json()["items"]}
    assert all_ids == {"vidFreshUnsub"}


def test_feed_new_only_pagination_cursor_works_in_filtered_set(client, db_session):
    now = datetime.now(timezone.utc)
    channel = Channel(youtube_channel_id="chanPage", title="Channel Page", subscribed=True)
    db_session.add(channel)
    db_session.commit()
    db_session.add_all(
        [
            Video(
                youtube_video_id=f"vidNew{i}",
                channel_id=channel.id,
                title=f"Video New {i}",
                published_at=now - timedelta(hours=i),
                youtube_url=f"https://www.youtube.com/watch?v=vidNew{i}",
            )
            for i in range(3)
        ]
    )
    db_session.commit()

    page1 = client.get("/api/feed?new_only=true&limit=2").json()
    assert [i["youtube_video_id"] for i in page1["items"]] == ["vidNew0", "vidNew1"]
    assert page1["next_cursor"] is not None

    page2 = client.get(f"/api/feed?new_only=true&limit=2&cursor={page1['next_cursor']}").json()
    assert [i["youtube_video_id"] for i in page2["items"]] == ["vidNew2"]
    assert page2["next_cursor"] is None


def test_feed_filters_downloaded(client, db_session):
    _, _, _, videos = _seed(db_session)

    completed = DownloadJob(video_id=videos[0].id, status="completed", media_url="http://x/f0.mp4")
    failed = DownloadJob(video_id=videos[1].id, status="failed")
    db_session.add_all([completed, failed])
    db_session.commit()

    resp = client.get("/api/feed?downloaded=true").json()

    ids = {i["youtube_video_id"] for i in resp["items"]}
    assert ids == {"vid0"}


def test_feed_filters_downloaded_uses_latest_job_per_video(client, db_session):
    _, _, _, videos = _seed(db_session)

    older = DownloadJob(video_id=videos[0].id, status="completed", media_url="http://x/f0.mp4")
    db_session.add(older)
    db_session.commit()
    newer = DownloadJob(video_id=videos[0].id, status="deleted")
    db_session.add(newer)
    db_session.commit()

    resp = client.get("/api/feed?downloaded=true").json()

    assert resp["items"] == []


def test_saved_counts(client, db_session):
    channel_a, channel_b, category, videos = _seed(db_session)

    # vid0 is on channel_a (in "Linux"), vid1 is on channel_b (uncategorized).
    db_session.add_all(
        [
            DownloadJob(video_id=videos[0].id, status="completed", media_url="http://x/0.mp4"),
            DownloadJob(video_id=videos[1].id, status="completed", media_url="http://x/1.mp4"),
            DownloadJob(video_id=videos[2].id, status="failed"),
        ]
    )
    db_session.commit()

    resp = client.get("/api/feed/saved-counts").json()

    assert resp["all"] == 2
    assert resp["uncategorized"] == 1
    assert resp["categories"] == {str(category.id): 1}


def test_feed_item_shape(client, db_session):
    channel_a, _, category, _ = _seed(db_session)

    resp = client.get("/api/feed?limit=1").json()
    item = resp["items"][0]

    assert item["channel"]["id"] == item["channel"]["id"]
    assert set(item["channel"].keys()) == {"id", "youtube_channel_id", "title", "thumbnail_url"}
    assert item["local"] == {
        "available": False,
        "status": "not_downloaded",
        "progress_percent": None,
        "media_url": None,
    }
    assert isinstance(item["categories"], list)