second-brain/tests/test_playlist.py

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