Trying to prevent the grpc setup from being flaky (#4233)

Automatically select a port and not set it by hand also make creating of
the channel more stable.

Author: antonpirker

tests/integrations/grpc/test_grpc.py

@@ -1,10 +1,8 @@
-import os
-
 import grpc
 import pytest
 
 from concurrent import futures
-from typing import List, Optional
+from typing import List, Optional, Tuple
 from unittest.mock import Mock
 
 from sentry_sdk import start_span, start_transaction
@@ -19,37 +17,48 @@
 )
 
 
-PORT = 50051
-PORT += os.getpid() % 100  # avoid port conflicts when running tests in parallel
-
-
-def _set_up(interceptors: Optional[List[grpc.ServerInterceptor]] = None):
+# Set up in-memory channel instead of network-based
+def _set_up(
+    interceptors: Optional[List[grpc.ServerInterceptor]] = None,
+) -> Tuple[grpc.Server, grpc.Channel]:
+    """
+    Sets up a gRPC server and returns both the server and a channel connected to it.
+    This eliminates network dependencies and makes tests more reliable.
+    """
+    # Create server with thread pool
     server = grpc.server(
         futures.ThreadPoolExecutor(max_workers=2),
         interceptors=interceptors,
     )
 
-    add_gRPCTestServiceServicer_to_server(TestService(), server)
-    server.add_insecure_port("[::]:{}".format(PORT))
+    # Add our test service to the server
+    servicer = TestService()
+    add_gRPCTestServiceServicer_to_server(servicer, server)
+
+    # Use dynamic port allocation instead of hardcoded port
+    port = server.add_insecure_port("[::]:0")  # Let gRPC choose an available port
     server.start()
 
-    return server
+    # Create channel connected to our server
+    channel = grpc.insecure_channel(f"localhost:{port}")  # noqa: E231
+
+    return server, channel
 
 
 def _tear_down(server: grpc.Server):
-    server.stop(None)
+    server.stop(grace=None)  # Immediate shutdown
 
 
 @pytest.mark.forked
 def test_grpc_server_starts_transaction(sentry_init, capture_events_forksafe):
     sentry_init(traces_sample_rate=1.0, integrations=[GRPCIntegration()])
     events = capture_events_forksafe()
 
-    server = _set_up()
+    server, channel = _set_up()
 
-    with grpc.insecure_channel("localhost:{}".format(PORT)) as channel:
-        stub = gRPCTestServiceStub(channel)
-        stub.TestServe(gRPCTestMessage(text="test"))
+    # Use the provided channel
+    stub = gRPCTestServiceStub(channel)
+    stub.TestServe(gRPCTestMessage(text="test"))
 
     _tear_down(server=server)
 
@@ -76,11 +85,11 @@ def test_grpc_server_other_interceptors(sentry_init, capture_events_forksafe):
     mock_interceptor = Mock()
     mock_interceptor.intercept_service.side_effect = mock_intercept
 
-    server = _set_up(interceptors=[mock_interceptor])
+    server, channel = _set_up(interceptors=[mock_interceptor])
 
-    with grpc.insecure_channel("localhost:{}".format(PORT)) as channel:
-        stub = gRPCTestServiceStub(channel)
-        stub.TestServe(gRPCTestMessage(text="test"))
+    # Use the provided channel
+    stub = gRPCTestServiceStub(channel)
+    stub.TestServe(gRPCTestMessage(text="test"))
 
     _tear_down(server=server)
 
@@ -103,30 +112,30 @@ def test_grpc_server_continues_transaction(sentry_init, capture_events_forksafe)
     sentry_init(traces_sample_rate=1.0, integrations=[GRPCIntegration()])
     events = capture_events_forksafe()
 
-    server = _set_up()
+    server, channel = _set_up()
 
-    with grpc.insecure_channel("localhost:{}".format(PORT)) as channel:
-        stub = gRPCTestServiceStub(channel)
+    # Use the provided channel
+    stub = gRPCTestServiceStub(channel)
 
