test(relay): Assert rejection outcomes instead of waiting out ingestion

src/sentry/testutils/relay.py

@@ -1,16 +1,24 @@
 from __future__ import annotations
 
-from collections.abc import Callable, Iterable, Mapping
+from collections.abc import Callable, Generator, Iterable, Mapping
+from contextlib import contextmanager
 from typing import IO, TYPE_CHECKING, Any, Protocol, cast
+from uuid import uuid4
 
 import pytest
 import requests
 import responses
-from django.conf import LazySettings
+from confluent_kafka import Consumer, KafkaException, TopicPartition
+from django.conf import LazySettings, settings
 
+from sentry.constants import DataCategory
 from sentry.models.eventattachment import PendingEventAttachment
 from sentry.services import eventstore
 from sentry.testutils.helpers import get_auth_header
+from sentry.testutils.pytest import xdist
+from sentry.utils import json
+from sentry.utils.batching_kafka_consumer import create_topics
+from sentry.utils.outcomes import Outcome
 
 if TYPE_CHECKING:
     from sentry.services.eventstore.models import Event
@@ -36,6 +44,49 @@ def _get_event(project_id: int, event_id: str) -> Event | None:
     )
 
 
+@contextmanager
+def _capture_outcomes() -> Generator[Callable[[], dict[str, Any] | None]]:
+    """Yield a non-blocking poll over outcomes Relay produces from now on."""
+    topic = xdist.get_kafka_topic("outcomes")
+    create_topics("default", [topic])
+    consumer = Consumer(
+        {
+            **settings.KAFKA_CLUSTERS["default"]["common"],
+            "group.id": f"relay-test-outcomes-{uuid4().hex}",
+            "enable.auto.commit": False,
+        }
+    )
+    try:
+        partitions = consumer.list_topics(topic=topic, timeout=10).topics[topic].partitions
+        assert partitions, f"No partitions for {topic}"
+        # Assign concrete high watermarks rather than OFFSET_END: OFFSET_END is
+        # resolved on first poll and could skip an outcome Relay already produced.
+        consumer.assign(
+            [
+                TopicPartition(
+                    topic,
+                    partition,
+                    consumer.get_watermark_offsets(TopicPartition(topic, partition), timeout=10)[1],
+                )
+                for partition in partitions
+            ]
+        )
+
+        def poll() -> dict[str, Any] | None:
+            message = consumer.poll(0)
+            if message is None:
+                return None
+            if error := message.error():
+                raise KafkaException(error)
+            value = message.value()
+            assert value is not None
+            return json.loads(value)
+
+        yield poll
+    finally:
+        consumer.close()
+
+
 class RelayStoreHelper(RequiredBaseclass):
     """
     Tests that post to the store entry point should use this helper class
@@ -109,13 +160,51 @@ def has_new_event() -> Event | None:
         assert event
         return event
 
-    def post_and_try_retrieve_event(
-        self, data: object, headers: Mapping[str, str] | None = None
-    ) -> Event | None:
-        try:
-            return self.post_and_retrieve_event(data, headers)
-        except AssertionError:
-            return None
+    def post_and_assert_rejected(
+        self,
+        data: Mapping[str, object],
+        *,
+        outcome: Outcome,
+        reason: str,
+        headers: Mapping[str, str | None] | None = None,
+    ) -> None:
+        """
+        Post an event that Relay accepts over HTTP and then drops asynchronously,
+        and wait for the outcome that records the drop instead of the ingestion timeout.
+        """
+        if headers is None:
+            headers = {"x-sentry-auth": self.auth_header, "content-type": "application/json"}
+        url = self.get_relay_store_url(self.project.id)
+        responses.add_passthru(url)
+        event_id = uuid4().hex
+
+        with _capture_outcomes() as poll:
+            resp = requests.post(url, headers=headers, json={**data, "event_id": event_id})
+            assert resp.status_code == 200, resp.text
+            assert resp.json()["id"] == event_id
+
+            # Relay aggregates outcomes and drops their event IDs. Snowflake project
+            # IDs are unique per test, and this is the only event posted for the key.
+            own_key = (self.project.organization_id, self.project.id, self.projectkey.id)
+
+            def poll_own_outcome() -> dict[str, Any] | None:
+                result = poll()
+                if (
+                    result is not None
+                    and (result.get("org_id"), result.get("project_id"), result.get("key_id"))
+                    == own_key
+                    and result.get("category") == DataCategory.ERROR
+                ):
+                    return result
+                return None
+
+            result = self.wait_for_ingest_consumer(poll_own_outcome)
+
+        assert result is not None, f"Relay produced no error outcome for event {event_id}"
+        assert result["quantity"] == 1, result
+        assert result["outcome"] == outcome, result
+        assert result["reason"] == reason, result
+        assert _get_event(self.project.id, event_id) is None
 
     def post_and_retrieve_attachment(
         self,

tests/relay_integration/test_integration.py

@@ -15,6 +15,7 @@
 from sentry.testutils.relay import RelayStoreHelper
 from sentry.testutils.skips import requires_kafka
 from sentry.testutils.thread_leaks.pytest import thread_leak_allowlist
+from sentry.utils.outcomes import Outcome
 
 pytestmark = [requires_kafka]
 
@@ -233,7 +234,7 @@ def test_accepted_with_valid_signature(self) -> None:
 
         signature = create_trusted_relay_signature(secret_key)
 
-        event = self.post_and_try_retrieve_event(
+        event = self.post_and_retrieve_event(
             event_data,
             headers={
                 "x-sentry-auth": self.auth_header,
@@ -243,7 +244,6 @@ def test_accepted_with_valid_signature(self) -> None:
             },
         )
 
-        assert event is not None
         assert event.message == "should be accepted with signature"
 
     def test_not_trusted_relay(self) -> None:
@@ -278,8 +278,9 @@ def test_not_trusted_relay(self) -> None:
             "x-sentry-relay-id": relay_id,
         }
 
-        event = self.post_and_try_retrieve_event(event_data, headers=headers)
-        assert event is None
+        self.post_and_assert_rejected(
+            event_data, headers=headers, outcome=Outcome.INVALID, reason="invalid_signature"
+        )
 
         url = self.get_relay_store_url(self.project.id)
         resp = requests.post(
@@ -312,8 +313,9 @@ def test_expired_signature(self) -> None:
             "x-sentry-relay-id": relay_id,
         }
 
-        event = self.post_and_try_retrieve_event(event_data, headers=headers)
-        assert event is None
+        self.post_and_assert_rejected(
+            event_data, headers=headers, outcome=Outcome.INVALID, reason="invalid_signature"
+        )
 
         url = self.get_relay_store_url(self.project.id)
         resp = requests.post(
@@ -343,11 +345,9 @@ def test_invalid_signature(self) -> None:
             "x-sentry-relay-id": relay_id,
         }
 
-        event = self.post_and_try_retrieve_event(
-            event_data,
-            headers=headers,
+        self.post_and_assert_rejected(
+            event_data, headers=headers, outcome=Outcome.INVALID, reason="invalid_signature"
         )
-        assert event is None
 
         url = self.get_relay_store_url(self.project.id)
         resp = requests.post(
@@ -376,11 +376,9 @@ def test_missing_signature(self) -> None:
             "x-sentry-relay-id": relay_id,
         }
 
-        event = self.post_and_try_retrieve_event(
-            event_data,
-            headers=headers,
+        self.post_and_assert_rejected(
+            event_data, headers=headers, outcome=Outcome.INVALID, reason="missing_signature"
         )
-        assert event is None
 
         url = self.get_relay_store_url(self.project.id)
         resp = requests.post(
@@ -410,11 +408,9 @@ def test_signature_header_is_none(self) -> None:
             "x-sentry-relay-id": relay_id,
         }
 
-        event = self.post_and_try_retrieve_event(
-            event_data,
-            headers=headers,
+        self.post_and_assert_rejected(
+            event_data, headers=headers, outcome=Outcome.INVALID, reason="missing_signature"
         )
-        assert event is None
 
         url = self.get_relay_store_url(self.project.id)
         resp = requests.post(

tests/relay_integration/test_message_filters.py

@@ -1,6 +1,7 @@
 from typing import Any
 
 from sentry.ingest.inbound_filters import (
+    FilterStatKeys,
     _browser_extensions_filter,
     _FilterSpec,
     _legacy_browsers_filter,
@@ -11,6 +12,7 @@
 from sentry.testutils.cases import TransactionTestCase
 from sentry.testutils.relay import RelayStoreHelper
 from sentry.testutils.skips import requires_kafka
+from sentry.utils.outcomes import Outcome
 from sentry.utils.safe import set_path
 
 pytestmark = [requires_kafka]
@@ -36,8 +38,9 @@ def _get_message_with_bad_ip(self) -> Any:
     def test_should_filter_local_ip_addresses_when_enabled(self) -> None:
         self._set_filter_state(_localhost_filter, "1")
         message = self._get_message_with_bad_ip()
-        event = self.post_and_try_retrieve_event(message)
-        assert event is None
+        self.post_and_assert_rejected(
+            message, outcome=Outcome.FILTERED, reason=FilterStatKeys.LOCALHOST
+        )
 
     def test_should_not_filter_bad_ip_addresses_when_disabled(self) -> None:
         self._set_filter_state(_localhost_filter, "0")
@@ -57,8 +60,9 @@ def _get_message_with_bad_extension(self) -> Any:
     def test_should_filter_browser_extensions_when_enabled(self) -> None:
         self._set_filter_state(_browser_extensions_filter, "1")
         message = self._get_message_with_bad_extension()
-        event = self.post_and_try_retrieve_event(message)
-        assert event is None
+        self.post_and_assert_rejected(
+            message, outcome=Outcome.FILTERED, reason=FilterStatKeys.BROWSER_EXTENSION
+        )
 
     def test_should_not_filter_browser_extensions_when_disabled(self) -> None:
         self._set_filter_state(_browser_extensions_filter, "0")
@@ -81,8 +85,9 @@ def _get_message_from_webcrawler(self) -> Any:
     def test_should_filter_web_crawlers_when_enabled(self) -> None:
         self._set_filter_state(_web_crawlers_filter, "1")
         message = self._get_message_from_webcrawler()
-        event = self.post_and_try_retrieve_event(message)
-        assert event is None
+        self.post_and_assert_rejected(
+            message, outcome=Outcome.FILTERED, reason=FilterStatKeys.WEB_CRAWLER
+        )
 
     def test_should_not_filter_web_crawlers_when_disabled(self) -> None:
         self._set_filter_state(_web_crawlers_filter, "0")
@@ -109,8 +114,9 @@ def _get_message_from_legacy_browser(self) -> Any:
     def test_should_filter_legacy_browsers(self) -> None:
         self._set_filter_state(_legacy_browsers_filter, "1")
         message = self._get_message_from_legacy_browser()
-        event = self.post_and_try_retrieve_event(message)
-        assert event is None
+        self.post_and_assert_rejected(
+            message, outcome=Outcome.FILTERED, reason=FilterStatKeys.LEGACY_BROWSER
+        )
 
     def test_should_not_filter_legacy_browsers_when_disabled(self) -> None:
         self._set_filter_state(_legacy_browsers_filter, "0")