thejayman77/upbeatBytes
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
cf5cbb33c011415edc50da2bc8d1ea62199f4696
upbeatBytes/tests/test_admin.py
T
thejayman77 cf5cbb33c0 Daily digest (opt-in) + finite "you're caught up" ending 
Reader-retention as ritual, not capture (Codex's framing). Opt-in calm morning
email of today's brief; the on-site twin is the finite end-of-feed nudge.

* Schema: users.digest_enabled + digest_unsub_token; digest_sends (dedupe +
  visibility). auth.get_user now returns the digest fields.
* goodnews/digest.py: build (dated calm subject, items w/ summary + "why it's
  here" + UB/source links + one-click unsubscribe, "you're caught up" sign-off)
  and send_due_digests (morning-window gated, >=4-item floor or skip quietly,
  deduped, reuses SMTP). No streaks/urgency/"you missed".
* API: /auth/me exposes digest_enabled; POST /api/account/digest toggle;
  GET /api/digest/unsubscribe (token, no login, calm confirmation page).
* CLI: cycle gains a morning-gated digest step (--no-digest) + a send-digests
  command (--force).
* Frontend: digest toggle on the Account profile; the Highlights end-cap now
  says "you're caught up — see you tomorrow" with a one-tap "Get tomorrow's
  brief by email" (signed-in → enable; anon → sign in).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-09 16:17:46 -04:00

234 lines
12 KiB
Python
Raw Blame History

import pytest
from fastapi.testclient import TestClient
def _make(tmp_path, monkeypatch, admin_email=""):
db = tmp_path / "t.sqlite3"
monkeypatch.setenv("GOODNEWS_DB", str(db))
monkeypatch.setenv("GOODNEWS_PUBLIC_BASE_URL", "http://testserver")
monkeypatch.setenv("GOODNEWS_ADMIN_EMAILS", admin_email)
import importlib
import goodnews.api as api
importlib.reload(api)
from goodnews.db import connect, init_db
c = connect(str(db)); init_db(c)
c.execute("INSERT INTO sources (id,name,feed_url,trust_score) VALUES (1,'S','http://s/f',5)")
c.execute("INSERT INTO articles (id,source_id,canonical_url,title,url_hash) VALUES (1,1,'http://s/1','t1','h1')")
c.execute("INSERT INTO article_scores (article_id,accepted,topic) VALUES (1,1,'science')")
c.execute("INSERT INTO article_tags (article_id,tag) VALUES (1,'science')")
c.commit(); c.close()
return api.create_app(), api
def _signin(app, api, email):
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": email})
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(tmp_path, monkeypatch):
app, api = _make(tmp_path, monkeypatch, admin_email="boss@x.com")
assert TestClient(app).get("/api/admin/stats").status_code == 401 # anon
nonadmin = _signin(app, api, "rando@x.com")
assert nonadmin.get("/api/admin/stats").status_code == 403 # signed in, not admin
assert nonadmin.get("/api/auth/me").json()["is_admin"] is False
admin = _signin(app, api, "Boss@X.com") # case-insensitive match
assert admin.get("/api/auth/me").json()["is_admin"] is True
assert admin.get("/api/admin/stats").status_code == 200
def test_admin_stats_shape(tmp_path, monkeypatch):
app, api = _make(tmp_path, monkeypatch, admin_email="boss@x.com")
admin = _signin(app, api, "boss@x.com")
# log a couple of events
admin.post("/api/events", json={"kind": "visit", "visitor": "v1"})
admin.post("/api/events", json={"kind": "open", "article_id": 1, "visitor": "v1"})
stats = admin.get("/api/admin/stats").json()
assert set(stats) >= {"visitors", "returning", "once", "top_articles", "top_groupings", "top_topics", "shares", "daily"}
assert stats["top_articles"][0]["id"] == 1 and stats["top_articles"][0]["opens"] == 1
assert any(g["tag"] == "science" for g in stats["top_groupings"])
def _src(tc, sid=1):
return next(s for s in tc.get("/api/admin/stats").json()["sources"] if s["id"] == sid)
def test_source_lifecycle_status_and_visibility(tmp_path, monkeypatch):
app, api = _make(tmp_path, monkeypatch, admin_email="boss@x.com")
tc = _signin(app, api, "boss@x.com")
# pause → status paused + active mirror 0
assert tc.post("/api/admin/sources/1/status", json={"status": "paused"}).json()["active"] == 0
assert _src(tc)["status"] == "paused" and _src(tc)["active"] == 0
# retire → status retired, still active=0, articles stay visible
tc.post("/api/admin/sources/1/status", json={"status": "retired"})
assert _src(tc)["status"] == "retired" and _src(tc)["active"] == 0
assert _src(tc)["content_visible"] == 1 # retire does NOT hide content
# restore → active again, mirror 1
tc.post("/api/admin/sources/1/status", json={"status": "active"})
assert _src(tc)["status"] == "active" and _src(tc)["active"] == 1
# hide content → feed excludes it; show → back
tc.post("/api/admin/sources/1/visibility", json={"visible": False})
assert _src(tc)["content_visible"] == 0
from goodnews import queries
import sqlite3, os
c = sqlite3.connect(os.environ["GOODNEWS_DB"]); c.row_factory = sqlite3.Row
assert queries.feed(c) == [] # hidden source's article drops out of the feed
tc.post("/api/admin/sources/1/visibility", json={"visible": True})
c2 = sqlite3.connect(os.environ["GOODNEWS_DB"]); c2.row_factory = sqlite3.Row
assert len(queries.feed(c2)) == 1 # back in the feed
# validation + 404
assert tc.post("/api/admin/sources/1/status", json={"status": "bogus"}).status_code == 422
assert tc.post("/api/admin/sources/999/status", json={"status": "paused"}).status_code == 404
def test_source_flag_and_gating(tmp_path, monkeypatch):
app, api = _make(tmp_path, monkeypatch, admin_email="boss@x.com")
tc = _signin(app, api, "boss@x.com")
tc.post("/api/admin/sources/1/review", json={"flag": True, "reason": "spammy lately"})
assert _src(tc)["review_flag"] == 1 and _src(tc)["review_reason"] == "spammy lately"
tc.post("/api/admin/sources/1/review", json={"flag": False})
assert _src(tc)["review_flag"] == 0 and _src(tc)["review_reason"] is None
anon = TestClient(app)
assert anon.post("/api/admin/sources/1/status", json={"status": "paused"}).status_code == 401
assert anon.post("/api/admin/sources/1/visibility", json={"visible": False}).status_code == 401
assert anon.post("/api/admin/sources/1/review", json={"flag": True}).status_code == 401
def test_source_health_includes_metrics(tmp_path, monkeypatch):
import sqlite3
from goodnews import queries
app, api = _make(tmp_path, monkeypatch, admin_email="boss@x.com")
import os
c = sqlite3.connect(os.environ["GOODNEWS_DB"]); c.row_factory = sqlite3.Row
sh = queries.source_health(c)
s = sh[0]
for key in ("active", "served", "accepted_total", "total_articles", "duplicates",
"acceptance_rate", "duplicate_rate", "review_reason", "next_due_at"):
assert key in s, f"missing {key}"
assert s["served"] == 1 and s["acceptance_rate"] == 100
def test_admin_stats_days_param_clamped(tmp_path, monkeypatch):
app, api = _make(tmp_path, monkeypatch, admin_email="boss@x.com")
tc = _signin(app, api, "boss@x.com")
assert tc.get("/api/admin/stats?days=7").json()["days"] == 7
assert tc.get("/api/admin/stats?days=90").json()["days"] == 90
assert tc.get("/api/admin/stats?days=999").json()["days"] == 30 # clamped
assert tc.get("/api/admin/stats").json()["days"] == 30 # default
def test_candidate_suggest_promote_paused(tmp_path, monkeypatch):
app, api = _make(tmp_path, monkeypatch, admin_email="boss@x.com")
monkeypatch.setattr(api.feeds, "preview_feed",
lambda url, **k: {"url": url, "sampled": 5, "accepted": 4, "examples_accepted": ["A", "B"]})
tc = _signin(app, api, "boss@x.com")
cand = tc.post("/api/admin/candidates", json={"feed_url": "http://good/feed", "name": "Good Feed"}).json()
assert cand["status"] == "suggested" and cand["preview"]["accepted"] == 4
cid = cand["id"]
assert any(c["id"] == cid for c in tc.get("/api/admin/candidates").json())
# promote defaults to paused (active-on-approval off) — no mirror drift
res = tc.post(f"/api/admin/candidates/{cid}/promote", json={}).json()
assert res["source"]["status"] == "paused" and res["source"]["active"] == 0
assert res["candidate"]["status"] == "promoted"
assert any(s["name"] == "Good Feed" for s in tc.get("/api/admin/stats").json()["sources"])
def test_candidate_reject_and_gating(tmp_path, monkeypatch):
app, api = _make(tmp_path, monkeypatch, admin_email="boss@x.com")
monkeypatch.setattr(api.feeds, "preview_feed", lambda url, **k: {"url": url, "sampled": 1, "accepted": 0})
tc = _signin(app, api, "boss@x.com")
cand = tc.post("/api/admin/candidates", json={"feed_url": "http://meh/feed"}).json()
assert tc.post(f"/api/admin/candidates/{cand['id']}/reject").json()["status"] == "rejected"
anon = TestClient(app)
assert anon.get("/api/admin/candidates").status_code == 401
assert anon.post("/api/admin/candidates", json={"feed_url": "http://x/f"}).status_code == 401
assert anon.post("/api/admin/candidates/1/promote", json={}).status_code == 401
def test_safe_fetch_feed_blocks_ssrf():
import pytest
from goodnews.feeds import safe_fetch_feed
for bad in ("http://127.0.0.1/x", "http://localhost/x", "file:///etc/passwd",
"http://169.254.169.254/latest", "ftp://x/y"):
with pytest.raises(RuntimeError):
safe_fetch_feed(bad, timeout=2)
def test_export_sources_csv(tmp_path, monkeypatch):
app, api = _make(tmp_path, monkeypatch, admin_email="boss@x.com")
tc = _signin(app, api, "boss@x.com")
r = tc.get("/api/admin/export/sources.csv")
assert r.status_code == 200 and r.headers["content-type"].startswith("text/csv")
assert 'attachment; filename="sources.csv"' in r.headers["content-disposition"]
lines = r.text.splitlines()
assert lines[0].startswith("name,feed_url,homepage,status,visible,served")
assert any("http://s/f" in ln for ln in lines[1:]) # the seeded source row
assert TestClient(app).get("/api/admin/export/sources.csv").status_code == 401 # gated
def test_export_audience_csv(tmp_path, monkeypatch):
app, api = _make(tmp_path, monkeypatch, admin_email="boss@x.com")
tc = _signin(app, api, "boss@x.com")
tc.post("/api/events", json={"kind": "visit", "visitor": "v1"})
r = tc.get("/api/admin/export/audience.csv?days=7")
assert r.status_code == 200 and r.headers["content-type"].startswith("text/csv")
body = r.text
assert "metric,value" in body and "window_days,7" in body
assert "date,visitors,opens" in body # daily time-series section
assert TestClient(app).get("/api/admin/export/audience.csv").status_code == 401 # gated
def test_export_sources_csv_escapes_formula_injection(tmp_path, monkeypatch):
import os, sqlite3
app, api = _make(tmp_path, monkeypatch, admin_email="boss@x.com")
c = sqlite3.connect(os.environ["GOODNEWS_DB"])
c.execute("UPDATE sources SET name = ?, review_flag = 1, review_reason = ? WHERE id = 1",
('=HYPERLINK("http://bad")', '+danger'))
c.commit(); c.close()
tc = _signin(app, api, "boss@x.com")
body = tc.get("/api/admin/export/sources.csv").text
assert "'=HYPERLINK" in body # leading apostrophe defuses the formula (CSV may quote the cell)
assert "'+danger" in body
assert ",=HYPERLINK" not in body # never written as a bare, evaluable formula cell
def test_source_check_preview_is_readonly(tmp_path, monkeypatch):
app, api = _make(tmp_path, monkeypatch, admin_email="boss@x.com")
monkeypatch.setattr(api.feeds, "preview_feed", lambda url, **k: {"url": url, "sampled": 8, "accepted": 6})
tc = _signin(app, api, "boss@x.com")
before = _src(tc)
r = tc.post("/api/admin/sources/1/preview").json()
assert r["sampled"] == 8 and r["accepted"] == 6
after = _src(tc)
# read-only: no state/health change, no poll attempt recorded
assert after["status"] == before["status"] and after["served"] == before["served"]
assert after["last_success_at"] == before["last_success_at"] and after["next_due_at"] == before["next_due_at"]
assert TestClient(app).post("/api/admin/sources/1/preview").status_code == 401 # gated
assert tc.post("/api/admin/sources/999/preview").status_code == 404
def test_digest_toggle_and_unsubscribe(tmp_path, monkeypatch):
import os, sqlite3
app, api = _make(tmp_path, monkeypatch, admin_email="boss@x.com")
tc = _signin(app, api, "reader@x.com")
assert tc.get("/api/auth/me").json()["digest_enabled"] is False
assert tc.post("/api/account/digest", json={"enabled": True}).json()["digest_enabled"] is True
assert tc.get("/api/auth/me").json()["digest_enabled"] is True
c = sqlite3.connect(os.environ["GOODNEWS_DB"])
uid, tok = c.execute("SELECT id, digest_unsub_token FROM users WHERE email='reader@x.com'").fetchone()
c.close()
assert tok # token generated on opt-in
# one-click unsubscribe: wrong token is rejected, right token disables
assert "invalid" in TestClient(app).get(f"/api/digest/unsubscribe?u={uid}&t=nope").text.lower()
assert "unsubscribed" in TestClient(app).get(f"/api/digest/unsubscribe?u={uid}&t={tok}").text.lower()
assert tc.get("/api/auth/me").json()["digest_enabled"] is False
assert TestClient(app).post("/api/account/digest", json={"enabled": True}).status_code == 401 # gated
Reference in New Issue View Git Blame Copy Permalink