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