fix(tests): Make test_failures a Queue (#4112)

Prevent manual `sleep` calls when waiting for
`mini_sentry.test_failures`. Because `sleep` blocks the tests
unconditionally, we tend to choose short timeouts which then produce
flakiness if the event we're waiting on does not occur fast enough.

By making `test_failures` a `queue.Queue`, we can set longer timeouts
while at the same time shortening the actual sleep time in most cases.

Author: jjbayer

tests/integration/fixtures/mini_sentry.py

@@ -6,7 +6,7 @@
 import re
 import uuid
 from copy import deepcopy
-from queue import Queue
+from queue import Empty, Queue
 
 import pytest
 
@@ -41,8 +41,7 @@ def __init__(self, server_address, app):
         self.captured_events = Queue()
         self.captured_outcomes = Queue()
         self.captured_metrics = Queue()
-        self.test_failures = []
-        self.reraise_test_failures = True
+        self.test_failures = Queue()
         self.hits = {}
         self.known_relays = {}
         self.fail_on_relay_error = True
@@ -64,9 +63,21 @@ def hit(self, path):
         self.hits.setdefault(path, 0)
         self.hits[path] += 1
 
+    def current_test_failures(self):
+        """Return current list of test failures without waiting for additional failures."""
+        try:
+            while failure := self.test_failures.get_nowait():
+                yield failure
+        except Empty:
+            return
+
+    def clear_test_failures(self):
+        """Reset test failures to an empty queue."""
+        self.test_failures = Queue()
+
     def format_failures(self):
         s = ""
-        for route, error in self.test_failures:
+        for route, error in self.current_test_failures():
             s += f"> {route}: {error}\n"
         return s
 
@@ -296,7 +307,7 @@ def store_internal_error_event():
 
         if event is not None and sentry.fail_on_relay_error:
             error = AssertionError("Relay sent us event: " + get_error_message(event))
-            sentry.test_failures.append(("/api/666/envelope/", error))
+            sentry.test_failures.put(("/api/666/envelope/", error))
 
         return jsonify({"event_id": uuid.uuid4().hex})
 
@@ -452,10 +463,10 @@ def fail(e):
         raise e
 
     def reraise_test_failures():
-        if sentry.test_failures and sentry.reraise_test_failures:
+        if not sentry.test_failures.empty():
             pytest.fail(
                 "{n} exceptions happened in mini_sentry:\n\n{failures}".format(
-                    n=len(sentry.test_failures), failures=sentry.format_failures()
+                    n=sentry.test_failures.qsize(), failures=sentry.format_failures()
                 )
             )
 

tests/integration/fixtures/relay.py

@@ -1,5 +1,6 @@
 import json
 import os
+from queue import Queue
 import sys
 import uuid
 import signal
@@ -211,11 +212,11 @@ def inner(
             relay.wait_relay_health_check()
 
             # Filter out health check failures, which can happen during startup
-            mini_sentry.test_failures = [
-                f
-                for f in mini_sentry.test_failures
-                if "Health check probe" not in str(f)
-            ]
+            filtered_test_failures = Queue()
+            for f in mini_sentry.current_test_failures():
+                if "Health check probe" not in str(f):
+                    filtered_test_failures.put(f)
+            mini_sentry.test_failures = filtered_test_failures
 
         return relay
 

tests/integration/test_basic.py

@@ -110,7 +110,7 @@ def get_project_config():
         relay.shutdown(sig=signal.SIGINT)
         pytest.raises(queue.Empty, lambda: mini_sentry.captured_events.get(timeout=1))
 
-        failures = mini_sentry.test_failures
+        failures = mini_sentry.current_test_failures()
         assert failures
 
         # we are expecting at least a dropped unfinished future error
@@ -121,7 +121,7 @@ def get_project_config():
                 dropped_unfinished_error_found = True
         assert dropped_unfinished_error_found
     finally:
-        mini_sentry.test_failures.clear()
+        mini_sentry.clear_test_failures()
 
 
 @pytest.mark.parametrize("trailing_slash", [True, False])

tests/integration/test_config.py

@@ -14,9 +14,9 @@ def test_invalid_kafka_config_should_fail(mini_sentry, relay_with_processing):
     relay = relay_with_processing(options=options, wait_health_check=False)
     assert relay.wait_for_exit() != 0
 
-    error = str(mini_sentry.test_failures.pop(0))
+    error = str(mini_sentry.test_failures.get_nowait())
     assert "__unknown" in error
-    error = str(mini_sentry.test_failures.pop(0))
+    error = str(mini_sentry.test_failures.get_nowait())
     assert "profiles" in error.lower()
 
 
@@ -26,5 +26,5 @@ def test_invalid_topics_raise_error(mini_sentry, relay_with_processing):
     relay = relay_with_processing(options=options, wait_health_check=False)
     assert relay.wait_for_exit() != 0
 
-    error = str(mini_sentry.test_failures.pop(0))
+    error = str(mini_sentry.test_failures.get_nowait())
     assert "failed to validate the topic with name" in error

tests/integration/test_envelope.py

@@ -465,12 +465,12 @@ def get_project_config():
 
         include_global = True
         # Clear errors because we log error when we request global config yet we dont receive it.
-        assert len(mini_sentry.test_failures) > 0
-        assert {str(e) for _, e in mini_sentry.test_failures} == {
+        assert not mini_sentry.test_failures.empty()
+        assert {str(e) for _, e in mini_sentry.current_test_failures()} == {
             "Relay sent us event: global config missing in upstream response"
         }
     finally:
-        mini_sentry.test_failures.clear()
+        mini_sentry.clear_test_failures()
 
     envelopes = []
     # Check that we received exactly {envelope_qty} envelopes.

tests/integration/test_healthchecks.py

@@ -53,7 +53,7 @@ def test_readiness(mini_sentry, relay):
         mini_sentry.app.view_functions["check_challenge"] = original_check_challenge
         relay.wait_relay_health_check()
     finally:
-        mini_sentry.test_failures.clear()
+        mini_sentry.clear_test_failures()
 
     response = relay.get("/api/relay/healthcheck/ready/")
     assert response.status_code == 200
@@ -69,7 +69,7 @@ def test_readiness_flag(mini_sentry, relay):
         response = wait_get(relay, "/api/relay/healthcheck/ready/")
         assert response.status_code == 200
     finally:
-        mini_sentry.test_failures.clear()
+        mini_sentry.clear_test_failures()
 
 
 def test_readiness_proxy(mini_sentry, relay):
@@ -88,12 +88,11 @@ def test_readiness_not_enough_memory_bytes(mini_sentry, relay):
     )
 
     response = wait_get(relay, "/api/relay/healthcheck/ready/")
-    time.sleep(1.0)  # Wait for error
-    error = str(mini_sentry.test_failures.pop(0))
+    error = str(mini_sentry.test_failures.get(timeout=2))
     assert "Not enough memory" in error and ">= 42" in error
-    error = str(mini_sentry.test_failures.pop(0))
+    error = str(mini_sentry.test_failures.get(timeout=1))
     assert "Health check probe 'system memory'" in error
-    error = str(mini_sentry.test_failures.pop(0))
+    error = str(mini_sentry.test_failures.get(timeout=1))
     assert "Health check probe 'spool health'" in error
     assert response.status_code == 503
 
@@ -105,12 +104,12 @@ def test_readiness_not_enough_memory_percent(mini_sentry, relay):
         wait_health_check=False,
     )
     response = wait_get(relay, "/api/relay/healthcheck/ready/")
-    time.sleep(1.0)  # Wait for error
-    error = str(mini_sentry.test_failures.pop(0))
+
+    error = str(mini_sentry.test_failures.get(timeout=2))
     assert "Not enough memory" in error and ">= 1.00%" in error
-    error = str(mini_sentry.test_failures.pop(0))
+    error = str(mini_sentry.test_failures.get(timeout=1))
     assert "Health check probe 'system memory'" in error
-    error = str(mini_sentry.test_failures.pop(0))
+    error = str(mini_sentry.test_failures.get(timeout=1))
     assert "Health check probe 'spool health'" in error
     assert response.status_code == 503
 
@@ -123,8 +122,8 @@ def test_readiness_depends_on_aggregator_being_full(mini_sentry, relay):
     )
 
     response = wait_get(relay, "/api/relay/healthcheck/ready/")
