thejayman77
35aaeece6d
Per Codex: upsert_sources() wrote `active` but not `status`, so a candidate promoted inactive (the pipeline default) became active=0 + status='active' — the exact mirror drift Phase 1 set out to avoid (scheduler won't poll, admin UI shows "active"). Now derive status from an explicit value or from active, mirror active off status, and write both columns together (insert + conflict update). Test: promote_candidate(active=False) → status='paused', active=0. Also fix stale source_health docstring (now includes retired). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
64 lines
2.3 KiB
Python
64 lines
2.3 KiB
Python
import pytest
|
|
|
|
from goodnews.db import connect, init_db
|
|
from goodnews.sources import (
|
|
list_candidates,
|
|
promote_candidate,
|
|
reject_candidate,
|
|
save_candidate,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def conn():
|
|
c = connect(":memory:")
|
|
init_db(c)
|
|
yield c
|
|
c.close()
|
|
|
|
|
|
def test_save_and_list_candidate(conn):
|
|
cand = save_candidate(conn, "http://x/feed", preview={"acceptance_rate": 0.8, "sampled": 10}, name="X")
|
|
assert cand["status"] == "quarantined"
|
|
rows = list_candidates(conn)
|
|
assert len(rows) == 1 and rows[0]["feed_url"] == "http://x/feed"
|
|
|
|
|
|
def test_re_preview_preserves_curator_status(conn):
|
|
save_candidate(conn, "http://x/feed")
|
|
reject_candidate(conn, list_candidates(conn)[0]["id"])
|
|
# Re-previewing must NOT revive a rejected feed.
|
|
save_candidate(conn, "http://x/feed", preview={"acceptance_rate": 0.9})
|
|
assert list_candidates(conn)[0]["status"] == "rejected"
|
|
|
|
|
|
def test_promote_creates_inactive_source_and_marks_promoted(conn):
|
|
cand = save_candidate(conn, "http://x/feed", name="Lovely Feed")
|
|
source_id = promote_candidate(conn, cand["id"]) # inactive by default
|
|
|
|
src = conn.execute("SELECT name, active, status FROM sources WHERE id = ?", (source_id,)).fetchone()
|
|
assert src["name"] == "Lovely Feed"
|
|
assert src["active"] == 0 # active-on-approval: not polled until activated
|
|
assert src["status"] == "paused" # status mirrors active — no drift (active=0 ⇒ paused)
|
|
|
|
status = conn.execute("SELECT status FROM source_candidates WHERE id = ?", (cand["id"],)).fetchone()["status"]
|
|
assert status == "promoted"
|
|
|
|
|
|
def test_promote_active_flag(conn):
|
|
cand = save_candidate(conn, "http://y/feed", name="Y")
|
|
sid = promote_candidate(conn, cand["id"], active=True)
|
|
src = conn.execute("SELECT active, status FROM sources WHERE id = ?", (sid,)).fetchone()
|
|
assert src["active"] == 1 and src["status"] == "active" # both set together
|
|
|
|
|
|
def test_promote_unknown_raises(conn):
|
|
with pytest.raises(ValueError):
|
|
promote_candidate(conn, 999)
|
|
|
|
|
|
def test_name_derived_from_url_when_missing(conn):
|
|
cand = save_candidate(conn, "https://news.example.org/rss")
|
|
sid = promote_candidate(conn, cand["id"])
|
|
assert conn.execute("SELECT name FROM sources WHERE id = ?", (sid,)).fetchone()["name"] == "news.example.org"
|