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


@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_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 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)