thermograph/backend/tests/core/test_events.py

"""Unit tests for the UI product-event schema (core/events.py).

The allowlist in ``events.SCHEMA`` is the entire security and privacy model of an
unauthenticated write endpoint, so most of what is worth asserting here is what
does NOT get through: free text, coordinates, unknown names, unknown values.
"""
import importlib

from core import events


def _fresh(monkeypatch, enabled=True):
    m = importlib.reload(events)
    monkeypatch.setattr(m, "ENABLED", enabled)
    return m


def test_known_event_keeps_only_declared_slots(monkeypatch):
    m = _fresh(monkeypatch)
    name, dims = m.normalize("view.open", {"view": "calendar", "prop": "nav", "value": "tmax"})
    assert name == "view.open"
    # view.open declares view + prop but not value, so value is dropped entirely
    # rather than stored — an undeclared slot can never carry anything.
    assert dims == {"view": "calendar", "prop": "nav", "value": ""}


def test_unknown_event_collapses_with_blank_dimensions(monkeypatch):
    m = _fresh(monkeypatch)
    name, dims = m.normalize("evil.event", {"view": "calendar", "prop": "nav"})
    assert name == m.EVENT_OTHER
    assert dims == {"view": "", "prop": "", "value": ""}


def test_unknown_value_in_a_known_slot_is_bucketed_never_stored(monkeypatch):
    """The property that makes the endpoint safe: nothing a client sends is ever
    persisted verbatim."""
    m = _fresh(monkeypatch)
    _, dims = m.normalize("place.pick", {"view": "home", "prop": "47.6,-122.3"})
    assert dims["prop"] == m.VALUE_OTHER
    _, dims = m.normalize("view.control", {"view": "home", "prop": "chart_metric",
                                           "value": "seattle wa"})
    assert dims["value"] == m.VALUE_OTHER


def test_non_string_props_do_not_crash_or_leak(monkeypatch):
    m = _fresh(monkeypatch)
    for props in (None, [], "x", {"view": 47.6}, {"view": {"lat": 1}}):
        name, dims = m.normalize("place.pick", props)
        assert name == "place.pick"
        assert set(dims) == {"view", "prop", "value"}
        assert all(isinstance(v, str) for v in dims.values())


def test_schema_slots_are_all_recognised(monkeypatch):
    """Guard against a typo'd slot name silently making an event dimensionless."""
    m = _fresh(monkeypatch)
    for name, spec in m.SCHEMA.items():
        assert spec, name
        assert set(spec) <= set(m.SLOTS), name
        for slot, allowed in spec.items():
            assert isinstance(allowed, frozenset) and allowed, (name, slot)


def test_no_slot_can_hold_free_text_by_construction(monkeypatch):
    """Every allowlisted value is a short, boring token. If this ever fails,
    something person-identifying has been added to the vocabulary."""
    m = _fresh(monkeypatch)
    for spec in m.SCHEMA.values():
        for allowed in spec.values():
            for v in allowed:
                assert v.replace("_", "").replace("-", "").isalnum(), v
                assert len(v) <= 20, v


def test_record_is_inert_when_the_flag_is_off(monkeypatch):
    m = _fresh(monkeypatch, enabled=False)
    calls = []
    monkeypatch.setattr(m, "_upsert", lambda *a: calls.append(a))
    monkeypatch.setattr(m.audit, "log_activity", lambda *a, **k: calls.append(a))
    m.record("view.open", {"view": "home"})
    assert calls == []


def test_record_writes_one_activity_line_with_only_allowlisted_fields(monkeypatch):
    m = _fresh(monkeypatch)
    lines = []
    monkeypatch.setattr(m, "_upsert", lambda *a: None)
    monkeypatch.setattr(m.audit, "log_activity", lambda tag, rec: lines.append((tag, rec)))
    m.record("view.open", {"view": "city", "prop": "nav"},
             referer="https://news.ycombinator.com/item?id=1", host="thermograph.org")
    (tag, rec), = lines
    assert tag == "ui.event"
    # The referrer is reduced to a bare domain — never the URL, which can carry a
    # search query or a private path.
    assert rec == {"event": "view.open", "view": "city", "prop": "nav",
                   "value": "", "referrer": "news.ycombinator.com"}


