"""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("