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