test(occurrences on eap): Create e2e test infrastructure (#108179)

Adds the `OccurrenceTestCase` test fixture with
`create_eap_occurrence()` & `store_occurrences()` for writing occurrence
trace items to EAP in tests, supporting both lightweight EAP-only tests
and dual-write tests via setting the `eventstream.eap_forwarding_rate`
option value. Using the former is preferred when dual-write
functionality is not needed in tests, since this option & the Snuba
write path will eventually be removed, and we will eventually be writing
only to EAP.

Extracts `build_occurrence_attributes()` in `item_helpers.py` as a
shared encoding function used by the production `encode_attributes()`
path and the test fixture.

Adds end-to-end tests for the `is_escalating` EAP read path
(`get_group_hourly_count`, `query_groups_past_counts`). Replaces the
mock-heavy unit tests which are no redundant with the new e2e coverage.

Follow-up items:
- [ ] Migrate all other EAP read path tests to use the e2e test mixin
- [ ] See if any logic common to all of the tests can be abstracted into
the mixin

Author: shashjar

src/sentry/eventstream/item_helpers.py

@@ -34,12 +34,46 @@ def serialize_event_data_as_item(
             Timestamp(seconds=int(event_data["received"])) if "received" in event_data else None
         ),
         retention_days=event_data.get("retention_days", 90),
-        attributes=encode_attributes(
+        attributes=_encode_attributes(
             event, event_data, ignore_fields={"event_id", "timestamp", "tags", "spans", "'spans'"}
         ),
     )
 
 
+def _encode_attributes(
+    event: Event | GroupEvent, event_data: Mapping[str, Any], ignore_fields: set[str] | None = None
+) -> Mapping[str, AnyValue]:
+    raw_tags = event_data.get("tags") or []
+    tags_dict = {kv[0]: kv[1] for kv in raw_tags if kv is not None and kv[1] is not None}
+
+    all_ignore_fields = (ignore_fields or set()) | {"tags"}
+    attributes = _build_occurrence_attributes(
+        event_data, tags=tags_dict, ignore_fields=all_ignore_fields
+    )
+
+    if event.group_id:
+        attributes["group_id"] = AnyValue(int_value=event.group_id)
+
+    return attributes
+
+
+def _build_occurrence_attributes(
+    data: Mapping[str, Any],
+    tags: Mapping[str, str] | None = None,
+    ignore_fields: set[str] | None = None,
+) -> dict[str, AnyValue]:
+    ignore_fields = ignore_fields or set()
+    attributes: dict[str, AnyValue] = {
+        k: _encode_value(v) for k, v in data.items() if k not in ignore_fields and v is not None
+    }
+
+    tag_attrs = {f"tags[{k}]": _encode_value(v) for k, v in (tags or {}).items()}
+    attributes.update(tag_attrs)
+    attributes["tag_keys"] = _encode_value(sorted(tag_attrs.keys()))
+
+    return attributes
+
+
 def _encode_value(value: Any, _depth: int = 0) -> AnyValue:
     if _depth > _ENCODE_MAX_DEPTH:
         # Beyond max depth, stringify to prevent protobuf nesting limit errors.
@@ -78,39 +112,3 @@ def _encode_value(value: Any, _depth: int = 0) -> AnyValue:
         )
     else:
         raise NotImplementedError(f"encode not supported for {type(value)}")
-
-
-def encode_attributes(
-    event: Event | GroupEvent, event_data: Mapping[str, Any], ignore_fields: set[str] | None = None
-) -> Mapping[str, AnyValue]:
-    attributes = {}
-    ignore_fields = ignore_fields or set()
-
-    for key, value in event_data.items():
-        if key in ignore_fields:
-            continue
-        if value is None:
-            continue
-        attributes[key] = _encode_value(value)
-
-    if event.group_id:
-        attributes["group_id"] = AnyValue(int_value=event.group_id)
-
-    format_tag_key = lambda key: f"tags[{key}]"
-
-    tag_keys = set()
-    tags = event_data.get("tags")
-    if tags is not None:
-        for tag in tags:
-            if tag is None:
-                continue
-            key, value = tag
-            if value is None:
-                continue
-            formatted_key = format_tag_key(key)
-            attributes[formatted_key] = _encode_value(value)
-            tag_keys.add(formatted_key)
-
-    attributes["tag_keys"] = _encode_value(sorted(tag_keys))
-
-    return attributes

