connor/news
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
161c518aa718580e4a963ecb18fed78d62ac1e87
news/tests/test_delivery.py
T
connor 161c518aa7 Sortable triage, and say which sort is active 
The list was ordered by max(interest, importance) and nothing on the page said
so, which makes it impossible to tell whether an odd ranking is a bug or the
intended order.

Five orders, with the active one explained in a line beneath the control:

  best       highest of the two scores (unchanged default)
  interest   purely the taste profile, importance ignored entirely
  important  purely the importance axis -- velocity, corroboration, prominence
  new        most recently seen, regardless of score
  covered    most independent newsrooms, syndicated copy collapsed

"My interests" is the one that was actually missing. The two-axis design means a
quiet story that is very much the reader's can sit below a big one that is not,
and until now there was no way to see past that.

An unknown sort falls back to the default rather than erroring.
2026-09-08 18:13:36 -04:00

357 lines
15 KiB
Python
Raw Blame History

"""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()))
# 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):
html = _triage_db(tmp_path, monkeypatch).get("/?sort=new&show=all").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
Reference in new issue View Git Blame Copy Permalink