[Backport release-1.19] test: stop fake servers leaking fixed ports and hanging the Windows unit tests

tests/actor/test_actor.py

@@ -23,7 +23,6 @@
 from dapr.actor.runtime.context import ActorRuntimeContext
 from dapr.actor.runtime.failure_policy import ActorReminderFailurePolicy
 from dapr.actor.runtime.runtime import ActorRuntime
-from dapr.conf import settings
 from dapr.serializers import DefaultJSONSerializer
 from tests.actor.fake_actor_classes import (
     FakeMultiInterfacesActor,
@@ -33,19 +32,16 @@
 )
 from tests.actor.fake_client import FakeDaprActorClient
 from tests.actor.utils import _async_mock, _run
-from tests.clients.fake_http_server import FakeHttpServer
+from tests.clients.fake_http_server import FakeHttpServer, point_settings_at_http_port
 
 
 class ActorTests(unittest.TestCase):
     @classmethod
     def setUpClass(cls):
-        cls.server = FakeHttpServer(3500)
+        cls.server = FakeHttpServer()
+        cls.addClassCleanup(cls.server.shutdown_server)
         cls.server.start()
-        settings.DAPR_HTTP_PORT = 3500
-
-    @classmethod
-    def tearDownClass(cls):
-        cls.server.shutdown_server()
+        point_settings_at_http_port(cls, cls.server.get_port())
 
     def setUp(self):
         ActorRuntime._actor_managers = {}

tests/actor/test_actor_reentrancy.py

@@ -19,27 +19,23 @@
 
 from dapr.actor.runtime.config import ActorReentrancyConfig, ActorRuntimeConfig
 from dapr.actor.runtime.runtime import ActorRuntime
-from dapr.conf import settings
 from dapr.serializers import DefaultJSONSerializer
 from tests.actor.fake_actor_classes import (
     FakeMultiInterfacesActor,
     FakeReentrantActor,
     FakeSlowReentrantActor,
 )
 from tests.actor.utils import _run
-from tests.clients.fake_http_server import FakeHttpServer
+from tests.clients.fake_http_server import FakeHttpServer, point_settings_at_http_port
 
 
 class ActorRuntimeTests(unittest.TestCase):
     @classmethod
     def setUpClass(cls):
-        cls.server = FakeHttpServer(3500)
+        cls.server = FakeHttpServer()
+        cls.addClassCleanup(cls.server.shutdown_server)
         cls.server.start()
-        settings.DAPR_HTTP_PORT = 3500
-
-    @classmethod
-    def tearDownClass(cls):
-        cls.server.shutdown_server()
+        point_settings_at_http_port(cls, cls.server.get_port())
 
     def setUp(self):
         ActorRuntime._actor_managers = {}
@@ -153,7 +149,7 @@ async def expected_return_value(*args, **kwargs):
 
             mocked.assert_called_with(
                 method='POST',
-                url=f'http://127.0.0.1:3500/v1.0/actors/{actor}/test-id/method/{method}',
+                url=f'http://127.0.0.1:{self.server.get_port()}/v1.0/actors/{actor}/test-id/method/{method}',
                 data=None,
                 headers={'Dapr-Reentrancy-Id': reentrancy_id},
             )
@@ -197,7 +193,7 @@ async def expected_return_value(*args, **kwargs):
 
             mocked.assert_called_with(
                 method='POST',
-                url=f'http://127.0.0.1:3500/v1.0/actors/{actor}/test-id/method/{method}',
+                url=f'http://127.0.0.1:{self.server.get_port()}/v1.0/actors/{actor}/test-id/method/{method}',
                 data=None,
                 headers={},
             )

tests/actor/test_actor_runtime.py

@@ -18,27 +18,23 @@
 
 from dapr.actor.runtime.config import ActorRuntimeConfig
 from dapr.actor.runtime.runtime import ActorRuntime
-from dapr.conf import settings
 from dapr.serializers import DefaultJSONSerializer
 from tests.actor.fake_actor_classes import (
     FakeMultiInterfacesActor,
     FakeSimpleActor,
     FakeSimpleTimerActor,
 )
 from tests.actor.utils import _run
