diff --git a/dapr/aio/clients/grpc/interceptors.py b/dapr/aio/clients/grpc/interceptors.py index 688559835..0174ca982 100644 --- a/dapr/aio/clients/grpc/interceptors.py +++ b/dapr/aio/clients/grpc/interceptors.py @@ -14,7 +14,7 @@ """ from collections import namedtuple -from typing import List, Tuple +from typing import List, Optional, Tuple from grpc.aio import ( # type: ignore ClientCallDetails, @@ -22,8 +22,11 @@ UnaryUnaryClientInterceptor, ) -from dapr.clients.grpc.interceptors import _is_unbounded_wait -from dapr.conf import settings +from dapr.clients.grpc.interceptors import ( + _is_unbounded_wait, + _resolve_timeout_seconds, + _validate_timeout_seconds, +) class _ClientCallDetailsAsync( @@ -41,19 +44,39 @@ class _ClientCallDetailsAsync( class DaprClientTimeoutInterceptorAsync(UnaryUnaryClientInterceptor): + """Applies a default deadline to unary calls that do not set their own. + + Calls that already carry a timeout keep it, and the workflow waits in + ``_UNBOUNDED_WAIT_METHODS`` are never given a default deadline. + """ + + def __init__(self, timeout_seconds: Optional[float] = None): + """Initializes the interceptor. + + Args: + timeout_seconds: Default deadline in seconds for calls made through this + interceptor. ``None`` falls back to ``settings.DAPR_API_TIMEOUT_SECONDS``, + read on every call; when that is also ``None`` no deadline is applied. + + Raises: + ValueError: If ``timeout_seconds`` is not a finite number greater than 0. + """ + self._timeout_seconds = _validate_timeout_seconds(timeout_seconds) + def intercept_unary_unary(self, continuation, client_call_details, request): - # Only apply a deadline when DAPR_API_TIMEOUT_SECONDS is explicitly configured. - # Without an explicit setting there is no SDK-level default deadline: Dapr's own - # resiliency policies and component timeouts act as the authoritative bounds for + # Only apply a deadline when one is explicitly configured, here or through + # DAPR_API_TIMEOUT_SECONDS. Without one there is no SDK-level default deadline: Dapr's + # own resiliency policies and component timeouts act as the authoritative bounds for # long-running operations such as LLM calls and workflow activities. + timeout_seconds = _resolve_timeout_seconds(self._timeout_seconds) if ( - settings.DAPR_API_TIMEOUT_SECONDS is not None + timeout_seconds is not None and client_call_details.timeout is None and not _is_unbounded_wait(client_call_details.method) ): new_client_call_details = _ClientCallDetailsAsync( client_call_details.method, - float(settings.DAPR_API_TIMEOUT_SECONDS), + timeout_seconds, client_call_details.metadata, client_call_details.credentials, client_call_details.wait_for_ready, diff --git a/dapr/clients/grpc/interceptors.py b/dapr/clients/grpc/interceptors.py index 3e654705a..6eba2d34c 100644 --- a/dapr/clients/grpc/interceptors.py +++ b/dapr/clients/grpc/interceptors.py @@ -1,5 +1,6 @@ +import math from collections import namedtuple -from typing import List, Tuple, Union +from typing import List, Optional, Tuple, Union from grpc import ( # type: ignore ClientCallDetails, @@ -40,20 +41,58 @@ def _is_unbounded_wait(method: Union[str, bytes]) -> bool: return method in _UNBOUNDED_WAIT_METHODS +def _validate_timeout_seconds(timeout_seconds: Optional[float]) -> Optional[float]: + if timeout_seconds is None: + return None + if not math.isfinite(timeout_seconds) or timeout_seconds <= 0: + raise ValueError( + f'timeout_seconds must be a finite number greater than 0, got {timeout_seconds}' + ) + return float(timeout_seconds) + + +def _resolve_timeout_seconds(timeout_seconds: Optional[float]) -> Optional[float]: + if timeout_seconds is not None: + return timeout_seconds + if settings.DAPR_API_TIMEOUT_SECONDS is None: + return None + return float(settings.DAPR_API_TIMEOUT_SECONDS) + + class DaprClientTimeoutInterceptor(UnaryUnaryClientInterceptor): + """Applies a default deadline to unary calls that do not set their own. + + Calls that already carry a timeout keep it, and the workflow waits in + ``_UNBOUNDED_WAIT_METHODS`` are never given a default deadline. + """ + + def __init__(self, timeout_seconds: Optional[float] = None): + """Initializes the interceptor. + + Args: + timeout_seconds: Default deadline in seconds for calls made through this + interceptor. ``None`` falls back to ``settings.DAPR_API_TIMEOUT_SECONDS``, + read on every call; when that is also ``None`` no deadline is applied. + + Raises: + ValueError: If ``timeout_seconds`` is not a finite number greater than 0. + """ + self._timeout_seconds = _validate_timeout_seconds(timeout_seconds) + def intercept_unary_unary(self, continuation, client_call_details, request): - # Only apply a deadline when DAPR_API_TIMEOUT_SECONDS is explicitly configured. - # Without an explicit setting there is no SDK-level default deadline: Dapr's own - # resiliency policies and component timeouts act as the authoritative bounds for + # Only apply a deadline when one is explicitly configured, here or through + # DAPR_API_TIMEOUT_SECONDS. Without one there is no SDK-level default deadline: Dapr's + # own resiliency policies and component timeouts act as the authoritative bounds for # long-running operations such as LLM calls and workflow activities. + timeout_seconds = _resolve_timeout_seconds(self._timeout_seconds) if ( - settings.DAPR_API_TIMEOUT_SECONDS is not None + timeout_seconds is not None and client_call_details.timeout is None and not _is_unbounded_wait(client_call_details.method) ): new_client_call_details = _ClientCallDetails( client_call_details.method, - float(settings.DAPR_API_TIMEOUT_SECONDS), + timeout_seconds, client_call_details.metadata, client_call_details.credentials, client_call_details.wait_for_ready, diff --git a/dapr/ext/workflow/AGENTS.md b/dapr/ext/workflow/AGENTS.md index e2465823b..56d2aa28c 100644 --- a/dapr/ext/workflow/AGENTS.md +++ b/dapr/ext/workflow/AGENTS.md @@ -140,6 +140,8 @@ Workflow (orchestrator) functions must remain generators (`def` with `yield`). T Client for workflow lifecycle management: +Constructor: `DaprWorkflowClient(host, port, logger_options, max_grpc_message_length, api_timeout_seconds)`. `api_timeout_seconds` is a default deadline for this client's unary calls; it overrides `DAPR_API_TIMEOUT_SECONDS` for this client only, through `DaprClientTimeoutInterceptor(timeout_seconds)`. `None` falls back to that setting, then to no deadline. Calls that set their own timeout keep it, and `wait_for_workflow_start` / `wait_for_workflow_completion` are never given it (`_UNBOUNDED_WAIT_METHODS`); their `timeout_in_seconds` bounds them. + - `schedule_new_workflow(workflow, *, input, instance_id, start_at, reuse_id_policy)` → returns `instance_id`. `reuse_id_policy` is deprecated (no effect, emits `DeprecationWarning`): an instance ID can always be reused once the existing instance is in a terminal state. - `get_workflow_state(instance_id, *, fetch_payloads=True)` → `Optional[WorkflowState]` - `wait_for_workflow_start(instance_id, *, fetch_payloads, timeout_in_seconds)` diff --git a/dapr/ext/workflow/aio/dapr_workflow_client.py b/dapr/ext/workflow/aio/dapr_workflow_client.py index 720bcccf2..5960187f7 100644 --- a/dapr/ext/workflow/aio/dapr_workflow_client.py +++ b/dapr/ext/workflow/aio/dapr_workflow_client.py @@ -59,6 +59,7 @@ def __init__( port: Optional[str] = None, logger_options: Optional[LoggerOptions] = None, max_grpc_message_length: Optional[int] = None, + api_timeout_seconds: Optional[float] = None, ): """Initializes the async workflow client. @@ -73,6 +74,17 @@ def __init__( ``DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTES`` env var (if non-zero), then the gRPC default (4 MiB). ``0`` in either source means "no opinion" and falls through to the next source. + api_timeout_seconds: Default deadline in seconds for this client's calls to the + sidecar, such as ``schedule_new_workflow`` and ``get_workflow_state``. It + overrides ``DAPR_API_TIMEOUT_SECONDS`` for this client only, so a hung call + fails with ``DEADLINE_EXCEEDED`` without bounding other Dapr calls in the + process. ``None`` (the default) falls back to ``DAPR_API_TIMEOUT_SECONDS``, + and to no deadline when that is unset. It does not apply to + ``wait_for_workflow_start`` and ``wait_for_workflow_completion``, which are + bounded only by their own ``timeout_in_seconds``. + + Raises: + ValueError: If ``api_timeout_seconds`` is not a finite number greater than 0. """ address = getAddress(host, port) @@ -88,13 +100,14 @@ def __init__( metadata = ((DAPR_API_TOKEN_HEADER, settings.DAPR_API_TOKEN),) options = self._logger.get_options() channel_options = get_grpc_channel_options(max_grpc_message_length) + timeout_interceptor = DaprClientTimeoutInterceptorAsync(api_timeout_seconds) self.__obj = aioclient.AsyncTaskHubGrpcClient( host_address=uri.endpoint, metadata=metadata, secure_channel=uri.tls, log_handler=options.log_handler, log_formatter=options.log_formatter, - interceptors=[DaprClientTimeoutInterceptorAsync()], + interceptors=[timeout_interceptor], channel_options=channel_options, ) diff --git a/dapr/ext/workflow/dapr_workflow_client.py b/dapr/ext/workflow/dapr_workflow_client.py index 975c5c292..71d2f367e 100644 --- a/dapr/ext/workflow/dapr_workflow_client.py +++ b/dapr/ext/workflow/dapr_workflow_client.py @@ -60,6 +60,7 @@ def __init__( port: Optional[str] = None, logger_options: Optional[LoggerOptions] = None, max_grpc_message_length: Optional[int] = None, + api_timeout_seconds: Optional[float] = None, ): """Initializes the sync workflow client. @@ -74,6 +75,17 @@ def __init__( ``DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTES`` env var (if non-zero), then the gRPC default (4 MiB). ``0`` in either source means "no opinion" and falls through to the next source. + api_timeout_seconds: Default deadline in seconds for this client's calls to the + sidecar, such as ``schedule_new_workflow`` and ``get_workflow_state``. It + overrides ``DAPR_API_TIMEOUT_SECONDS`` for this client only, so a hung call + fails with ``DEADLINE_EXCEEDED`` without bounding other Dapr calls in the + process. ``None`` (the default) falls back to ``DAPR_API_TIMEOUT_SECONDS``, + and to no deadline when that is unset. It does not apply to + ``wait_for_workflow_start`` and ``wait_for_workflow_completion``, which are + bounded only by their own ``timeout_in_seconds``. + + Raises: + ValueError: If ``api_timeout_seconds`` is not a finite number greater than 0. """ address = getAddress(host, port) @@ -89,13 +101,14 @@ def __init__( metadata = ((DAPR_API_TOKEN_HEADER, settings.DAPR_API_TOKEN),) options = self._logger.get_options() channel_options = get_grpc_channel_options(max_grpc_message_length) + timeout_interceptor = DaprClientTimeoutInterceptor(api_timeout_seconds) self.__obj = client.TaskHubGrpcClient( host_address=uri.endpoint, metadata=metadata, secure_channel=uri.tls, log_handler=options.log_handler, log_formatter=options.log_formatter, - interceptors=[DaprClientTimeoutInterceptor()], + interceptors=[timeout_interceptor], channel_options=channel_options, ) diff --git a/tests/clients/test_timeout_interceptor.py b/tests/clients/test_timeout_interceptor.py index 7676f33b3..3a55a34e8 100644 --- a/tests/clients/test_timeout_interceptor.py +++ b/tests/clients/test_timeout_interceptor.py @@ -14,7 +14,8 @@ """ import unittest -from unittest.mock import Mock, patch +from typing import Optional, Union +from unittest.mock import ANY, Mock, patch from dapr.clients.grpc.interceptors import DaprClientTimeoutInterceptor from dapr.conf import settings @@ -159,3 +160,79 @@ def test_workflow_waits_keep_an_explicit_timeout(self): ) continuation.assert_called_once_with(client_call_details, request) self.assertEqual(5, continuation.call_args[0][0].timeout) + + +def _call_details(method: Union[str, bytes], timeout: Optional[float]) -> Mock: + client_call_details = Mock() + client_call_details.method = method + client_call_details.timeout = timeout + client_call_details.metadata = 'metadata' + client_call_details.credentials = 'credentials' + client_call_details.wait_for_ready = 'wait_for_ready' + client_call_details.compression = 'compression' + return client_call_details + + +class DaprClientTimeoutInterceptorExplicitTimeoutTests(unittest.TestCase): + @patch.object(settings, 'DAPR_API_TIMEOUT_SECONDS', None) + def test_explicit_timeout_applies_without_global_timeout(self): + continuation = Mock() + client_call_details = _call_details('method', None) + + DaprClientTimeoutInterceptor(3).intercept_unary_unary( + continuation, client_call_details, Mock() + ) + self.assertEqual(3.0, continuation.call_args[0][0].timeout) + + @patch.object(settings, 'DAPR_API_TIMEOUT_SECONDS', 7) + def test_explicit_timeout_overrides_global_timeout(self): + continuation = Mock() + client_call_details = _call_details('method', None) + + DaprClientTimeoutInterceptor(3).intercept_unary_unary( + continuation, client_call_details, Mock() + ) + self.assertEqual(3.0, continuation.call_args[0][0].timeout) + + @patch.object(settings, 'DAPR_API_TIMEOUT_SECONDS', 7) + def test_explicit_timeout_keeps_per_call_timeout(self): + continuation = Mock() + client_call_details = _call_details('method', 10) + + DaprClientTimeoutInterceptor(3).intercept_unary_unary( + continuation, client_call_details, Mock() + ) + continuation.assert_called_once_with(client_call_details, ANY) + self.assertEqual(10, continuation.call_args[0][0].timeout) + + @patch.object(settings, 'DAPR_API_TIMEOUT_SECONDS', None) + def test_explicit_timeout_skips_workflow_waits(self): + for method in ( + '/TaskHubSidecarService/WaitForInstanceStart', + b'/TaskHubSidecarService/WaitForInstanceCompletion', + ): + with self.subTest(method=method): + continuation = Mock() + client_call_details = _call_details(method, None) + + DaprClientTimeoutInterceptor(3).intercept_unary_unary( + continuation, client_call_details, Mock() + ) + self.assertIsNone(continuation.call_args[0][0].timeout) + + def test_default_interceptor_reads_global_timeout_on_every_call(self): + interceptor = DaprClientTimeoutInterceptor() + for global_timeout, expected_timeout in ((None, None), (7, 7.0)): + with self.subTest(global_timeout=global_timeout): + continuation = Mock() + client_call_details = _call_details('method', None) + + with patch.object(settings, 'DAPR_API_TIMEOUT_SECONDS', global_timeout): + interceptor.intercept_unary_unary(continuation, client_call_details, Mock()) + self.assertEqual(expected_timeout, continuation.call_args[0][0].timeout) + + def test_non_positive_timeout_is_rejected(self): + for timeout in (0, -1, float('nan'), float('inf')): + with self.subTest(timeout=timeout): + with self.assertRaises(ValueError): + DaprClientTimeoutInterceptor(timeout) diff --git a/tests/clients/test_timeout_interceptor_async.py b/tests/clients/test_timeout_interceptor_async.py index 503466f2c..b043a1467 100644 --- a/tests/clients/test_timeout_interceptor_async.py +++ b/tests/clients/test_timeout_interceptor_async.py @@ -14,7 +14,8 @@ """ import unittest -from unittest.mock import Mock, patch +from typing import Optional, Union +from unittest.mock import ANY, Mock, patch from dapr.aio.clients.grpc.interceptors import DaprClientTimeoutInterceptorAsync from dapr.conf import settings @@ -133,3 +134,67 @@ def test_workflow_waits_keep_an_explicit_timeout(self): ) continuation.assert_called_once_with(client_call_details, request) self.assertEqual(5, continuation.call_args[0][0].timeout) + + +def _call_details(method: Union[str, bytes], timeout: Optional[float]) -> Mock: + client_call_details = Mock() + client_call_details.method = method + client_call_details.timeout = timeout + client_call_details.metadata = 'metadata' + client_call_details.credentials = 'credentials' + client_call_details.wait_for_ready = 'wait_for_ready' + return client_call_details + + +class DaprClientTimeoutInterceptorAsyncExplicitTimeoutTests(unittest.TestCase): + @patch.object(settings, 'DAPR_API_TIMEOUT_SECONDS', None) + def test_explicit_timeout_applies_without_global_timeout(self): + continuation = Mock() + client_call_details = _call_details('method', None) + + DaprClientTimeoutInterceptorAsync(3).intercept_unary_unary( + continuation, client_call_details, Mock() + ) + self.assertEqual(3.0, continuation.call_args[0][0].timeout) + + @patch.object(settings, 'DAPR_API_TIMEOUT_SECONDS', 7) + def test_explicit_timeout_overrides_global_timeout(self): + continuation = Mock() + client_call_details = _call_details('method', None) + + DaprClientTimeoutInterceptorAsync(3).intercept_unary_unary( + continuation, client_call_details, Mock() + ) + self.assertEqual(3.0, continuation.call_args[0][0].timeout) + + @patch.object(settings, 'DAPR_API_TIMEOUT_SECONDS', 7) + def test_explicit_timeout_keeps_per_call_timeout(self): + continuation = Mock() + client_call_details = _call_details('method', 10) + + DaprClientTimeoutInterceptorAsync(3).intercept_unary_unary( + continuation, client_call_details, Mock() + ) + continuation.assert_called_once_with(client_call_details, ANY) + self.assertEqual(10, continuation.call_args[0][0].timeout) + + @patch.object(settings, 'DAPR_API_TIMEOUT_SECONDS', None) + def test_explicit_timeout_skips_workflow_waits(self): + for method in ( + b'/TaskHubSidecarService/WaitForInstanceStart', + b'/TaskHubSidecarService/WaitForInstanceCompletion', + ): + with self.subTest(method=method): + continuation = Mock() + client_call_details = _call_details(method, None) + + DaprClientTimeoutInterceptorAsync(3).intercept_unary_unary( + continuation, client_call_details, Mock() + ) + self.assertIsNone(continuation.call_args[0][0].timeout) + + def test_non_positive_timeout_is_rejected(self): + for timeout in (0, -1, float('nan'), float('inf')): + with self.subTest(timeout=timeout): + with self.assertRaises(ValueError): + DaprClientTimeoutInterceptorAsync(timeout) diff --git a/tests/ext/workflow/test_workflow_client.py b/tests/ext/workflow/test_workflow_client.py index 10461c4b6..00d1d09f6 100644 --- a/tests/ext/workflow/test_workflow_client.py +++ b/tests/ext/workflow/test_workflow_client.py @@ -13,13 +13,16 @@ limitations under the License. """ +import socket +import threading +import time import unittest import warnings from datetime import datetime from typing import Any, Union from unittest import mock -from grpc import RpcError +from grpc import RpcError, StatusCode from dapr.conf import settings from dapr.ext.workflow._durabletask import client @@ -155,6 +158,73 @@ def test_timeout_interceptor_is_passed_to_client(self): self.assertIsInstance(interceptors[0], DaprClientTimeoutInterceptor) +class SilentTcpServer: + """Accepts TCP connections and never answers, like a sidecar lost behind a network change.""" + + def __init__(self): + self._listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self._listener.bind(('127.0.0.1', 0)) + self._listener.listen() + self.port = self._listener.getsockname()[1] + self._connections: list[socket.socket] = [] + self._accept_thread = threading.Thread(target=self._accept_forever, daemon=True) + self._accept_thread.start() + + def _accept_forever(self) -> None: + while True: + try: + connection, _ = self._listener.accept() + except OSError: + return + self._connections.append(connection) + + def close(self) -> None: + self._listener.close() + for connection in self._connections: + connection.close() + self._accept_thread.join(timeout=5) + + +class WorkflowClientApiTimeoutTest(unittest.TestCase): + def test_api_timeout_seconds_is_passed_to_interceptor(self): + with mock.patch( + 'dapr.ext.workflow._durabletask.client.TaskHubGrpcClient' + ) as mock_client_cls: + DaprWorkflowClient(api_timeout_seconds=2.5) + interceptor = mock_client_cls.call_args[1]['interceptors'][0] + self.assertEqual(2.5, interceptor._timeout_seconds) + + def test_api_timeout_seconds_defaults_to_none(self): + with mock.patch( + 'dapr.ext.workflow._durabletask.client.TaskHubGrpcClient' + ) as mock_client_cls: + DaprWorkflowClient() + interceptor = mock_client_cls.call_args[1]['interceptors'][0] + self.assertIsNone(interceptor._timeout_seconds) + + def test_non_positive_api_timeout_seconds_is_rejected(self): + with mock.patch('dapr.ext.workflow._durabletask.client.TaskHubGrpcClient'): + with self.assertRaises(ValueError): + DaprWorkflowClient(api_timeout_seconds=0) + + @mock.patch.object(settings, 'DAPR_API_TIMEOUT_SECONDS', None) + def test_call_to_unresponsive_sidecar_hits_the_deadline(self): + server = SilentTcpServer() + self.addCleanup(server.close) + wf_client = DaprWorkflowClient( + host='127.0.0.1', port=str(server.port), api_timeout_seconds=0.5 + ) + self.addCleanup(wf_client.close) + + started_at = time.monotonic() + with self.assertRaises(RpcError) as raised: + wf_client.schedule_new_workflow(workflow='wf', instance_id=mock_instance_id) + elapsed_seconds = time.monotonic() - started_at + + self.assertEqual(StatusCode.DEADLINE_EXCEEDED, raised.exception.code()) + self.assertLess(elapsed_seconds, 5) + + class WorkflowClientChannelOptionsTest(unittest.TestCase): @mock.patch.object(settings, 'DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTES', 0) def test_explicit_kwarg_sets_symmetric_channel_options(self): diff --git a/tests/ext/workflow/test_workflow_client_aio.py b/tests/ext/workflow/test_workflow_client_aio.py index b0e50ba45..c3fed6d4e 100644 --- a/tests/ext/workflow/test_workflow_client_aio.py +++ b/tests/ext/workflow/test_workflow_client_aio.py @@ -13,12 +13,16 @@ limitations under the License. """ +import socket +import threading +import time import unittest import warnings from datetime import datetime from typing import Any, Union from unittest import mock +from grpc import StatusCode from grpc.aio import AioRpcError from dapr.conf import settings @@ -158,6 +162,66 @@ async def test_timeout_interceptor_is_passed_to_client(self): self.assertIsInstance(interceptors[0], DaprClientTimeoutInterceptorAsync) +class SilentTcpServer: + """Accepts TCP connections and never answers, like a sidecar lost behind a network change.""" + + def __init__(self): + self._listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self._listener.bind(('127.0.0.1', 0)) + self._listener.listen() + self.port = self._listener.getsockname()[1] + self._connections: list[socket.socket] = [] + self._accept_thread = threading.Thread(target=self._accept_forever, daemon=True) + self._accept_thread.start() + + def _accept_forever(self) -> None: + while True: + try: + connection, _ = self._listener.accept() + except OSError: + return + self._connections.append(connection) + + def close(self) -> None: + self._listener.close() + for connection in self._connections: + connection.close() + self._accept_thread.join(timeout=5) + + +class WorkflowClientAioApiTimeoutTest(unittest.IsolatedAsyncioTestCase): + async def test_api_timeout_seconds_is_passed_to_interceptor(self): + with mock.patch( + 'dapr.ext.workflow._durabletask.aio.client.AsyncTaskHubGrpcClient' + ) as mock_client_cls: + DaprWorkflowClient(api_timeout_seconds=2.5) + interceptor = mock_client_cls.call_args[1]['interceptors'][0] + self.assertEqual(2.5, interceptor._timeout_seconds) + + async def test_non_positive_api_timeout_seconds_is_rejected(self): + with mock.patch('dapr.ext.workflow._durabletask.aio.client.AsyncTaskHubGrpcClient'): + with self.assertRaises(ValueError): + DaprWorkflowClient(api_timeout_seconds=-1) + + @mock.patch.object(settings, 'DAPR_API_TIMEOUT_SECONDS', None) + async def test_call_to_unresponsive_sidecar_hits_the_deadline(self): + server = SilentTcpServer() + self.addCleanup(server.close) + wf_client = DaprWorkflowClient( + host='127.0.0.1', port=str(server.port), api_timeout_seconds=0.5 + ) + # The async workflow client has no public close; close its channel directly. + self.addAsyncCleanup(wf_client._DaprWorkflowClient__obj.aclose) + + started_at = time.monotonic() + with self.assertRaises(AioRpcError) as raised: + await wf_client.get_workflow_state(mock_instance_id) + elapsed_seconds = time.monotonic() - started_at + + self.assertEqual(StatusCode.DEADLINE_EXCEEDED, raised.exception.code()) + self.assertLess(elapsed_seconds, 5) + + class WorkflowClientAioChannelOptionsTest(unittest.TestCase): @mock.patch.object(settings, 'DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTES', 0) def test_explicit_kwarg_sets_symmetric_channel_options(self):