thermograph/backend/tests/web/test_event_beacon.py

"""Route-level tests for the public product-event beacon (POST /api/v2/event).

The endpoint is unauthenticated and reachable by anything on the internet — on a
site where ~60% of inbound traffic is crawlers and scanners — so these tests are
mostly about what it REFUSES to do: grow storage, echo anything back, or accept
a payload big enough to be worth abusing.
"""
import json

import pytest
from fastapi.testclient import TestClient

from core import events
from web import app as appmod

EVENT_URL = "/thermograph/api/v2/event"
# Browsers set this on fetch/sendBeacon; a bare scanner POST does not.
FIRST_PARTY = {"sec-fetch-site": "same-origin"}


@pytest.fixture(autouse=True)
def _fresh_rate_limiter():
    """TestClient always presents the same client IP, so without this one test's
    beacon traffic silently rate-limits the next one's."""
    from core import metrics
    metrics.reset_rate_limiter()
    yield
    metrics.reset_rate_limiter()


@pytest.fixture
def client():
    return TestClient(appmod.app)


@pytest.fixture
def recorded(monkeypatch):
    """Capture what reaches the recorder, with the flag forced on."""
    seen = []
    monkeypatch.setattr(events, "ENABLED", True)
    monkeypatch.setattr(events, "_upsert", lambda *a: None)
    monkeypatch.setattr(events.audit, "log_activity", lambda tag, rec: seen.append(rec))
    return seen


def test_beacon_always_answers_204_and_returns_no_body(client, recorded):
    """Counted, rejected, or rate-limited must be indistinguishable to a prober."""
    for body in ({"event": "view.open", "props": {"view": "home"}},
                 {"event": "not-a-real-event"},
                 {"nonsense": True},
                 [1, 2, 3],
                 "plain string"):
        r = client.post(EVENT_URL, content=json.dumps(body), headers=FIRST_PARTY)
        assert r.status_code == 204
        assert r.content == b""


def test_valid_event_is_recorded_with_allowlisted_dimensions(client, recorded):
    client.post(EVENT_URL, json={"event": "view.control",
                                 "props": {"view": "compare", "prop": "unit", "value": "C"}},
                headers=FIRST_PARTY)
    assert recorded == [{"event": "view.control", "view": "compare", "prop": "unit",
                         "value": "C", "referrer": ""}]


def test_batched_payload_is_accepted_and_capped(client, recorded):
    items = [{"event": "share", "props": {"prop": "link"}}] * (events.MAX_BATCH + 5)
    client.post(EVENT_URL, json={"events": items}, headers=FIRST_PARTY)
    assert len(recorded) == events.MAX_BATCH


def test_a_full_batch_of_real_events_fits_under_the_byte_cap(client, recorded):
    """Regression: if MAX_BODY_BYTES ever drops below a full MAX_BATCH payload,
    the byte cap silently eats legitimate batches instead of abusive ones."""
    items = [{"event": "view.control",
              "props": {"view": "compare", "prop": "chart_metric", "value": "precip"}}
             ] * events.MAX_BATCH
    assert len(json.dumps({"events": items})) < events.MAX_BODY_BYTES
    client.post(EVENT_URL, json={"events": items}, headers=FIRST_PARTY)
    assert len(recorded) == events.MAX_BATCH


def test_oversized_body_is_dropped_without_being_parsed(client, recorded):
    """The endpoint must not become storage: a body past the cap is never read
    into a record at all."""
    fat = {"event": "view.open", "props": {"view": "home"},
           "junk": "x" * (events.MAX_BODY_BYTES * 4)}
    r = client.post(EVENT_URL, content=json.dumps(fat), headers=FIRST_PARTY)
    assert r.status_code == 204
    assert recorded == []


def test_free_text_and_coordinates_never_survive(client, recorded):
    """Anything a client puts in a dimension slot is allowlisted or bucketed —
    a leaked query string or lat/lon cannot reach a sink."""
    client.post(EVENT_URL, json={"event": "place.pick",
                                 "props": {"view": "home", "prop": "47.6062,-122.3321"}},
                headers=FIRST_PARTY)
    client.post(EVENT_URL, json={"event": "view.control",
                                 "props": {"view": "home", "prop": "chart_metric",
                                           "value": "user@example.com"}},
                headers=FIRST_PARTY)
    stored = {v for rec in recorded for v in rec.values()}
    assert not any("47.6" in s or "@" in s for s in stored)


