test(integration): Grab captured events with default timeout (#5354)

Author: Dav1dde

tests/integration/fixtures/mini_sentry.py

@@ -203,8 +203,11 @@ def set_global_config_option(self, option_name, value):
         # must be called before initializing relay fixture
         self.global_config["options"][option_name] = value
 
+    def get_captured_event(self, *, timeout=None):
+        return self.captured_events.get(timeout=timeout or self.timeout)
+
     def get_client_report(self, timeout=None):
-        envelope = self.captured_events.get(timeout=timeout or self.timeout)
+        envelope = self.get_captured_event(timeout=timeout)
         items = envelope.items
         assert len(items) == 1
         item = items[0]
@@ -213,7 +216,7 @@ def get_client_report(self, timeout=None):
         return json.loads(item.payload.bytes)
 
     def get_metrics(self, timeout=None):
-        envelope = self.captured_events.get(timeout=timeout or self.timeout)
+        envelope = self.get_captured_event(timeout=timeout)
         items = envelope.items
         assert len(items) == 1
         item = items[0]

tests/integration/test_attachments.py

@@ -172,8 +172,7 @@ def test_attachments_pii(mini_sentry, relay):
         relay.send_attachments(project_id, event_id, [attachment])
 
     payloads = {
-        mini_sentry.captured_events.get(timeout=5).items[0].payload.bytes
-        for _ in range(2)
+        mini_sentry.get_captured_event().items[0].payload.bytes for _ in range(2)
     }
     assert payloads == {
         b"here's an IP that should get masked -> ********* <-",
@@ -217,9 +216,7 @@ def test_view_hierarchy_scrubbing(mini_sentry, relay, feature_flags, expected):
     relay.send_envelope(project_id, envelope)
 
     relay.send_envelope(project_id, envelope)
-    payload = json.loads(
-        mini_sentry.captured_events.get(timeout=5).items[0].payload.bytes
-    )
+    payload = json.loads(mini_sentry.get_captured_event().items[0].payload.bytes)
     assert payload == {"rendering_system": "UIKIT", "identifier": expected}
 
 
@@ -270,7 +267,7 @@ def test_attachment_scrubbing_with_fallback(
 
     relay.send_envelope(project_id, envelope)
 
-    payload = mini_sentry.captured_events.get(timeout=5).items[0].payload.bytes
+    payload = mini_sentry.get_captured_event().items[0].payload.bytes
     assert payload == expected
 
 
@@ -294,9 +291,7 @@ def test_view_hierarchy_not_scrubbed_without_config(mini_sentry, relay):
     )
 
     relay.send_envelope(project_id, envelope)
-    payload = json.loads(
-        mini_sentry.captured_events.get(timeout=5).items[0].payload.bytes
-    )
+    payload = json.loads(mini_sentry.get_captured_event().items[0].payload.bytes)
     assert payload == json_payload
 
 
@@ -330,7 +325,7 @@ def test_attachments_pii_logfile(mini_sentry, relay):
 
     relay.send_envelope(project_id, envelope)
 
-    scrubbed_payload = mini_sentry.captured_events.get(timeout=5).items[0].payload.bytes
+    scrubbed_payload = mini_sentry.get_captured_event().items[0].payload.bytes
 
     assert (
         scrubbed_payload

tests/integration/test_basic.py

@@ -1,5 +1,4 @@
 import datetime
-import queue
 import os
 import gzip
 import sqlite3
@@ -34,7 +33,7 @@ def get_project_config():
 
     # When using the memory envelope buffer, we optimistically do not do anything on shutdown, which means that the
     # buffer will try and pop as always as long as it can (within the shutdown timeout).
-    event = mini_sentry.captured_events.get(timeout=0).get_event()
+    event = mini_sentry.get_captured_event().get_event()
     assert event["logentry"] == {"formatted": "Hello, World!"}
 
 
@@ -137,7 +136,7 @@ def get_project_config():
         sleep(0.5)  # Give the event time to get stuck
 
         relay.shutdown(sig=signal.SIGINT)
-        pytest.raises(queue.Empty, lambda: mini_sentry.captured_events.get(timeout=1))
+        assert mini_sentry.captured_events.empty()
 
         failures = mini_sentry.current_test_failures()
         assert failures
@@ -177,7 +176,7 @@ def test_store_pixel_gif(mini_sentry, relay, input, trailing_slash):
     response.raise_for_status()
     assert response.headers["content-type"] == "image/gif"
 
-    event = mini_sentry.captured_events.get(timeout=1).get_event()
+    event = mini_sentry.get_captured_event().get_event()
     assert event["logentry"]["formatted"] == "im in ur query params"
 
 
@@ -197,7 +196,7 @@ def test_store_post_trailing_slash(mini_sentry, relay, route):
     )
     response.raise_for_status()
 
-    event = mini_sentry.captured_events.get(timeout=1).get_event()
+    event = mini_sentry.get_captured_event().get_event()
     assert event["logentry"]["formatted"] == "hi"
 
 
@@ -238,7 +237,7 @@ def test_store_allowed_origins_passes(mini_sentry, relay, allowed_origins):
     )
 
     if should_be_allowed:
-        mini_sentry.captured_events.get(timeout=1).get_event()
+        assert mini_sentry.get_captured_event().get_event() is not None
     assert mini_sentry.captured_events.empty()
 
 
@@ -360,7 +359,7 @@ def send_transaction_with_dsc(mini_sentry, relay, project_id, sampling_project_k
         },
     )
 