src/sentry/testutils/cases.py

@@ -50,7 +50,10 @@
     CheckStatus,
     CheckStatusReason,
 )
-from sentry_protos.snuba.v1.request_common_pb2 import TraceItemType
+from sentry_protos.snuba.v1.request_common_pb2 import (
+    TRACE_ITEM_TYPE_OCCURRENCE,
+    TraceItemType,
+)
 from sentry_protos.snuba.v1.trace_item_pb2 import AnyValue, TraceItem
 from sentry_relay.consts import SPAN_STATUS_NAME_TO_CODE
 from slack_sdk.web import SlackResponse
@@ -62,7 +65,9 @@
 from sentry.auth.authenticators.totp import TotpInterface
 from sentry.auth.provider import Provider
 from sentry.auth.providers.dummy import DummyProvider
-from sentry.auth.providers.saml2.activedirectory.apps import ACTIVE_DIRECTORY_PROVIDER_NAME
+from sentry.auth.providers.saml2.activedirectory.apps import (
+    ACTIVE_DIRECTORY_PROVIDER_NAME,
+)
 from sentry.auth.staff import COOKIE_DOMAIN as STAFF_COOKIE_DOMAIN
 from sentry.auth.staff import COOKIE_NAME as STAFF_COOKIE_NAME
 from sentry.auth.staff import COOKIE_PATH as STAFF_COOKIE_PATH
@@ -77,6 +82,7 @@
 from sentry.auth.superuser import SUPERUSER_ORG_ID, Superuser
 from sentry.conf.types.kafka_definition import Topic, get_topic_codec
 from sentry.event_manager import EventManager
+from sentry.eventstream.item_helpers import _build_occurrence_attributes
 from sentry.eventstream.snuba import SnubaEventStream
 from sentry.issue_detection.performance_detection import detect_performance_problems
 from sentry.issues.grouptype import (
@@ -108,8 +114,12 @@
 from sentry.models.rule import RuleSource
 from sentry.monitors.models import Monitor, MonitorEnvironment, ScheduleType
 from sentry.new_migrations.monkey.state import SentryProjectState
-from sentry.notifications.models.notificationsettingoption import NotificationSettingOption
-from sentry.notifications.models.notificationsettingprovider import NotificationSettingProvider
+from sentry.notifications.models.notificationsettingoption import (
+    NotificationSettingOption,
+)
+from sentry.notifications.models.notificationsettingprovider import (
+    NotificationSettingProvider,
+)
 from sentry.notifications.notifications.base import alert_page_needs_org_id
 from sentry.notifications.types import FineTuningAPIKey
 from sentry.organizations.services.organization.serial import serialize_rpc_organization
@@ -1162,6 +1172,17 @@ def store_ourlogs(self, ourlogs):
         )
         assert response.status_code == 200
 