def test_referrer_is_kept_only_for_view_open(monkeypatch):
    """Acquisition needs it; nothing else does, and dropping it everywhere else is
    what keeps the key space small enough to eyeball."""
    m = _fresh(monkeypatch)
    lines = []
    monkeypatch.setattr(m, "_upsert", lambda *a: None)
    monkeypatch.setattr(m.audit, "log_activity", lambda tag, rec: lines.append(rec))
    m.record("share", {"view": "home", "prop": "link"},
             referer="https://reddit.com/r/weather", host="thermograph.org")
    assert lines[0]["referrer"] == ""


def test_record_never_raises(monkeypatch):
    m = _fresh(monkeypatch)

    def boom(*a, **k):
        raise RuntimeError("sink down")

    monkeypatch.setattr(m, "_upsert", boom)
    monkeypatch.setattr(m.audit, "log_activity", boom)
    m.record("view.open", {"view": "home"})   # must not propagate


# --- Tier B: the ephemeral session id ------------------------------------------
# The session field is the one place a client could put arbitrary text into
# storage, so it is validated by SHAPE and dropped whole on any mismatch. These
# are the tests that keep it from quietly becoming a free-text column.

def test_session_id_accepts_only_the_exact_minted_shape(monkeypatch):
    m = _fresh(monkeypatch)
    good = "AbCd1234_-efGH5678ijK"     # 21 chars: too short
    assert m.validate_session(good) == ""
    assert m.validate_session("A" * 22) == "A" * 22
    assert m.validate_session("aZ0_-" + "b" * 17) == "aZ0_-" + "b" * 17


def test_session_id_rejects_anything_that_could_carry_text(monkeypatch):
    m = _fresh(monkeypatch)
    for bad in ("user@example.com", "47.6062,-122.3321", "a" * 200, "", None, 12345,
                "AAAAAAAAAAAAAAAAAAAA<>", "has space here aaaaaaa", {"id": "x"},
                "user-id-42", "AAAAAAAAAAAAAAAAAAAAA;DROP"):
        assert m.validate_session(bad) == "", bad


def test_sequence_number_is_bounded(monkeypatch):
    m = _fresh(monkeypatch)
    assert m.validate_seq(0) == 0
    assert m.validate_seq(m.MAX_SEQ) == m.MAX_SEQ
    for bad in (-1, m.MAX_SEQ + 1, 10**9, "3", 3.0, True, None):
        assert m.validate_seq(bad) == -1, bad


def test_session_rows_need_their_own_flag(monkeypatch):
    """Tier A must record with Tier B off — the anonymous signal never depends on
    consent rates or on the session tier being enabled."""
    m = _fresh(monkeypatch)
    monkeypatch.setattr(m, "SESSIONS_ENABLED", False)
    agg, rows = [], []
    monkeypatch.setattr(m, "_upsert", lambda *a: agg.append(a))
    monkeypatch.setattr(m, "_insert_session", lambda *a: rows.append(a))
    monkeypatch.setattr(m.audit, "log_activity", lambda *a: None)
    m.record("view.open", {"view": "home"}, session="A" * 22, seq=1)
    assert len(agg) == 1 and rows == []


def test_session_row_written_when_consented_and_enabled(monkeypatch):
    m = _fresh(monkeypatch)
    monkeypatch.setattr(m, "SESSIONS_ENABLED", True)
    agg, rows = [], []
    monkeypatch.setattr(m, "_upsert", lambda *a: agg.append(a))
    monkeypatch.setattr(m, "_insert_session", lambda *a: rows.append(a))
    monkeypatch.setattr(m.audit, "log_activity", lambda *a: None)
    m.record("place.pick", {"view": "home", "prop": "map"}, session="B" * 22, seq=4)
    assert len(agg) == 1
    (sid, seq, event, dims, _ref), = rows
    assert (sid, seq, event) == ("B" * 22, 4, "place.pick")
    assert dims["prop"] == "map"