-        with start_transaction() as transaction:
-            metadata = (
-                (
-                    "baggage",
-                    "sentry-trace_id={trace_id},sentry-environment=test,"
-                    "sentry-transaction=test-transaction,sentry-sample_rate=1.0".format(
-                        trace_id=transaction.trace_id
-                    ),
+    with start_transaction() as transaction:
+        metadata = (
+            (
+                "baggage",
+                "sentry-trace_id={trace_id},sentry-environment=test,"
+                "sentry-transaction=test-transaction,sentry-sample_rate=1.0".format(
+                    trace_id=transaction.trace_id
                 ),
-                (
-                    "sentry-trace",
-                    "{trace_id}-{parent_span_id}-{sampled}".format(
-                        trace_id=transaction.trace_id,
-                        parent_span_id=transaction.span_id,
-                        sampled=1,
-                    ),
+            ),
+            (
+                "sentry-trace",
+                "{trace_id}-{parent_span_id}-{sampled}".format(
+                    trace_id=transaction.trace_id,
+                    parent_span_id=transaction.span_id,
+                    sampled=1,
                 ),
-            )
-            stub.TestServe(gRPCTestMessage(text="test"), metadata=metadata)
+            ),
+        )
+        stub.TestServe(gRPCTestMessage(text="test"), metadata=metadata)
 
     _tear_down(server=server)
 
@@ -148,13 +157,13 @@ def test_grpc_client_starts_span(sentry_init, capture_events_forksafe):
     sentry_init(traces_sample_rate=1.0, integrations=[GRPCIntegration()])
     events = capture_events_forksafe()
 
-    server = _set_up()
+    server, channel = _set_up()
 
-    with grpc.insecure_channel("localhost:{}".format(PORT)) as channel:
-        stub = gRPCTestServiceStub(channel)
+    # Use the provided channel
+    stub = gRPCTestServiceStub(channel)
 
-        with start_transaction():
-            stub.TestServe(gRPCTestMessage(text="test"))
+    with start_transaction():
+        stub.TestServe(gRPCTestMessage(text="test"))
 
     _tear_down(server=server)
 
@@ -183,13 +192,13 @@ def test_grpc_client_unary_stream_starts_span(sentry_init, capture_events_forksa
     sentry_init(traces_sample_rate=1.0, integrations=[GRPCIntegration()])
     events = capture_events_forksafe()
 
-    server = _set_up()
+    server, channel = _set_up()
 
-    with grpc.insecure_channel("localhost:{}".format(PORT)) as channel:
-        stub = gRPCTestServiceStub(channel)
+    # Use the provided channel
+    stub = gRPCTestServiceStub(channel)
 
-        with start_transaction():
-            [el for el in stub.TestUnaryStream(gRPCTestMessage(text="test"))]
+    with start_transaction():
+        [el for el in stub.TestUnaryStream(gRPCTestMessage(text="test"))]
 
     _tear_down(server=server)
 
@@ -227,14 +236,14 @@ def test_grpc_client_other_interceptor(sentry_init, capture_events_forksafe):
     sentry_init(traces_sample_rate=1.0, integrations=[GRPCIntegration()])
     events = capture_events_forksafe()
 
-    server = _set_up()
+    server, channel = _set_up()
 
-    with grpc.insecure_channel("localhost:{}".format(PORT)) as channel:
-        channel = grpc.intercept_channel(channel, MockClientInterceptor())
-        stub = gRPCTestServiceStub(channel)
+    # Intercept the channel
+    channel = grpc.intercept_channel(channel, MockClientInterceptor())
+    stub = gRPCTestServiceStub(channel)
 
-        with start_transaction():
-            stub.TestServe(gRPCTestMessage(text="test"))
+    with start_transaction():
+        stub.TestServe(gRPCTestMessage(text="test"))
 
     _tear_down(server=server)
 
@@ -267,13 +276,13 @@ def test_grpc_client_and_servers_interceptors_integration(
     sentry_init(traces_sample_rate=1.0, integrations=[GRPCIntegration()])
     events = capture_events_forksafe()
 
-    server = _set_up()
+    server, channel = _set_up()
 
-    with grpc.insecure_channel("localhost:{}".format(PORT)) as channel:
-        stub = gRPCTestServiceStub(channel)
+    # Use the provided channel
+    stub = gRPCTestServiceStub(channel)
 
-        with start_transaction():
-            stub.TestServe(gRPCTestMessage(text="test"))
+    with start_transaction():
+        stub.TestServe(gRPCTestMessage(text="test"))
 
     _tear_down(server=server)
 
@@ -290,13 +299,13 @@ def test_grpc_client_and_servers_interceptors_integration(
 @pytest.mark.forked
 def test_stream_stream(sentry_init):
     sentry_init(traces_sample_rate=1.0, integrations=[GRPCIntegration()])
-    server = _set_up()
+    server, channel = _set_up()
 
-    with grpc.insecure_channel("localhost:{}".format(PORT)) as channel:
-        stub = gRPCTestServiceStub(channel)
-        response_iterator = stub.TestStreamStream(iter((gRPCTestMessage(text="test"),)))
-        for response in response_iterator:
-            assert response.text == "test"
+    # Use the provided channel
+    stub = gRPCTestServiceStub(channel)
+    response_iterator = stub.TestStreamStream(iter((gRPCTestMessage(text="test"),)))
+    for response in response_iterator:
+        assert response.text == "test"
 
     _tear_down(server=server)
 
@@ -308,12 +317,12 @@ def test_stream_unary(sentry_init):
     Tracing not supported for it yet.
     """
     sentry_init(traces_sample_rate=1.0, integrations=[GRPCIntegration()])
-    server = _set_up()
+    server, channel = _set_up()
 
-    with grpc.insecure_channel("localhost:{}".format(PORT)) as channel:
-        stub = gRPCTestServiceStub(channel)
-        response = stub.TestStreamUnary(iter((gRPCTestMessage(text="test"),)))
-        assert response.text == "test"
+    # Use the provided channel
+    stub = gRPCTestServiceStub(channel)
+    response = stub.TestStreamUnary(iter((gRPCTestMessage(text="test"),)))
+    assert response.text == "test"
 
     _tear_down(server=server)
 
@@ -323,13 +332,13 @@ def test_span_origin(sentry_init, capture_events_forksafe):
     sentry_init(traces_sample_rate=1.0, integrations=[GRPCIntegration()])
     events = capture_events_forksafe()
 
-    server = _set_up()
+    server, channel = _set_up()
 
-    with grpc.insecure_channel("localhost:{}".format(PORT)) as channel:
-        stub = gRPCTestServiceStub(channel)
+    # Use the provided channel
+    stub = gRPCTestServiceStub(channel)
 
-        with start_transaction(name="custom_transaction"):
-            stub.TestServe(gRPCTestMessage(text="test"))
+    with start_transaction(name="custom_transaction"):
+        stub.TestServe(gRPCTestMessage(text="test"))
 
     _tear_down(server=server)