-    return mini_sentry.captured_events.get(timeout=1).get_transaction_event()
+    return mini_sentry.get_captured_event().get_transaction_event()
 
 
 def test_root_project_disabled(mini_sentry, relay):

tests/integration/test_clock_drift.py

@@ -37,9 +37,7 @@ def test_clock_drift_applied_when_timestamp_is_too_old(mini_sentry, relay):
 
     relay.send_envelope(project_id, envelope, headers={"Accept-Encoding": "gzip"})
 
-    transaction_event = mini_sentry.captured_events.get(
-        timeout=1
-    ).get_transaction_event()
+    transaction_event = mini_sentry.get_captured_event().get_transaction_event()
 
     error_name, error_metadata = transaction_event["_meta"]["timestamp"][""]["err"][0]
     assert error_name == "clock_drift"
@@ -63,9 +61,7 @@ def test_clock_drift_not_applied_when_timestamp_is_recent(mini_sentry, relay):
 
     relay.send_envelope(project_id, envelope, headers={"Accept-Encoding": "gzip"})
 
-    transaction_event = mini_sentry.captured_events.get(
-        timeout=1
-    ).get_transaction_event()
+    transaction_event = mini_sentry.get_captured_event().get_transaction_event()
     assert "_meta" not in transaction_event
     assert transaction_event["timestamp"] == five_minutes_ago.timestamp()
 