def test_bad_session_id_does_not_cost_the_aggregate(monkeypatch):
    m = _fresh(monkeypatch)
    monkeypatch.setattr(m, "SESSIONS_ENABLED", True)
    agg, rows = [], []
    monkeypatch.setattr(m, "_upsert", lambda *a: agg.append(a))
    monkeypatch.setattr(m, "_insert_session", lambda *a: rows.append(a))
    monkeypatch.setattr(m.audit, "log_activity", lambda *a: None)
    m.record("view.open", {"view": "home"}, session="not-a-valid-id", seq=1)
    assert len(agg) == 1 and rows == []


def test_session_id_never_reaches_the_log_stream(monkeypatch):
    """The activity JSONL goes to Loki with its own retention and access rules;
    a behavioural sequence must not be duplicated into it."""
    m = _fresh(monkeypatch)
    monkeypatch.setattr(m, "SESSIONS_ENABLED", True)
    lines = []
    monkeypatch.setattr(m, "_upsert", lambda *a: None)
    monkeypatch.setattr(m, "_insert_session", lambda *a: None)
    monkeypatch.setattr(m.audit, "log_activity", lambda tag, rec: lines.append(rec))
    m.record("view.open", {"view": "home"}, session="C" * 22, seq=2)
    assert "C" * 22 not in str(lines)
    assert "session" not in lines[0] and "seq" not in lines[0]


# --- the one free-text field: zero-result search queries ------------------------

def test_search_miss_normalisation(monkeypatch):
    m = _fresh(monkeypatch)
    assert m.normalize_search_miss("  Saint-Étienne  ") == "saint-étienne"
    assert m.normalize_search_miss("NEW\tYORK\n CITY") == "new york city"
    assert m.normalize_search_miss("Königsberg") == m.normalize_search_miss("KÖNIGSBERG")
    # A postcode-length digit run is a legitimate place query and survives.
    assert m.normalize_search_miss("98101") == "98101"


def test_search_miss_rejects_anything_with_a_personal_data_smell(monkeypatch):
    m = _fresh(monkeypatch)
    for bad in (
        "someone@example.com",                     # email
        "https://example.com/x",                   # URL
        "www.example.com",                         # hostname
        "call me on 07700900123",                  # long digit run
        "x" * 60,                                  # too long
        "my house is the third one on the left",   # a sentence
        "$$$ ???",                                 # punctuation soup
        "47.6062,-122.3321",                       # coordinates
        None, 42, "",
    ):
        assert m.normalize_search_miss(bad) == "", bad


def test_search_miss_needs_its_own_flag(monkeypatch):
    m = _fresh(monkeypatch)
    monkeypatch.setattr(m, "SEARCH_MISS_ENABLED", False)
    seen = []
    monkeypatch.setattr(m, "_upsert_search_miss", lambda q: seen.append(q))
    m.record_search_miss("atlantis")
    assert seen == []
    monkeypatch.setattr(m, "SEARCH_MISS_ENABLED", True)
    m.record_search_miss("atlantis")
    m.record_search_miss("someone@example.com")
    assert seen == ["atlantis"]


def test_search_miss_prune_enforces_a_k_anonymity_floor(monkeypatch):
    """The strongest mitigation on the free-text field: a string only one or two
    people typed is deleted; one several people typed is a product signal."""
    m = _fresh(monkeypatch)
    assert m.SEARCH_MISS_MIN_COUNT >= 3
    assert f"count < {m.SEARCH_MISS_MIN_COUNT}" in m.PRUNE_SEARCH_MISS_SQL
    assert f"{m.SEARCH_MISS_PRUNE_DAYS} days" in m.PRUNE_SEARCH_MISS_SQL