news/tests/test_delivery.py

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


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