upbeatBytes/tests/test_publishing_api.py

"""Publishing Desk Phase 1 — admin API: gating, background build (deterministic
fallback), lifecycle enforcement, snooze validation, draft preservation, restore."""
from datetime import datetime, timedelta, timezone

import pytest
from fastapi.testclient import TestClient

from goodnews.db import connect, init_db


def _future(hours: int = 24) -> str:
    return (datetime.now(timezone.utc) + timedelta(hours=hours)).strftime("%Y-%m-%d %H:%M:%S")


def _recent() -> str:
    return (datetime.now(timezone.utc) - timedelta(hours=1)).strftime("%Y-%m-%d %H:%M:%S")


@pytest.fixture
def api_app(tmp_path, monkeypatch):
    db = tmp_path / "t.sqlite3"
    monkeypatch.setenv("GOODNEWS_DB", str(db))
    # http (not https) so the session cookie isn't Secure-only — TestClient runs over http
    monkeypatch.setenv("GOODNEWS_PUBLIC_BASE_URL", "http://testserver")
    monkeypatch.setenv("GOODNEWS_ADMIN_EMAILS", "admin@b.com")
    monkeypatch.setenv("GOODNEWS_LLM_BASE_URL", "http://127.0.0.1:9")  # dead → deterministic fallback, fast
    import importlib
    import goodnews.api as api
    importlib.reload(api)
    c = connect(str(db)); init_db(c)
    # one eligible article (accepted, visible, complete summary, recent, readable)
    c.execute("INSERT INTO sources (id,name,feed_url,trust_score,content_visible) VALUES (1,'S','http://s/f',5,1)")
    c.execute("INSERT INTO articles (id,source_id,canonical_url,title,url_hash,image_url,published_at) "
              "VALUES (1,1,'https://ex.com/a','Title','h1','http://img/x.jpg',?)", (_recent(),))
    c.execute("INSERT INTO article_scores (article_id,accepted,novelty_score,constructive_score,topic,reason_code) "
              "VALUES (1,1,7,7,'science','ok')")
    c.execute("INSERT INTO article_summaries (article_id,summary,what_happened,why_matters,why_belongs) "
              "VALUES (1,'Sum','wh','wm','wb')")
    c.commit(); c.close()
    return api.create_app()


def _admin(app):
    tc = TestClient(app)
    sent = {}
    import goodnews.email_send as es
    orig = es.send_magic_link
    es.send_magic_link = lambda to, link: sent.update(link=link)
    try:
        tc.post("/api/auth/email/start", json={"email": "admin@b.com"})
        tc.post("/api/auth/email/verify", json={"token": sent["link"].split("token=")[1]})
    finally:
        es.send_magic_link = orig
    return tc


def test_admin_gating(api_app):
    anon = TestClient(api_app)
    assert anon.get("/api/admin/publishing/queue").status_code == 401
    assert anon.post("/api/admin/publishing/build").status_code == 401


def test_build_then_queue_deterministic(api_app):
    tc = _admin(api_app)
    assert tc.post("/api/admin/publishing/build").json() == {"building": True}
    # TestClient runs the background task before returning; LLM URL is dead → fallback.
    q = tc.get("/api/admin/publishing/queue").json()
    assert q["building"] is False and q["last"]["ranked_by"] == "deterministic"
    assert len(q["items"]) == 1 and q["items"][0]["article_id"] == 1
    assert "utm_source=x" in q["items"][0]["share_url"]
    # a second build is a no-op (already full) — never duplicates
    tc.post("/api/admin/publishing/build")
    assert len(tc.get("/api/admin/publishing/queue").json()["items"]) == 1


def _one(tc):
    tc.post("/api/admin/publishing/build")
    return tc.get("/api/admin/publishing/queue").json()["items"][0]["id"]


def test_invalid_transition_rejected(api_app):
    tc = _admin(api_app)
    sid = _one(tc)
    assert tc.post(f"/api/admin/publishing/{sid}/status", json={"status": "posted"}).status_code == 200
    # posted is terminal — resurrection refused
    assert tc.post(f"/api/admin/publishing/{sid}/status", json={"status": "queued"}).status_code == 400


def test_snooze_validation(api_app):
    tc = _admin(api_app)
    sid = _one(tc)
    assert tc.post(f"/api/admin/publishing/{sid}/status", json={"status": "snoozed"}).status_code == 400  # null
    assert tc.post(f"/api/admin/publishing/{sid}/status",
                   json={"status": "snoozed", "snooze_until": "2000-01-01 00:00:00"}).status_code == 400  # past
    assert tc.post(f"/api/admin/publishing/{sid}/status",
                   json={"status": "snoozed", "snooze_until": _future()}).status_code == 200


def test_draft_preserved_through_skip_and_restore(api_app):
    tc = _admin(api_app)
    sid = _one(tc)
    assert tc.post(f"/api/admin/publishing/{sid}/draft", json={"draft_text": "my blurb"}).status_code == 200
    assert tc.post(f"/api/admin/publishing/{sid}/status", json={"status": "skipped"}).status_code == 200
    assert sid not in {i["id"] for i in tc.get("/api/admin/publishing/queue").json()["items"]}  # left the queue
    assert tc.post(f"/api/admin/publishing/{sid}/restore").status_code == 200
    items = tc.get("/api/admin/publishing/queue").json()["items"]
    back = next(i for i in items if i["id"] == sid)
    assert back["draft_text"] == "my blurb"     # work survived skip→restore


def test_save_handle_validates(api_app):
    tc = _admin(api_app)
    assert tc.post("/api/admin/publishing/handles",
                   json={"entity_name": "NASA", "handle": "@not a handle"}).status_code == 400
    assert tc.post("/api/admin/publishing/handles",
                   json={"entity_name": "NASA", "handle": "https://x.com/NASA"}).status_code == 400
    assert tc.post("/api/admin/publishing/handles",
                   json={"entity_name": "NASA", "handle": "@NASA"}).status_code == 200