thejayman77
84fd61bf3f
Reply to a reader from the admin inbox instead of a mailto. Per Codex: keep v1
plain text (no rich editor — defers the user's bold/bullets ask as a fast-follow).
* DB: feedback_replies table (feedback_id, user_id, message, sent_to, sent_at),
created on the live DB.
* email_send.send_feedback_reply: plain-text "Re: Your Upbeat Bytes feedback"
with a quoted context block, no analytics/account details.
* API: POST /api/admin/feedback/{id}/reply — admin-gated, requires the feedback
exists (404) and has a contact_email (400), trims+caps the message; sends via
SMTP and only records the reply on success (502 on send failure so the UI keeps
the draft); marks the item read. Feedback list now includes each item's replies.
* Frontend: inline composer (Send/Cancel, sending state, error keeps draft) +
reply thread under the message; Reply only shows when there's an address,
else "No reply address".
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
129 lines
6.5 KiB
Python
129 lines
6.5 KiB
Python
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
|
|
def _app(tmp_path, monkeypatch, admin=""):
|
|
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)
|
|
import importlib
|
|
import goodnews.api as api
|
|
importlib.reload(api)
|
|
from goodnews.db import connect, init_db
|
|
connect(str(db)).close()
|
|
c = connect(str(db)); init_db(c); c.close()
|
|
return api.create_app(), api, db
|
|
|
|
|
|
def _count(db):
|
|
from goodnews.db import connect
|
|
c = connect(str(db)); n = c.execute("SELECT COUNT(*) FROM feedback").fetchone()[0]; c.close()
|
|
return n
|
|
|
|
|
|
def test_feedback_stored_and_validated(tmp_path, monkeypatch):
|
|
app, api, db = _app(tmp_path, monkeypatch)
|
|
monkeypatch.setattr(api.email_send, "send_feedback", lambda *a, **k: None)
|
|
tc = TestClient(app)
|
|
assert tc.post("/api/feedback", json={"message": " "}).status_code == 422 # empty
|
|
assert tc.post("/api/feedback", json={"category": "idea", "message": "Love it", "visitor": "v"}).json() == {"ok": True}
|
|
assert _count(db) == 1
|
|
# honeypot tripped → accepted but NOT stored
|
|
tc.post("/api/feedback", json={"message": "spam", "hp": "i am a bot", "visitor": "v"})
|
|
assert _count(db) == 1
|
|
|
|
|
|
def test_admin_feedback_gated(tmp_path, monkeypatch):
|
|
app, api, db = _app(tmp_path, monkeypatch, admin="boss@x.com")
|
|
monkeypatch.setattr(api.email_send, "send_feedback", lambda *a, **k: None)
|
|
TestClient(app).post("/api/feedback", json={"message": "hi", "visitor": "v"})
|
|
assert TestClient(app).get("/api/admin/feedback").status_code == 401 # anon
|
|
|
|
# sign in as admin
|
|
tc = TestClient(app); sent = {}
|
|
monkeypatch.setattr(api.email_send, "send_magic_link", lambda to, link: sent.update(link=link))
|
|
tc.post("/api/auth/email/start", json={"email": "boss@x.com"})
|
|
tc.post("/api/auth/email/verify", json={"token": sent["link"].split("token=")[1]})
|
|
fb = tc.get("/api/admin/feedback").json()
|
|
assert len(fb) == 1 and fb[0]["message"] == "hi"
|
|
stats = tc.get("/api/admin/stats").json()
|
|
assert {"funnel", "accounts", "retention", "emotional_mix", "paywall", "replace"} <= set(stats)
|
|
assert stats["accounts"]["total"] >= 1
|
|
|
|
|
|
def _admin_client(tmp_path, monkeypatch):
|
|
app, api, db = _app(tmp_path, monkeypatch, admin="boss@x.com")
|
|
monkeypatch.setattr(api.email_send, "send_feedback", lambda *a, **k: None)
|
|
TestClient(app).post("/api/feedback", json={"message": "hello there", "visitor": "v"})
|
|
tc = TestClient(app); sent = {}
|
|
monkeypatch.setattr(api.email_send, "send_magic_link", lambda to, link: sent.update(link=link))
|
|
tc.post("/api/auth/email/start", json={"email": "boss@x.com"})
|
|
tc.post("/api/auth/email/verify", json={"token": sent["link"].split("token=")[1]})
|
|
return tc
|
|
|
|
|
|
def test_feedback_read_unread_and_delete(tmp_path, monkeypatch):
|
|
tc = _admin_client(tmp_path, monkeypatch)
|
|
fb = tc.get("/api/admin/feedback").json()
|
|
fid = fb[0]["id"]
|
|
assert fb[0]["read_at"] is None # starts unread
|
|
assert tc.get("/api/admin/stats").json()["feedback_unread"] == 1
|
|
|
|
tc.post(f"/api/admin/feedback/{fid}/read", json={"read": True})
|
|
assert tc.get("/api/admin/feedback").json()[0]["read_at"] is not None
|
|
assert tc.get("/api/admin/stats").json()["feedback_unread"] == 0
|
|
|
|
tc.post(f"/api/admin/feedback/{fid}/read", json={"read": False}) # back to unread
|
|
assert tc.get("/api/admin/stats").json()["feedback_unread"] == 1
|
|
|
|
assert tc.delete(f"/api/admin/feedback/{fid}").json() == {"ok": True}
|
|
assert tc.get("/api/admin/feedback").json() == []
|
|
|
|
|
|
def test_feedback_admin_actions_gated(tmp_path, monkeypatch):
|
|
app, api, db = _app(tmp_path, monkeypatch, admin="boss@x.com")
|
|
monkeypatch.setattr(api.email_send, "send_feedback", lambda *a, **k: None)
|
|
TestClient(app).post("/api/feedback", json={"message": "hi", "visitor": "v"})
|
|
anon = TestClient(app)
|
|
assert anon.post("/api/admin/feedback/1/read", json={"read": True}).status_code == 401
|
|
assert anon.delete("/api/admin/feedback/1").status_code == 401
|
|
assert anon.post("/api/admin/feedback/1/reply", json={"message": "hi"}).status_code == 401
|
|
|
|
|
|
def test_feedback_actions_404_on_missing_id(tmp_path, monkeypatch):
|
|
tc = _admin_client(tmp_path, monkeypatch)
|
|
assert tc.post("/api/admin/feedback/9999/read", json={"read": True}).status_code == 404
|
|
assert tc.delete("/api/admin/feedback/9999").status_code == 404
|
|
|
|
|
|
def test_feedback_reply_sends_stores_marks_read(tmp_path, monkeypatch):
|
|
app, api, db = _app(tmp_path, monkeypatch, admin="boss@x.com")
|
|
monkeypatch.setattr(api.email_send, "send_feedback", lambda *a, **k: None)
|
|
sent = {}
|
|
monkeypatch.setattr(api.email_send, "send_feedback_reply",
|
|
lambda to, msg, orig: sent.update(to=to, msg=msg, orig=orig))
|
|
TestClient(app).post("/api/feedback", json={"message": "is there an app?", "email": "reader@x.com", "visitor": "v"})
|
|
tc = TestClient(app); link = {}
|
|
monkeypatch.setattr(api.email_send, "send_magic_link", lambda to, l: link.update(l=l))
|
|
tc.post("/api/auth/email/start", json={"email": "boss@x.com"})
|
|
tc.post("/api/auth/email/verify", json={"token": link["l"].split("token=")[1]})
|
|
fid = tc.get("/api/admin/feedback").json()[0]["id"]
|
|
r = tc.post(f"/api/admin/feedback/{fid}/reply", json={"message": "Yes — a companion app is coming."})
|
|
assert r.status_code == 200
|
|
assert sent["to"] == "reader@x.com" and "companion app" in sent["msg"] and "is there an app?" in sent["orig"]
|
|
fb = tc.get("/api/admin/feedback").json()[0]
|
|
assert fb["read_at"] is not None # marked read on reply
|
|
assert len(fb["replies"]) == 1 and fb["replies"][0]["sent_to"] == "reader@x.com"
|
|
|
|
|
|
def test_feedback_reply_requires_address_and_message(tmp_path, monkeypatch):
|
|
tc = _admin_client(tmp_path, monkeypatch) # posts anonymous feedback (no email)
|
|
monkeypatch.setattr("goodnews.email_send.send_feedback_reply",
|
|
lambda *a, **k: (_ for _ in ()).throw(AssertionError("should not send")))
|
|
fid = tc.get("/api/admin/feedback").json()[0]["id"]
|
|
assert tc.post(f"/api/admin/feedback/{fid}/reply", json={"message": ""}).status_code == 422
|
|
assert tc.post(f"/api/admin/feedback/{fid}/reply", json={"message": "hi"}).status_code == 400 # no address
|
|
assert tc.post("/api/admin/feedback/9999/reply", json={"message": "hi"}).status_code == 404
|
|
import goodnews.api as api # already reloaded by _admin_client's app
|