Merge "PyExecutor ResourceResoluton helper class."

6 files changed, 601 insertions, 6 deletions

diff --git a/ms/py-executor/resource_resolution/README b/ms/py-executor/resource_resolution/README
index 222dae499..353600445 100644
--- a/ms/py-executor/resource_resolution/README
+++ b/ms/py-executor/resource_resolution/README
@@ -77,4 +77,67 @@ if __name__ == "__main__":
             for response in client.process(generate_messages()):
                 print(response)
 
+```
+
+### Authorizarion header
+
+```
+from proto.BluePrintCommon_pb2 import ActionIdentifiers, CommonHeader
+from proto.BluePrintProcessing_pb2 import ExecutionServiceInput
+from resource_resolution.client import Client as ResourceResolutionClient
+
+
+def generate_messages():
+    commonHeader = CommonHeader()
+    commonHeader.requestId = "1234"
+    commonHeader.subRequestId = "1234-1"
+    commonHeader.originatorId = "CDS"
+
+    actionIdentifiers = ActionIdentifiers()
+    actionIdentifiers.blueprintName = "sample-cba"
+    actionIdentifiers.blueprintVersion = "1.0.0"
+    actionIdentifiers.actionName = "SampleScript"
+
+    input = ExecutionServiceInput(commonHeader=commonHeader, actionIdentifiers=actionIdentifiers)
+
+    commonHeader2 = CommonHeader()
+    commonHeader2.requestId = "1235"
+    commonHeader2.subRequestId = "1234-2"
+    commonHeader2.originatorId = "CDS"
+
+    input2 = ExecutionServiceInput(commonHeader=commonHeader2, actionIdentifiers=actionIdentifiers)
+
+    yield from [input, input2]
+
+
+if __name__ == "__main__":
+    with ResourceResolutionClient("127.0.0.1:9111", use_header_auth=True, header_auth_token="Token test") as client:
+        for response in client.process(generate_messages()):
+            print(response)
+
+```
+
+# ResourceResoulution helper class
+
+## How to use examples
+
+### Insecure channel
+
+```
+from resource_resolution.resource_resolution import ResourceResolution, WorkflowExecution, WorkflowExecutionResult
+
+
+if __name__ == "__main__":
+    with ResourceResolution(use_header_auth=True, header_auth_token="Basic token") as rr:
+        for response in rr.execute_workflows(  # type: WorkflowExecutionResult
+            WorkflowExecution(
+                blueprint_name="blueprintName",
+                blueprint_version="1.0",
+                workflow_name="resource-assignment"
+            )
+        ):
+            if response.has_error:
+                print(response.error_message)
+            else:
+                print(response.payload)
 ```
\ No newline at end of file
diff --git a/ms/py-executor/resource_resolution/authorization.py b/ms/py-executor/resource_resolution/authorization.py
new file mode 100644
index 000000000..ae5954ecc
--- /dev/null
+++ b/ms/py-executor/resource_resolution/authorization.py
@@ -0,0 +1,64 @@
+"""Copyright 2020 Deutsche Telekom.
+
+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.
+"""
+from collections import namedtuple
+from typing import Any, Callable, List
+
+from grpc import ClientCallDetails, StreamStreamClientInterceptor
+
+
+class NewClientCallDetails(
+    namedtuple("_ClientCallDetails", ("method", "timeout", "metadata", "credentials")), ClientCallDetails
+):
+    """Namedtuple class to store metadata.
+
+    It's impossible to change original metadata in ClientCallDetails object
+    passed as a parameter to intercept method, so this class is going to get
+    original metadata tuple and add the authorization one.
+    """
+
+    pass
+
+
+class AuthTokenInterceptor(StreamStreamClientInterceptor):
+    """Interceptor class to set authorization header.
+
+    Set authorization header (but it can be any header also) for a gRPC call.
+    """
+
+    def __init__(self, token: str, header: str = "authorization") -> None:
+        """Initialize interceptor.
+
+        Set token and header which should be set into call. By default header is "authorization".
+        Header have to be lowercase.
+
+        Args:
+            token (str): Token value to be set.
+            header (str, optional): Header name. It must be lowercase. Defaults to "authorization".
+        """
+        self.token: str = token
+        if not header.islower():
+            raise ValueError("Header must be lowercase.")
+        self.header: str = header
+
+    def intercept_stream_stream(
+        self, continuation: Callable, client_call_details: ClientCallDetails, request_iterator: Any
+    ) -> Any:
+        """Add header into metadata."""
+        metadata: List = list(client_call_details.metadata) if client_call_details.metadata is not None else []
+        metadata.append((self.header, self.token,))
+        new_client_call_details: NewClientCallDetails = NewClientCallDetails(
+            client_call_details.method, client_call_details.timeout, metadata, client_call_details.credentials
+        )
+        return continuation(new_client_call_details, request_iterator)
diff --git a/ms/py-executor/resource_resolution/client.py b/ms/py-executor/resource_resolution/client.py
index 89087745c..fee168628 100644
--- a/ms/py-executor/resource_resolution/client.py
+++ b/ms/py-executor/resource_resolution/client.py
@@ -14,12 +14,20 @@ limitations under the License.
 """
 from logging import Logger, getLogger
 from types import TracebackType
-from typing import Iterable, List, Optional, Type
-
-from grpc import Channel, insecure_channel, secure_channel, ssl_channel_credentials
+from typing import Iterable, Optional, Type
+
+from grpc import (
+    Channel,
+    insecure_channel,
+    intercept_channel,
+    secure_channel,
+    ssl_channel_credentials,
+)
 from proto.BluePrintProcessing_pb2 import ExecutionServiceInput, ExecutionServiceOutput
 from proto.BluePrintProcessing_pb2_grpc import BluePrintProcessingServiceStub
 
+from .authorization import AuthTokenInterceptor
+
 
 class Client:
     """Resource resoulution client class."""
@@ -28,20 +36,29 @@ class Client:
         self,
         server_address: str,
         *,
+        # TLS/SSL configuration
         use_ssl: bool = False,
         root_certificates: bytes = None,
         private_key: bytes = None,
         certificate_chain: bytes = None,
+        # Authentication header configuration
+        use_header_auth: bool = False,
+        header_auth_token: str = None,
     ) -> None:
         """Client class initialization.
 
         :param server_address: Address to server to connect.
         :param use_ssl: Boolean flag to determine if secure channel should be created or not. Keyword argument.
         :param root_certificates: The PEM-encoded root certificates. None if it shouldn't be used. Keyword argument.
-        :param private_key: The PEM-encoded private key as a byte string, or None if no private key should be used. Keyword argument.
-        :param certificate_chain: The PEM-encoded certificate chain as a byte string to use or or None if no certificate chain should be used. Keyword argument.
+        :param private_key: The PEM-encoded private key as a byte string, or None if no private key should be used.
+            Keyword argument.
+        :param certificate_chain: The PEM-encoded certificate chain as a byte string to use or or None if
+            no certificate chain should be used. Keyword argument.
+        :param use_header_auth: Boolean flag to determine if authorization headed shoud be added for every call or not.
+            Keyword argument.
+        :param header_auth_token: Authorization token value. Keyword argument.
         """
-        self.logger = getLogger(__name__)
+        self.logger: Logger = getLogger(__name__)
         if use_ssl:
             self.channel: Channel = secure_channel(
                 server_address, ssl_channel_credentials(root_certificates, private_key, certificate_chain)
@@ -50,6 +67,8 @@ class Client:
         else:
             self.channel: Channel = insecure_channel(server_address)
             self.logger.debug(f"Create insecure channel to connect to {server_address}")
+        if use_header_auth:
+            self.channel: Channel = intercept_channel(self.channel, AuthTokenInterceptor(header_auth_token))
         self.stub: BluePrintProcessingServiceStub = BluePrintProcessingServiceStub(self.channel)
 
     def close(self) -> None:
diff --git a/ms/py-executor/resource_resolution/resource_resolution.py b/ms/py-executor/resource_resolution/resource_resolution.py
new file mode 100644
index 000000000..e4f162f8f
--- /dev/null
+++ b/ms/py-executor/resource_resolution/resource_resolution.py
@@ -0,0 +1,294 @@
+"""Copyright 2020 Deutsche Telekom.
+
+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.
+"""
+
+from enum import Enum, unique
+from logging import Logger, getLogger
+from types import TracebackType
+from typing import Any, Dict, Generator, Optional, Type
+
+from google.protobuf import json_format
+
+from proto.BluePrintProcessing_pb2 import ExecutionServiceInput, ExecutionServiceOutput
+
+from .client import Client
+
+
+@unique
+class WorkflowMode(Enum):
+    """Workflow mode enumerator.
+
+    Workflow can be executed in two modes: synchronously and asynchronously.
+    This enumerator stores valid values to set the mode: SYNC for synchronously mode and ASYNC for asynchronously.
+    """
+
+    SYNC = "sync"
+    ASYNC = "async"
+
+
+class WorkflowExecution:
+    """Wokflow execution class.
+
+    Describes workflow to call. Set blueprint name and version and workflow name to execute.
+    Workflow inputs are optional, by default set to empty directory.
+    Workflow mode is also optional. It is set by default to call workflow synchronously.
+    """
+
+    def __init__(
+        self,
+        blueprint_name: str,
+        blueprint_version: str,
+        workflow_name: str,
+        workflow_inputs: Dict[str, Any] = None,
+        workflow_mode: WorkflowMode = WorkflowMode.SYNC,
+    ) -> None:
+        """Initialize workflow execution.
+
+        Get all needed information to execute workflow.
+
+        Args:
+            blueprint_name (str): Blueprint name to execute workflow from.
+            blueprint_version (str): Blueprint version.
+            workflow_name (str): Name of the workflow to execute
+            workflow_inputs (Dict[str, Any], optional): Key-value workflow inputs. Defaults to None.
+            workflow_mode (WorkflowMode, optional): Workflow execution mode. It can be run synchronously or
+                asynchronously. Defaults to WorkflowMode.SYNC.
+        """
+        self.blueprint_name: str = blueprint_name
+        self.blueprint_version: str = blueprint_version
+        self.workflow_name: str = workflow_name
+        if workflow_inputs is None:
+            workflow_inputs = {}
+        self.workflow_inputs: Dict[str, Any] = workflow_inputs
+        self.workflow_mode: WorkflowMode = workflow_mode
+
+    @property
+    def message(self) -> ExecutionServiceInput:
+        """Workflow execution protobuf message.
+
+        This message is going to be sent to gRPC server to execute workflow.
+
+        Returns:
+            ExecutionServiceInput: Properly filled protobuf message.
+        """
+        execution_msg: ExecutionServiceInput = ExecutionServiceInput()
+        execution_msg.actionIdentifiers.mode = self.workflow_mode.value
+        execution_msg.actionIdentifiers.blueprintName = self.blueprint_name
+        execution_msg.actionIdentifiers.blueprintVersion = self.blueprint_version
+        execution_msg.actionIdentifiers.actionName = self.workflow_name
+        execution_msg.payload.update({f"{self.workflow_name}-request": self.workflow_inputs})
+        return execution_msg
+
+
+class WorkflowExecutionResult:
+    """Result of workflow execution.
+
+    Store both workflow data and the result returns by server.
+    """
+
+    def __init__(self, workflow_execution: WorkflowExecution, execution_output: ExecutionServiceOutput) -> None:
+        """Initialize workflow execution result object.
+
+        Stores workflow execution data and execution result.
+
+        Args:
+            workflow_execution (WorkflowExecution): WorkflowExecution object which was used to call request.
+            execution_output (ExecutionServiceOutput): gRPC server response.
+        """
+        self.workflow_execution: WorkflowExecution = workflow_execution
+        self.execution_output: ExecutionServiceOutput = execution_output
+
+    @property
+    def blueprint_name(self) -> str:
+        """Name of blueprint used to call workflow.
+
+        This value is taken from server response not request (should be the same).
+
+        Returns:
+            str: Blueprint name
+        """
+        return self.execution_output.actionIdentifiers.blueprintName
+
+    @property
+    def blueprint_version(self) -> str:
+        """Blueprint version.
+
+        This value is taken from server response not request (should be the same).
+
+        Returns:
+            str: Blueprint version
+        """
+        return self.execution_output.actionIdentifiers.blueprintVersion
+
+    @property
+    def workflow_name(self) -> str:
+        """Workflow name.
+
+        This value is taken from server response not request (should be the same).
+
+        Returns:
+            str: Workflow name
+        """
+        return self.execution_output.actionIdentifiers.actionName
+
+    @property
+    def has_error(self) -> bool:
+        """Returns bool if request returns error or not.
+
+        Returns:
+            bool: True if response has status code different than 200
+        """
+        return self.execution_output.status.code != 200
+
+    @property
+    def error_message(self) -> str:
+        """Error message.
+
+        This property is available only if response has error. Otherwise AttributeError will be raised.
+
+        Raises:
+            AttributeError: Response has 200 response code and hasn't error message.
+
+        Returns:
+            str: Error message returned by server
+        """
+        if self.has_error:
+            return self.execution_output.status.errorMessage
+        raise AttributeError("Execution does not finish with error")
+
+    @property
+    def payload(self) -> dict:
+        """Response payload.
+
+        Payload retured by the server is migrated to Python dict.
+
+        Returns:
+            dict: Response's payload.
+        """
+        return json_format.MessageToDict(self.execution_output.payload)
+
+
+class ResourceResolution:
+    """Resource resolution class.
+
+    Helper class to connect to blueprintprocessor's gRPC server, send request to execute workflow and parse responses.
+    Blueprint with workflow must be deployed before workflow request.
+    It's possible to create both secre or unsecure connection (without SSL/TLS).
+    """
+
+    def __init__(
+        self,
+        *,
+        server_address: str = "127.0.0.1",
+        server_port: int = "9111",
+        use_ssl: bool = False,
+        root_certificates: bytes = None,
+        private_key: bytes = None,
+        certificate_chain: bytes = None,
+        # Authentication header configuration
+        use_header_auth: bool = False,
+        header_auth_token: str = None,
+    ) -> None:
+        """Resource resolution object initialization.
+
+        Args:
+            server_address (str, optional): gRPC server address. Defaults to "127.0.0.1".
+            server_port (int, optional): gRPC server address port. Defaults to "9111".
+            use_ssl (bool, optional): Boolean flag to determine if secure channel should be created or not.
+                Defaults to False.
+            root_certificates (bytes, optional): The PEM-encoded root certificates. None if it shouldn't be used.
+                Defaults to None.
+            private_key (bytes, optional): The PEM-encoded private key as a byte string, or None if no private key
+                should be used. Defaults to None.
+            certificate_chain (bytes, optional): The PEM-encoded certificate chain as a byte string to use or or None if
+                no certificate chain should be used. Defaults to None.
+            use_header_auth (bool, optional): Boolean flag to determine if authorization headed shoud be added for
+                every call or not. Defaults to False.
+            header_auth_token (str, optional): Authorization token value. Defaults to None.
+        """
+        # Logger
+        self.logger: Logger = getLogger(__name__)
+        # Client settings
+        self.client_server_address: str = server_address
+        self.client_server_port: str = server_port
+        self.client_use_ssl: bool = use_ssl
+        self.client_root_certificates: bytes = root_certificates
+        self.client_private_key: bytes = private_key
+        self.client_certificate_chain: bytes = certificate_chain
+        self.client_use_header_auth: bool = use_header_auth
+        self.client_header_auth_token: str = header_auth_token
+        self.client: Client = None
+
+    def __enter__(self) -> "ResourceResolution":
+        """Enter ResourceResolution instance context.
+
+        Client connection is created.
+        """
+        self.client = Client(
+            server_address=f"{self.client_server_address}:{self.client_server_port}",
+            use_ssl=self.client_use_ssl,
+            root_certificates=self.client_root_certificates,
+            private_key=self.client_private_key,
+            certificate_chain=self.client_certificate_chain,
+            use_header_auth=self.client_use_header_auth,
+            header_auth_token=self.client_header_auth_token,
+        )
+        return self
+
+    def __exit__(
+        self,
+        unused_exc_type: Optional[Type[BaseException]],
+        unused_exc_value: Optional[BaseException],
+        unused_traceback: Optional[TracebackType],
+    ) -> None:
+        """Exit ResourceResolution instance context.
+
+        Client connection is closed.
+        """
+        self.client.close()
+
+    def execute_workflows(self, *workflows: WorkflowExecution) -> Generator[WorkflowExecutionResult, None, None]:
+        """Execute provided workflows.
+
+        Workflows are going to be execured using one gRPC API call. Depends of implementation that may has
+        some consequences. In some cases if any request fails all requests after that won't be called.
+
+        Responses and zipped with workflows and WorkflowExecutionResult object is initialized and yielded.
+
+        Raises:
+            AttributeError: Raises if client object is not created. It occurs only if you not uses context manager.
+                Then user have to create client instance for ResourceResolution object by himself calling:
+                ```
+                resource_resoulution.client = Client(
+                    server_address=f"{resource_resoulution.client_server_address}:{resource_resoulution.client_server_port}",
+                    use_ssl=resource_resoulution.client_use_ssl,
+                    root_certificates=resource_resoulution.client_root_certificates,
+                    private_key=resource_resoulution.client_private_key,
+                    certificate_chain=resource_resoulution.client_certificate_chain,
+                    use_header_auth=resource_resoulution.client_use_header_auth,
+                    header_auth_token=resource_resoulution.client_header_auth_token,
+                )
+                ```
+                Remeber also to close client connection.
+
+        Returns:
+            Generator[WorkflowExecutionResult, None, None]: WorkflowExecutionResult object
+                with both WorkflowExection object and server response for it's request.
+        """
+        self.logger.debug("Execute workflows")
+        if not self.client:
+            raise AttributeError("gRPC client not connected")
+
+        for response, workflow in zip(self.client.process((workflow.message for workflow in workflows)), workflows):
+            yield WorkflowExecutionResult(workflow, response)
diff --git a/ms/py-executor/resource_resolution/tests/authorization_interceptor_test.py b/ms/py-executor/resource_resolution/tests/authorization_interceptor_test.py
new file mode 100644
index 000000000..4b03f0b36
--- /dev/null
+++ b/ms/py-executor/resource_resolution/tests/authorization_interceptor_test.py
@@ -0,0 +1,50 @@
+"""Copyright 2020 Deutsche Telekom.
+
+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.
+"""
+
+from unittest.mock import MagicMock, _Call
+
+import pytest
+
+from resource_resolution.authorization import AuthTokenInterceptor, NewClientCallDetails
+
+
+def test_resource_resolution_auth_token_interceptor():
+    """Test AuthTokenInterceptor class.
+
+     - Checks if it's correctly set default value.
+     - Checks if it's correctly set passed values.
+     - Checks if it's correctly checked if all header characters are lowercase.
+     - Checks if continuation function is called with headers setted 
+    """
+    interceptor: AuthTokenInterceptor = AuthTokenInterceptor("test_token", header="header")
+    assert interceptor.token == "test_token"
+    assert interceptor.header == "header"
+
+    interceptor: AuthTokenInterceptor = AuthTokenInterceptor("test_token")
+    assert interceptor.token == "test_token"
+    assert interceptor.header == "authorization"
+
+    with pytest.raises(ValueError):
+        AuthTokenInterceptor("test_token", header="Auth")
+
+    continuation_mock: MagicMock = MagicMock()
+    client_call_details: MagicMock = MagicMock()
+    request_iterator: MagicMock = MagicMock()
+
+    interceptor.intercept_stream_stream(continuation_mock, client_call_details, request_iterator)
+    continuation_mock.assert_called_once()
+    client_call_details_argument: _Call = continuation_mock.call_args_list[0][0][0]  # Get NewClientCallDetails instance
+    assert isinstance(client_call_details_argument, NewClientCallDetails)
+    assert client_call_details_argument.metadata[0] == (interceptor.header, interceptor.token)
diff --git a/ms/py-executor/resource_resolution/tests/resource_resolution_test.py b/ms/py-executor/resource_resolution/tests/resource_resolution_test.py
new file mode 100644
index 000000000..8a41357e6
--- /dev/null
+++ b/ms/py-executor/resource_resolution/tests/resource_resolution_test.py
@@ -0,0 +1,105 @@
+"""Copyright 2020 Deutsche Telekom.
+
+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.
+"""
+
+from google.protobuf import json_format
+from pytest import raises
+
+from resource_resolution.resource_resolution import (
+    ExecutionServiceInput,
+    ExecutionServiceOutput,
+    WorkflowExecution,
+    WorkflowExecutionResult,
+    WorkflowMode,
+)
+
+
+def test_workflow_execution_class():
+    """Workflow execution class tests.
+
+    - Test initialization and default values
+    - Test request message formatting
+    """
+    # Without inputs
+    workflow_execution: WorkflowExecution = WorkflowExecution("test blueprint", "test version", "test workflow")
+    assert workflow_execution.blueprint_name == "test blueprint"
+    assert workflow_execution.blueprint_version == "test version"
+    assert workflow_execution.workflow_name == "test workflow"
+    assert workflow_execution.workflow_inputs == {}
+    assert workflow_execution.workflow_mode == WorkflowMode.SYNC
+
+    msg: ExecutionServiceInput = workflow_execution.message
+    msg_dict: dict = json_format.MessageToDict(msg)
+    assert msg_dict["actionIdentifiers"]["blueprintName"] == "test blueprint"
+    assert msg_dict["actionIdentifiers"]["blueprintVersion"] == "test version"
+    assert msg_dict["actionIdentifiers"]["actionName"] == "test workflow"
+    assert msg_dict["actionIdentifiers"]["mode"] == "sync"
+    assert list(msg_dict["payload"].keys())[0] == "test workflow-request"
+    assert msg_dict["payload"]["test workflow-request"] == {}
+
+    # With inputs
+    workflow_execution: WorkflowExecution = WorkflowExecution(
+        "test blueprint2",
+        "test version2",
+        "test workflow2",
+        workflow_inputs={"test": "test"},
+        workflow_mode=WorkflowMode.ASYNC,
+    )
+    assert workflow_execution.blueprint_name == "test blueprint2"
+    assert workflow_execution.blueprint_version == "test version2"
+    assert workflow_execution.workflow_name == "test workflow2"
+    assert workflow_execution.workflow_inputs == {"test": "test"}
+    assert workflow_execution.workflow_mode == WorkflowMode.ASYNC
+
+    msg: ExecutionServiceInput = workflow_execution.message
+    msg_dict: dict = json_format.MessageToDict(msg)
+    assert msg_dict["actionIdentifiers"]["blueprintName"] == "test blueprint2"
+    assert msg_dict["actionIdentifiers"]["blueprintVersion"] == "test version2"
+    assert msg_dict["actionIdentifiers"]["actionName"] == "test workflow2"
+    assert msg_dict["actionIdentifiers"]["mode"] == "async"
+    assert list(msg_dict["payload"].keys())[0] == "test workflow2-request"
+    assert msg_dict["payload"]["test workflow2-request"] == {"test": "test"}
+
+
+def test_workflow_execution_result_class():
+    """Workflow execution result class tests.
+
+    - Test initizalization and default values
+    - Test `has_error` property
+    - Test `error_message` property
+    - Test payload formatting
+    """
+    workflow_execution: WorkflowExecution = WorkflowExecution("test blueprint", "test version", "test workflow")
+    execution_output: ExecutionServiceOutput = ExecutionServiceOutput()
+    execution_output.actionIdentifiers.blueprintName = "test blueprint"
+    execution_output.actionIdentifiers.blueprintVersion = "test version"
+    execution_output.actionIdentifiers.actionName = "test workflow"
+    execution_output.status.code = 200
+
+    execution_result: WorkflowExecutionResult = WorkflowExecutionResult(workflow_execution, execution_output)
+    assert not execution_result.has_error
+    with raises(AttributeError):
+        execution_result.error_message
+    assert execution_result.payload == {}
+    assert execution_result.blueprint_name == "test blueprint"
+    assert execution_result.blueprint_version == "test version"
+    assert execution_result.workflow_name == "test workflow"
+
+    execution_output.payload.update({"test_key": "test_value"})
+    execution_result: WorkflowExecutionResult = WorkflowExecutionResult(workflow_execution, execution_output)
+    assert execution_result.payload == {"test_key": "test_value"}
+
+    execution_output.status.code = 500
+    assert execution_result.has_error
+    assert execution_result.error_message == ""