From 0ce069591f65940f888231a94c446f24889be523 Mon Sep 17 00:00:00 2001 From: Charlie Stanley Date: Thu, 28 Jan 2021 15:33:34 -0800 Subject: [PATCH] Add an HTTP client for service invocation and default to using it --- .github/workflows/validate_examples.yaml | 10 +- dapr/clients/__init__.py | 2 +- dapr/clients/grpc/_response.py | 5 + dapr/clients/grpc/client.py | 4 +- dapr/clients/http/client.py | 86 ++++++ dapr/clients/http/dapr_actor_http_client.py | 78 ++---- .../http/dapr_invocation_http_client.py | 76 ++++++ dapr/clients/main.py | 64 +++++ dapr/conf/global_settings.py | 2 + dapr/serializers/json.py | 4 +- examples/invoke-custom-data/README.md | 47 +++- examples/invoke-simple/README.md | 45 +++- tests/clients/fake_http_server.py | 79 ++++++ tests/clients/test_dapr_grpc_client.py | 40 +-- .../test_http_service_invocation_client.py | 253 ++++++++++++++++++ tox.ini | 2 + 16 files changed, 710 insertions(+), 87 deletions(-) create mode 100644 dapr/clients/http/client.py create mode 100644 dapr/clients/http/dapr_invocation_http_client.py create mode 100644 dapr/clients/main.py create mode 100644 tests/clients/fake_http_server.py create mode 100644 tests/clients/test_http_service_invocation_client.py diff --git a/.github/workflows/validate_examples.yaml b/.github/workflows/validate_examples.yaml index f12f5d046..31e3e945e 100644 --- a/.github/workflows/validate_examples.yaml +++ b/.github/workflows/validate_examples.yaml @@ -19,11 +19,11 @@ jobs: GOOS: linux GOARCH: amd64 GOPROXY: https://proxy.golang.org - DAPR_CLI_VER: 1.0.0-rc.3 - DAPR_RUNTIME_VER: 1.0.0-rc.2 - DAPR_INSTALL_URL: https://raw.githubusercontent.com/dapr/cli/3dacfb672d55f1436c249057aaebbe597e1066f3/install/install.sh + DAPR_CLI_VER: 1.0.0-rc.4 + DAPR_RUNTIME_VER: 1.0.0-rc.3 + DAPR_INSTALL_URL: https://raw.githubusercontent.com/dapr/cli/v1.0.0-rc.4/install/install.sh DAPR_CLI_REF: - DAPR_REF: f4327d91c3cd921b26077d99abe0f237c9a38f5d + DAPR_REF: steps: - uses: actions/checkout@v2 - name: Set up Python ${{ env.PYTHON_VER }} @@ -82,4 +82,4 @@ jobs: ./dist/linux_amd64/release/placement --healthz-port 9091 & - name: Check Examples run: | - tox -e examples \ No newline at end of file + tox -e examples diff --git a/dapr/clients/__init__.py b/dapr/clients/__init__.py index 2e26babc8..43237a41e 100644 --- a/dapr/clients/__init__.py +++ b/dapr/clients/__init__.py @@ -7,7 +7,7 @@ from dapr.clients.base import DaprActorClientBase from dapr.clients.exceptions import DaprInternalError, ERROR_CODE_UNKNOWN -from dapr.clients.grpc.client import DaprClient +from dapr.clients.main import DaprClient from dapr.clients.http.dapr_actor_http_client import DaprActorHttpClient __all__ = [ diff --git a/dapr/clients/grpc/_response.py b/dapr/clients/grpc/_response.py index 4ec1299df..182b8e0e4 100644 --- a/dapr/clients/grpc/_response.py +++ b/dapr/clients/grpc/_response.py @@ -191,6 +191,11 @@ def unpack(self, message: GrpcMessage) -> None: ValueError: message is not protocol buffer message object or message's type is not matched with the response data type """ + + if self.content_type is not None and self.content_type.lower() == "application/x-protobuf": + message.ParseFromString(self.data) + return + unpack(self.proto, message) diff --git a/dapr/clients/grpc/client.py b/dapr/clients/grpc/client.py index a0c105cd7..0b8a0af5f 100644 --- a/dapr/clients/grpc/client.py +++ b/dapr/clients/grpc/client.py @@ -38,7 +38,7 @@ from opencensus.trace.tracers.base import Tracer # type: ignore -class DaprClient: +class DaprGrpcClient: """The convenient layer implementation of Dapr gRPC APIs. This provides the wrappers and helpers to allows developers to use Dapr runtime gRPC API @@ -86,7 +86,7 @@ def close(self): def __del__(self): self.close() - def __enter__(self) -> 'DaprClient': + def __enter__(self) -> 'DaprGrpcClient': return self def __exit__(self, exc_type, exc_value, traceback) -> None: diff --git a/dapr/clients/http/client.py b/dapr/clients/http/client.py new file mode 100644 index 000000000..818aa7a37 --- /dev/null +++ b/dapr/clients/http/client.py @@ -0,0 +1,86 @@ +# -*- coding: utf-8 -*- + +""" +Copyright (c) Microsoft Corporation. +Licensed under the MIT License. +""" + +import aiohttp + +from typing import Dict, Optional, Union, Tuple, TYPE_CHECKING +if TYPE_CHECKING: + from dapr.serializers import Serializer + +from dapr.conf import settings +from dapr.clients.base import DEFAULT_JSON_CONTENT_TYPE +from dapr.clients.exceptions import DaprInternalError, ERROR_CODE_DOES_NOT_EXIST, ERROR_CODE_UNKNOWN +from opencensus.trace.tracers.base import Tracer # type: ignore + +CONTENT_TYPE_HEADER = 'content-type' +DAPR_API_TOKEN_HEADER = 'dapr-api-token' + + +class DaprHttpClient: + """A Dapr Http API client""" + + def __init__(self, + message_serializer: 'Serializer', + timeout: int = 60, + tracer: Optional[Tracer] = None): + + self._timeout = aiohttp.ClientTimeout(total=timeout) + self._serializer = message_serializer + self._tracer = tracer + + def get_api_url(self) -> str: + return 'http://{}:{}/{}'.format( + settings.DAPR_RUNTIME_HOST, + settings.DAPR_HTTP_PORT, + settings.DAPR_API_VERSION) + + async def send_bytes( + self, method: str, url: str, + data: Optional[bytes], + headers: Dict[str, Union[bytes, str]] = {}, + query_params: Dict[str, Union[bytes, str]] = {} + ) -> Tuple[bytes, aiohttp.ClientResponse]: + if not headers.get(CONTENT_TYPE_HEADER): + headers[CONTENT_TYPE_HEADER] = DEFAULT_JSON_CONTENT_TYPE + + if settings.DAPR_API_TOKEN is not None: + headers[DAPR_API_TOKEN_HEADER] = settings.DAPR_API_TOKEN + + if self._tracer is not None: + trace_headers = self._tracer.propagator.to_headers(self._tracer.span_context) + headers.update(trace_headers) + + r = None + async with aiohttp.ClientSession(timeout=self._timeout) as session: + r = await session.request( + method=method, + url=url, + data=data, + headers=headers, + params=query_params) + + if r.status >= 200 and r.status < 300: + return await r.read(), r + + raise (await self.convert_to_error(r)) + + async def convert_to_error(self, response) -> DaprInternalError: + error_info = None + try: + error_body = await response.read() + if (error_body is None or len(error_body) == 0) and response.status == 404: + return DaprInternalError("Not Found", ERROR_CODE_DOES_NOT_EXIST) + error_info = self._serializer.deserialize(error_body) + except Exception: + return DaprInternalError(f'Unknown Dapr Error. HTTP status code: {response.status}') + + if error_info and isinstance(error_info, dict): + message = error_info.get('message') + error_code = error_info.get('errorCode') or ERROR_CODE_UNKNOWN + return DaprInternalError(message, error_code) + + return DaprInternalError(f'Unknown Dapr Error. HTTP status code: {response.status}') diff --git a/dapr/clients/http/dapr_actor_http_client.py b/dapr/clients/http/dapr_actor_http_client.py index d6810b27e..1f70e2366 100644 --- a/dapr/clients/http/dapr_actor_http_client.py +++ b/dapr/clients/http/dapr_actor_http_client.py @@ -4,26 +4,27 @@ Copyright (c) Microsoft Corporation. Licensed under the MIT License. """ -import aiohttp -from typing import Dict, Optional, TYPE_CHECKING +from typing import Optional, TYPE_CHECKING + if TYPE_CHECKING: from dapr.serializers import Serializer -from dapr.conf import settings -from dapr.clients.base import DaprActorClientBase, DEFAULT_JSON_CONTENT_TYPE -from dapr.clients.exceptions import DaprInternalError, ERROR_CODE_DOES_NOT_EXIST, ERROR_CODE_UNKNOWN - - -CONTENT_TYPE_HEADER = 'content-type' +from dapr.clients.http.client import DaprHttpClient +from dapr.clients.base import DaprActorClientBase +from opencensus.trace.tracers.base import Tracer # type: ignore class DaprActorHttpClient(DaprActorClientBase): """A Dapr Actor http client implementing :class:`DaprActorClientBase`""" - def __init__(self, message_serializer: 'Serializer', timeout: int = 60): - self._timeout = aiohttp.ClientTimeout(total=timeout) - self._serializer = message_serializer + def __init__( + self, + message_serializer: 'Serializer', + timeout: int = 60, + tracer: Optional[Tracer] = None): + + self._client = DaprHttpClient(message_serializer, timeout, tracer) async def invoke_method( self, actor_type: str, actor_id: str, @@ -40,7 +41,9 @@ async def invoke_method( bytes: the response from the actor. """ url = f'{self._get_base_url(actor_type, actor_id)}/method/{method}' - return await self._send_bytes(method='POST', url=url, data=data) + + body, r = await self._client.send_bytes(method='POST', url=url, data=data) + return body async def save_state_transactionally( self, actor_type: str, actor_id: str, @@ -53,7 +56,7 @@ async def save_state_transactionally( data (bytes): Json-serialized the transactional state operations. """ url = f'{self._get_base_url(actor_type, actor_id)}/state' - await self._send_bytes(method='PUT', url=url, data=data) + await self._client.send_bytes(method='PUT', url=url, data=data) async def get_state( self, actor_type: str, actor_id: str, name: str) -> bytes: @@ -68,7 +71,8 @@ async def get_state( bytes: the value of the state. """ url = f'{self._get_base_url(actor_type, actor_id)}/state/{name}' - return await self._send_bytes(method='GET', url=url, data=None) + body, r = await self._client.send_bytes(method='GET', url=url, data=None) + return body async def register_reminder( self, actor_type: str, actor_id: str, name: str, data: bytes) -> None: @@ -81,7 +85,7 @@ async def register_reminder( data (bytes): Reminder request json body. """ url = f'{self._get_base_url(actor_type, actor_id)}/reminders/{name}' - await self._send_bytes(method='PUT', url=url, data=data) + await self._client.send_bytes(method='PUT', url=url, data=data) async def unregister_reminder( self, actor_type: str, actor_id: str, name: str) -> None: @@ -93,7 +97,7 @@ async def unregister_reminder( name (str): the name of reminder. """ url = f'{self._get_base_url(actor_type, actor_id)}/reminders/{name}' - await self._send_bytes(method='DELETE', url=url, data=None) + await self._client.send_bytes(method='DELETE', url=url, data=None) async def register_timer( self, actor_type: str, actor_id: str, name: str, data: bytes) -> None: @@ -106,7 +110,7 @@ async def register_timer( data (bytes): Timer request json body. """ url = f'{self._get_base_url(actor_type, actor_id)}/timers/{name}' - await self._send_bytes(method='PUT', url=url, data=data) + await self._client.send_bytes(method='PUT', url=url, data=data) async def unregister_timer( self, actor_type: str, actor_id: str, name: str) -> None: @@ -118,44 +122,10 @@ async def unregister_timer( name (str): The name of timer """ url = f'{self._get_base_url(actor_type, actor_id)}/timers/{name}' - await self._send_bytes(method='DELETE', url=url, data=None) + await self._client.send_bytes(method='DELETE', url=url, data=None) def _get_base_url(self, actor_type: str, actor_id: str) -> str: - return 'http://{}:{}/{}/actors/{}/{}'.format( - settings.DAPR_RUNTIME_HOST, - settings.DAPR_HTTP_PORT, - settings.DAPR_API_VERSION, + return '{}/actors/{}/{}'.format( + self._client.get_api_url(), actor_type, actor_id) - - async def _send_bytes( - self, method: str, url: str, - data: Optional[bytes], headers: Dict[str, str] = {}) -> bytes: - if not headers.get(CONTENT_TYPE_HEADER): - headers[CONTENT_TYPE_HEADER] = DEFAULT_JSON_CONTENT_TYPE - - r = None - async with aiohttp.ClientSession(timeout=self._timeout) as session: - r = await session.request(method=method, url=url, data=data, headers=headers) - - if r.status >= 200 and r.status < 300: - return await r.read() - - raise (await self.convert_to_error(r)) - - async def convert_to_error(self, response) -> DaprInternalError: - error_info = None - try: - error_body = await response.read() - if (error_body is None or len(error_body) == 0) and response.status == 404: - return DaprInternalError("Not Found", ERROR_CODE_DOES_NOT_EXIST) - error_info = self._serializer.deserialize(error_body) - except Exception: - return DaprInternalError(f'Unknown Dapr Error. HTTP status code: {response.status}') - - if error_info and isinstance(error_info, dict): - message = error_info.get('message') - error_code = error_info.get('errorCode') or ERROR_CODE_UNKNOWN - return DaprInternalError(message, error_code) - - return DaprInternalError(f'Unknown Dapr Error. HTTP status code: {response.status}') diff --git a/dapr/clients/http/dapr_invocation_http_client.py b/dapr/clients/http/dapr_invocation_http_client.py new file mode 100644 index 000000000..a403eb0e4 --- /dev/null +++ b/dapr/clients/http/dapr_invocation_http_client.py @@ -0,0 +1,76 @@ +# -*- coding: utf-8 -*- + +""" +Copyright (c) Microsoft Corporation. +Licensed under the MIT License. +""" + +import asyncio + +from typing import Optional, Union + +from dapr.clients.http.client import DaprHttpClient, CONTENT_TYPE_HEADER +from dapr.clients.grpc._helpers import MetadataTuple, GrpcMessage +from dapr.clients.grpc._response import InvokeMethodResponse +from dapr.serializers import DefaultJSONSerializer +from opencensus.trace.tracers.base import Tracer # type: ignore + + +class DaprInvocationHttpClient: + """Service Invocation HTTP Client""" + + def __init__(self, timeout: int = 60, tracer: Optional[Tracer] = None): + self._client = DaprHttpClient(DefaultJSONSerializer(), timeout, tracer) + + def invoke_method( + self, + app_id: str, + method_name: str, + data: Union[bytes, str, GrpcMessage], + content_type: Optional[str] = None, + metadata: Optional[MetadataTuple] = None, + http_verb: Optional[str] = None, + http_querystring: Optional[MetadataTuple] = None) -> InvokeMethodResponse: + """This implementation is designed to seemlessly mimic the behavior of the grpc + service invoke implementation. Please see the grpc version for parameter definitions + and samples""" + + verb = 'GET' + if http_verb is not None: + verb = http_verb + + headers = {} + if metadata is not None: + for key, value in metadata: + headers[key] = value + query_params = {} + if http_querystring is not None: + for key, value in http_querystring: + query_params[key] = value + + if content_type is not None: + headers[CONTENT_TYPE_HEADER] = content_type + + url = f'{self._client.get_api_url()}/invoke/{app_id}/method/{method_name}' + + if isinstance(data, GrpcMessage): + body = data.SerializeToString() + elif isinstance(data, str): + body = data.encode('utf-8') + else: + body = data + + async def make_request(): + resp_body, r = await self._client.send_bytes( + method=verb, + headers=headers, + url=url, + data=body, + query_params=query_params) + + resp_data = InvokeMethodResponse(resp_body, r.content_type) + for key in r.headers: + resp_data.headers[key] = r.headers.getall(key) # type: ignore + return resp_data + + return asyncio.run(make_request()) diff --git a/dapr/clients/main.py b/dapr/clients/main.py new file mode 100644 index 000000000..dfe046845 --- /dev/null +++ b/dapr/clients/main.py @@ -0,0 +1,64 @@ +# -*- coding: utf-8 -*- + +""" +Copyright (c) Microsoft Corporation. +Licensed under the MIT License. +""" + +from typing import Optional, Union +from dapr.clients.exceptions import DaprInternalError +from dapr.clients.grpc.client import DaprGrpcClient, MetadataTuple, InvokeMethodResponse +from dapr.clients.http.dapr_invocation_http_client import DaprInvocationHttpClient +from dapr.conf import settings +from google.protobuf.message import Message as GrpcMessage +from opencensus.trace.tracers.base import Tracer # type: ignore + + +class DaprClient(DaprGrpcClient): + """The Dapr python-sdk uses gRPC for most operations. The exception being + service invocation which needs to support HTTP to HTTP invocations. The sdk defaults + to HTTP but can be overridden with the DAPR_API_METHOD_INVOCATION_PROTOCOL environment + variable. See: https://github.com/dapr/python-sdk/issues/176 for more details""" + + def __init__(self, address: Optional[str] = None, tracer: Optional[Tracer] = None): + super().__init__(address, tracer) + self.invocation_client = None + + invocation_protocol = settings.DAPR_API_METHOD_INVOCATION_PROTOCOL.upper() + + if invocation_protocol == 'HTTP': + self.invocation_client = DaprInvocationHttpClient(tracer=tracer) + elif invocation_protocol == 'GRPC': + pass + else: + raise DaprInternalError( + f'Unknown value for DAPR_API_METHOD_INVOCATION_PROTOCOL: {invocation_protocol}') + + def invoke_method( + self, + app_id: str, + method_name: str, + data: Union[bytes, str, GrpcMessage], + content_type: Optional[str] = None, + metadata: Optional[MetadataTuple] = None, + http_verb: Optional[str] = None, + http_querystring: Optional[MetadataTuple] = None) -> InvokeMethodResponse: + + if self.invocation_client: + return self.invocation_client.invoke_method( + app_id, + method_name, + data, + content_type=content_type, + metadata=metadata, + http_verb=http_verb, + http_querystring=http_querystring) + else: + return super().invoke_method( + app_id, + method_name, + data, + content_type=content_type, + metadata=metadata, + http_verb=http_verb, + http_querystring=http_querystring) diff --git a/dapr/conf/global_settings.py b/dapr/conf/global_settings.py index 457c2e4b9..e5a72c367 100644 --- a/dapr/conf/global_settings.py +++ b/dapr/conf/global_settings.py @@ -15,3 +15,5 @@ DAPR_HTTP_PORT = 3500 DAPR_GRPC_PORT = 50001 DAPR_API_VERSION = 'v1.0' + +DAPR_API_METHOD_INVOCATION_PROTOCOL = 'http' diff --git a/dapr/serializers/json.py b/dapr/serializers/json.py index 2be45cb22..fefa7ce67 100644 --- a/dapr/serializers/json.py +++ b/dapr/serializers/json.py @@ -13,7 +13,6 @@ from typing import Any, Callable, Optional, Type from dateutil import parser -from dapr.actor.runtime.config import ActorRuntimeConfig from dapr.serializers.base import Serializer from dapr.serializers.util import ( convert_from_dapr_duration, @@ -28,6 +27,9 @@ def serialize( custom_hook: Optional[Callable[[object], bytes]] = None) -> bytes: dict_obj = obj + + # importing this from top scope creates a circular import + from dapr.actor.runtime.config import ActorRuntimeConfig if callable(custom_hook): dict_obj = custom_hook(obj) elif isinstance(obj, bytes): diff --git a/examples/invoke-custom-data/README.md b/examples/invoke-custom-data/README.md index 2002db22a..fbb9ecd14 100644 --- a/examples/invoke-custom-data/README.md +++ b/examples/invoke-custom-data/README.md @@ -11,6 +11,8 @@ This example utilizes a receiver and a caller for the OnInvoke / Invoke function ## Install Dapr python-SDK + + ```bash pip3 install dapr dapr-ext-grpc ``` @@ -28,16 +30,38 @@ To run this example, the following steps should be followed: 2. Start Receiver (expose gRPC server receiver on port 50051) + + ```bash - dapr run --app-id invoke-receiver --app-protocol grpc --app-port 50051 python3 invoke-receiver.py + dapr run --app-id invoke-receiver --app-protocol grpc --app-port 50051 python3 invoke-receiver.py ``` + + 3. Start Caller + + ```bash - dapr run --app-id invoke-caller --app-protocol grpc python3 invoke-caller. py + dapr run --app-id invoke-caller --app-protocol grpc python3 invoke-caller.py ``` + + Expected output from caller: ``` @@ -52,4 +76,21 @@ Expected output from receiver: ``` == APP == {'user-agent': ['grpc-go/1.33.1'], 'x-forwarded-host': ['MyPC'], 'x-forwarded-for': ['192. 168.1.3'], 'forwarded': ['for=192.168.1.3;by=192.168.1.3; host=MyPC'], 'grpc-trace-bin': [b'\x00\x00\x90Zc\x17\xaav?5)L\xcd]>. \x88>\x01\x81\xe9\x9c\xbd\x01x\xfc\xc5\x02\x01']} == APP == SOME_DATA - ``` \ No newline at end of file + ``` + +4. Cleanup + + + +```bash +dapr stop --app-id invoke-caller +dapr stop --app-id invoke-receiver +``` + + \ No newline at end of file diff --git a/examples/invoke-simple/README.md b/examples/invoke-simple/README.md index a9c22ebb3..c1f7b345b 100644 --- a/examples/invoke-simple/README.md +++ b/examples/invoke-simple/README.md @@ -11,6 +11,8 @@ This example utilizes a receiver and a caller for the OnInvoke / Invoke function ## Install Dapr python-SDK + + ```bash pip3 install dapr dapr-ext-grpc ``` @@ -19,18 +21,59 @@ pip3 install dapr dapr-ext-grpc Run the following command in a terminal/command-prompt: + + ```bash # 1. Start Receiver (expose gRPC server receiver on port 50051) dapr run --app-id invoke-receiver --app-protocol grpc --app-port 50051 python3 invoke-receiver.py ``` + + In another terminal/command prompt run: + + ```bash # 2. Start Caller -dapr run --app-id invoke-caller --app-protocol grpc python3 invoke-caller.py +dapr run --app-id invoke-caller --app-protocol grpc --dapr-http-port 3500 python3 invoke-caller.py ``` + + +## Cleanup + + + +```bash +dapr stop --app-id invoke-caller +dapr stop --app-id invoke-receiver +``` + + + ## Running in Kubernetes mode 1. Build docker image diff --git a/tests/clients/fake_http_server.py b/tests/clients/fake_http_server.py new file mode 100644 index 000000000..0c3c0fa59 --- /dev/null +++ b/tests/clients/fake_http_server.py @@ -0,0 +1,79 @@ + + +from threading import Thread +from http.server import BaseHTTPRequestHandler, HTTPServer + + +class DaprHandler(BaseHTTPRequestHandler): + protocol_version = 'HTTP/1.1' + + def serve_forever(self): + while not self.running: + self.handle_request() + + def do_request(self, verb): + self.received_verb = verb + self.server.request_headers = self.headers + if 'Content-Length' in self.headers: + content_length = int(self.headers['Content-Length']) + + self.server.request_body += self.rfile.read(content_length) + + self.send_response(self.server.response_code) + for key, value in self.server.response_header_list: + self.send_header(key, value) + self.send_header('Content-Length', str(len(self.server.response_body))) + self.end_headers() + + self.server.path = self.path + + self.wfile.write(self.server.response_body) + + def do_GET(self): + self.do_request('GET') + + def do_POST(self): + self.do_request('POST') + + def do_PUT(self): + self.do_request('PUT') + + def do_DELETE(self): + self.do_request('DELETE') + + +class FakeHttpServer(Thread): + def __init__(self): + super().__init__() + self.server = HTTPServer(('localhost', 0), DaprHandler) + self.server.response_body = b'' + self.server.response_code = 200 + self.server.response_header_list = [] + self.server.request_body = b'' + + def get_port(self): + return self.server.socket.getsockname()[1] + + def reply_header(self, key, value): + self.server.response_header_list.append((key, value)) + + def get_request_headers(self): + return self.server.request_headers + + def shutdown_server(self): + self.server.shutdown() + self.server.socket.close() + self.join() + + def request_path(self): + return self.server.path + + def set_response(self, body: bytes, code=200): + self.server.response_body = body + self.server.response_code = code + + def get_request_body(self): + return self.server.request_body + + def run(self): + self.server.serve_forever() diff --git a/tests/clients/test_dapr_grpc_client.py b/tests/clients/test_dapr_grpc_client.py index e1f72ec6e..fbfb4003c 100644 --- a/tests/clients/test_dapr_grpc_client.py +++ b/tests/clients/test_dapr_grpc_client.py @@ -11,7 +11,7 @@ from unittest.mock import patch -from dapr.clients.grpc.client import DaprClient +from dapr.clients.grpc.client import DaprGrpcClient from dapr.proto import common_v1 from .fake_dapr_server import FakeDaprSidecar from dapr.conf import settings @@ -31,7 +31,7 @@ def tearDown(self): self._fake_dapr_server.stop() def test_http_extension(self): - dapr = DaprClient(f'localhost:{self.server_port}') + dapr = DaprGrpcClient(f'localhost:{self.server_port}') # Test POST verb without querystring ext = dapr._get_http_extension('POST') @@ -53,7 +53,7 @@ def test_http_extension(self): self.assertEqual(val, ext.querystring[key]) def test_invoke_method_bytes_data(self): - dapr = DaprClient(f'localhost:{self.server_port}') + dapr = DaprGrpcClient(f'localhost:{self.server_port}') resp = dapr.invoke_method( app_id='targetId', method_name='bytes', @@ -71,7 +71,7 @@ def test_invoke_method_bytes_data(self): self.assertEqual(['value1'], resp.headers['hkey1']) def test_invoke_method_proto_data(self): - dapr = DaprClient(f'localhost:{self.server_port}') + dapr = DaprGrpcClient(f'localhost:{self.server_port}') req = common_v1.StateItem(key='test') resp = dapr.invoke_method( app_id='targetId', @@ -93,7 +93,7 @@ def test_invoke_method_proto_data(self): self.assertEqual('test', new_resp.key) def test_invoke_binding_bytes_data(self): - dapr = DaprClient(f'localhost:{self.server_port}') + dapr = DaprGrpcClient(f'localhost:{self.server_port}') resp = dapr.invoke_binding( binding_name='binding', operation='create', @@ -110,7 +110,7 @@ def test_invoke_binding_bytes_data(self): self.assertEqual(['value1'], resp.headers['hkey1']) def test_invoke_binding_no_metadata(self): - dapr = DaprClient(f'localhost:{self.server_port}') + dapr = DaprGrpcClient(f'localhost:{self.server_port}') resp = dapr.invoke_binding( binding_name='binding', operation='create', @@ -122,7 +122,7 @@ def test_invoke_binding_no_metadata(self): self.assertEqual(0, len(resp.headers)) def test_invoke_binding_no_create(self): - dapr = DaprClient(f'localhost:{self.server_port}') + dapr = DaprGrpcClient(f'localhost:{self.server_port}') resp = dapr.invoke_binding( binding_name='binding', operation='delete', @@ -134,7 +134,7 @@ def test_invoke_binding_no_create(self): self.assertEqual(0, len(resp.headers)) def test_publish_event(self): - dapr = DaprClient(f'localhost:{self.server_port}') + dapr = DaprGrpcClient(f'localhost:{self.server_port}') resp = dapr.publish_event( pubsub_name='pubsub', topic_name='example', @@ -145,7 +145,7 @@ def test_publish_event(self): self.assertEqual(['haha'], resp.headers['hdata']) def test_publish_error(self): - dapr = DaprClient(f'localhost:{self.server_port}') + dapr = DaprGrpcClient(f'localhost:{self.server_port}') with self.assertRaisesRegex(ValueError, "invalid type for data "): dapr.publish_event( pubsub_name='pubsub', @@ -155,7 +155,7 @@ def test_publish_error(self): @patch.object(settings, 'DAPR_API_TOKEN', 'test-token') def test_dapr_api_token_insertion(self): - dapr = DaprClient(f'localhost:{self.server_port}') + dapr = DaprGrpcClient(f'localhost:{self.server_port}') resp = dapr.invoke_method( app_id='targetId', method_name='bytes', @@ -174,7 +174,7 @@ def test_dapr_api_token_insertion(self): self.assertEqual(['test-token'], resp.headers['hdapr-api-token']) def test_get_save_delete_state(self): - dapr = DaprClient(f'localhost:{self.server_port}') + dapr = DaprGrpcClient(f'localhost:{self.server_port}') key = "key_1" value = "value_1" options = StateOptions( @@ -219,7 +219,7 @@ def test_get_save_delete_state(self): self.assertTrue('delete failed' in str(context.exception)) def test_get_save_state_etag_none(self): - dapr = DaprClient(f'localhost:{self.server_port}') + dapr = DaprGrpcClient(f'localhost:{self.server_port}') value = 'test' no_etag_key = 'no_etag' @@ -246,7 +246,7 @@ def test_get_save_state_etag_none(self): self.assertEqual(resp.etag, "") def test_transaction_then_get_states(self): - dapr = DaprClient(f'localhost:{self.server_port}') + dapr = DaprGrpcClient(f'localhost:{self.server_port}') key = str(uuid.uuid4()) value = str(uuid.uuid4()) @@ -280,7 +280,7 @@ def test_transaction_then_get_states(self): self.assertEqual(resp.items[1].data, to_bytes(another_value.upper())) def test_save_then_get_states(self): - dapr = DaprClient(f'localhost:{self.server_port}') + dapr = DaprGrpcClient(f'localhost:{self.server_port}') key = str(uuid.uuid4()) value = str(uuid.uuid4()) @@ -316,7 +316,7 @@ def test_save_then_get_states(self): self.assertEqual(resp.items[1].data, to_bytes(another_value.upper())) def test_get_secret(self): - dapr = DaprClient(f'localhost:{self.server_port}') + dapr = DaprGrpcClient(f'localhost:{self.server_port}') key1 = 'key_1' resp = dapr.get_secret( store_name='store_1', @@ -332,7 +332,7 @@ def test_get_secret(self): self.assertEqual({key1: "val"}, resp._secret) def test_get_secret_metadata_absent(self): - dapr = DaprClient(f'localhost:{self.server_port}') + dapr = DaprGrpcClient(f'localhost:{self.server_port}') key1 = 'key_1' resp = dapr.get_secret( store_name='store_1', @@ -344,7 +344,7 @@ def test_get_secret_metadata_absent(self): self.assertEqual({key1: "val"}, resp._secret) def test_get_bulk_secret(self): - dapr = DaprClient(f'localhost:{self.server_port}') + dapr = DaprGrpcClient(f'localhost:{self.server_port}') resp = dapr.get_bulk_secret( store_name='store_1', metadata=( @@ -358,7 +358,7 @@ def test_get_bulk_secret(self): self.assertEqual({"keya": {"keyb": "val"}}, resp._secrets) def test_get_bulk_secret_metadata_absent(self): - dapr = DaprClient(f'localhost:{self.server_port}') + dapr = DaprGrpcClient(f'localhost:{self.server_port}') resp = dapr.get_bulk_secret(store_name='store_1') self.assertEqual(1, len(resp.headers)) @@ -366,7 +366,7 @@ def test_get_bulk_secret_metadata_absent(self): self.assertEqual({"keya": {"keyb": "val"}}, resp._secrets) def test_wait_ok(self): - dapr = DaprClient(f'localhost:{self.server_port}') + dapr = DaprGrpcClient(f'localhost:{self.server_port}') dapr.wait(0.1) def test_wait_timeout(self): @@ -375,7 +375,7 @@ def test_wait_timeout(self): with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.bind(('', 0)) port = s.getsockname()[1] - dapr = DaprClient(f'localhost:{port}') + dapr = DaprGrpcClient(f'localhost:{port}') with self.assertRaises(Exception) as context: dapr.wait(0.1) self.assertTrue('Connection refused' in str(context.exception)) diff --git a/tests/clients/test_http_service_invocation_client.py b/tests/clients/test_http_service_invocation_client.py new file mode 100644 index 000000000..f8c881bc8 --- /dev/null +++ b/tests/clients/test_http_service_invocation_client.py @@ -0,0 +1,253 @@ +# -*- coding: utf-8 -*- + +""" +Copyright (c) Microsoft Corporation. +Licensed under the MIT License. +""" + +import json +import unittest + +from .fake_http_server import FakeHttpServer +from dapr.conf import settings +from dapr.clients.main import DaprClient +from dapr.clients.exceptions import DaprInternalError +from dapr.proto import common_v1 +from opencensus.trace.tracer import Tracer # type: ignore +from opencensus.trace import print_exporter, samplers + + +class DaprInvocationHttpClientTests(unittest.TestCase): + def setUp(self): + self.server = FakeHttpServer() + self.server_port = self.server.get_port() + self.server.start() + settings.DAPR_HTTP_PORT = self.server_port + settings.DAPR_API_METHOD_INVOCATION_PROTOCOL = 'http' + self.client = DaprClient() + self.app_id = 'fakeapp' + self.method_name = 'fakemethod' + self.invoke_url = f'/v1.0/invoke/{self.app_id}/method/{self.method_name}' + + def tearDown(self): + self.server.shutdown_server() + settings.DAPR_API_TOKEN = None + settings.DAPR_API_METHOD_INVOCATION_PROTOCOL = 'http' + + def test_basic_invoke(self): + self.server.set_response(b"STRING_BODY") + + response = self.client.invoke_method(self.app_id, self.method_name, "") + + self.assertEqual(b"STRING_BODY", response.data) + self.assertEqual(self.invoke_url, self.server.request_path()) + + def test_invoke_PUT_with_body(self): + self.server.set_response(b"STRING_BODY") + + response = self.client.invoke_method(self.app_id, self.method_name, b"FOO", http_verb='PUT') + + self.assertEqual(b"STRING_BODY", response.data) + self.assertEqual(self.invoke_url, self.server.request_path()) + self.assertEqual(b"FOO", self.server.get_request_body()) + + def test_invoke_PUT_with_bytes_body(self): + self.server.set_response(b"STRING_BODY") + + response = self.client.invoke_method(self.app_id, self.method_name, b"FOO", http_verb='PUT') + + self.assertEqual(b"STRING_BODY", response.data) + self.assertEqual(self.invoke_url, self.server.request_path()) + self.assertEqual(b"FOO", self.server.get_request_body()) + + def test_invoke_GET_with_query_params(self): + self.server.set_response(b"STRING_BODY") + query_params = (('key1', 'value1'), + ('key2', 'value2')) + + response = self.client.invoke_method( + self.app_id, + self.method_name, + '', + http_querystring=query_params) + + self.assertEqual(b"STRING_BODY", response.data) + self.assertEqual(f"{self.invoke_url}?key1=value1&key2=value2", self.server.request_path()) + + def test_invoke_PUT_with_content_type(self): + self.server.set_response(b"STRING_BODY") + + sample_object = {'foo': ['val1', 'val2']} + + response = self.client.invoke_method( + self.app_id, + self.method_name, + json.dumps(sample_object), + content_type='application/json') + + self.assertEqual(b"STRING_BODY", response.data) + self.assertEqual(b'{"foo": ["val1", "val2"]}', self.server.get_request_body()) + + def test_invoke_method_proto_data(self): + self.server.set_response(b"\x0a\x04resp") + self.server.reply_header('Content-Type', 'application/x-protobuf') + + req = common_v1.StateItem(key='test') + resp = self.client.invoke_method( + self.app_id, + self.method_name, + http_verb='PUT', + data=req + ) + + self.assertEqual(b"\x0a\x04test", self.server.get_request_body()) + # unpack to new protobuf object + new_resp = common_v1.StateItem() + resp.unpack(new_resp) + self.assertEqual('resp', new_resp.key) + + def test_invoke_method_metadata(self): + self.server.set_response(b"FOO") + + req = common_v1.StateItem(key='test') + resp = self.client.invoke_method( + self.app_id, + self.method_name, + http_verb='PUT', + data=req, + metadata=(('header1', 'value1'), + ('header2', 'value2')) + ) + + request_headers = self.server.get_request_headers() + + self.assertEqual(b'FOO', resp.data) + + self.assertEqual('value1', request_headers['header1']) + self.assertEqual('value2', request_headers['header2']) + + def test_invoke_method_protobuf_response_with_suffix(self): + self.server.set_response(b"\x0a\x04resp") + self.server.reply_header('Content-Type', 'application/x-protobuf; gzip') + + req = common_v1.StateItem(key='test') + resp = self.client.invoke_method( + self.app_id, + self.method_name, + http_verb='PUT', + data=req, + metadata=(('header1', 'value1'), + ('header2', 'value2')) + ) + self.assertEqual(b"\x0a\x04test", self.server.get_request_body()) + # unpack to new protobuf object + new_resp = common_v1.StateItem() + resp.unpack(new_resp) + self.assertEqual('resp', new_resp.key) + + def test_invoke_method_protobuf_response_case_insensitive(self): + self.server.set_response(b"\x0a\x04resp") + self.server.reply_header('Content-Type', 'apPlicaTion/x-protobuf; gzip') + + req = common_v1.StateItem(key='test') + resp = self.client.invoke_method( + self.app_id, + self.method_name, + http_verb='PUT', + data=req, + metadata=(('header1', 'value1'), + ('header2', 'value2')) + ) + + self.assertEqual(b"\x0a\x04test", self.server.get_request_body()) + # unpack to new protobuf object + new_resp = common_v1.StateItem() + resp.unpack(new_resp) + self.assertEqual('resp', new_resp.key) + + def test_invoke_method_error_returned(self): + error_response = b'{"errorCode":"ERR_DIRECT_INVOKE","message":"Something bad happend"}' + self.server.set_response(error_response, 500) + + expected_msg = "('Something bad happend', 'ERR_DIRECT_INVOKE')" + + with self.assertRaises(DaprInternalError) as ctx: + self.client.invoke_method( + self.app_id, + self.method_name, + http_verb='PUT', + data='FOO', + ) + self.assertEqual(expected_msg, str(ctx.exception)) + + def test_invoke_method_non_dapr_error(self): + error_response = b'UNPARSABLE_ERROR' + self.server.set_response(error_response, 500) + + expected_msg = "Unknown Dapr Error. HTTP status code: 500" + + with self.assertRaises(DaprInternalError) as ctx: + self.client.invoke_method( + self.app_id, + self.method_name, + http_verb='PUT', + data='FOO', + ) + self.assertEqual(expected_msg, str(ctx.exception)) + + def test_generic_client_unknown_protocol(self): + settings.DAPR_API_METHOD_INVOCATION_PROTOCOL = 'unknown' + + expected_msg = "Unknown value for DAPR_API_METHOD_INVOCATION_PROTOCOL: UNKNOWN" + + with self.assertRaises(DaprInternalError) as ctx: + client = DaprClient() + + self.assertEqual(expected_msg, str(ctx.exception)) + + settings.DAPR_API_METHOD_INVOCATION_PROTOCOL = 'grpc' + client = DaprClient() + + self.assertIsNotNone(client) + + settings.DAPR_API_METHOD_INVOCATION_PROTOCOL = 'http' + client = DaprClient() + + self.assertIsNotNone(client) + + def test_invoke_method_with_api_token(self): + self.server.set_response(b"FOO") + settings.DAPR_API_TOKEN = 'c29saSBkZW8gZ2xvcmlhCg==' + + req = common_v1.StateItem(key='test') + resp = self.client.invoke_method( + self.app_id, + self.method_name, + http_verb='PUT', + data=req, + ) + + request_headers = self.server.get_request_headers() + + self.assertEqual('c29saSBkZW8gZ2xvcmlhCg==', request_headers['dapr-api-token']) + self.assertEqual(b'FOO', resp.data) + + def test_invoke_method_with_tracer(self): + tracer = Tracer(sampler=samplers.AlwaysOnSampler(), exporter=print_exporter.PrintExporter()) + + self.client = DaprClient(tracer=tracer) + self.server.set_response(b"FOO") + + with tracer.span(name="test"): + req = common_v1.StateItem(key='test') + resp = self.client.invoke_method( + self.app_id, + self.method_name, + http_verb='PUT', + data=req, + ) + + request_headers = self.server.get_request_headers() + + self.assertIn('Traceparent', request_headers) + self.assertEqual(b'FOO', resp.data) diff --git a/tox.ini b/tox.ini index 2122489a3..2e020d1a1 100644 --- a/tox.ini +++ b/tox.ini @@ -37,6 +37,8 @@ commands = ./validate.sh pubsub-simple ./validate.sh state_store ./validate.sh secret_store + ./validate.sh invoke-simple + ./validate.sh invoke-custom-data commands_pre = pip3 install -e {toxinidir}/ pip3 install -e {toxinidir}/ext/dapr-ext-grpc/