diff --git a/tests/actor/test_actor.py b/tests/actor/test_actor.py index 99bae2e7d..bd6ec9d39 100644 --- a/tests/actor/test_actor.py +++ b/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 = {} diff --git a/tests/actor/test_actor_reentrancy.py b/tests/actor/test_actor_reentrancy.py index 69f11160e..aef5e56f5 100644 --- a/tests/actor/test_actor_reentrancy.py +++ b/tests/actor/test_actor_reentrancy.py @@ -19,7 +19,6 @@ 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, @@ -27,19 +26,16 @@ 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={}, ) diff --git a/tests/actor/test_actor_runtime.py b/tests/actor/test_actor_runtime.py index c07085ef1..b227d9fd4 100644 --- a/tests/actor/test_actor_runtime.py +++ b/tests/actor/test_actor_runtime.py @@ -18,7 +18,6 @@ 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, @@ -26,19 +25,16 @@ 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 = {} diff --git a/tests/actor/test_grpc_host.py b/tests/actor/test_grpc_host.py index fe9e7e7e6..c5f5c668c 100644 --- a/tests/actor/test_grpc_host.py +++ b/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: diff --git a/tests/clients/fake_dapr_server.py b/tests/clients/fake_dapr_server.py index dd8b6a2e6..23d6d06cc 100644 --- a/tests/clients/fake_dapr_server.py +++ b/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): """ diff --git a/tests/clients/fake_http_server.py b/tests/clients/fake_http_server.py index 8476b18ba..097488593 100644 --- a/tests/clients/fake_http_server.py +++ b/tests/clients/fake_http_server.py @@ -3,8 +3,11 @@ from ssl import PROTOCOL_TLS_SERVER, SSLContext from threading import Thread +from dapr.conf import settings from tests.clients.certs import HttpCerts +LOCALHOST = '127.0.0.1' + class DaprHandler(BaseHTTPRequestHandler): protocol_version = 'HTTP/1.1' @@ -51,14 +54,25 @@ def do_DELETE(self): self.do_request('DELETE') +class _TestHTTPServer(HTTPServer): + # HTTPServer turns on SO_REUSEADDR. On Windows that lets a second socket bind a port + # another socket is still listening on, so a new test server can end up sharing a + # port with an abandoned one and connections land on the socket nobody serves. On + # POSIX the flag only affects TIME_WAIT reuse, which does not matter with port 0. + allow_reuse_address = False + + class FakeHttpServer(Thread): secure = False - def __init__(self, port: int = 8080): - super().__init__() + def __init__(self, port: int = 0): + """Bind a local HTTP server. Port 0 (the default) lets the OS pick a free port; + read it back with get_port().""" + super().__init__(daemon=True) - self.port = port - self.server = HTTPServer(('localhost', self.port), DaprHandler) + self.server = _TestHTTPServer((LOCALHOST, port), DaprHandler) + self.port = self.get_port() + self._closed = False self.server.response_body = b'' self.server.response_code = 200 @@ -76,11 +90,22 @@ def get_request_headers(self): return self.server.request_headers def shutdown_server(self): - self.server.shutdown() - self.server.socket.close() - self.join() - if self.secure: - HttpCerts.delete_certificates() + """Stop serving and close the socket. Safe to call more than once, and safe to + call when the server thread was never started.""" + if self._closed: + return + self._closed = True + try: + # shutdown() waits for serve_forever() to return, so only call it when the + # thread is running. If serve_forever() has not been entered yet it sees the + # shutdown request and returns straight away. + if self.is_alive(): + self.server.shutdown() + self.join() + finally: + self.server.server_close() + if self.secure: + HttpCerts.delete_certificates() def request_path(self): return self.server.path @@ -98,12 +123,16 @@ def set_server_delay(self, delay_seconds): def start_secure(self): self.secure = True - HttpCerts.create_certificates() - ssl_context = SSLContext(PROTOCOL_TLS_SERVER) - ssl_context.load_cert_chain(HttpCerts.get_cert_path(), HttpCerts.get_pk_path()) - self.server.socket = ssl_context.wrap_socket(self.server.socket, server_side=True) + try: + HttpCerts.create_certificates() + ssl_context = SSLContext(PROTOCOL_TLS_SERVER) + ssl_context.load_cert_chain(HttpCerts.get_cert_path(), HttpCerts.get_pk_path()) + self.server.socket = ssl_context.wrap_socket(self.server.socket, server_side=True) - self.start() + self.start() + except BaseException: + self.shutdown_server() + raise def run(self): self.server.serve_forever() @@ -114,3 +143,16 @@ def reset(self): self.server.response_header_list = [] self.server.request_body = b'' self.server.sleep_time = None + + +def point_settings_at_http_port(cls, port: int, scheme: str = 'http') -> None: + """Point settings.DAPR_HTTP_PORT and DAPR_HTTP_ENDPOINT at a fake server for the + duration of a test class, restoring the previous values when the class finishes.""" + saved = (settings.DAPR_HTTP_PORT, settings.DAPR_HTTP_ENDPOINT) + + def restore() -> None: + settings.DAPR_HTTP_PORT, settings.DAPR_HTTP_ENDPOINT = saved + + cls.addClassCleanup(restore) + settings.DAPR_HTTP_PORT = port + settings.DAPR_HTTP_ENDPOINT = f'{scheme}://{LOCALHOST}:{port}' diff --git a/tests/clients/test_conversation.py b/tests/clients/test_conversation.py index 105c7b291..798d9595b 100644 --- a/tests/clients/test_conversation.py +++ b/tests/clients/test_conversation.py @@ -62,8 +62,8 @@ from dapr.clients.grpc.conversation import ( tool as tool_decorator, ) -from dapr.conf import settings from tests.clients.fake_dapr_server import FakeDaprSidecar +from tests.clients.fake_http_server import point_settings_at_http_port """ Comprehensive tests for Dapr conversation API functionality. @@ -127,21 +127,19 @@ def create_calculate_tool(): class ConversationTestBase: """Base class for conversation tests with common setup.""" - grpc_port = 50011 - http_port = 3510 + grpc_port: int # set in setUpClass from the port the fake sidecar bound + http_port: int scheme = '' @classmethod def setUpClass(cls): - cls._fake_dapr_server = FakeDaprSidecar(grpc_port=cls.grpc_port, http_port=cls.http_port) + cls._fake_dapr_server = FakeDaprSidecar() + cls.addClassCleanup(cls._fake_dapr_server.stop) cls._fake_dapr_server.start() + cls.grpc_port = cls._fake_dapr_server.grpc_port + cls.http_port = cls._fake_dapr_server.http_port # Configure health check to use fake server's HTTP port - settings.DAPR_HTTP_PORT = cls.http_port - settings.DAPR_HTTP_ENDPOINT = f'http://127.0.0.1:{cls.http_port}' - - @classmethod - def tearDownClass(cls): - cls._fake_dapr_server.stop() + point_settings_at_http_port(cls, cls.http_port) class ConversationTestBaseSync(ConversationTestBase, unittest.TestCase): diff --git a/tests/clients/test_dapr_actor_grpc_client.py b/tests/clients/test_dapr_actor_grpc_client.py index 49bd65c8e..4fc169d91 100644 --- a/tests/clients/test_dapr_actor_grpc_client.py +++ b/tests/clients/test_dapr_actor_grpc_client.py @@ -26,20 +26,21 @@ from dapr.serializers import DefaultJSONSerializer from .fake_dapr_server import FakeDaprSidecar +from .fake_http_server import point_settings_at_http_port class DaprActorGrpcClientTests(unittest.IsolatedAsyncioTestCase): - grpc_port = 50001 - http_port = 3500 + grpc_port: int # set in setUpClass from the port the fake sidecar bound + http_port: int @classmethod def setUpClass(cls): - cls._fake_dapr_server = FakeDaprSidecar(grpc_port=cls.grpc_port, http_port=cls.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 + cls.http_port = cls._fake_dapr_server.http_port + point_settings_at_http_port(cls, cls.http_port) def setUp(self): self._fake_dapr_server.actor_requests.clear() diff --git a/tests/clients/test_dapr_grpc_client.py b/tests/clients/test_dapr_grpc_client.py index 58b76400f..527b9994a 100644 --- a/tests/clients/test_dapr_grpc_client.py +++ b/tests/clients/test_dapr_grpc_client.py @@ -57,24 +57,23 @@ from dapr.proto import common_v1 from .fake_dapr_server import FakeDaprSidecar +from .fake_http_server import point_settings_at_http_port class DaprGrpcClientTests(unittest.TestCase): - grpc_port = 50001 - http_port = 3500 + grpc_port: int # set in setUpClass from the port the fake sidecar bound + http_port: int scheme = '' error = None @classmethod def setUpClass(cls): - cls._fake_dapr_server = FakeDaprSidecar(grpc_port=cls.grpc_port, http_port=cls.http_port) + cls._fake_dapr_server = FakeDaprSidecar() + cls.addClassCleanup(cls._fake_dapr_server.stop) cls._fake_dapr_server.start() - settings.DAPR_HTTP_PORT = cls.http_port - settings.DAPR_HTTP_ENDPOINT = 'http://127.0.0.1:{}'.format(cls.http_port) - - @classmethod - def tearDownClass(cls): - cls._fake_dapr_server.stop() + cls.grpc_port = cls._fake_dapr_server.grpc_port + cls.http_port = cls._fake_dapr_server.http_port + point_settings_at_http_port(cls, cls.http_port) def test_http_extension(self): dapr = DaprGrpcClient(f'{self.scheme}localhost:{self.grpc_port}') diff --git a/tests/clients/test_dapr_grpc_client_async.py b/tests/clients/test_dapr_grpc_client_async.py index 3c0c24254..49d365487 100644 --- a/tests/clients/test_dapr_grpc_client_async.py +++ b/tests/clients/test_dapr_grpc_client_async.py @@ -50,24 +50,22 @@ from dapr.proto import common_v1 from .fake_dapr_server import FakeDaprSidecar +from .fake_http_server import point_settings_at_http_port class DaprGrpcClientAsyncTests(unittest.IsolatedAsyncioTestCase): - grpc_port = 50001 - http_port = 3500 + grpc_port: int # set in setUpClass from the port the fake sidecar bound + http_port: int scheme = '' @classmethod def setUpClass(cls): - cls._fake_dapr_server = FakeDaprSidecar(grpc_port=cls.grpc_port, http_port=cls.http_port) + cls._fake_dapr_server = FakeDaprSidecar() + cls.addClassCleanup(cls._fake_dapr_server.stop) cls._fake_dapr_server.start() - - settings.DAPR_HTTP_PORT = cls.http_port - settings.DAPR_HTTP_ENDPOINT = 'http://127.0.0.1:{}'.format(cls.http_port) - - @classmethod - def tearDownClass(cls): - cls._fake_dapr_server.stop() + cls.grpc_port = cls._fake_dapr_server.grpc_port + cls.http_port = cls._fake_dapr_server.http_port + point_settings_at_http_port(cls, cls.http_port) async def test_client_installs_grpc_aio_poller_noise_filter(self): asyncio_logger = logging.getLogger('asyncio') diff --git a/tests/clients/test_dapr_grpc_client_async_secure.py b/tests/clients/test_dapr_grpc_client_async_secure.py index a76a0449f..56f115364 100644 --- a/tests/clients/test_dapr_grpc_client_async_secure.py +++ b/tests/clients/test_dapr_grpc_client_async_secure.py @@ -20,31 +20,30 @@ from dapr.aio.clients.health import DaprHealth as DaprHealthAsync from dapr.clients.health import DaprHealth from dapr.conf import settings +from tests.clients import test_dapr_grpc_client_async as base from tests.clients.certs import replacement_get_credentials_func, replacement_get_health_context -from tests.clients.test_dapr_grpc_client_async import DaprGrpcClientAsyncTests from .fake_dapr_server import FakeDaprSidecar +from .fake_http_server import point_settings_at_http_port DaprGrpcClientAsync.get_credentials = replacement_get_credentials_func DaprHealthAsync.get_ssl_context = replacement_get_health_context DaprHealth.get_ssl_context = replacement_get_health_context -class DaprSecureGrpcClientAsyncTests(DaprGrpcClientAsyncTests): - grpc_port = 50001 - http_port = 4443 # The http server is used for health checks only, and doesn't need TLS +# The base class is reached through its module so pytest does not collect it a second +# time in this file. +class DaprSecureGrpcClientAsyncTests(base.DaprGrpcClientAsyncTests): scheme = 'https://' @classmethod def setUpClass(cls): - cls._fake_dapr_server = FakeDaprSidecar(grpc_port=cls.grpc_port, http_port=cls.http_port) + cls._fake_dapr_server = FakeDaprSidecar() + cls.addClassCleanup(cls._fake_dapr_server.stop_secure) cls._fake_dapr_server.start_secure() - settings.DAPR_HTTP_PORT = cls.http_port - settings.DAPR_HTTP_ENDPOINT = 'https://127.0.0.1:{}'.format(cls.http_port) - - @classmethod - def tearDownClass(cls): - cls._fake_dapr_server.stop_secure() + cls.grpc_port = cls._fake_dapr_server.grpc_port + cls.http_port = cls._fake_dapr_server.http_port + point_settings_at_http_port(cls, cls.http_port, scheme='https') @patch.object(settings, 'DAPR_GRPC_ENDPOINT', 'dns:domain1.com:5000') def test_init_with_DAPR_GRPC_ENDPOINT(self): diff --git a/tests/clients/test_dapr_grpc_client_secure.py b/tests/clients/test_dapr_grpc_client_secure.py index 2a6710403..f71b8f2ba 100644 --- a/tests/clients/test_dapr_grpc_client_secure.py +++ b/tests/clients/test_dapr_grpc_client_secure.py @@ -19,15 +19,16 @@ from dapr.clients.grpc.client import DaprGrpcClient from dapr.clients.health import DaprHealth from dapr.conf import settings +from tests.clients import test_dapr_grpc_client as base from tests.clients.certs import replacement_get_credentials_func, replacement_get_health_context -from tests.clients.test_dapr_grpc_client import DaprGrpcClientTests from .fake_dapr_server import FakeDaprSidecar +from .fake_http_server import point_settings_at_http_port -class DaprSecureGrpcClientTests(DaprGrpcClientTests): - grpc_port = 50001 - http_port = 4443 # The http server is used for health checks only, and doesn't need TLS +# The base class is reached through its module so pytest does not collect it a second +# time in this file. +class DaprSecureGrpcClientTests(base.DaprGrpcClientTests): scheme = 'https://' DaprGrpcClient.get_credentials = replacement_get_credentials_func @@ -35,14 +36,12 @@ class DaprSecureGrpcClientTests(DaprGrpcClientTests): @classmethod def setUpClass(cls): - cls._fake_dapr_server = FakeDaprSidecar(grpc_port=cls.grpc_port, http_port=cls.http_port) + cls._fake_dapr_server = FakeDaprSidecar() + cls.addClassCleanup(cls._fake_dapr_server.stop_secure) cls._fake_dapr_server.start_secure() - settings.DAPR_HTTP_PORT = cls.http_port - settings.DAPR_HTTP_ENDPOINT = 'https://127.0.0.1:{}'.format(cls.http_port) - - @classmethod - def tearDownClass(cls): - cls._fake_dapr_server.stop_secure() + cls.grpc_port = cls._fake_dapr_server.grpc_port + cls.http_port = cls._fake_dapr_server.http_port + point_settings_at_http_port(cls, cls.http_port, scheme='https') @patch.object(settings, 'DAPR_GRPC_ENDPOINT', 'https://domain1.com:5000') def test_init_with_DAPR_GRPC_ENDPOINT(self): diff --git a/tests/clients/test_exceptions.py b/tests/clients/test_exceptions.py index e8b4c6d9f..c0439ee8d 100644 --- a/tests/clients/test_exceptions.py +++ b/tests/clients/test_exceptions.py @@ -9,9 +9,9 @@ from dapr.clients import DaprGrpcClient from dapr.clients.exceptions import DaprGrpcError, DaprInternalError -from dapr.conf import settings from .fake_dapr_server import FakeDaprSidecar +from .fake_http_server import point_settings_at_http_port def create_expected_status(): @@ -92,21 +92,19 @@ def create_expected_status(): class DaprExceptionsTestCase(unittest.TestCase): - _grpc_port = 50001 - _http_port = 3500 + _grpc_port: int # set in setUpClass from the port the fake sidecar bound + _http_port: int @classmethod def setUpClass(cls): - cls._fake_dapr_server = FakeDaprSidecar(grpc_port=cls._grpc_port, http_port=cls._http_port) - settings.DAPR_HTTP_PORT = cls._http_port - settings.DAPR_HTTP_ENDPOINT = 'http://127.0.0.1:{}'.format(cls._http_port) + cls._fake_dapr_server = FakeDaprSidecar() + cls.addClassCleanup(cls._fake_dapr_server.stop) cls._fake_dapr_server.start() + cls._grpc_port = cls._fake_dapr_server.grpc_port + cls._http_port = cls._fake_dapr_server.http_port + point_settings_at_http_port(cls, cls._http_port) cls._expected_status = create_expected_status() - @classmethod - def tearDownClass(cls): - cls._fake_dapr_server.stop() - def test_exception_status_parsing(self): dapr = DaprGrpcClient(f'localhost:{self._grpc_port}') diff --git a/tests/clients/test_fake_dapr_server.py b/tests/clients/test_fake_dapr_server.py new file mode 100644 index 000000000..6ae21ec63 --- /dev/null +++ b/tests/clients/test_fake_dapr_server.py @@ -0,0 +1,257 @@ +# +# Copyright 2026 The Dapr Authors +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# http://www.apache.org/licenses/LICENSE-2.0 +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +"""Tests for the fake sidecar used by the client and actor tests. + +A fake that fails to start must not leave a socket listening: unittest skips +tearDownClass when setUpClass raises, so an abandoned socket would stay open, +unserved, for the rest of the session and later tests could connect to it and hang. +""" + +import os +import socket +import threading +import unittest +from typing import Callable +from unittest import mock + +from tests.clients.certs import GrpcCerts, HttpCerts +from tests.clients.fake_dapr_server import FakeDaprSidecar +from tests.clients.fake_http_server import LOCALHOST, FakeHttpServer +from tests.wait_utils import wait_until + + +def _free_port() -> int: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + sock.bind((LOCALHOST, 0)) + return sock.getsockname()[1] + + +def _can_bind(family: socket.AddressFamily, host: str, port: int) -> bool: + # No SO_REUSEADDR: the bind must fail while any socket is still listening on the port. + with socket.socket(family, socket.SOCK_STREAM) as sock: + try: + sock.bind((host, port)) + except OSError: + return False + return True + + +def _can_bind_http(port: int) -> bool: + return _can_bind(socket.AF_INET, LOCALHOST, port) + + +def _can_bind_grpc(port: int) -> bool: + return _can_bind(socket.AF_INET6, '::', port) + + +def _returns_within(seconds: float, fn: Callable[[], None]) -> bool: + """Run fn in a daemon thread so a hang fails the test instead of the session.""" + worker = threading.Thread(target=fn, daemon=True) + worker.start() + worker.join(seconds) + return not worker.is_alive() + + +class FakeDaprSidecarStartFailureTests(unittest.TestCase): + def test_grpc_bind_failure_releases_http_port(self): + http_port = _free_port() + sidecar = FakeDaprSidecar(http_port=http_port) + self.addCleanup(sidecar.stop) + + with mock.patch.object( + sidecar._grpc_server, + 'add_insecure_port', + side_effect=RuntimeError('Failed to bind to address [::]:0'), + ): + with self.assertRaises(RuntimeError): + sidecar.start() + + self.assertTrue(_can_bind_http(http_port), 'HTTP port still held after failed start()') + + def test_grpc_bind_failure_on_held_port_releases_http_port(self): + http_port = _free_port() + with socket.socket(socket.AF_INET6, socket.SOCK_STREAM) as holder: + holder.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0) + holder.bind(('::', 0)) + holder.listen() + held_grpc_port = holder.getsockname()[1] + + sidecar = FakeDaprSidecar(grpc_port=held_grpc_port, http_port=http_port) + self.addCleanup(sidecar.stop) + with self.assertRaises(RuntimeError): + sidecar.start() + + self.assertTrue(_can_bind_http(http_port), 'HTTP port still held after failed start()') + + def test_secure_grpc_bind_failure_releases_http_port_and_certs(self): + http_port = _free_port() + sidecar = FakeDaprSidecar(http_port=http_port) + self.addCleanup(sidecar.stop_secure) + + with mock.patch.object( + sidecar._grpc_server, + 'add_secure_port', + side_effect=RuntimeError('Failed to bind to address [::]:0'), + ): + with self.assertRaises(RuntimeError): + sidecar.start_secure() + + self.assertTrue(_can_bind_http(http_port), 'HTTP port still held after failed start()') + self.assertFalse(os.path.exists(GrpcCerts.get_cert_path())) + self.assertFalse(os.path.exists(HttpCerts.get_cert_path())) + + def test_secure_http_start_failure_after_certs_releases_ports_and_certs(self): + sidecar = FakeDaprSidecar() + self.addCleanup(sidecar.stop_secure) + ssl_context = mock.Mock() + ssl_context.load_cert_chain.side_effect = OSError('cannot load cert chain') + + with mock.patch('tests.clients.fake_http_server.SSLContext', return_value=ssl_context): + with self.assertRaises(OSError): + sidecar.start_secure() + + ssl_context.load_cert_chain.assert_called_once_with( + HttpCerts.get_cert_path(), HttpCerts.get_pk_path() + ) + grpc_port, http_port = sidecar.grpc_port, sidecar.http_port + self.assertNotEqual(0, grpc_port) + self.assertNotEqual(0, http_port) + self.assertFalse(os.path.exists(HttpCerts.get_cert_path())) + self.assertFalse(os.path.exists(GrpcCerts.get_cert_path())) + self.assertTrue(_can_bind_http(http_port), 'HTTP port still held after failed start()') + wait_until(lambda: _can_bind_grpc(grpc_port), timeout=5) + + def test_http_bind_failure_releases_grpc_port(self): + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as holder: + holder.bind((LOCALHOST, 0)) + holder.listen() + held_http_port = holder.getsockname()[1] + + sidecar = FakeDaprSidecar(http_port=held_http_port) + self.addCleanup(sidecar.stop) + with self.assertRaises(OSError): + sidecar.start() + + grpc_port = sidecar.grpc_port + self.assertNotEqual(0, grpc_port) + wait_until(lambda: _can_bind_grpc(grpc_port), timeout=5) + + def test_stop_after_failed_grpc_bind_returns_promptly(self): + # The HTTP server thread never started, so socketserver.shutdown() would wait + # forever for serve_forever() to finish. + sidecar = FakeDaprSidecar() + with mock.patch.object( + sidecar._grpc_server, + 'add_insecure_port', + side_effect=RuntimeError('Failed to bind to address [::]:0'), + ): + with self.assertRaises(RuntimeError): + sidecar.start() + + self.assertTrue(_returns_within(5, sidecar.stop), 'stop() hung after a failed start()') + self.assertTrue(_returns_within(5, sidecar.stop_secure)) + + def test_stop_after_failed_http_bind_returns_promptly(self): + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as holder: + holder.bind((LOCALHOST, 0)) + holder.listen() + sidecar = FakeDaprSidecar(http_port=holder.getsockname()[1]) + with self.assertRaises(OSError): + sidecar.start() + + self.assertTrue(_returns_within(5, sidecar.stop), 'stop() hung after a failed start()') + + +class FakeDaprSidecarLifecycleTests(unittest.TestCase): + def test_start_reports_bound_ports_and_stop_releases_them(self): + sidecar = FakeDaprSidecar() + sidecar.start() + grpc_port, http_port = sidecar.grpc_port, sidecar.http_port + + self.assertNotEqual(0, grpc_port) + self.assertNotEqual(0, http_port) + with socket.create_connection((LOCALHOST, http_port), timeout=5): + pass + + sidecar.stop() + sidecar.stop() # a second stop is a no-op + + self.assertTrue(_can_bind_http(http_port)) + wait_until(lambda: _can_bind_grpc(grpc_port), timeout=5) + + def test_second_stop_does_not_release_again(self): + sidecar = FakeDaprSidecar() + sidecar.start() + http_server = sidecar._http_server + assert http_server is not None + + with ( + mock.patch.object( + sidecar._grpc_server, 'stop', wraps=sidecar._grpc_server.stop + ) as grpc_stop, + mock.patch.object( + http_server, 'shutdown_server', wraps=http_server.shutdown_server + ) as http_shutdown, + ): + sidecar.stop() + sidecar.stop_secure() + + grpc_stop.assert_called_once_with(None) + http_shutdown.assert_called_once_with() + + def test_stop_without_start_returns_promptly(self): + self.assertTrue(_returns_within(5, FakeDaprSidecar().stop)) + + +class FakeHttpServerTests(unittest.TestCase): + def test_shutdown_without_start_releases_port(self): + server = FakeHttpServer() + port = server.get_port() + + self.assertTrue(_returns_within(5, server.shutdown_server), 'shutdown_server() hung') + self.assertTrue(_returns_within(5, server.shutdown_server)) + + self.assertTrue(_can_bind_http(port)) + + def test_second_shutdown_does_not_close_again(self): + server = FakeHttpServer() + server.start() + + with ( + mock.patch.object(server.server, 'shutdown', wraps=server.server.shutdown) as shutdown, + mock.patch.object( + server.server, 'server_close', wraps=server.server.server_close + ) as server_close, + ): + server.shutdown_server() + server.shutdown_server() + + shutdown.assert_called_once_with() + server_close.assert_called_once_with() + + def test_second_server_cannot_share_a_listening_port(self): + # Only Windows can catch a regression here. HTTPServer's SO_REUSEADDR lets a + # second socket bind a listening port on Windows, which is the bug this guards. + # Linux and macOS refuse that bind even with SO_REUSEADDR, so this test passes + # there whether or not allow_reuse_address is off. The Windows CI job runs it. + server = FakeHttpServer() + self.addCleanup(server.shutdown_server) + server.start() + + with self.assertRaises(OSError): + FakeHttpServer(server.get_port()) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/clients/test_http_service_invocation_client.py b/tests/clients/test_http_service_invocation_client.py index a0a7aadd6..b462b0531 100644 --- a/tests/clients/test_http_service_invocation_client.py +++ b/tests/clients/test_http_service_invocation_client.py @@ -29,25 +29,25 @@ from dapr.conf import settings from dapr.proto import common_v1 -from .fake_http_server import FakeHttpServer +from .fake_http_server import FakeHttpServer, point_settings_at_http_port class DaprInvocationHttpClientTests(unittest.TestCase): - server_port = 3500 + server_port: int # set in setUpClass from the port the fake server bound @classmethod def setUpClass(cls): - cls.server = FakeHttpServer(cls.server_port) + cls.server = FakeHttpServer() + cls.addClassCleanup(cls.server.shutdown_server) + cls.server_port = cls.server.get_port() + # setUp repoints the settings at the server; this restores them afterwards. + point_settings_at_http_port(cls, cls.server_port) cls.server.start() cls.app_id = 'fakeapp' cls.method_name = 'fakemethod' cls.invoke_url = f'/v1.0/invoke/{cls.app_id}/method/{cls.method_name}' - @classmethod - def tearDownClass(cls): - cls.server.shutdown_server() - def setUp(self): settings.DAPR_API_TOKEN = None settings.DAPR_HTTP_PORT = self.server_port diff --git a/tests/clients/test_secure_http_service_invocation_client.py b/tests/clients/test_secure_http_service_invocation_client.py index df13d8197..95fa3229e 100644 --- a/tests/clients/test_secure_http_service_invocation_client.py +++ b/tests/clients/test_secure_http_service_invocation_client.py @@ -30,9 +30,9 @@ from dapr.conf import settings from dapr.proto import common_v1 +from . import test_http_service_invocation_client as base from .certs import GrpcCerts, replacement_get_credentials_func, replacement_get_health_context -from .fake_http_server import FakeHttpServer -from .test_http_service_invocation_client import DaprInvocationHttpClientTests +from .fake_http_server import FakeHttpServer, point_settings_at_http_port def replacement_get_client_ssl_context(a): @@ -52,12 +52,15 @@ def replacement_get_client_ssl_context(a): DaprHealth.get_ssl_context = replacement_get_health_context -class DaprSecureInvocationHttpClientTests(DaprInvocationHttpClientTests): - server_port = 4443 - +# The base class is reached through its module so pytest does not collect it a second +# time in this file. +class DaprSecureInvocationHttpClientTests(base.DaprInvocationHttpClientTests): @classmethod def setUpClass(cls): - cls.server = FakeHttpServer(cls.server_port) + cls.server = FakeHttpServer() + cls.addClassCleanup(cls.server.shutdown_server) + cls.server_port = cls.server.get_port() + point_settings_at_http_port(cls, cls.server_port, scheme='https') cls.server.start_secure() cls.app_id = 'fakeapp' @@ -66,13 +69,9 @@ def setUpClass(cls): # We need to set up the certificates for the gRPC server # because the DaprGrpcClient will try to create a connection + cls.addClassCleanup(GrpcCerts.delete_certificates) GrpcCerts.create_certificates() - @classmethod - def tearDownClass(cls): - GrpcCerts.delete_certificates() - cls.server.shutdown_server() - def setUp(self): settings.DAPR_API_TOKEN = None settings.DAPR_HTTP_PORT = self.server_port diff --git a/tests/clients/test_subscription.py b/tests/clients/test_subscription.py index fd34a2dd0..525a1d3c8 100644 --- a/tests/clients/test_subscription.py +++ b/tests/clients/test_subscription.py @@ -4,8 +4,10 @@ import threading import unittest from collections.abc import Iterator +from typing import Any, List from unittest.mock import AsyncMock, MagicMock, patch +import grpc from google.protobuf.struct_pb2 import Struct from grpc import RpcError, StatusCode from grpc.aio import AioRpcError @@ -13,7 +15,14 @@ from dapr.aio.clients.grpc import subscription as subscription_async_module from dapr.aio.clients.grpc.subscription import Subscription as SubscriptionAsync from dapr.clients.grpc import subscription as subscription_module -from dapr.clients.grpc.subscription import StreamInactiveError, Subscription, SubscriptionMessage +from dapr.clients.grpc._response import TopicEventResponse +from dapr.clients.grpc.client import DaprGrpcClient +from dapr.clients.grpc.subscription import ( + StreamCancelledError, + StreamInactiveError, + Subscription, + SubscriptionMessage, +) from dapr.common.pubsub.subscription import StreamReconnectError from dapr.proto import api_v1 from dapr.proto.runtime.v1.appcallback_pb2 import TopicEventRequest @@ -485,6 +494,84 @@ def cancel(self) -> bool: return True +class _CancelledRpcError(grpc.RpcError): + def code(self) -> grpc.StatusCode: + return grpc.StatusCode.CANCELLED + + def details(self) -> str: + return 'Locally cancelled by application!' + + +class _CancelledStreamCall: + """Fake bidi call whose reads fail the way gRPC reports a cancelled stream.""" + + def __next__(self) -> api_v1.SubscribeTopicEventsResponseAlpha1: + raise _CancelledRpcError() + + def cancel(self) -> bool: + return True + + +class _ThreadCapturingThreading: + """Stands in for the threading module as seen by the client module only. + + Records the threads the client creates without replacing threading.Thread for the + rest of the process. Every other attribute resolves to the real threading module. + """ + + def __init__(self) -> None: + self.started: List[threading.Thread] = [] + + def Thread(self, *args: Any, **kwargs: Any) -> threading.Thread: # noqa: N802 + thread = threading.Thread(*args, **kwargs) + self.started.append(thread) + return thread + + def __getattr__(self, name: str) -> Any: + return getattr(threading, name) + + +def _cancelled_subscription() -> Subscription: + subscription = Subscription(stub=None, pubsub_name='pubsub', topic='topic') + subscription._stream = _CancelledStreamCall() # type: ignore[assignment] + subscription._stream_active = True + return subscription + + +class SubscriptionCancelledStreamTests(unittest.TestCase): + def test_next_message_raises_stream_cancelled_error_on_cancelled_rpc(self): + subscription = _cancelled_subscription() + + with self.assertRaises(StreamCancelledError): + subscription.next_message() + + def test_subscribe_with_handler_stops_streaming_when_stream_is_cancelled(self): + subscription = _cancelled_subscription() + + def handler(message: SubscriptionMessage) -> TopicEventResponse: + return TopicEventResponse('success') + + threading_capturing = _ThreadCapturingThreading() + + with patch('dapr.clients.health.DaprHealth.wait_for_sidecar'): + client = DaprGrpcClient('localhost:50001') + try: + with ( + patch.object(client, 'subscribe', return_value=subscription), + patch('dapr.clients.grpc.client.threading', threading_capturing), + ): + close_fn = client.subscribe_with_handler('pubsub', 'topic', handler) + # Closing marks the stream inactive, so a regressed thread still exits. + self.addCleanup(close_fn) + + started = threading_capturing.started + self.assertEqual(1, len(started)) + started[0].join(timeout=_TEST_TIMEOUT_SECONDS) + self.assertFalse(started[0].is_alive(), 'streaming thread kept running after cancel') + finally: + client.close() + + class _InitialReadThenUnavailableAsync: def __init__(self) -> None: self._initial_read_done = False