+ {# min-w-0 keeps this flex child shrinkable; break-words / break-all
+ make the text wrap inside instead of forcing the parent wider. #}
-
+
{{ source.title }}
-
{{ source.url }}
+
{{ source.url }}
{% if source.focus %}
-
Focus: {{ source.focus }}
+
Focus: {{ source.focus }}
{% endif %}
diff --git a/tests/test_playlist.py b/tests/test_playlist.py
new file mode 100644
index 0000000..d5d3e78
--- /dev/null
+++ b/tests/test_playlist.py
@@ -0,0 +1,252 @@
+"""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
diff --git a/tests/test_transcribe.py b/tests/test_transcribe.py
index d75a624..ba239d3 100644
--- a/tests/test_transcribe.py
+++ b/tests/test_transcribe.py
@@ -63,6 +63,95 @@ def test_write_srt_roundtrip(tmp_path):
assert "2\n00:00:01,500 --> 00:00:03,250\nworld" in body
+# ---------------------------------------------------------------------------
+# Segment / word conversion — exercised on faked faster-whisper objects so
+# we don't need a real model + audio to validate the JSONB-bound shape.
+# ---------------------------------------------------------------------------
+
+
+class _FakeWord:
+ def __init__(self, start, end, word, probability):
+ self.start = start
+ self.end = end
+ self.word = word
+ self.probability = probability
+
+
+class _FakeSegment:
+ def __init__(self, id, start, end, text, words=None):
+ self.id = id
+ self.start = start
+ self.end = end
+ self.text = text
+ self.words = words
+
+
+def test_segment_to_dict_with_words():
+ """Happy path: faster-whisper segment + word_timestamps → JSON-safe dict."""
+ import json
+
+ from second_brain.transcribe import segment_to_dict
+
+ seg = _FakeSegment(
+ id=3,
+ start=1.25,
+ end=3.5,
+ text=" hello world ",
+ words=[
+ _FakeWord(1.25, 1.6, " hello", 0.92),
+ _FakeWord(1.6, 3.5, " world", 0.87),
+ ],
+ )
+ d = segment_to_dict(seg)
+
+ assert d == {
+ "id": 3,
+ "start": 1.25,
+ "end": 3.5,
+ "text": "hello world", # stripped
+ "words": [
+ {"start": 1.25, "end": 1.6, "word": " hello", "probability": 0.92},
+ {"start": 1.6, "end": 3.5, "word": " world", "probability": 0.87},
+ ],
+ }
+ # Must be JSON-serialisable for the JSONB column.
+ json.dumps(d)
+
+
+def test_segment_to_dict_without_words():
+ """word_timestamps=False (or older faster-whisper) → segment.words is None."""
+ from second_brain.transcribe import segment_to_dict
+
+ seg = _FakeSegment(id=0, start=0.0, end=1.0, text="just text", words=None)
+ d = segment_to_dict(seg)
+ assert d["text"] == "just text"
+ assert d["words"] == []
+
+
+def test_segment_to_dict_skips_malformed_words():
+ """A single bad word entry must not drop the whole segment."""
+ from second_brain.transcribe import segment_to_dict
+
+ class _BadWord:
+ # Missing start → float(None) raises in the helper, must be skipped.
+ start = None
+ end = 1.0
+ word = "broken"
+ probability = 0.5
+
+ seg = _FakeSegment(
+ id=1,
+ start=0.0,
+ end=1.0,
+ text="hi",
+ words=[_FakeWord(0.0, 0.5, "hi", 0.99), _BadWord()],
+ )
+ d = segment_to_dict(seg)
+ # Good word survives; bad word is filtered.
+ assert len(d["words"]) == 1
+ assert d["words"][0]["word"] == "hi"
+
+
# ---------------------------------------------------------------------------
# Integration: synthesized audio → CPU/int8 faster-whisper
# ---------------------------------------------------------------------------