+
+
+ {% if source.status == 'analyzed' %}
+ {% endif %}
+
+
+
+
+ {% if source.has_transcript and source.source_type == 'video' %}
+
+
+ {% endif %}
{% endif %}
diff --git a/tests/test_source_edit_retry.py b/tests/test_source_edit_retry.py
new file mode 100644
index 0000000..c3177f6
--- /dev/null
+++ b/tests/test_source_edit_retry.py
@@ -0,0 +1,258 @@
+"""Tests for the source edit-metadata + retry-reset service helpers.
+
+Live-DB pattern matches `tests/test_claim.py` / `tests/test_playlist.py`:
+both groups auto-skip when no petalbrain connection is configured. The
+service layer is where edit/retry logic lives (the web routes + CLI are
+thin wrappers around it), so this is where the meaningful coverage is.
+"""
+
+from __future__ import annotations
+
+import os
+import uuid
+
+import psycopg
+import pytest
+
+
+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 = pytest.mark.skipif(
+ not _db_reachable(),
+ reason="needs SECOND_BRAIN_DATABASE_URL pointed at a real petalbrain",
+)
+
+
+def _seed_source(db, **overrides):
+ """Insert a Source row and return (id, url). Caller cleans up by id."""
+ from second_brain.models import Source, SourceStatus, SourceType, utcnow
+
+ tag = uuid.uuid4().hex[:8]
+ defaults = dict(
+ url=f"https://example.test/edit-retry/{tag}",
+ title=f"seed-{tag}",
+ domain="development",
+ source_type=SourceType.ARTICLE,
+ status=SourceStatus.PENDING,
+ ingested_at=utcnow(),
+ )
+ defaults.update(overrides)
+ with db.session() as sess:
+ src = Source(**defaults)
+ sess.add(src)
+ sess.flush()
+ return src.id, src.url
+
+
+def _cleanup(db, source_id):
+ from second_brain.models import Source
+
+ with db.session() as sess:
+ sess.query(Source).filter(Source.id == source_id).delete(
+ synchronize_session=False
+ )
+
+
+# ---------------------------------------------------------------------------
+# update_source_metadata
+# ---------------------------------------------------------------------------
+
+
+def test_update_metadata_writes_domain_title_focus():
+ from second_brain.database import Database, reset_database_singleton
+ from second_brain.models import Source
+ from second_brain.sources_service import update_source_metadata
+
+ reset_database_singleton()
+ db = Database()
+ sid, _ = _seed_source(db)
+ try:
+ with db.session() as sess:
+ update_source_metadata(
+ sess,
+ source_id=sid,
+ domain="homelab",
+ title="new title",
+ focus="new focus directive",
+ )
+ with db.session() as sess:
+ row = sess.query(Source).filter(Source.id == sid).first()
+ assert row.domain == "homelab"
+ assert row.title == "new title"
+ assert row.focus == "new focus directive"
+ assert row.updated_at is not None
+ finally:
+ _cleanup(db, sid)
+ reset_database_singleton()
+
+
+def test_update_metadata_blank_strings_become_null():
+ from second_brain.database import Database, reset_database_singleton
+ from second_brain.models import Source
+ from second_brain.sources_service import update_source_metadata
+
+ reset_database_singleton()
+ db = Database()
+ sid, _ = _seed_source(db)
+ try:
+ with db.session() as sess:
+ update_source_metadata(
+ sess, source_id=sid, domain="development", title=" ", focus=""
+ )
+ with db.session() as sess:
+ row = sess.query(Source).filter(Source.id == sid).first()
+ assert row.title is None
+ assert row.focus is None
+ finally:
+ _cleanup(db, sid)
+ reset_database_singleton()
+
+
+def test_update_metadata_rejects_unknown_domain():
+ from second_brain.database import Database, reset_database_singleton
+ from second_brain.sources_service import update_source_metadata
+
+ reset_database_singleton()
+ db = Database()
+ sid, _ = _seed_source(db)
+ try:
+ with db.session() as sess:
+ with pytest.raises(ValueError, match="unknown domain"):
+ update_source_metadata(sess, source_id=sid, domain="not-a-domain")
+ finally:
+ _cleanup(db, sid)
+ reset_database_singleton()
+
+
+def test_update_metadata_missing_source_raises_lookup():
+ from second_brain.database import Database, reset_database_singleton
+ from second_brain.sources_service import update_source_metadata
+
+ reset_database_singleton()
+ db = Database()
+ try:
+ with db.session() as sess:
+ with pytest.raises(LookupError):
+ update_source_metadata(
+ sess, source_id=-99999, domain="development"
+ )
+ finally:
+ reset_database_singleton()
+
+
+# ---------------------------------------------------------------------------
+# reset_source_for_retry
+# ---------------------------------------------------------------------------
+
+
+def test_retry_full_resets_to_pending_and_clears_claim():
+ from second_brain.database import Database, reset_database_singleton
+ from second_brain.models import Source, SourceStatus, utcnow
+ from second_brain.sources_service import reset_source_for_retry
+
+ reset_database_singleton()
+ db = Database()
+ sid, _ = _seed_source(
+ db,
+ status=SourceStatus.FAILED,
+ error_message="boom",
+ claimed_by="worker-x",
+ claimed_at=utcnow(),
+ )
+ try:
+ with db.session() as sess:
+ reset_source_for_retry(sess, source_id=sid, mode="full")
+ with db.session() as sess:
+ row = sess.query(Source).filter(Source.id == sid).first()
+ assert row.status == SourceStatus.PENDING
+ assert row.claimed_by is None
+ assert row.claimed_at is None
+ assert row.error_message is None
+ finally:
+ _cleanup(db, sid)
+ reset_database_singleton()
+
+
+def test_retry_extract_sets_transcribed_when_transcript_present():
+ from second_brain.database import Database, reset_database_singleton
+ from second_brain.models import Source, SourceStatus
+ from second_brain.sources_service import reset_source_for_retry
+
+ reset_database_singleton()
+ db = Database()
+ sid, _ = _seed_source(
+ db,
+ status=SourceStatus.ANALYZED,
+ transcript_text="word " * 50,
+ error_message="prior failure",
+ )
+ try:
+ with db.session() as sess:
+ reset_source_for_retry(sess, source_id=sid, mode="extract")
+ with db.session() as sess:
+ row = sess.query(Source).filter(Source.id == sid).first()
+ assert row.status == SourceStatus.TRANSCRIBED
+ assert row.error_message is None
+ assert row.transcript_text # transcript preserved
+ finally:
+ _cleanup(db, sid)
+ reset_database_singleton()
+
+
+def test_retry_extract_without_transcript_raises():
+ from second_brain.database import Database, reset_database_singleton
+ from second_brain.models import SourceStatus
+ from second_brain.sources_service import reset_source_for_retry
+
+ reset_database_singleton()
+ db = Database()
+ sid, _ = _seed_source(db, status=SourceStatus.PENDING, transcript_text=None)
+ try:
+ with db.session() as sess:
+ with pytest.raises(ValueError, match="requires an existing transcript"):
+ reset_source_for_retry(sess, source_id=sid, mode="extract")
+ finally:
+ _cleanup(db, sid)
+ reset_database_singleton()
+
+
+def test_retry_unknown_mode_raises():
+ from second_brain.database import Database, reset_database_singleton
+ from second_brain.sources_service import reset_source_for_retry
+
+ reset_database_singleton()
+ db = Database()
+ sid, _ = _seed_source(db)
+ try:
+ with db.session() as sess:
+ with pytest.raises(ValueError, match="unknown retry mode"):
+ reset_source_for_retry(sess, source_id=sid, mode="bogus")
+ finally:
+ _cleanup(db, sid)
+ reset_database_singleton()
+
+
+def test_retry_missing_source_raises_lookup():
+ from second_brain.database import Database, reset_database_singleton
+ from second_brain.sources_service import reset_source_for_retry
+
+ reset_database_singleton()
+ db = Database()
+ try:
+ with db.session() as sess:
+ with pytest.raises(LookupError):
+ reset_source_for_retry(sess, source_id=-99999, mode="full")
+ finally:
+ reset_database_singleton()