-    time.sleep(0.3)  # Wait for error
-    error = str(mini_sentry.test_failures.pop())
+
+    error = str(mini_sentry.test_failures.get(timeout=1))
     assert "Health check probe 'aggregator'" in error
     assert response.status_code == 503
 
@@ -140,20 +139,19 @@ def test_readiness_depends_on_aggregator_being_full_after_metrics(mini_sentry, r
 
     for _ in range(100):
         response = wait_get(relay, "/api/relay/healthcheck/ready/")
-        print(response, response.status_code)
         if response.status_code == 503:
-            error = str(mini_sentry.test_failures.pop())
-            assert "Health check probe 'aggregator'" in error
-            error = str(mini_sentry.test_failures.pop())
+            error = str(mini_sentry.test_failures.get(timeout=1))
             assert "aggregator limit exceeded" in error
+            error = str(mini_sentry.test_failures.get(timeout=1))
+            assert "Health check probe 'aggregator'" in error
             return
         time.sleep(0.1)
 
     assert False, "health check never failed"
 
 
 def test_readiness_disk_spool(mini_sentry, relay):
-    mini_sentry.reraise_test_failures = False
+    mini_sentry.fail_on_relay_error = False
     temp = tempfile.mkdtemp()
     dbfile = os.path.join(temp, "buffer.db")
 

tests/integration/test_metrics.py

@@ -1182,7 +1182,7 @@ def test_transaction_metrics_not_extracted_on_unsupported_version(
     tx_consumer.assert_empty()
 
     if unsupported_version < TRANSACTION_EXTRACT_MIN_SUPPORTED_VERSION:
-        error = str(mini_sentry.test_failures.pop(0))
+        error = str(mini_sentry.test_failures.get_nowait())
         assert "Processing Relay outdated" in error
 
     metrics_consumer.assert_empty()

tests/integration/test_projectconfigs.py

@@ -260,11 +260,11 @@ def test_unparsable_project_config(mini_sentry, relay):
 
     def assert_clear_test_failures():
         try:
-            assert {str(e) for _, e in mini_sentry.test_failures} == {
+            assert {str(e) for _, e in mini_sentry.current_test_failures()} == {
                 f"Relay sent us event: error fetching project state {public_key}: invalid type: integer `99`, expected a string",
             }
         finally:
-            mini_sentry.test_failures.clear()
+            mini_sentry.clear_test_failures()
 
     # Event is not propagated, relay logs an error:
     relay.send_event(project_key)
@@ -343,11 +343,11 @@ def test_cached_project_config(mini_sentry, relay):
     try:
         relay.send_event(project_key)
         time.sleep(0.5)
-        assert {str(e) for _, e in mini_sentry.test_failures} == {
+        assert {str(e) for _, e in mini_sentry.current_test_failures()} == {
             f"Relay sent us event: error fetching project state {public_key}: invalid type: integer `99`, expected a string",
         }
     finally:
-        mini_sentry.test_failures.clear()
+        mini_sentry.clear_test_failures()
 
 
 def test_get_global_config(mini_sentry, relay):

tests/integration/test_query.py

@@ -38,7 +38,7 @@ def test_local_project_config(mini_sentry, relay):
     dsn_key = config["publicKeys"][0]["publicKey"]
 
     relay.wait_relay_health_check()
-    mini_sentry.test_failures.clear()
+    mini_sentry.clear_test_failures()
 
     relay.send_event(project_id, dsn_key=dsn_key)
     event = mini_sentry.captured_events.get(timeout=1).get_event()
@@ -148,11 +148,10 @@ def get_project_config():
         assert event["logentry"] == {"formatted": "Hello, World!"}
         assert retry_count == 3
 
-        if mini_sentry.test_failures:
-            for _, error in mini_sentry.test_failures:
-                assert isinstance(error, (socket.error, AssertionError))
+        for _, error in mini_sentry.current_test_failures():
+            assert isinstance(error, (socket.error, AssertionError))
     finally:
-        mini_sentry.test_failures.clear()
+        mini_sentry.clear_test_failures()
 
 
 def test_query_retry_maxed_out(mini_sentry, relay_with_processing, events_consumer):
@@ -190,18 +189,18 @@ def get_project_config():
     )
 
     # No error messages yet
-    assert not mini_sentry.test_failures
+    assert mini_sentry.test_failures.empty()
 
     try:
         relay.send_event(42)
         time.sleep(query_timeout)
 
         assert request_count == 1 + RETRIES
-        assert {str(e) for _, e in mini_sentry.test_failures} == {
+        assert {str(e) for _, e in mini_sentry.current_test_failures()} == {
             "Relay sent us event: error fetching project states: upstream request returned error 500 Internal Server Error: no error details",
         }
     finally:
-        mini_sentry.test_failures.clear()
+        mini_sentry.clear_test_failures()
 
 
 @pytest.mark.parametrize("disabled", (True, False))

tests/integration/test_store.py

@@ -193,7 +193,7 @@ def configure_static_project(dir):
     assert event["logentry"] == {"formatted": "Hello, World!"}
 
     sleep(1)  # Regression test: Relay tried to issue a request for 0 states
-    if mini_sentry.test_failures:
+    if not mini_sentry.test_failures.empty():
         raise AssertionError(
             f"Exceptions happened in mini_sentry: {mini_sentry.format_failures()}"
         )
@@ -230,15 +230,15 @@ def test_store_with_low_memory(mini_sentry, relay):
         pytest.raises(queue.Empty, lambda: mini_sentry.captured_events.get(timeout=1))
 
         found_queue_error = False
-        for _, error in mini_sentry.test_failures:
+        for _, error in mini_sentry.current_test_failures():
             assert isinstance(error, AssertionError)
             if "failed to queue envelope" in str(error):
                 found_queue_error = True
                 break
 
         assert found_queue_error
     finally:
-        mini_sentry.test_failures.clear()
+        mini_sentry.clear_test_failures()
 
 
 def test_store_max_concurrent_requests(mini_sentry, relay):
@@ -954,7 +954,7 @@ def server_error(*args, **kwargs):
         assert event["logentry"] == {"formatted": "Hello, World!"}
     finally:
         # Relay reports authentication errors, which is fine.
-        mini_sentry.test_failures.clear()
+        mini_sentry.clear_test_failures()
 
 
 def test_events_are_retried(relay, mini_sentry):
@@ -1181,7 +1181,7 @@ def counted_check_challenge(*args, **kwargs):
     evt.clear()
     assert evt.wait(2)
     # clear authentication errors accumulated until now
-    mini_sentry.test_failures.clear()
+    mini_sentry.clear_test_failures()
     # check that we have had some auth that succeeded
     auth_count_3 = counter[0]
     assert auth_count_2 < auth_count_3
@@ -1250,7 +1250,7 @@ def counted_check_challenge(*args, **kwargs):
     # to be sure verify that we have only been called once (after failing)
     assert counter[1] == 1
     # clear authentication errors accumulated until now
-    mini_sentry.test_failures.clear()
+    mini_sentry.clear_test_failures()
 
 
 def test_buffer_events_during_outage(relay, mini_sentry):