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