"""Delivery: ntfy actions, the feedback endpoint, and the digest page. The feedback endpoint is the most important route in the system -- it is the only way the taste model ever learns. The old system collected 32 labels in a week because it did not work. """ from __future__ import annotations import os from datetime import UTC, datetime import pytest from starlette.testclient import TestClient from news import store from news.digest import _extract_json_object, push_summary from news.notify import NtfyConfig, build_actions, missing_button_reason, sign, verify SECRET = "test-secret" # --------------------------------------------------------------- signing ---- def test_token_is_scoped_to_one_cluster_and_one_kind(): """A leaked URL must authorise one vote on one story, not act as a bearer token for the whole model.""" t = sign(SECRET, "c1", "interested") assert verify(SECRET, "c1", "interested", t) assert not verify(SECRET, "c1", "not_interested", t) assert not verify(SECRET, "c2", "interested", t) assert not verify("other-secret", "c1", "interested", t) def test_verify_rejects_empty_and_none_tokens(): assert not verify(SECRET, "c1", "interested", "") assert not verify(SECRET, "c1", "interested", None) # ---------------------------------------------------------------- actions --- def test_buttons_are_built_when_the_api_is_reachable_and_signed(): cfg = NtfyConfig(url="https://n", topic="t", api_base="https://news.example", secret=SECRET) actions = build_actions(cfg, "c1", "https://story") assert actions[0].startswith("view, Read, https://story") assert any("More like this" in a and "/interested/" in a for a in actions) assert any("Less" in a and "/not_interested/" in a for a in actions) assert missing_button_reason(cfg) is None @pytest.mark.parametrize("cfg,needle", [ (NtfyConfig(url="https://n", topic="t", secret=SECRET), "NEWS_API_BASE_URL"), (NtfyConfig(url="https://n", topic="t", api_base="https://x"), "NEWS_FEEDBACK_SECRET"), ]) def test_buttons_are_dropped_loudly_not_silently(cfg, needle): """v0.2 dropped them behind one boot WARN and nobody noticed for a week.""" assert build_actions(cfg, "c1", "https://story") == ["view, Read, https://story"] assert needle in missing_button_reason(cfg) # ----------------------------------------------------------------- digest --- def test_push_summary_is_short_and_points_elsewhere(): """The push must never carry the digest: ntfy turns a long message into a .md attachment, which is what made the old report unreadable on a phone.""" stories = [{"headline": "Lead story", "topic": "urbanism"}] + [ {"headline": f"s{i}", "topic": "space-aerospace"} for i in range(20)] title, body = push_summary(stories) assert title == "Daily briefing" assert body.count("\n") <= 2 assert len(body) < 300 assert "20 more" in body def test_push_summary_handles_an_empty_day(): title, body = push_summary([]) assert "No digest" in title @pytest.mark.parametrize("raw", [ '{"briefing":"hello"}', '```json\n{"briefing":"hello"}\n```', 'Sure! Here you go: {"briefing":"hello"} — hope that helps', ]) def test_json_survives_code_fences_and_prose(raw): assert _extract_json_object(raw)["briefing"] == "hello" def test_malformed_model_output_raises_rather_than_returning_garbage(): with pytest.raises(ValueError): _extract_json_object("no object here") # -------------------------------------------------------------------- web --- @pytest.fixture def client(tmp_path, monkeypatch): monkeypatch.setenv("NEWS_DB", str(tmp_path / "w.db")) monkeypatch.setenv("NEWS_FEEDBACK_SECRET", SECRET) con = store.connect(tmp_path / "w.db") store.upsert_source(con, id="s1", name="BBC", url="http://b", kind="rss", bucket="bbc", weight=1.0, prominence=0, enabled=1, poll_interval_secs=900) now = datetime.now(UTC) con.execute("""INSERT INTO items (id,source_id,title,summary,url,canonical_url, published_at,first_seen_at,content_hash) VALUES ('i1','s1','Council votes on transit','','http://b/1','http://b/1',?,?,'h')""", (now.isoformat(), now.isoformat())) con.execute("""INSERT INTO clusters (id,centroid,size,first_seen_at,last_seen_at,lead_item_id) VALUES ('c1',?,1,?,?,'i1')""", (store.to_blob(__import__("numpy").zeros(3, dtype="float32")), now.isoformat(), now.isoformat())) con.execute("INSERT INTO cluster_members VALUES ('c1','i1',?)", (now.isoformat(),)) con.execute("""INSERT INTO notification_log (cluster_id,lane,score,created_at) VALUES ('c1','alert:interest',3.0,?)""", (now.isoformat(),)) con.execute("""INSERT INTO digest_prose (day,briefing,model,created_at) VALUES (?,'The council met.','m',?)""", (now.date().isoformat(), now.isoformat())) from news import web return TestClient(web.app), now def test_feedback_button_records_a_label(client): c, _ = client token = sign(SECRET, "c1", "interested") r = c.post(f"/api/feedback/c1/interested/{token}") assert r.status_code == 200 and r.json()["ok"] is True con = store.connect(os.environ["NEWS_DB"]) row = con.execute("SELECT cluster_id,kind,source FROM feedback").fetchone() assert (row["cluster_id"], row["kind"], row["source"]) == ("c1", "interested", "ntfy") def test_feedback_rejects_a_forged_token(client): c, _ = client assert c.post("/api/feedback/c1/interested/deadbeef").status_code == 403 con = store.connect(os.environ["NEWS_DB"]) assert con.execute("SELECT COUNT(*) n FROM feedback").fetchone()["n"] == 0 def test_feedback_rejects_an_unknown_kind(client): c, _ = client token = sign(SECRET, "c1", "sideways") assert c.post(f"/api/feedback/c1/sideways/{token}").status_code == 400 def test_digest_page_renders_stories_and_briefing(client): c, now = client r = c.get(f"/digest/{now.date().isoformat()}") assert r.status_code == 200 assert "Council votes on transit" in r.text assert "The council met." in r.text assert "BBC" in r.text def test_digest_page_handles_a_day_with_nothing(client): c, _ = client r = c.get("/digest/2020-01-01") assert r.status_code == 200 and "Nothing cleared the bar" in r.text def test_bad_date_is_rejected(client): c, _ = client assert c.get("/digest/not-a-date").status_code == 400