@@ -84,9 +80,7 @@ def test_clock_drift_not_applied_when_sent_at_is_not_supplied(mini_sentry, relay
 
     relay.send_envelope(project_id, envelope, headers={"Accept-Encoding": "gzip"})
 
-    transaction_event = mini_sentry.captured_events.get(
-        timeout=1
-    ).get_transaction_event()
+    transaction_event = mini_sentry.get_captured_event().get_transaction_event()
     error_name, error_metadata = transaction_event["_meta"]["timestamp"][""]["err"][0]
     # In case clock drift is not run, we expect timestamps normalization to go into effect to mark timestamps as
     # past or future if they surpass a certain threshold.

tests/integration/test_dynamic_sampling.py

@@ -254,9 +254,8 @@ def test_it_removes_events(mini_sentry, relay):
     # send the event, the transaction should be removed.
     relay.send_envelope(project_id, envelope)
     # the event should be removed by Relay sampling
-    assert mini_sentry.captured_events.get(timeout=2) == only_items("metric_buckets")
-    with pytest.raises(queue.Empty):
-        mini_sentry.captured_events.get(timeout=1)
+    assert mini_sentry.get_captured_event() == only_items("metric_buckets")
+    assert mini_sentry.captured_events.empty()
 
     outcomes = mini_sentry.captured_outcomes.get(timeout=2)
     assert outcomes is not None
@@ -285,7 +284,7 @@ def test_it_does_not_sample_error(mini_sentry, relay):
     # send the event, the transaction should be removed.
     relay.send_envelope(project_id, envelope)
     # test that error is kept by Relay
-    envelope = mini_sentry.captured_events.get(timeout=1)
+    envelope = mini_sentry.get_captured_event()
     assert envelope is not None
     # double check that we get back our object
     # we put the id in extra since Relay overrides the initial event_id
@@ -325,7 +324,7 @@ def test_it_tags_error(mini_sentry, relay, expected_sampled, sample_rate):
     # send the event, the transaction should be removed.
     relay.send_envelope(project_id, envelope)
     # test that error is kept by Relay
-    envelope = mini_sentry.captured_events.get(timeout=1)
+    envelope = mini_sentry.get_captured_event()
     assert envelope is not None
     # double check that we get back our object
     # we put the id in extra since Relay overrides the initial event_id
@@ -426,7 +425,7 @@ def test_it_keeps_events(mini_sentry, relay):
     # send the event, the transaction should be removed.
     relay.send_envelope(project_id, envelope)
     # the event should be left alone by Relay sampling
-    envelope = mini_sentry.captured_events.get(timeout=1)
+    envelope = mini_sentry.get_captured_event()
     assert envelope is not None
     # double check that we get back our object
     # we put the id in extra since Relay overrides the initial event_id
@@ -491,9 +490,8 @@ def test_uses_trace_public_key(mini_sentry, relay):
     # send the event, the transaction should be removed.
     relay.send_envelope(project_id2, envelope)
     # the event should be removed by Relay sampling
-    assert mini_sentry.captured_events.get(timeout=2) == only_items("metric_buckets")
-    with pytest.raises(queue.Empty):
-        mini_sentry.captured_events.get(timeout=1)
+    assert mini_sentry.get_captured_event() == only_items("metric_buckets")
+    assert mini_sentry.captured_events.empty()
 
     # and it should create an outcome
     outcomes = mini_sentry.captured_outcomes.get(timeout=2)  # Spans
@@ -514,7 +512,7 @@ def test_uses_trace_public_key(mini_sentry, relay):
     relay.send_envelope(project_id1, envelope)
 
     # the event should be passed along to upstream (with the transaction unchanged)
-    evt = mini_sentry.captured_events.get(timeout=1).get_transaction_event()
+    evt = mini_sentry.get_captured_event().get_transaction_event()
     assert evt is not None
 
     # no outcome should be generated (since the event is passed along to the upstream)
@@ -570,11 +568,8 @@ def test_multi_item_envelope(mini_sentry, relay, rule_type, event_factory):
         # send the event, the transaction should be removed.
         relay.send_envelope(project_id, envelope)
         # the event should be removed by Relay sampling
-        assert mini_sentry.captured_events.get(timeout=2) == only_items(
-            "metric_buckets"
-        )
-        with pytest.raises(queue.Empty):
-            mini_sentry.captured_events.get(timeout=1)
+        assert mini_sentry.get_captured_event() == only_items("metric_buckets")
+        assert mini_sentry.captured_events.empty()
 
         outcomes = mini_sentry.captured_outcomes.get(timeout=2)
         assert outcomes is not None
@@ -602,7 +597,7 @@ def collect_events_batch(expected_count, timeout_per_event=0.1):
 
         for _ in range(expected_count * 3):  # Allow for some extra attempts
             try:
-                received_envelope = mini_sentry.captured_events.get(
+                received_envelope = mini_sentry.get_captured_event(
                     timeout=timeout_per_event
                 )
                 if (
@@ -696,8 +691,7 @@ def test_relay_chain(
     )
     relay.send_envelope(project_id, envelope)
 
-    envelope = mini_sentry.captured_events.get(timeout=1)
-    envelope.get_transaction_event()
+    assert mini_sentry.get_captured_event().get_transaction_event() is not None
 
 
 @pytest.mark.parametrize("mode", ["default", "chain"])
@@ -892,4 +886,4 @@ def test_invalid_global_generic_filters_skip_dynamic_sampling(mini_sentry, relay
     envelope, _, _ = _create_transaction_envelope(public_key, client_sample_rate=0)
 
     relay.send_envelope(project_id, envelope)
-    assert mini_sentry.captured_events.get(timeout=1)
+    assert mini_sentry.get_captured_event()

tests/integration/test_envelope.py

@@ -15,7 +15,7 @@ def test_envelope(mini_sentry, relay_chain):
         project_id, envelope, headers={"Accept-Encoding": "gzip"}
     )
 
-    event = mini_sentry.captured_events.get(timeout=1).get_event()
+    event = mini_sentry.get_captured_event().get_event()
     assert event["logentry"] == {"formatted": "Hello, World!"}
     # The response should not be compressed
     assert "content-encoding" not in response.headers
@@ -51,7 +51,7 @@ def test_envelope_close_connection(mini_sentry, relay):
         sock.close()  # Close the connection
 
     for _ in range(num_iterations):
-        envelope = mini_sentry.captured_events.get(timeout=1)
+        envelope = mini_sentry.get_captured_event()
         assert envelope
     assert mini_sentry.captured_events.empty()
 
@@ -66,7 +66,7 @@ def test_envelope_empty(mini_sentry, relay):
 
     # there is nothing sent to the upstream
     with pytest.raises(queue.Empty):
-        mini_sentry.captured_events.get(timeout=1)
+        mini_sentry.get_captured_event(timeout=1)
 
 
 def test_envelope_without_header(mini_sentry, relay):
@@ -82,7 +82,7 @@ def test_envelope_without_header(mini_sentry, relay):
         headers={"X-Sentry-Auth": ""},  # Empty auth header is ignored by Relay
     )
 
-    event = mini_sentry.captured_events.get(timeout=1).get_event()
+    event = mini_sentry.get_captured_event().get_event()
     assert event["logentry"] == {"formatted": "Hello, World!"}
 
 
@@ -101,8 +101,8 @@ def test_unknown_item(mini_sentry, relay):
     relay.send_envelope(PROJECT_ID, envelope)
 
     envelopes = [  # non-event items are split into separate envelopes, so fetch 2x here
-        mini_sentry.captured_events.get(timeout=1),
-        mini_sentry.captured_events.get(timeout=1),
+        mini_sentry.get_captured_event(),
+        mini_sentry.get_captured_event(),
     ]
 
     types = {
@@ -129,12 +129,11 @@ def test_drop_unknown_item(mini_sentry, relay):
     envelope.add_item(attachment)
     relay.send_envelope(PROJECT_ID, envelope)
 
-    envelope = mini_sentry.captured_events.get(timeout=1)
+    envelope = mini_sentry.get_captured_event()
     assert len(envelope.items) == 1
     assert envelope.items[0].type == "attachment"
 
-    with pytest.raises(queue.Empty):
-        mini_sentry.captured_events.get(timeout=1)
+    assert mini_sentry.captured_events.empty()
 
 
 def generate_transaction_item():
@@ -218,8 +217,7 @@ def test_empty_measurement_interface(mini_sentry, relay_chain):
     envelope.add_transaction(transaction_item)
     relay.send_envelope(42, envelope)
 
-    envelope = mini_sentry.captured_events.get(timeout=1)
-    event = envelope.get_transaction_event()
+    event = mini_sentry.get_captured_event().get_transaction_event()
 
     assert event["transaction"] == "/organizations/:orgId/performance/:eventSlug/"
     assert "measurements" not in event, event
@@ -509,7 +507,7 @@ def test_sample_rate_propagates_across_orgs(
     envelope.add_transaction(transaction_item)
     relay.send_envelope(downstream_project_id, envelope)
 
-    dsc = mini_sentry.captured_events.get(timeout=2).headers["trace"]
+    dsc = mini_sentry.get_captured_event().headers["trace"]
 
     # Because the organization IDs mismatch, the DSC should be regenerated
     assert dsc["public_key"] == downstream_public_key

tests/integration/test_ephemeral_config.py

@@ -87,7 +87,7 @@ def test_store_via_ephemeral_relay(
 
     raw_payload = {"message": "Hello, World!"}
     relay.send_event(project_id, payload=raw_payload)
-    event = mini_sentry.captured_events.get(timeout=1).get_event()
+    event = mini_sentry.get_captured_event().get_event()
 
     if mode == "proxy":
         assert event == raw_payload

tests/integration/test_feedback.py

@@ -130,7 +130,7 @@ def test_feedback_events_without_processing(mini_sentry, relay_chain):
     relay = relay_chain(min_relay_version="latest")
     relay.send_user_feedback(project_id, replay_item)
 
-    envelope = mini_sentry.captured_events.get(timeout=20)
+    envelope = mini_sentry.get_captured_event()
     assert len(envelope.items) == 1
 
     userfeedback = envelope.items[0]

tests/integration/test_filters.py

@@ -413,8 +413,7 @@ def test_localhost_filter_no_local_header_ips(mini_sentry, relay, headers):
     event = {"user": {"ip_address": None}, "request": {"headers": headers}}
     relay.send_event(project_id, event)
 
-    envelope = mini_sentry.captured_events.get(timeout=1)
-    event = envelope.get_event()
+    event = mini_sentry.get_captured_event().get_event()
     assert event is not None
 
 

tests/integration/test_flags.py

@@ -30,7 +30,7 @@ def test_event_with_flags(relay, mini_sentry):
 
     relay.send_envelope(42, envelope)
 
-    envelope = mini_sentry.captured_events.get(timeout=1)
+    envelope = mini_sentry.get_captured_event()
     assert envelope
     assert envelope.items[0].payload.json["contexts"] == {
         "flags": {
@@ -63,7 +63,7 @@ def test_event_with_flags_malformed(relay, mini_sentry):
     relay.send_envelope(42, envelope)
 
     # Malformed fields are removed. Unknown fields are passed through.
-    envelope = mini_sentry.captured_events.get(timeout=1)
+    envelope = mini_sentry.get_captured_event()
     assert envelope
     assert envelope.items[0].payload.json["contexts"] == {
         "flags": {"values": None, "key": "value", "type": "flags"}
@@ -98,7 +98,7 @@ def test_event_with_flags_malformed_inner_object(relay, mini_sentry):
     relay.send_envelope(42, envelope)
 
     # Malformed fields are removed. Unknown fields are passed through.
-    envelope = mini_sentry.captured_events.get(timeout=1)
+    envelope = mini_sentry.get_captured_event()
     assert envelope
     assert envelope.items[0].payload.json["contexts"] == {
         "flags": {"values": [{"flag": None, "key": "key"}, None], "type": "flags"}