+    def store_occurrences(self, occurrences: Sequence[TraceItem]):
+        files = {
+            f"occurrence_{i}": occurrence.SerializeToString()
+            for i, occurrence in enumerate(occurrences)
+        }
+        response = requests.post(
+            settings.SENTRY_SNUBA + EAP_ITEMS_INSERT_ENDPOINT,
+            files=files,
+        )
+        assert response.status_code == 200
+
     def store_trace_metrics(self, trace_metrics):
         files = {
             f"trace_metric_{i}": trace_metric.SerializeToString()
@@ -3488,6 +3509,68 @@ def create_ourlog(
         )
 
 
+class OccurrenceTestCase(BaseTestCase, TraceItemTestCase):
+    def create_eap_occurrence(
+        self,
+        *,
+        organization: Organization | None = None,
+        project: Project | None = None,
+        group_id: int | None = None,
+        event_id: str | None = None,
+        trace_id: str | None = None,
+        timestamp: datetime | None = None,
+        level: str = "error",
+        environment: str | None = None,
+        title: str = "some error",
+        transaction: str | None = None,
+        occurrence_type: str = "error",
+        tags: dict[str, str] | None = None,
+        attributes: dict[str, Any] | None = None,
+        retention_days: int = 90,
+    ) -> TraceItem:
+        if organization is None:
+            organization = self.organization
+        if project is None:
+            project = self.project
+        if timestamp is None:
+            timestamp = datetime.now() - timedelta(minutes=1)
+        if event_id is None:
+            event_id = uuid4().hex
+        if trace_id is None:
+            trace_id = uuid4().hex
+
+        timestamp_proto = Timestamp()
+        timestamp_proto.FromDatetime(timestamp)
+
+        data: dict[str, Any] = {
+            "level": level,
+            "title": title,
+            "type": occurrence_type,
+        }
+        if group_id is not None:
+            data["group_id"] = group_id
+        if environment is not None:
+            data["environment"] = environment
+        if transaction is not None:
+            data["transaction"] = transaction
+        if attributes:
+            data.update(attributes)
+
+        attributes_proto = _build_occurrence_attributes(data, tags=tags)
+
+        return TraceItem(
+            organization_id=organization.id,
+            project_id=project.id,
+            item_type=TRACE_ITEM_TYPE_OCCURRENCE,
+            timestamp=timestamp_proto,
+            trace_id=trace_id,
+            item_id=hex_to_item_id(event_id),
+            received=timestamp_proto,
+            retention_days=retention_days,
+            attributes=attributes_proto,
+        )
+
+
 class TraceMetricsTestCase(BaseTestCase, TraceItemTestCase):
     def create_trace_metric(
         self,

tests/sentry/eventstream/test_item_helpers.py

@@ -6,7 +6,7 @@
 from sentry.db.models import NodeData
 from sentry.eventstream.item_helpers import (
     _ENCODE_MAX_DEPTH,
-    encode_attributes,
+    _encode_attributes,
     serialize_event_data_as_item,
 )
 from sentry.services.eventstore.models import Event, GroupEvent
@@ -41,7 +41,7 @@ def test_encode_attributes_basic(self) -> None:
             data=event_data,
             project_id=self.project.id,
         )
-        result = encode_attributes(event, event_data)
+        result = _encode_attributes(event, event_data)
 
         assert result["string_field"] == AnyValue(string_value="test")
         assert result["int_field"] == AnyValue(int_value=123)
@@ -60,7 +60,7 @@ def test_encode_attributes_with_ignore_fields(self) -> None:
             data=event_data,
             project_id=self.project.id,
         )
-        result = encode_attributes(event, event_data, ignore_fields={"ignore_field"})
+        result = _encode_attributes(event, event_data, ignore_fields={"ignore_field"})
 
         assert "keep_field" in result
         assert "another_keep" in result
@@ -78,7 +78,7 @@ def test_encode_attributes_with_multiple_ignore_fields(self) -> None:
             data=event_data,
             project_id=self.project.id,
         )
-        result = encode_attributes(event, event_data, ignore_fields={"field1", "field3"})
+        result = _encode_attributes(event, event_data, ignore_fields={"field1", "field3"})
 
         assert "field1" not in result
         assert "field2" in result
@@ -88,7 +88,7 @@ def test_encode_attributes_with_group_id(self) -> None:
         event_data = {"field": "value", "tags": []}
 
         event = self.create_group_event(event_data)
-        result = encode_attributes(event, event_data)
+        result = _encode_attributes(event, event_data)
 
         assert result["group_id"] == AnyValue(int_value=event.group.id)
         assert result["field"] == AnyValue(string_value="value")
@@ -102,7 +102,7 @@ def test_encode_attributes_without_group_id(self) -> None:
             project_id=self.project.id,
         )
 
-        result = encode_attributes(event, event_data)
+        result = _encode_attributes(event, event_data)
 
         assert "group_id" not in result
         assert result["field"] == AnyValue(string_value="value")
@@ -123,7 +123,7 @@ def test_encode_attributes_with_tags(self) -> None:
             project_id=self.project.id,
         )
 
-        result = encode_attributes(event, event_data)
+        result = _encode_attributes(event, event_data)
 
         assert result["tags[environment]"] == AnyValue(string_value="production")
         assert result["tags[release]"] == AnyValue(string_value="1.0.0")
@@ -147,7 +147,7 @@ def test_encode_attributes_with_empty_tags(self) -> None:
             project_id=self.project.id,
         )
 
-        result = encode_attributes(event, event_data)
+        result = _encode_attributes(event, event_data)
 
         assert result["field"] == AnyValue(string_value="value")
         # No tags[] keys should be present
