connor
23a2f75299
A headline with a curly quote in the Title header made httpx raise UnicodeEncodeError, which failed news-cycle and dropped every alert after it in the run. Non-ASCII header values are now sent as UTF-8 encoded-words, which ntfy decodes.
733 lines
33 KiB
Python
733 lines
33 KiB
Python
"""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 json
|
|
import os
|
|
from datetime import UTC, datetime
|
|
|
|
import httpx
|
|
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, publish, 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_publish_sends_a_non_ascii_title():
|
|
# A headline with a curly quote raised UnicodeEncodeError inside httpx and
|
|
# blocked every alert behind it, 2026-10-18 to 2026-10-27.
|
|
sent = {}
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
sent.update(request.headers)
|
|
return httpx.Response(200)
|
|
|
|
cfg = NtfyConfig(url="http://ntfy", topic="news")
|
|
with httpx.Client(transport=httpx.MockTransport(handler)) as client:
|
|
publish(client, cfg, lane="interest", title="It\u2019s here", body="b",
|
|
link="https://example.com", cluster_id="c1")
|
|
assert sent["title"] == "=?UTF-8?B?SXTigJlzIGhlcmU=?="
|
|
|
|
|
|
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()))
|
|
# 384 dims: the triage page scores against the real profile probes, so a
|
|
# short centroid blows up in the matmul.
|
|
import numpy as np
|
|
con.execute("""INSERT INTO clusters (id,centroid,size,first_seen_at,last_seen_at,lead_item_id)
|
|
VALUES ('c1',?,1,?,?,'i1')""",
|
|
(store.to_blob(np.zeros(384, 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
|
|
web._VECTORS = None # module-level cache must not leak across tests
|
|
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
|
|
|
|
|
|
def test_feedback_survives_a_concurrent_writer(tmp_path, monkeypatch):
|
|
"""Regression: a feedback button pressed during a poll returned HTTP 500.
|
|
|
|
SQLite's default busy timeout is zero, so the second writer failed instantly
|
|
with "database is locked". Polls take minutes and run every 15, so this hit
|
|
the one endpoint that must never drop a write.
|
|
"""
|
|
import threading
|
|
import time
|
|
|
|
monkeypatch.setenv("NEWS_DB", str(tmp_path / "c.db"))
|
|
monkeypatch.setenv("NEWS_FEEDBACK_SECRET", SECRET)
|
|
store.connect(tmp_path / "c.db")
|
|
|
|
stop = threading.Event()
|
|
|
|
def hammer():
|
|
# Bursts with gaps, like a real poll cycle: it writes a batch, releases,
|
|
# writes the next. A thread that NEVER releases would defeat any finite
|
|
# timeout, which is a different failure and not the one being tested.
|
|
con = store.connect(tmp_path / "c.db")
|
|
while not stop.is_set():
|
|
con.execute("BEGIN IMMEDIATE")
|
|
con.execute(
|
|
"INSERT INTO digest_prose (day,briefing,model,created_at) VALUES (?,?,?,?)",
|
|
(str(time.time()), "x", "m", "t"))
|
|
time.sleep(0.4) # hold the write lock
|
|
con.execute("COMMIT")
|
|
time.sleep(0.1) # ... then let others through
|
|
|
|
t = threading.Thread(target=hammer, daemon=True)
|
|
t.start()
|
|
time.sleep(0.1)
|
|
try:
|
|
from news import web
|
|
client = TestClient(web.app)
|
|
token = sign(SECRET, "c9", "interested")
|
|
r = client.post(f"/api/feedback/c9/interested/{token}")
|
|
assert r.status_code == 200, f"feedback must not fail under write contention: {r.text}"
|
|
finally:
|
|
stop.set()
|
|
t.join(timeout=5)
|
|
|
|
|
|
def test_browser_labels_use_the_same_signed_public_endpoint(client):
|
|
"""The browser must not post to an Authelia-gated path.
|
|
|
|
fetch() follows redirects, so with an expired session a gated POST would go
|
|
302 -> login page -> HTTP 200, and the UI would report success having
|
|
recorded nothing. Both surfaces use the signed public endpoint instead.
|
|
"""
|
|
c, _ = client
|
|
html = c.get("/").text
|
|
assert "/api/feedback/" in html
|
|
assert "/ui/feedback/" not in html, "the gated browser route must be gone"
|
|
assert "redirect: 'error'" in html, "a redirect must fail loudly, not look like success"
|
|
assert html.count("data-yes=") > 0
|
|
|
|
|
|
def test_labelling_is_disabled_when_no_secret_is_configured(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("NEWS_DB", str(tmp_path / "n.db"))
|
|
monkeypatch.delenv("NEWS_FEEDBACK_SECRET", raising=False)
|
|
store.connect(tmp_path / "n.db")
|
|
from news import web
|
|
web._VECTORS = None
|
|
assert "labelling is disabled" in TestClient(web.app).get("/").text
|
|
|
|
|
|
@pytest.mark.parametrize("path,active", [
|
|
("/", "Triage"), ("/tune", "Tune"), ("/config", "Config"), ("/digest", "Digest")])
|
|
def test_every_page_carries_the_nav(client, path, active):
|
|
"""Navigation must not require knowing the URLs."""
|
|
c, _ = client
|
|
html = c.get(path).text
|
|
for target in ("/tune", "/config", "/digest"):
|
|
assert f'href="{target}"' in html, f"{path} is missing a link to {target}"
|
|
assert f'class="here">{active}<' in html, f"{path} does not mark itself active"
|
|
|
|
|
|
@pytest.mark.parametrize("path", ["/", "/tune", "/config", "/digest"])
|
|
def test_every_page_has_a_working_theme_toggle(client, path):
|
|
"""Three states (system/light/dark), no flash, and the guard that makes the
|
|
light choice win over a dark OS setting."""
|
|
c, _ = client
|
|
html = c.get(path).text
|
|
assert 'id="theme"' in html, "theme button missing"
|
|
assert '[data-theme="dark"]' in html, "explicit dark rules missing"
|
|
assert ':root:not([data-theme="light"])' in html, (
|
|
"without this guard, choosing light does nothing under a dark OS theme")
|
|
# The theme must be applied before first paint, so its script has to appear
|
|
# ahead of any rendered content.
|
|
assert html.index("localStorage.getItem('theme')") < html.index("<main")
|
|
|
|
|
|
@pytest.mark.parametrize("path", ["/", "/tune", "/config", "/digest"])
|
|
def test_palette_is_defined_once_from_the_shared_partial(client, path):
|
|
"""A page that redeclares the palette locally would drift from the others."""
|
|
c, _ = client
|
|
html = c.get(path).text
|
|
assert html.count(".sitenav {") == 1
|
|
assert html.count("--accent:#7c3aed") == 1, "light palette declared more than once"
|
|
|
|
|
|
def test_triage_cards_expand_to_every_report(tmp_path, monkeypatch):
|
|
"""A card must show all the reports behind a story, not a truncated slice."""
|
|
import numpy as np
|
|
from datetime import UTC, datetime
|
|
|
|
monkeypatch.setenv("NEWS_DB", str(tmp_path / "e.db"))
|
|
monkeypatch.setenv("NEWS_FEEDBACK_SECRET", SECRET)
|
|
con = store.connect(tmp_path / "e.db")
|
|
now = datetime.now(UTC)
|
|
for n in range(7):
|
|
store.upsert_source(con, id=f"s{n}", name=f"Outlet{n}", url=f"http://s{n}",
|
|
kind="rss", bucket=f"b{n}", weight=1.0, prominence=0,
|
|
enabled=1, poll_interval_secs=900)
|
|
con.execute("""INSERT INTO items (id,source_id,title,summary,url,canonical_url,
|
|
published_at,first_seen_at,content_hash)
|
|
VALUES (?,?,?,'',?,?,?,?,?)""",
|
|
(f"i{n}", f"s{n}", f"Report number {n}", f"http://x/{n}",
|
|
f"http://x/{n}", now.isoformat(), now.isoformat(), f"h{n}"))
|
|
con.execute("""INSERT INTO clusters (id,centroid,size,first_seen_at,last_seen_at,lead_item_id)
|
|
VALUES ('big',?,7,?,?,'i0')""",
|
|
(store.to_blob(np.zeros(384, dtype="float32")),
|
|
now.isoformat(), now.isoformat()))
|
|
for n in range(7):
|
|
con.execute("INSERT INTO cluster_members VALUES ('big',?,?)", (f"i{n}", now.isoformat()))
|
|
|
|
from news import web
|
|
web._VECTORS = None
|
|
html = TestClient(web.app).get("/").text
|
|
|
|
for n in range(7):
|
|
assert f"Report number {n}" in html, f"report {n} missing from the card"
|
|
assert "<details" in html and "4 more reports" in html
|
|
assert "across 7 outlets" in html
|
|
|
|
|
|
def _triage_db(tmp_path, monkeypatch):
|
|
"""Three clusters that rank differently under each sort order."""
|
|
import numpy as np
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
monkeypatch.setenv("NEWS_DB", str(tmp_path / "s.db"))
|
|
monkeypatch.setenv("NEWS_FEEDBACK_SECRET", SECRET)
|
|
con = store.connect(tmp_path / "s.db")
|
|
now = datetime.now(UTC)
|
|
for n in range(3):
|
|
store.upsert_source(con, id=f"s{n}", name=f"O{n}", url=f"http://s{n}", kind="rss",
|
|
bucket=f"b{n}", weight=1.0, prominence=0, enabled=1,
|
|
poll_interval_secs=900)
|
|
# c_old is oldest, c_new is newest.
|
|
for cid, age_h in (("c_old", 30), ("c_mid", 10), ("c_new", 1)):
|
|
when = (now - timedelta(hours=age_h)).isoformat()
|
|
con.execute("""INSERT INTO items (id,source_id,title,summary,url,canonical_url,
|
|
published_at,first_seen_at,content_hash)
|
|
VALUES (?,?,?,'',?,?,?,?,?)""",
|
|
(cid, "s0", f"Story {cid}", f"http://x/{cid}", f"http://x/{cid}",
|
|
when, when, cid))
|
|
con.execute("""INSERT INTO clusters (id,centroid,size,first_seen_at,last_seen_at,
|
|
lead_item_id) VALUES (?,?,1,?,?,?)""",
|
|
(cid, store.to_blob(np.zeros(384, dtype="float32")), when, when, cid))
|
|
con.execute("INSERT INTO cluster_members VALUES (?,?,?)", (cid, cid, when))
|
|
from news import web
|
|
web._VECTORS = None
|
|
return TestClient(web.app)
|
|
|
|
|
|
def test_newest_first_orders_by_recency(tmp_path, monkeypatch):
|
|
# window=48h, not the default: c_old is 30 hours back and the default
|
|
# window is 24, by design.
|
|
html = _triage_db(tmp_path, monkeypatch).get("/?sort=new&show=all&window=48h").text
|
|
order = [t for t in ("c_new", "c_mid", "c_old") if f"Story {t}" in html]
|
|
positions = [html.index(f"Story {t}") for t in order]
|
|
assert positions == sorted(positions), "newest-first must be strictly by recency"
|
|
assert html.index("Story c_new") < html.index("Story c_old")
|
|
|
|
|
|
def test_sort_control_offers_every_order_and_marks_the_active_one(tmp_path, monkeypatch):
|
|
c = _triage_db(tmp_path, monkeypatch)
|
|
html = c.get("/?sort=interest").text
|
|
for value in ("best", "interest", "important", "new", "covered"):
|
|
assert f'value="{value}"' in html, f"sort option {value} missing"
|
|
assert 'value="interest" selected' in html
|
|
assert "Ignores importance entirely" in html, "the active sort must explain itself"
|
|
|
|
|
|
def test_unknown_sort_falls_back_rather_than_erroring(tmp_path, monkeypatch):
|
|
html = _triage_db(tmp_path, monkeypatch).get("/?sort=../../etc/passwd").text
|
|
assert 'value="best" selected' in html
|
|
|
|
|
|
def test_a_paywalled_source_is_never_the_link_when_a_readable_one_exists(tmp_path, monkeypatch):
|
|
"""Knowing the NYT covered a story is real corroboration; being handed its
|
|
unreadable link is not useful. Readable reports lead."""
|
|
import numpy as np
|
|
from datetime import UTC, datetime
|
|
|
|
monkeypatch.setenv("NEWS_DB", str(tmp_path / "p.db"))
|
|
monkeypatch.setenv("NEWS_FEEDBACK_SECRET", SECRET)
|
|
con = store.connect(tmp_path / "p.db")
|
|
now = datetime.now(UTC)
|
|
store.upsert_source(con, id="nyt", name="NYT", url="http://nyt", kind="rss",
|
|
bucket="nyt", weight=1.0, prominence=0, enabled=1,
|
|
poll_interval_secs=900, paywalled=1)
|
|
store.upsert_source(con, id="free", name="Gothamist", url="http://g", kind="rss",
|
|
bucket="g", weight=1.0, prominence=0, enabled=1,
|
|
poll_interval_secs=900, paywalled=0)
|
|
# The paywalled item is NEWER, so recency alone would put it first.
|
|
for iid, sid, mins in (("free_i", "free", 30), ("nyt_i", "nyt", 1)):
|
|
when = now.replace(microsecond=0).isoformat()
|
|
con.execute("""INSERT INTO items (id,source_id,title,summary,url,canonical_url,
|
|
published_at,first_seen_at,content_hash)
|
|
VALUES (?,?,?,'',?,?,?,?,?)""",
|
|
(iid, sid, f"Report from {sid}", f"http://x/{iid}", f"http://x/{iid}",
|
|
when if sid == "nyt" else "2026-01-01T00:00:00+00:00",
|
|
when, iid))
|
|
con.execute("""INSERT INTO clusters (id,centroid,size,first_seen_at,last_seen_at,
|
|
lead_item_id) VALUES ('c',?,2,?,?,'free_i')""",
|
|
(store.to_blob(np.zeros(384, dtype="float32")),
|
|
now.isoformat(), now.isoformat()))
|
|
for iid in ("free_i", "nyt_i"):
|
|
con.execute("INSERT INTO cluster_members VALUES ('c',?,?)", (iid, now.isoformat()))
|
|
|
|
from news import web
|
|
web._VECTORS = None
|
|
html = TestClient(web.app).get("/?show=all").text
|
|
assert html.index("Report from free") < html.index("Report from nyt"), (
|
|
"the readable report must be listed first even though it is older")
|
|
assert "🔒" in html, "paywalled outlets should be marked"
|
|
|
|
|
|
def test_all_pages_share_one_container(client):
|
|
"""The four views had 38 / 52 / 54 / 54rem containers with three different
|
|
padding sets, so they visibly jumped when navigating between them."""
|
|
import re
|
|
c, _ = client
|
|
seen = set()
|
|
for path in ("/", "/tune", "/config", "/digest"):
|
|
html = c.get(path).text
|
|
rules = re.findall(r"main \{ max-width:([0-9.]+rem); margin:0 auto; padding:([^;}]+)",
|
|
html)
|
|
assert len(rules) == 1, f"{path} declares {len(rules)} container rules"
|
|
seen.add(rules[0])
|
|
assert len(seen) == 1, f"containers differ across pages: {seen}"
|
|
|
|
|
|
def test_digest_prose_is_bounded_and_centred(client):
|
|
"""The prose may be wider than the classic measure, but not unbounded, and
|
|
it should sit centred in the shared container."""
|
|
c, _ = client
|
|
html = c.get("/digest").text
|
|
assert ".briefing" in html
|
|
# Wider than a classic measure but still bounded, and centred rather than
|
|
# pinned to the left edge of a 54rem container.
|
|
assert "max-width:44rem" in html
|
|
assert "margin:0 auto 2.25rem" in html
|
|
|
|
|
|
def test_favicon_is_served_at_the_root_and_linked_from_every_page(client):
|
|
"""Browsers request /favicon.ico from the root regardless of link tags, so
|
|
both the root route and the explicit links need to work."""
|
|
c, _ = client
|
|
ico = c.get("/favicon.ico")
|
|
assert ico.status_code == 200
|
|
assert ico.headers["content-type"] == "image/x-icon"
|
|
assert len(ico.content) > 100
|
|
|
|
for asset in ("favicon-32.png", "favicon-16.png", "apple-touch-icon.png"):
|
|
r = c.get(f"/static/{asset}")
|
|
assert r.status_code == 200, f"{asset} not served"
|
|
assert r.headers["content-type"] == "image/png"
|
|
|
|
for path in ("/", "/tune", "/config", "/digest"):
|
|
html = c.get(path).text
|
|
assert 'rel="icon"' in html, f"{path} has no icon link"
|
|
assert "apple-touch-icon" in html
|
|
|
|
|
|
def test_the_triage_queue_ages_stories_out(tmp_path, monkeypatch):
|
|
"""Regression: the queue was permanent.
|
|
|
|
Scoring covers every cluster ever stored, deliberately -- the z-score
|
|
population should describe what news looks like. But the PAGE rendered that
|
|
whole archive ranked by score, so on day three the top of the queue was
|
|
still day one's stories, plus feed backfill published before the system
|
|
existed. Nothing aged out except the handful that had been labelled.
|
|
"""
|
|
import numpy as np
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
monkeypatch.setenv("NEWS_DB", str(tmp_path / "age.db"))
|
|
monkeypatch.setenv("NEWS_FEEDBACK_SECRET", SECRET)
|
|
con = store.connect(tmp_path / "age.db")
|
|
now = datetime.now(UTC)
|
|
store.upsert_source(con, id="s0", name="O", url="http://s", kind="rss", bucket="b",
|
|
weight=1.0, prominence=0, enabled=1, poll_interval_secs=900)
|
|
ages = {"fresh": 2, "yesterday": 30, "lastweek": 120, "backfill": 24 * 40}
|
|
for cid, age_h in ages.items():
|
|
when = (now - timedelta(hours=age_h)).isoformat()
|
|
con.execute("""INSERT INTO items (id,source_id,title,summary,url,canonical_url,
|
|
published_at,first_seen_at,content_hash)
|
|
VALUES (?,'s0',?,'',?,?,?,?,?)""",
|
|
(cid, f"Story {cid}", f"http://x/{cid}", f"http://x/{cid}",
|
|
when, when, cid))
|
|
con.execute("""INSERT INTO clusters (id,centroid,size,first_seen_at,last_seen_at,
|
|
lead_item_id) VALUES (?,?,1,?,?,?)""",
|
|
(cid, store.to_blob(np.zeros(384, dtype="float32")), when, when, cid))
|
|
con.execute("INSERT INTO cluster_members VALUES (?,?,?)", (cid, cid, when))
|
|
|
|
from news import web
|
|
web._VECTORS = None
|
|
c = TestClient(web.app)
|
|
|
|
# The default window matches the routing staleness rule -- nothing older can
|
|
# be notified, so nothing older needs triaging -- and both are 24h, the
|
|
# daily cadence the app is built on.
|
|
html = c.get("/?show=all").text
|
|
assert "Story fresh" in html
|
|
assert "Story yesterday" not in html, "a 30-hour-old story is outside a day"
|
|
assert "Story lastweek" not in html, "a five-day-old story is still in the queue"
|
|
assert "Story backfill" not in html, "feed backfill is still in the queue"
|
|
|
|
# Widening is one click, for catching up after a few days away.
|
|
assert "Story yesterday" in c.get("/?show=all&window=48h").text
|
|
assert "Story lastweek" in c.get("/?show=all&window=7d").text
|
|
assert "Story backfill" in c.get("/?show=all&window=all").text
|
|
# A bad value must fall back, not 500 and not silently widen to everything.
|
|
assert "Story backfill" not in c.get("/?show=all&window=../../etc").text
|
|
|
|
|
|
def test_the_window_control_offers_every_span_and_marks_the_active_one(tmp_path, monkeypatch):
|
|
html = _triage_db(tmp_path, monkeypatch).get("/?window=7d").text
|
|
for value in ("24h", "48h", "7d", "all"):
|
|
assert f'value="{value}"' in html, f"window option {value} missing"
|
|
assert 'value="7d" selected' in html
|
|
|
|
|
|
def test_digest_page_renders_the_recorded_digest_not_only_the_alerts(tmp_path, monkeypatch):
|
|
"""Regression: only one story reached the page.
|
|
|
|
The digest was computed correctly -- 28 stories, and the briefing was
|
|
written from all of them -- but nothing recorded which ones. The page read
|
|
notification_log, the ALERT ledger, so it showed the single story that had
|
|
been pushed that day beside prose discussing twenty-eight.
|
|
"""
|
|
import numpy as np
|
|
from datetime import UTC, datetime
|
|
|
|
monkeypatch.setenv("NEWS_DB", str(tmp_path / "d.db"))
|
|
monkeypatch.setenv("NEWS_FEEDBACK_SECRET", SECRET)
|
|
con = store.connect(tmp_path / "d.db")
|
|
now = datetime.now(UTC)
|
|
store.upsert_source(con, id="s0", name="O", url="http://s", kind="rss", bucket="b",
|
|
weight=1.0, prominence=0, enabled=1, poll_interval_secs=900)
|
|
for cid in ("alerted", "d1", "d2", "d3"):
|
|
con.execute("""INSERT INTO items (id,source_id,title,summary,url,canonical_url,
|
|
published_at,first_seen_at,content_hash)
|
|
VALUES (?,'s0',?,'',?,?,?,?,?)""",
|
|
(cid, f"Story {cid}", f"http://x/{cid}", f"http://x/{cid}",
|
|
now.isoformat(), now.isoformat(), cid))
|
|
con.execute("""INSERT INTO clusters (id,centroid,size,first_seen_at,last_seen_at,
|
|
lead_item_id) VALUES (?,?,1,?,?,?)""",
|
|
(cid, store.to_blob(np.zeros(384, dtype="float32")),
|
|
now.isoformat(), now.isoformat(), cid))
|
|
con.execute("INSERT INTO cluster_members VALUES (?,?,?)", (cid, cid, now.isoformat()))
|
|
con.execute("""INSERT INTO notification_log (cluster_id,lane,score,created_at)
|
|
VALUES ('alerted','alert:important',4.0,?)""", (now.isoformat(),))
|
|
day = now.date().isoformat()
|
|
for rank, cid in enumerate(("d1", "d2", "d3")):
|
|
con.execute("""INSERT INTO digest_items (day,cluster_id,rank,score,reason)
|
|
VALUES (?,?,?,?,'digest')""", (day, cid, rank, 3.0 - rank))
|
|
|
|
from news import web
|
|
web._VECTORS = None
|
|
html = TestClient(web.app).get(f"/digest/{day}").text
|
|
for cid in ("alerted", "d1", "d2", "d3"):
|
|
assert f"Story {cid}" in html, f"{cid} missing from the day's page"
|
|
# Alerts lead -- they cleared a higher bar -- then the digest in rank order.
|
|
assert html.index("Story alerted") < html.index("Story d1") < html.index("Story d3")
|
|
|
|
|
|
def test_a_story_both_alerted_and_digested_appears_once(tmp_path, monkeypatch):
|
|
import numpy as np
|
|
from datetime import UTC, datetime
|
|
|
|
monkeypatch.setenv("NEWS_DB", str(tmp_path / "dup.db"))
|
|
monkeypatch.setenv("NEWS_FEEDBACK_SECRET", SECRET)
|
|
con = store.connect(tmp_path / "dup.db")
|
|
now = datetime.now(UTC)
|
|
store.upsert_source(con, id="s0", name="O", url="http://s", kind="rss", bucket="b",
|
|
weight=1.0, prominence=0, enabled=1, poll_interval_secs=900)
|
|
con.execute("""INSERT INTO items (id,source_id,title,summary,url,canonical_url,
|
|
published_at,first_seen_at,content_hash)
|
|
VALUES ('c','s0','Only once','','http://x','http://x',?,?,'h')""",
|
|
(now.isoformat(), now.isoformat()))
|
|
con.execute("""INSERT INTO clusters (id,centroid,size,first_seen_at,last_seen_at,
|
|
lead_item_id) VALUES ('c',?,1,?,?,'c')""",
|
|
(store.to_blob(np.zeros(384, dtype="float32")),
|
|
now.isoformat(), now.isoformat()))
|
|
con.execute("INSERT INTO cluster_members VALUES ('c','c',?)", (now.isoformat(),))
|
|
con.execute("""INSERT INTO notification_log (cluster_id,lane,score,created_at)
|
|
VALUES ('c','alert:interest',4.0,?)""", (now.isoformat(),))
|
|
day = now.date().isoformat()
|
|
con.execute("""INSERT INTO digest_items (day,cluster_id,rank,score,reason)
|
|
VALUES (?,'c',0,3.0,'digest')""", (day,))
|
|
|
|
from news import web
|
|
web._VECTORS = None
|
|
html = TestClient(web.app).get(f"/digest/{day}").text
|
|
# Count the card headings, not the string: a single-item cluster prints its
|
|
# title twice by design -- once as the headline, once in its source list.
|
|
assert html.count("<h2>Only once</h2>") == 1
|
|
|
|
|
|
# ------------------------------------------------------------ summary text ---
|
|
|
|
def test_clean_summary_strips_feed_furniture():
|
|
from news.digest import clean_summary
|
|
got = clean_summary(
|
|
"<p>Report Submitted by MEE staff on Tue, 09/15/2026 - 21:30 "
|
|
"The transfer of 'bunker busters' is under review.</p>", "Report")
|
|
assert "Submitted by" not in got and "'" not in got and "<p>" not in got
|
|
assert "bunker busters" in got and "'" in got
|
|
|
|
|
|
def test_clean_summary_drops_a_headline_echo():
|
|
from news.digest import clean_summary
|
|
title = "Council approves the downtown transit plan"
|
|
got = clean_summary(f"{title} — the vote was 7-2 after two hours of testimony.", title)
|
|
assert got.startswith("the vote was 7-2")
|
|
|
|
|
|
@pytest.mark.parametrize("junk", [
|
|
"Real content here. Read more about it on our site.",
|
|
"Real content here. The post Something appeared first on Blog.",
|
|
"Real content here. [+2431 chars]",
|
|
])
|
|
def test_clean_summary_removes_trailers(junk):
|
|
from news.digest import clean_summary
|
|
assert clean_summary(junk, "T").rstrip(" .…") == "Real content here"
|
|
|
|
|
|
def test_clean_summary_is_bounded_and_survives_nothing():
|
|
from news.digest import clean_summary
|
|
assert clean_summary(None) == "" and clean_summary("") == ""
|
|
long = clean_summary("word " * 500, "T")
|
|
assert len(long) <= 601, f"summary not truncated: {len(long)}"
|
|
|
|
|
|
def test_the_digest_model_is_given_the_summary_not_only_the_headline():
|
|
"""The reader's complaint: an opinion piece arrived as its headline alone,
|
|
so the briefing described the provocation rather than the argument."""
|
|
import inspect
|
|
from news import cli, digest
|
|
src = inspect.getsource(cli.cmd_digest)
|
|
assert '"summary": i["summary"]' in src, "cmd_digest must pass the summary through"
|
|
assert "summary" in digest.SYSTEM_PROMPT.lower()
|
|
assert "headline" in digest.SYSTEM_PROMPT.lower()
|
|
|
|
|
|
def test_reasoning_effort_is_nested_under_extra_body():
|
|
"""LiteLLM validates TOP-LEVEL parameters against a per-provider allowlist
|
|
and rejects this one with HTTP 400:
|
|
|
|
UnsupportedParamsError: zai does not support parameters:
|
|
['reasoning_effort']
|
|
|
|
`extra_body` is the documented passthrough and reaches the provider intact.
|
|
Sending it at the top level fails the whole digest.
|
|
"""
|
|
import httpx
|
|
from news.digest import LlmConfig, write_prose
|
|
|
|
seen = {}
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
seen.update(json.loads(request.content))
|
|
return httpx.Response(200, json={"choices": [
|
|
{"message": {"content": '{"briefing": "ok"}'}}]})
|
|
|
|
cfg = LlmConfig(enabled=True, base_url="http://litellm", api_key_env="K",
|
|
reasoning_effort="low")
|
|
os.environ["K"] = "x"
|
|
with httpx.Client(transport=httpx.MockTransport(handler)) as c:
|
|
assert write_prose(c, cfg, [{"headline": "h", "topic": "t", "summary": "s"}]) == "ok"
|
|
assert seen.get("extra_body") == {"reasoning_effort": "low"}
|
|
assert "reasoning_effort" not in seen, "top-level reasoning_effort is a 400"
|
|
|
|
|
|
def test_reasoning_effort_is_omitted_when_unset():
|
|
"""A provider that does not know the parameter must not receive it."""
|
|
import httpx
|
|
from news.digest import LlmConfig, write_prose
|
|
|
|
seen = {}
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
seen.update(json.loads(request.content))
|
|
return httpx.Response(200, json={"choices": [
|
|
{"message": {"content": '{"briefing": "ok"}'}}]})
|
|
|
|
os.environ["K"] = "x"
|
|
cfg = LlmConfig(enabled=True, base_url="http://litellm", api_key_env="K")
|
|
with httpx.Client(transport=httpx.MockTransport(handler)) as c:
|
|
write_prose(c, cfg, [{"headline": "h", "topic": "t", "summary": "s"}])
|
|
assert "extra_body" not in seen
|
|
|
|
|
|
@pytest.mark.parametrize("raw,want", [
|
|
# The contract: plain prose, returned as-is.
|
|
("Two paragraphs here.\n\nAnd the second.", "Two paragraphs here.\n\nAnd the second."),
|
|
("```\nFenced anyway.\n```", "Fenced anyway."),
|
|
("```markdown\nFenced with a tag.\n```", "Fenced with a tag."),
|
|
# Tolerated, not required: the model may still reach for the old envelope.
|
|
('{"briefing": "Legacy shape."}', "Legacy shape."),
|
|
# The failure that retired the envelope: an unescaped quote inside the
|
|
# value. json.loads dies at the character and the whole briefing went with
|
|
# it -- intermittently, since the same prompt at temperature 0 parsed on one
|
|
# run and not the next.
|
|
('{"briefing": "A minister said "protect families" and moved on."}',
|
|
'A minister said "protect families" and moved on.'),
|
|
('{"briefing": "Line one.\\nLine two."}', "Line one.\nLine two."),
|
|
])
|
|
def test_the_briefing_survives_whatever_shape_the_model_replies_in(raw, want):
|
|
from news.digest import extract_briefing
|
|
assert extract_briefing(raw) == want
|
|
|
|
|
|
def test_extract_briefing_gives_up_cleanly_on_nothing_usable():
|
|
from news.digest import extract_briefing
|
|
assert extract_briefing("") == ""
|
|
assert extract_briefing(None) == ""
|
|
# An object we cannot read is not prose; better the templated list than
|
|
# rendering raw JSON into the reader's morning briefing.
|
|
assert extract_briefing('{"something_else": 1}') == ""
|
|
|
|
|
|
def test_the_prompt_asks_for_prose_not_an_envelope():
|
|
"""The INPUT is still JSON data -- that is what stops a headline reading as
|
|
an instruction. Only the reply format changed."""
|
|
from news.digest import SYSTEM_PROMPT
|
|
assert "Reply with the briefing text itself" in SYSTEM_PROMPT
|
|
assert '{"briefing"' not in SYSTEM_PROMPT, "still asking for the old envelope"
|
|
assert "Respond with JSON" not in SYSTEM_PROMPT
|
|
assert "JSON array" in SYSTEM_PROMPT, "the input contract must still be JSON"
|