-from tests.clients.fake_http_server import FakeHttpServer
+from tests.clients.fake_http_server import FakeHttpServer, point_settings_at_http_port
 
 
 class ActorRuntimeTests(unittest.TestCase):
     @classmethod
     def setUpClass(cls):
-        cls.server = FakeHttpServer(3500)
+        cls.server = FakeHttpServer()
+        cls.addClassCleanup(cls.server.shutdown_server)
         cls.server.start()
-        settings.DAPR_HTTP_PORT = 3500
-
-    @classmethod
-    def tearDownClass(cls):
-        cls.server.shutdown_server()
+        point_settings_at_http_port(cls, cls.server.get_port())
 
     def setUp(self):
         ActorRuntime._actor_managers = {}

tests/actor/test_grpc_host.py

@@ -27,6 +27,7 @@
 import asyncio
 import base64
 import json
+import socket
 import time
 import unittest
 from datetime import timedelta
@@ -48,8 +49,11 @@
 from dapr.serializers import DefaultJSONSerializer
 from tests.clients.fake_dapr_server import FakeDaprSidecar
 
-GRPC_PORT = 50010
-HTTP_PORT = 3510
+
+def _free_port() -> int:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+        sock.bind(('127.0.0.1', 0))
+        return sock.getsockname()[1]
 
 
 class HostTestActorInterface(ActorInterface):
@@ -218,12 +222,10 @@ class ActorGrpcHostTests(unittest.IsolatedAsyncioTestCase):
 
     @classmethod
     def setUpClass(cls):
-        cls._fake_dapr_server = FakeDaprSidecar(grpc_port=GRPC_PORT, http_port=HTTP_PORT)
+        cls._fake_dapr_server = FakeDaprSidecar()
+        cls.addClassCleanup(cls._fake_dapr_server.stop)
         cls._fake_dapr_server.start()
-
-    @classmethod
-    def tearDownClass(cls):
-        cls._fake_dapr_server.stop()
+        cls.grpc_port = cls._fake_dapr_server.grpc_port
 
     def setUp(self):
         server = self._fake_dapr_server
@@ -242,7 +244,7 @@ def setUp(self):
         # app_port=0 disables the host-managed app-channel listener: the fake
         # sidecar never dials the app, and it keeps tests deterministic if
         # APP_PORT happens to be set in the environment.
-        self.host = ActorGrpcHost(address=f'localhost:{GRPC_PORT}', app_port=0)
+        self.host = ActorGrpcHost(address=f'localhost:{self.grpc_port}', app_port=0)
 
     async def asyncTearDown(self):
         await self.host.close()
@@ -571,7 +573,7 @@ def test_resolve_app_port_precedence(self):
     async def test_owns_app_channel_listener(self):
         # When an app port is set, the host starts the empty app-channel
         # listener daprd needs and tears it down on close.
-        host = ActorGrpcHost(address=f'localhost:{GRPC_PORT}', app_port=50099)
+        host = ActorGrpcHost(address=f'localhost:{self.grpc_port}', app_port=_free_port())
         self._fake_dapr_server.actor_stream_plans.append(_plan())
         await host.register_actor(HostTestActor)
         try:

tests/clients/fake_dapr_server.py

@@ -15,7 +15,7 @@
 
 import json
 from concurrent import futures
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Callable, Dict, List, Optional, Tuple
 
 import grpc
 from google.protobuf import empty_pb2, struct_pb2
@@ -32,11 +32,17 @@
 
 
 class FakeDaprSidecar(api_service_v1.DaprServicer):
-    def __init__(self, grpc_port: int = 50001, http_port: int = 8080):
+    def __init__(self, grpc_port: int = 0, http_port: int = 0):
+        """Ports default to 0 so the OS picks free ones. start()/start_secure() replace
+        grpc_port and http_port with the ports actually bound."""
         self.grpc_port = grpc_port
         self.http_port = http_port
         self._grpc_server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