def test_cross_site_post_is_dropped(client, recorded):
    r = client.post(EVENT_URL, json={"event": "view.open", "props": {"view": "home"}},
                    headers={"sec-fetch-site": "cross-site"})
    assert r.status_code == 204
    assert recorded == []


def test_flag_off_records_nothing(client, monkeypatch):
    """The whole feature ships inert: with THERMOGRAPH_EVENTS unset the beacon
    still answers, and nothing reaches the durable sink."""
    seen = []
    monkeypatch.setattr(events, "ENABLED", False)
    monkeypatch.setattr(events, "_upsert", lambda *a: seen.append(a))
    monkeypatch.setattr(events.audit, "log_activity", lambda tag, rec: seen.append(rec))
    r = client.post(EVENT_URL, json={"event": "view.open", "props": {"view": "home"}},
                    headers=FIRST_PARTY)
    assert r.status_code == 204
    assert seen == []


def test_legacy_single_event_shape_still_counts(client, monkeypatch):
    """The four home.* aggregate counters already in production keep working —
    they are unflagged and predate the schema."""
    from core import metrics
    seen = []
    monkeypatch.setattr(metrics, "_store", type("S", (), {
        "record_event": lambda self, *a: seen.append(a)})())
    client.post(EVENT_URL, json={"event": "home.locate"}, headers=FIRST_PARTY)
    assert seen and seen[0][0] == "home.locate"


def test_beacon_is_not_counted_as_inbound_traffic(client):
    """It reports traffic; counting it would double every interaction."""
    from core import metrics
    assert metrics.classify_inbound(EVENT_URL, "/thermograph") == "event"


# --- Tier B over the wire -------------------------------------------------------

@pytest.fixture
def session_rows(monkeypatch, recorded):
    rows = []
    monkeypatch.setattr(events, "SESSIONS_ENABLED", True)
    monkeypatch.setattr(events, "_insert_session", lambda *a: rows.append(a))
    return rows


def test_session_and_seq_travel_with_a_batch(client, session_rows):
    client.post(EVENT_URL, json={
        "session": "D" * 22,
        "events": [{"event": "view.open", "props": {"view": "home"}, "seq": 1},
                   {"event": "place.pick", "props": {"view": "home", "prop": "map"}, "seq": 2}],
    }, headers=FIRST_PARTY)
    assert [(r[0], r[1], r[2]) for r in session_rows] == [
        ("D" * 22, 1, "view.open"), ("D" * 22, 2, "place.pick")]


def test_a_forged_session_field_cannot_smuggle_text(client, session_rows, recorded):
    """The session field is the only client-controlled string that reaches a
    sink, so a malformed one must lose its payload entirely — while the
    anonymous aggregate still records."""
    for forged in ("'; DROP TABLE ui_event_session; --", "user@example.com",
                   "47.6062,-122.3321", "x" * 500):
        client.post(EVENT_URL, json={"session": forged, "seq": 1,
                                     "event": "view.open", "props": {"view": "home"}},
                    headers=FIRST_PARTY)
    assert session_rows == []
    assert len(recorded) == 4          # Tier A recorded every one of them


def test_sessions_flag_off_records_only_the_aggregate(client, monkeypatch, recorded):
    rows = []
    monkeypatch.setattr(events, "SESSIONS_ENABLED", False)
    monkeypatch.setattr(events, "_insert_session", lambda *a: rows.append(a))
    client.post(EVENT_URL, json={"session": "E" * 22, "seq": 1,
                                 "event": "view.open", "props": {"view": "home"}},
                headers=FIRST_PARTY)
    assert rows == [] and len(recorded) == 1


def test_beacon_never_reads_the_account_session_cookie(client, session_rows, recorded):
    """Even a signed-in visitor's events stay unlinked to the account: the route
    does not consult the auth cookie, so there is nothing to join on."""
    client.cookies.set("thermographauth", "pretend-jwt-value")
    client.post(EVENT_URL, json={"event": "view.open", "props": {"view": "home"}},
                headers=FIRST_PARTY)
    assert "pretend-jwt" not in str(recorded) + str(session_rows)
    assert recorded and all("user" not in k for rec in recorded for k in rec)