From d35703a3d9222a65e02bbbe78c391d517df8c647 Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Sat, 14 Nov 2026 01:15:54 +0000 Subject: [PATCH] test(relay): Assert rejection outcomes instead of waiting out ingestion Negative Relay integration tests caught the AssertionError from the successful-ingestion helper, so every expected rejection waited the full 16s ingest timeout and any unrelated assertion also counted as a rejection. Read the per-worker outcomes topic from its current high watermarks and wait for the error outcome Relay emits for the test's org, project, and key, then assert its outcome type, reason, and quantity. Co-Authored-By: Joshua Li <14209781+joshuarli@users.noreply.github.com> --- src/sentry/testutils/relay.py | 107 ++++++++++++++++-- tests/relay_integration/test_integration.py | 32 +++--- .../relay_integration/test_message_filters.py | 22 ++-- 3 files changed, 126 insertions(+), 35 deletions(-) diff --git a/src/sentry/testutils/relay.py b/src/sentry/testutils/relay.py index c1d51a584f43..1988fc2253b4 100644 --- a/src/sentry/testutils/relay.py +++ b/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, diff --git a/tests/relay_integration/test_integration.py b/tests/relay_integration/test_integration.py index 1821ccb82463..511edc57fbe2 100644 --- a/tests/relay_integration/test_integration.py +++ b/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( diff --git a/tests/relay_integration/test_message_filters.py b/tests/relay_integration/test_message_filters.py index 22756e1e94a8..fa59cde25b86 100644 --- a/tests/relay_integration/test_message_filters.py +++ b/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")