@@ -164,7 +164,7 @@ def test_encode_attributes_with_integer_tag_values(self) -> None:
         }
 
         event = self.create_group_event(event_data)
-        result = encode_attributes(event, event_data)
+        result = _encode_attributes(event, event_data)
 
         assert result["tags[numeric_tag]"] == AnyValue(string_value="42")
         assert result["tags[string_tag]"] == AnyValue(string_value="value")
@@ -178,12 +178,11 @@ def test_encode_attributes_empty_event_data(self) -> None:
             data=event_data,
             project_id=self.project.id,
         )
-        result = encode_attributes(event, event_data)
+        result = _encode_attributes(event, event_data)
 
-        # "tags" field itself gets encoded as a non-scalar value in the loop
-        # Then tags are processed separately, but empty list adds no tag attributes
-        assert len(result) == 2
-        assert result["tags"] == AnyValue(array_value=ArrayValue(values=[]))
+        assert len(result) == 1
+        assert "tag_keys" in result
+        assert result["tag_keys"] == AnyValue(array_value=ArrayValue(values=[]))
 
     def test_encode_attributes_with_complex_types(self) -> None:
         event_data = {
@@ -198,7 +197,7 @@ def test_encode_attributes_with_complex_types(self) -> None:
             project_id=self.project.id,
         )
 
-        result = encode_attributes(event, event_data)
+        result = _encode_attributes(event, event_data)
 
         assert result["list_field"] == AnyValue(
             array_value=ArrayValue(
@@ -218,7 +217,7 @@ def test_encode_attributes_with_complex_types(self) -> None:
         )
 
     def test_encode_attributes_with_none_tags(self) -> None:
-        """Test that encode_attributes handles None tags gracefully."""
+        """Test that _encode_attributes handles None tags gracefully."""
         event_data = {
             "field": "value",
             "tags": None,
@@ -230,7 +229,7 @@ def test_encode_attributes_with_none_tags(self) -> None:
             project_id=self.project.id,
         )
 
-        result = encode_attributes(event, event_data)
+        result = _encode_attributes(event, event_data)
 
         assert result["field"] == AnyValue(string_value="value")
         # No tags[] keys should be present when tags is None
@@ -239,7 +238,7 @@ def test_encode_attributes_with_none_tags(self) -> None:
         assert result["tag_keys"] == AnyValue(array_value=ArrayValue(values=[]))
 
     def test_encode_attributes_with_none_elements_in_tags(self) -> None:
-        """Test that encode_attributes handles tags list containing None values."""
+        """Test that _encode_attributes handles tags list containing None values."""
         event_data = {
             "field": "value",
             "tags": [
@@ -256,7 +255,7 @@ def test_encode_attributes_with_none_elements_in_tags(self) -> None:
             project_id=self.project.id,
         )
 
-        result = encode_attributes(event, event_data)
+        result = _encode_attributes(event, event_data)
 
         assert result["field"] == AnyValue(string_value="value")
         assert result["tags[tag1]"] == AnyValue(string_value="value1")
@@ -440,7 +439,7 @@ def test_encode_attributes_at_max_depth_boundary(self) -> None:
 
         event_data = {"field": nested, "tags": []}
         event = Event(event_id="a" * 32, data=event_data, project_id=self.project.id)
-        result = encode_attributes(event, event_data)
+        result = _encode_attributes(event, event_data)
 
         current = result["field"]
         for _ in range(_ENCODE_MAX_DEPTH):
@@ -457,7 +456,7 @@ def test_encode_attributes_deeply_nested_list_stringifies(self) -> None:
 
         event_data = {"field": nested, "tags": []}
         event = Event(event_id="a" * 32, data=event_data, project_id=self.project.id)
-        result = encode_attributes(event, event_data)
+        result = _encode_attributes(event, event_data)
 
         # Walk down to the stringification boundary
         current = result["field"]
@@ -476,7 +475,7 @@ def test_encode_attributes_deeply_nested_dict_stringifies(self) -> None:
 
         event_data = {"field": nested, "tags": []}
         event = Event(event_id="a" * 32, data=event_data, project_id=self.project.id)
-        result = encode_attributes(event, event_data)
+        result = _encode_attributes(event, event_data)
 
         current = result["field"]
         for _ in range(_ENCODE_MAX_DEPTH + 1):