tests/test_supadata_first_pipeline.py

import os
import sys
import types
import unittest
from pathlib import Path
from unittest import mock


sys.path.insert(0, str(Path(__file__).parent.parent))

fake_auth_pkg = types.ModuleType("src.auth")
fake_auth_pkg.__path__ = []
fake_auth_dependencies = types.ModuleType("src.auth.dependencies")
fake_auth_dependencies.get_current_user = lambda: None
fake_auth_pkg.dependencies = fake_auth_dependencies
sys.modules["src.auth"] = fake_auth_pkg
sys.modules["src.auth.dependencies"] = fake_auth_dependencies

fake_db_pkg = types.ModuleType("src.db")
fake_db_pkg.__path__ = []
fake_db_models = types.ModuleType("src.db.models")


class _User:
    id = "test-user"


fake_db_models.User = _User
fake_db_pkg.models = fake_db_models
sys.modules["src.db"] = fake_db_pkg
sys.modules["src.db.models"] = fake_db_models

fake_groq = types.ModuleType("groq")


class _Groq:
    def __init__(self, *args, **kwargs):
        pass


fake_groq.Groq = _Groq
sys.modules["groq"] = fake_groq

from src.api import downloader, notes_routes  # noqa: E402


TEST_URL = "https://www.youtube.com/watch?v=abcdefghijk"
TEST_VIDEO_ID = "abcdefghijk"


class DownloaderStrategyTests(unittest.TestCase):
    def test_supadata_first_uses_supadata_before_other_providers(self):
        with mock.patch.object(downloader.settings, "youtube_transcript_strategy", "supadata_first"):
            worker = downloader.YouTubeDownloader()

        with (
            mock.patch.object(worker, "_get_transcript_via_supadata", return_value="supadata transcript") as supadata,
            mock.patch.object(worker, "_get_transcript_via_youtube") as youtube,
            mock.patch.object(worker, "_get_transcript_via_assemblyai") as assemblyai,
        ):
            transcript = worker.get_transcript(TEST_URL)

        self.assertEqual(transcript, "supadata transcript")
        supadata.assert_called_once_with(TEST_VIDEO_ID)
        youtube.assert_not_called()
        assemblyai.assert_not_called()

    def test_youtube_first_falls_back_to_supadata(self):
        with mock.patch.object(downloader.settings, "youtube_transcript_strategy", "youtube_first"):
            worker = downloader.YouTubeDownloader()

        with (
            mock.patch.object(
                worker,
                "_get_transcript_via_youtube",
                side_effect=downloader.TranscriptProviderError("youtube blocked"),
            ) as youtube,
            mock.patch.object(worker, "_get_transcript_via_supadata", return_value="fallback transcript") as supadata,
            mock.patch.object(worker, "_get_transcript_via_assemblyai") as assemblyai,
        ):
            transcript = worker.get_transcript(TEST_URL)

        self.assertEqual(transcript, "fallback transcript")
        youtube.assert_called_once_with(TEST_VIDEO_ID)
        supadata.assert_called_once_with(TEST_VIDEO_ID)
        assemblyai.assert_not_called()


class NotesRouteStrategyTests(unittest.IsolatedAsyncioTestCase):
    async def test_process_video_task_uses_summary_title_in_supadata_first_mode(self):
        task_id = "task-supadata-title"
        notes_routes.tasks[task_id] = {}

        note_generator = mock.Mock()
        note_generator.generateSummary.return_value = {
            "title": "Inferred Title",
            "detected_language": "English",
            "summary": "Summary",
            "segments": [],
            "conclusion": "Done",
            "topics": [],
        }
        note_generator.format_notes_to_markdown.return_value = "body"
        note_generator.format_final_notes.return_value = "final markdown"

        try:
            with (
                mock.patch.object(notes_routes.settings, "youtube_transcript_strategy", "supadata_first"),
                mock.patch.object(notes_routes, "_fetch_video_info_via_ytdlp") as metadata_fetch,
                mock.patch.object(notes_routes, "YouTubeDownloader") as downloader_cls,
                mock.patch.object(notes_routes, "NoteGenerator", return_value=note_generator),
                mock.patch.object(notes_routes, "get_youtube_duration", return_value=274) as duration_fetch,
                mock.patch(
                    "src.summarization.topic_classifier.classify_topics",
                    return_value=["Education & Science"],
                ),
            ):
                downloader_cls.return_value.get_transcript.return_value = "transcript text"

                await notes_routes.process_video_task(
                    task_id=task_id,
                    youtube_url=TEST_URL,
                    language="en",
                    user_id="user-1",
                )

            metadata_fetch.assert_not_called()
            duration_fetch.assert_called_once()
            self.assertEqual(notes_routes.tasks[task_id]["status"], "completed")
            self.assertEqual(notes_routes.tasks[task_id]["videoTitle"], "Inferred Title")
            self.assertEqual(note_generator.format_final_notes.call_args.args[1], "Inferred Title")
        finally:
            notes_routes.tasks.pop(task_id, None)

    def test_supadata_first_duration_skips_ytdlp_when_supadata_succeeds(self):
        with (
            mock.patch.object(notes_routes.settings, "youtube_transcript_strategy", "supadata_first"),
            mock.patch.object(notes_routes, "_duration_via_supadata", return_value=274) as supadata_duration,
            mock.patch.object(notes_routes, "_duration_via_html_scrape") as html_duration,
            mock.patch.object(notes_routes, "_duration_via_ytdlp") as ytdlp_duration,
        ):
            duration = notes_routes.get_youtube_duration(TEST_URL)

        self.assertEqual(duration, 274)
        supadata_duration.assert_called_once_with(TEST_VIDEO_ID)
        html_duration.assert_not_called()
        ytdlp_duration.assert_not_called()


if __name__ == "__main__":
    unittest.main()