Gan, Jimmy
187 lines
7.3 KiB
Python
187 lines
7.3 KiB
Python
"""Tests for fetcher.py — mocks YouTube Data API v3 + RSS."""
|
|
import pytest
|
|
import os
|
|
import json
|
|
import aiosqlite
|
|
import fetcher
|
|
from unittest.mock import patch, AsyncMock, MagicMock
|
|
|
|
|
|
def _clean_rotation_file():
|
|
"""Remove rotation state file if it exists."""
|
|
path = getattr(fetcher, "_ROTATION_FILE", None)
|
|
if path and os.path.exists(path):
|
|
os.remove(path)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def auto_clean_rotation():
|
|
_clean_rotation_file()
|
|
yield
|
|
_clean_rotation_file()
|
|
|
|
|
|
def _mock_youtube_build(sub_channels=None, video_data=None):
|
|
"""Build a mock YouTube API service that returns canned responses."""
|
|
sub_channels = sub_channels or [
|
|
{"snippet": {"resourceId": {"channelId": "chan001"}}},
|
|
{"snippet": {"resourceId": {"channelId": "chan002"}}},
|
|
]
|
|
video_data = video_data or {}
|
|
|
|
def mock_subscriptions_list(**kwargs):
|
|
return MagicMock(execute=MagicMock(return_value={"items": sub_channels}))
|
|
|
|
def mock_videos_list(**kwargs):
|
|
vid_str = kwargs.get("id", "")
|
|
ids = [v for v in vid_str.split(",") if v]
|
|
items = []
|
|
for vid in ids:
|
|
data = video_data.get(vid, {
|
|
"title": f"Video {vid}",
|
|
"channelTitle": "Test Channel",
|
|
"channelId": "chan001",
|
|
"thumbnails": {"high": {"url": f"https://img.youtube.com/vi/{vid}/hqdefault.jpg"}},
|
|
"publishTime": "2026-06-01T00:00:00Z",
|
|
})
|
|
items.append({"id": vid, "snippet": data})
|
|
return MagicMock(execute=MagicMock(return_value={"items": items}))
|
|
|
|
svc = MagicMock()
|
|
svc.subscriptions = MagicMock(return_value=MagicMock(list=mock_subscriptions_list, list_next=MagicMock(return_value=None)))
|
|
svc.videos = MagicMock(return_value=MagicMock(list=mock_videos_list))
|
|
return svc
|
|
|
|
|
|
def _mock_rss_response(videos):
|
|
"""Build RSS XML string for a list of (video_id, title) tuples."""
|
|
entries = []
|
|
for vid, title in videos:
|
|
entries.append(f"""<entry>
|
|
<id>yt:video:{vid}</id>
|
|
<link href="https://www.youtube.com/watch?v={vid}"/>
|
|
<title>{title}</title>
|
|
</entry>""")
|
|
return f"""<?xml version="1.0" encoding="UTF-8"?>
|
|
<feed xmlns="http://www.w3.org/2005/Atom">
|
|
{''.join(entries)}
|
|
</feed>"""
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_zero_channels(test_db):
|
|
"""When no subscriptions exist, should return 0."""
|
|
mock_svc = _mock_youtube_build(sub_channels=[])
|
|
with patch("fetcher.build", return_value=mock_svc), \
|
|
patch("fetcher.get_credentials", return_value="mock_creds"):
|
|
count = await fetcher.fetch_subscriptions()
|
|
assert count == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_rss_empty(test_db):
|
|
"""When RSS returns nothing, count should be 0."""
|
|
mock_svc = _mock_youtube_build()
|
|
with patch("fetcher.build", return_value=mock_svc), \
|
|
patch("fetcher.get_credentials", return_value="mock_creds"), \
|
|
patch("httpx.AsyncClient.get", AsyncMock(return_value=MagicMock(status_code=404))):
|
|
count = await fetcher.fetch_subscriptions()
|
|
assert count == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_new_videos(test_db):
|
|
"""New videos from RSS should be inserted into DB."""
|
|
mock_svc = _mock_youtube_build(video_data={
|
|
"vid001": {"title": "First Video", "channelTitle": "Chan One",
|
|
"channelId": "chan001", "thumbnails": {"high": {"url": ""}},
|
|
"publishTime": "2026-06-01T00:00:00Z"},
|
|
"vid002": {"title": "Second Video", "channelTitle": "Chan Two",
|
|
"channelId": "chan002", "thumbnails": {"high": {"url": ""}},
|
|
"publishTime": "2026-06-02T00:00:00Z"},
|
|
})
|
|
rss_xml = _mock_rss_response([("vid001", "First Video"), ("vid002", "Second Video")])
|
|
mock_response = MagicMock(status_code=200, text=rss_xml)
|
|
|
|
with patch("fetcher.build", return_value=mock_svc), \
|
|
patch("fetcher.get_credentials", return_value="mock_creds"), \
|
|
patch("httpx.AsyncClient.get", AsyncMock(return_value=mock_response)):
|
|
count = await fetcher.fetch_subscriptions()
|
|
assert count == 2
|
|
|
|
# Verify DB
|
|
cursor = await test_db.execute("SELECT * FROM videos ORDER BY video_id")
|
|
rows = await cursor.fetchall()
|
|
assert len(rows) == 2
|
|
assert rows[0]["video_id"] == "vid001"
|
|
assert rows[0]["title"] == "First Video"
|
|
assert rows[1]["video_id"] == "vid002"
|
|
assert rows[1]["title"] == "Second Video"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_skip_duplicates(test_db):
|
|
"""Existing videos should be skipped."""
|
|
# Pre-insert one video
|
|
await test_db.execute(
|
|
"INSERT INTO videos (video_id, title, url, status) VALUES ('vid001', 'Old', 'https://youtube.com/watch?v=vid001', 'pending')"
|
|
)
|
|
await test_db.commit()
|
|
|
|
mock_svc = _mock_youtube_build(video_data={
|
|
"vid001": {"title": "First Video (updated)", "channelTitle": "Chan One",
|
|
"channelId": "chan001"},
|
|
"vid002": {"title": "Second Video", "channelTitle": "Chan Two",
|
|
"channelId": "chan002"},
|
|
})
|
|
rss_xml = _mock_rss_response([("vid001", "First Video"), ("vid002", "Second Video")])
|
|
mock_response = MagicMock(status_code=200, text=rss_xml)
|
|
|
|
with patch("fetcher.build", return_value=mock_svc), \
|
|
patch("fetcher.get_credentials", return_value="mock_creds"), \
|
|
patch("httpx.AsyncClient.get", AsyncMock(return_value=mock_response)):
|
|
count = await fetcher.fetch_subscriptions()
|
|
assert count == 1 # Only vid002 is new
|
|
|
|
# Verify DB — vid001 still has original title (not updated)
|
|
cursor = await test_db.execute("SELECT title FROM videos WHERE video_id='vid001'")
|
|
row = await cursor.fetchone()
|
|
assert row["title"] == "Old"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rotation_advances(test_db):
|
|
"""Rotation slice should advance after each sync."""
|
|
# Mock 150 channels
|
|
sub_channels = [{"snippet": {"resourceId": {"channelId": f"chan{i:03d}"}}} for i in range(150)]
|
|
mock_svc = _mock_youtube_build(sub_channels=sub_channels)
|
|
|
|
rss_xml = _mock_rss_response([]) # No videos, just test rotation
|
|
mock_response = MagicMock(status_code=200, text=rss_xml)
|
|
|
|
with patch("fetcher.build", return_value=mock_svc), \
|
|
patch("fetcher.get_credentials", return_value="mock_creds"), \
|
|
patch("httpx.AsyncClient.get", AsyncMock(return_value=mock_response)):
|
|
# First sync — should process chan000-chan049
|
|
count = await fetcher.fetch_subscriptions()
|
|
# Second sync — chan050-chan099
|
|
count2 = await fetcher.fetch_subscriptions()
|
|
# Third sync — chan100-chan149
|
|
count3 = await fetcher.fetch_subscriptions()
|
|
# Fourth sync — wrap around, chan000-chan049 again
|
|
count4 = await fetcher.fetch_subscriptions()
|
|
|
|
# Verify rotation file
|
|
with open(fetcher._ROTATION_FILE) as f:
|
|
state = json.load(f)
|
|
assert state["offset"] == 50 # After 4th sync, offset = 50 (wrapped from 150)
|
|
assert state["total"] == 150
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_quietly_handles_missing_credentials(test_db):
|
|
"""When OAuth is missing, should return 0 gracefully."""
|
|
with patch("fetcher.get_credentials", side_effect=FileNotFoundError("No client_secret")):
|
|
count = await fetcher.fetch_subscriptions()
|
|
assert count == 0
|