"""Tests for YouTube playlist fan-out. URL-detection + count-aggregation are pure-Python and run anywhere. The DB-backed `add_playlist` tests need a live petalbrain connection (same auto-skip pattern as `tests/test_claim.py`). One opt-in live-network test exercises `expand_youtube_playlist` against a real small public playlist; it stays gated behind `SECOND_BRAIN_LIVE_NETWORK_TESTS=1` so CI doesn't depend on YouTube reachability. """ from __future__ import annotations import os import uuid import psycopg import pytest # --------------------------------------------------------------------------- # Pure-Python detection # --------------------------------------------------------------------------- @pytest.mark.parametrize( "url", [ "https://www.youtube.com/playlist?list=PLxxxxxxxxxx", "https://youtube.com/playlist?list=PLxxxxxxxxxx", "https://music.youtube.com/playlist?list=PLxxxxxxxxxx", "https://m.youtube.com/playlist?list=PLxxxxxxxxxx", "https://www.youtube.com/playlist/?list=PLxxxxxxxxxx", # trailing slash ], ) def test_is_youtube_playlist_url_positives(url): from second_brain.sources_service import is_youtube_playlist_url assert is_youtube_playlist_url(url) is True @pytest.mark.parametrize( "url", [ # Watch URL with list= — ambiguous; we treat as single video. "https://www.youtube.com/watch?v=abc123&list=PLxxxxxxxxxx", "https://www.youtube.com/watch?v=abc123", "https://youtu.be/abc123", "https://www.youtube.com/@somechannel", "https://www.youtube.com/playlist", # no list param "https://example.com/playlist?list=PL", # not a YouTube host "https://marktechpost.com/some/article", "", "not-a-url", "ftp://www.youtube.com/playlist?list=PLxxxxxxxxxx", ], ) def test_is_youtube_playlist_url_negatives(url): from second_brain.sources_service import is_youtube_playlist_url assert is_youtube_playlist_url(url) is False # --------------------------------------------------------------------------- # DB-backed: add_playlist with a mocked expander # --------------------------------------------------------------------------- def _db_reachable() -> bool: url = os.environ.get("SECOND_BRAIN_DATABASE_URL") or os.environ.get( "HERBYLAB_DATABASE_URL" ) if not url: return False raw = url.replace("postgresql+psycopg://", "postgresql://", 1) try: with psycopg.connect(raw, connect_timeout=2): return True except Exception: return False pytestmark_db = pytest.mark.skipif( not _db_reachable(), reason="needs SECOND_BRAIN_DATABASE_URL pointed at a real petalbrain", ) @pytestmark_db def test_add_playlist_aggregates_counts(): """5 entries; 2 are pre-seeded duplicates → added=3, duplicates=2, failed=0.""" from second_brain.database import Database, reset_database_singleton from second_brain.models import Source, SourceStatus, SourceType, utcnow from second_brain.sources_service import add_playlist reset_database_singleton() db = Database() # Unique URL set per run so re-execution is idempotent. tag = uuid.uuid4().hex[:8] urls = [ f"https://www.youtube.com/watch?v=playlist-{tag}-{i}" for i in range(5) ] # Pre-seed entries 0 and 1 so they're already in the queue. pre_seeded_ids: list[int] = [] with db.session() as sess: for u in urls[:2]: src = Source( url=u, title=f"pre-{tag}", domain="development", source_type=SourceType.VIDEO, status=SourceStatus.PENDING, ingested_at=utcnow(), ) sess.add(src) sess.flush() pre_seeded_ids.append(src.id) def fake_expander(playlist_url, *, max_items): # noqa: ARG001 return [{"url": u, "title": f"vid-{tag}-{i}"} for i, u in enumerate(urls)] try: with db.session() as sess: result = add_playlist( sess, url=f"https://www.youtube.com/playlist?list=FAKE-{tag}", domain="development", expander=fake_expander, ) assert result.expanded == 5 assert result.added == 3 assert result.duplicates == 2 assert result.failed == 0 assert result.failures == [] with db.session() as sess: count = sess.query(Source).filter(Source.url.in_(urls)).count() assert count == 5 # all 5 URLs now in the queue (2 pre-existing + 3 new) finally: with db.session() as sess: sess.query(Source).filter(Source.url.in_(urls)).delete( synchronize_session=False ) reset_database_singleton() @pytestmark_db def test_add_playlist_tolerates_bad_entries(): """A bad entry mid-batch must not abort the rest.""" from second_brain.database import Database, reset_database_singleton from second_brain.models import Source from second_brain.sources_service import add_playlist reset_database_singleton() db = Database() tag = uuid.uuid4().hex[:8] entries = [ {"url": f"https://www.youtube.com/watch?v=good-{tag}-1", "title": "ok 1"}, {"url": "not-a-url", "title": "bad"}, # validate_url raises {"url": f"https://www.youtube.com/watch?v=good-{tag}-2", "title": "ok 2"}, ] urls_added = [entries[0]["url"], entries[2]["url"]] def fake_expander(playlist_url, *, max_items): # noqa: ARG001 return entries try: with db.session() as sess: result = add_playlist( sess, url=f"https://www.youtube.com/playlist?list=FAKE-{tag}", expander=fake_expander, ) assert result.expanded == 3 assert result.added == 2 assert result.duplicates == 0 assert result.failed == 1 assert len(result.failures) == 1 assert result.failures[0][0] == "not-a-url" with db.session() as sess: count = sess.query(Source).filter(Source.url.in_(urls_added)).count() assert count == 2 finally: with db.session() as sess: sess.query(Source).filter(Source.url.in_(urls_added)).delete( synchronize_session=False ) reset_database_singleton() @pytestmark_db def test_add_playlist_empty_expansion(): """Empty playlist → expanded=0, no rows touched.""" from second_brain.database import Database, reset_database_singleton from second_brain.sources_service import add_playlist reset_database_singleton() db = Database() def fake_expander(playlist_url, *, max_items): # noqa: ARG001 return [] try: with db.session() as sess: result = add_playlist( sess, url="https://www.youtube.com/playlist?list=EMPTY", expander=fake_expander, ) assert result.expanded == 0 assert result.added == 0 assert result.duplicates == 0 assert result.failed == 0 finally: reset_database_singleton() # --------------------------------------------------------------------------- # Opt-in live-network test: hit a real (small) public playlist. # Gate with SECOND_BRAIN_LIVE_NETWORK_TESTS=1 so CI doesn't depend on YouTube. # --------------------------------------------------------------------------- @pytest.mark.skipif( os.environ.get("SECOND_BRAIN_LIVE_NETWORK_TESTS") != "1", reason="set SECOND_BRAIN_LIVE_NETWORK_TESTS=1 to hit yt-dlp against YouTube", ) def test_expand_youtube_playlist_live(): """Sanity-check the yt-dlp call shape against a real playlist. Uses Google's "YouTube Developers" playlist which has been stable for years. We only assert the returned shape (list of dicts with `url` and `title`), not the count — playlist contents change. """ from second_brain.sources_service import expand_youtube_playlist url = ( "https://www.youtube.com/playlist?" "list=PLOU2XLYxmsIKpaV8h0AGE05so0fAwwfTw" ) entries = expand_youtube_playlist(url, max_items=5) assert isinstance(entries, list) assert len(entries) > 0 assert len(entries) <= 5 for e in entries: assert e["url"].startswith("https://www.youtube.com/watch?v=") assert "title" in e