Add automated unit and mock integration test suite
- Configure conftest.py with pytest tmp_path DB isolation fixtures - Add test_db.py for schema connection validations - Add test_fetcher.py with mocked yt-dlp subprocess parser checks - Add test_summarizer.py with mocked transcripts and Claude API requests - Add test_main.py for FastAPI router endpoints and state updates Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,125 @@
|
||||
import pytest
|
||||
import os
|
||||
import httpx
|
||||
import aiosqlite
|
||||
import summarizer
|
||||
from unittest.mock import patch, MagicMock, AsyncMock
|
||||
|
||||
class MockTranscript:
|
||||
def fetch(self):
|
||||
return [
|
||||
{"text": "Welcome to this video tutorial.", "start": 0.0, "duration": 3.0},
|
||||
{"text": "Today we are learning about Svelte 5 and runes.", "start": 3.0, "duration": 4.0}
|
||||
]
|
||||
|
||||
class MockTranscriptList:
|
||||
def __init__(self, transcript):
|
||||
self.transcript = transcript
|
||||
|
||||
def find_manually_created_transcript(self, languages):
|
||||
return self.transcript
|
||||
|
||||
def find_generated_transcript(self, languages):
|
||||
return self.transcript
|
||||
|
||||
def __iter__(self):
|
||||
return iter([self.transcript])
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_download_transcript_success():
|
||||
mock_trans = MockTranscript()
|
||||
mock_list = MockTranscriptList(mock_trans)
|
||||
|
||||
with patch("youtube_transcript_api.YouTubeTranscriptApi.list_transcripts", return_value=mock_list):
|
||||
text = summarizer.download_transcript("test_vid")
|
||||
assert text == "Welcome to this video tutorial. Today we are learning about Svelte 5 and runes."
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_summarize_video_success(test_db):
|
||||
# Insert a pending video in db first
|
||||
await test_db.execute("""
|
||||
INSERT INTO videos (video_id, title, url, status)
|
||||
VALUES ('summarize_vid', 'Video to Summarize', 'https://www.youtube.com/watch?v=summarize_vid', 'pending')
|
||||
""")
|
||||
await test_db.commit()
|
||||
|
||||
mock_trans = MockTranscript()
|
||||
mock_list = MockTranscriptList(mock_trans)
|
||||
|
||||
# Mock Claude HTTP response
|
||||
mock_response = MagicMock(spec=httpx.Response)
|
||||
mock_response.status_code = 200
|
||||
mock_response.json.return_value = {
|
||||
"content": [
|
||||
{
|
||||
"text": "**Mock Core Hook**\n- Takeaway 1\n- Takeaway 2\nVerdict line."
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
# Patch both transcript api and httpx AsyncClient post call
|
||||
with patch("youtube_transcript_api.YouTubeTranscriptApi.list_transcripts", return_value=mock_list), \
|
||||
patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=mock_response) as mock_post:
|
||||
|
||||
summary = await summarizer.summarize_video("summarize_vid")
|
||||
|
||||
# Verify API request details
|
||||
mock_post.assert_called_once()
|
||||
url = mock_post.call_args[0][0]
|
||||
assert "bytecatcode.org" in url
|
||||
|
||||
headers = mock_post.call_args[1]["headers"]
|
||||
assert "x-api-key" in headers
|
||||
assert headers["User-Agent"] is not None
|
||||
|
||||
payload = mock_post.call_args[1]["json"]
|
||||
assert payload["model"] == "claude-haiku-4-5-20251001"
|
||||
assert len(payload["messages"]) == 1
|
||||
|
||||
assert summary == "**Mock Core Hook**\n- Takeaway 1\n- Takeaway 2\nVerdict line."
|
||||
|
||||
# Query db to verify transcript & summary were saved
|
||||
cursor = await test_db.execute("SELECT * FROM videos WHERE video_id = 'summarize_vid'")
|
||||
video = await cursor.fetchone()
|
||||
assert video is not None
|
||||
assert video["transcript"] == "Welcome to this video tutorial. Today we are learning about Svelte 5 and runes."
|
||||
assert video["summary"] == "**Mock Core Hook**\n- Takeaway 1\n- Takeaway 2\nVerdict line."
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_summarize_all_pending(test_db):
|
||||
# Insert multiple videos, one pending and one already summarized
|
||||
await test_db.execute("""
|
||||
INSERT INTO videos (video_id, title, url, summary, status)
|
||||
VALUES
|
||||
('pending_vid', 'Pending Video', 'https://youtube.com/watch?v=pending_vid', '', 'pending'),
|
||||
('done_vid', 'Done Video', 'https://youtube.com/watch?v=done_vid', 'Existing summary', 'pending')
|
||||
""")
|
||||
await test_db.commit()
|
||||
|
||||
mock_trans = MockTranscript()
|
||||
mock_list = MockTranscriptList(mock_trans)
|
||||
|
||||
mock_response = MagicMock(spec=httpx.Response)
|
||||
mock_response.status_code = 200
|
||||
mock_response.json.return_value = {
|
||||
"content": [{"text": "Fresh summary"}]
|
||||
}
|
||||
|
||||
with patch("youtube_transcript_api.YouTubeTranscriptApi.list_transcripts", return_value=mock_list), \
|
||||
patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=mock_response) as mock_post:
|
||||
|
||||
await summarizer.summarize_all_pending()
|
||||
|
||||
# Should only call API once for 'pending_vid'
|
||||
mock_post.assert_called_once()
|
||||
payload = mock_post.call_args[1]["json"]
|
||||
assert "Welcome to this video tutorial." in payload["messages"][0]["content"]
|
||||
|
||||
# Verify db states
|
||||
cursor = await test_db.execute("SELECT video_id, summary FROM videos ORDER BY video_id")
|
||||
videos = await cursor.fetchall()
|
||||
assert videos[0]["video_id"] == "done_vid"
|
||||
assert videos[0]["summary"] == "Existing summary" # Untouched
|
||||
|
||||
assert videos[1]["video_id"] == "pending_vid"
|
||||
assert videos[1]["summary"] == "Fresh summary" # Updated
|
||||
Reference in New Issue
Block a user