5131f3ccf6
Adds per-record edit mode and retry to the source detail page, plus a
CLI `second-brain retry <id>` for parity.
Edit mode exposes domain (select), title, focus on /sources/{id}.
URL stays read-only — it's the UNIQUE dedupe key and the embeddings
identity tuple. Save goes through a new service helper
`update_source_metadata` so the web route stays a thin wrapper.
Retry exposes two modes, both available on any record:
- full: status -> PENDING (re-pull + re-process)
- extract: status -> TRANSCRIBED (re-extract only, requires an
existing transcript; offered when source_type=video and a
transcript is present, to avoid the expensive re-download path)
Both modes clear claimed_by/claimed_at and error_message so the queue
dance picks it up cleanly.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
259 lines
8.3 KiB
Python
259 lines
8.3 KiB
Python
"""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()
|