-        self._http_server = FakeHttpServer(self.http_port)  # Needed for the healthcheck endpoint
+        # Serves the healthcheck endpoint. Created in start()/start_secure() so that no
+        # socket is open until the sidecar is started.
+        self._http_server: Optional[FakeHttpServer] = None
+        self._secure = False
+        self._stopped = False
         api_service_v1.add_DaprServicer_to_server(self, self._grpc_server)
         self.store = {}
         self.transaction_operation_metadata: Dict[str, Dict[str, str]] = {}
@@ -92,38 +98,66 @@ def set_bulk_publish_failed_entries_on_next_call(
         self._bulk_publish_fail_next = (failed_entry_count, error_message)
 
     def start(self):
-        self._grpc_server.add_insecure_port(f'[::]:{self.grpc_port}')
-        self._grpc_server.start()
-        self._http_server.start()
+        self._start(self._grpc_server.add_insecure_port, secure=False)
 
     def start_secure(self):
-        GrpcCerts.create_certificates()
-
-        private_key_file = open(GrpcCerts.get_pk_path(), 'rb')
-        private_key_content = private_key_file.read()
-        private_key_file.close()
-
-        certificate_chain_file = open(GrpcCerts.get_cert_path(), 'rb')
-        certificate_chain_content = certificate_chain_file.read()
-        certificate_chain_file.close()
-
-        credentials = grpc.ssl_server_credentials(
-            [(private_key_content, certificate_chain_content)]
-        )
-
-        self._grpc_server.add_secure_port(f'[::]:{self.grpc_port}', credentials)
-        self._grpc_server.start()
-
-        self._http_server.start_secure()
+        self._secure = True
+        try:
+            GrpcCerts.create_certificates()
+            with open(GrpcCerts.get_pk_path(), 'rb') as private_key_file:
+                private_key_content = private_key_file.read()
+            with open(GrpcCerts.get_cert_path(), 'rb') as certificate_chain_file:
+                certificate_chain_content = certificate_chain_file.read()
+            credentials = grpc.ssl_server_credentials(
+                [(private_key_content, certificate_chain_content)]
+            )
+        except BaseException:
+            self._shutdown()
+            raise
+
+        self._start(lambda address: self._grpc_server.add_secure_port(address, credentials), True)
+
+    def _start(self, add_port: Callable[[str], int], secure: bool) -> None:
+        """Bind and start the gRPC server, then the HTTP server. If any step fails,
+        release everything that was opened and re-raise, so a failed setUpClass (after
+        which unittest skips tearDownClass) does not leave a listening socket behind."""
+        try:
+            bound_port = add_port(f'[::]:{self.grpc_port}')
+            if not bound_port:
+                # Older grpc versions return 0 instead of raising when the bind fails.
+                raise RuntimeError(f'Failed to bind to address [::]:{self.grpc_port}')
+            self.grpc_port = bound_port
+            self._grpc_server.start()
+
+            self._http_server = FakeHttpServer(self.http_port)
+            self.http_port = self._http_server.get_port()
+            if secure:
+                self._http_server.start_secure()
+            else:
+                self._http_server.start()
+        except BaseException:
+            self._shutdown()
+            raise
 
     def stop(self):
-        self._http_server.shutdown_server()
-        self._grpc_server.stop(None)
+        self._shutdown()
 
     def stop_secure(self):
-        self._http_server.shutdown_server()
-        self._grpc_server.stop(None)
-        GrpcCerts.delete_certificates()
+        self._shutdown()
+
+    def _shutdown(self) -> None:
+        """Release the HTTP socket, the gRPC server and any generated certificates.
+        Safe to call more than once."""
+        if self._stopped:
+            return
+        self._stopped = True
+        try:
+            if self._http_server is not None:
+                self._http_server.shutdown_server()
+        finally:
+            self._grpc_server.stop(None)
+            if self._secure:
+                GrpcCerts.delete_certificates()
 
     def raise_exception_on_next_call(self, exception):
         """