From 73da1d8c326341e36f3ec795ef2fd3adb0a673fc Mon Sep 17 00:00:00 2001 From: Casper Nielsen Date: Thu, 15 Oct 2026 14:49:18 +0000 Subject: [PATCH] test: stop fake servers leaking fixed ports and hanging the Windows unit tests (#1242) * test: bind fake Dapr servers to free ports and release them on failure The client and actor unit tests started their fake sidecars on fixed ports (3500, 4443, 3510, 50001, 50010, 50011). FakeDaprSidecar opened its HTTP socket in __init__ and then bound gRPC in start(). When the gRPC port was taken (50001 sits in Windows' dynamic port range) start() raised, unittest skipped tearDownClass, and the HTTP socket stayed listening, unserved, for the rest of the session. HTTPServer sets SO_REUSEADDR, so on Windows the next test could bind the same port next to it, connect to the dead socket, and block forever in the health check. - FakeHttpServer binds 127.0.0.1 on port 0 by default, turns off address reuse, runs as a daemon thread, and its shutdown works (and does not hang) when the thread never started. It always closes the socket. - FakeDaprSidecar defaults both ports to 0, reads the real ports back from grpc and the socket, creates the HTTP server only after gRPC is up, and releases everything (sockets, grpc server, generated certs) if any step of start()/start_secure() fails. stop()/stop_secure() are idempotent. - Every test class using the fakes registers addClassCleanup right after creating the server, reads the bound ports back, and points DAPR_HTTP_PORT/DAPR_HTTP_ENDPOINT at them, restoring the previous values when the class finishes. - The three secure test modules reach their parent class through its module, so pytest no longer collects and runs the insecure parent a second time in each of them (213 duplicate runs). - Add tests for the fakes: a failed start releases every port and returns from stop() promptly. Signed-off-by: Casper Nielsen Co-Authored-By: Claude Opus 5.5 * test: assert fake server cleanup runs once and mark the reuse test as Windows-only Signed-off-by: Casper Nielsen Co-Authored-By: Claude Opus 5.5 * test: cover cancelled subscription streams directly The CANCELLED branch in Subscription.next_message and the matching break in subscribe_with_handler were only covered by chance, when a test's fake sidecar shut down under a live stream. With the fake servers now stopped cleanly, those lines went uncovered. Test both paths against a fake call instead of relying on teardown timing. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Casper Nielsen * test: capture the streaming thread without patching threading.Thread globally Co-Authored-By: Claude Opus 5.5 Signed-off-by: Casper Nielsen * test: cover secure HTTP start failing after certificates are created Co-Authored-By: Claude Opus 5.5 Signed-off-by: Casper Nielsen --------- Signed-off-by: Casper Nielsen Co-authored-by: Claude Opus 5.5 (cherry picked from commit 5a556c2c815116cf1bc48599e631c566cef8f353) Signed-off-by: dapr-bot --- tests/actor/test_actor.py | 12 +- tests/actor/test_actor_reentrancy.py | 16 +- tests/actor/test_actor_runtime.py | 12 +- tests/actor/test_grpc_host.py | 20 +- tests/clients/fake_dapr_server.py | 92 +++++-- tests/clients/fake_http_server.py | 70 ++++- tests/clients/test_conversation.py | 18 +- tests/clients/test_dapr_actor_grpc_client.py | 15 +- tests/clients/test_dapr_grpc_client.py | 17 +- tests/clients/test_dapr_grpc_client_async.py | 18 +- .../test_dapr_grpc_client_async_secure.py | 21 +- tests/clients/test_dapr_grpc_client_secure.py | 21 +- tests/clients/test_exceptions.py | 18 +- tests/clients/test_fake_dapr_server.py | 257 ++++++++++++++++++ .../test_http_service_invocation_client.py | 14 +- ...t_secure_http_service_invocation_client.py | 21 +- tests/clients/test_subscription.py | 89 +++++- 17 files changed, 566 insertions(+), 165 deletions(-) create mode 100644 tests/clients/test_fake_dapr_server.py 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