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