fix-hydration

Signed-off-by: Antonio Mendoza Pérez <[email protected]>

Author: antmendoza

serverlessworkflow/sdk/databased_switch_state.py

@@ -6,7 +6,7 @@
 from serverlessworkflow.sdk.default_condition_def import DefaultConditionDef
 from serverlessworkflow.sdk.end_data_condition import EndDataCondition
 from serverlessworkflow.sdk.error import Error
-from serverlessworkflow.sdk.hydration import HydratableParameter, UnionTypeOf, ComplexTypeOf, ArrayTypeOf, \
+from serverlessworkflow.sdk.hydration import HydratableParameter, ComplexTypeOf, ArrayTypeOf, \
 Fields
 from serverlessworkflow.sdk.metadata import Metadata
 from serverlessworkflow.sdk.state import State
@@ -52,7 +52,8 @@ def f_hydration(p_key, p_value):
 return HydratableParameter(value=p_value).hydrateAs(ComplexTypeOf(DataBasedSwitchStateTime0ut))
 
 if p_key == 'dataConditions':
-return [DataBasedSwitchState.hydrate_state(v) if type(v) is not (TransitionDataCondition or EndDataCondition) else v for v in p_value]
+return [DataBasedSwitchState.hydrate_state(v) if not (
+isinstance(v, TransitionDataCondition or EndDataCondition)) else v for v in p_value]
 
 if p_key == 'onErrors':
 return HydratableParameter(value=p_value).hydrateAs(ArrayTypeOf(Error))

serverlessworkflow/sdk/event_based_switch_state.py

@@ -6,7 +6,7 @@
 from serverlessworkflow.sdk.end_event_condition import EndEventCondition
 from serverlessworkflow.sdk.error import Error
 from serverlessworkflow.sdk.event_based_switch_state_timeout import EventBasedSwitchStateTimeOut
-from serverlessworkflow.sdk.hydration import HydratableParameter, ComplexTypeOf, UnionTypeOf, ArrayTypeOf, \
+from serverlessworkflow.sdk.hydration import HydratableParameter, ComplexTypeOf, ArrayTypeOf, \
 Fields
 from serverlessworkflow.sdk.metadata import Metadata
 from serverlessworkflow.sdk.state import State
@@ -52,7 +52,8 @@ def f_hydration(p_key, p_value):
 return HydratableParameter(value=p_value).hydrateAs(ComplexTypeOf(EventBasedSwitchStateTimeOut))
 
 if p_key == 'eventConditions':
-return [EventBasedSwitchState.hydrate_state(v) if type(v) is not (TransitionEventCondition or EndEventCondition) else v for v in p_value]
+return [EventBasedSwitchState.hydrate_state(v) if not (
+isinstance(v, TransitionEventCondition or EndEventCondition)) else v for v in p_value]
 
 if p_key == 'onErrors':
 return HydratableParameter(value=p_value).hydrateAs(ArrayTypeOf(Error))

serverlessworkflow/sdk/function_ref.py

@@ -11,7 +11,7 @@ class FunctionRef:
 
 def __init__(self,
 refName: str = None,
-arguments: dict[str, dict] = None,
+arguments: dict[str, any] = None,
 selectionSet: str = None,
 invoke: str = None,
 **kwargs):

serverlessworkflow/sdk/hydration.py

@@ -1,10 +1,12 @@
 from __future__ import annotations
 
 import dataclasses
+from abc import ABC, abstractmethod
 from typing import Any
 
 
-class HydratableType:
+class HydratableType(ABC):
+@abstractmethod
 def hydrate(self, value):
 pass
 

serverlessworkflow/sdk/workflow.py

@@ -142,7 +142,7 @@ def f_hydration(p_key, p_value):
 ArrayTypeOf(AuthDef)]))
 
 if p_key == 'states':
-return [Workflow.hydrate_state(v) if type(v) is not State else v for v in p_value]
+return [Workflow.hydrate_state(v) if not isinstance(v,State) else v for v in p_value]
 
 if p_key == 'functions':
 return HydratableParameter(value=p_value).hydrateAs(UnionTypeOf([SimpleTypeOf(str),

tests/serverlessworkflow/sdk/test_event_based_switch_state.py

@@ -0,0 +1,16 @@
+import unittest
+
+from serverlessworkflow.sdk.event_based_switch_state import EventBasedSwitchState
+from serverlessworkflow.sdk.transition_event_condition import TransitionEventCondition
+
+
+class TestEventBasedSwitchState(unittest.TestCase):
+def test_programmatically_create_object(self):
+event_based_switch_state = EventBasedSwitchState(eventConditions=[TransitionEventCondition(
+name="Hold Book",
+eventRef="Hold Book Event",
+transition="Request Hold"
+
+)])
+
+self.assertTrue(isinstance(event_based_switch_state.eventConditions[0], TransitionEventCondition))

tests/serverlessworkflow/sdk/test_workflow.py

@@ -4,45 +4,46 @@
 from os import listdir
 
 from serverlessworkflow.sdk.action import Action
+from serverlessworkflow.sdk.action_data_filter import ActionDataFilter
 from serverlessworkflow.sdk.function import Function
 from serverlessworkflow.sdk.function_ref import FunctionRef
+from serverlessworkflow.sdk.operation_state import OperationState
 from serverlessworkflow.sdk.workflow import Workflow
 
 
 class TestWorkflow(unittest.TestCase):
-workflow = Workflow(id_="greeting",
-name="Greeting Workflow",
-description="Greet Someone",
-version='1.0',
-specVersion='0.8',
-start="Greet",
-states=[
-{
-"name": "Greet",
-"type": "operation",
-"actions": [
-{
-"functionRef": {
-"refName": "greetingFunction",
-"arguments": {
-"name": "${ .person.name }"
-}
-},
-"actionDataFilter": {
-"results": "${ .greeting }"
-}
-}
-],
-"end": True
+workflow = Workflow(
+id_="greeting",
+name="Greeting Workflow",
+description="Greet Someone",
+version='1.0',
+specVersion='0.8',
+start="Greet",
+states=[
+OperationState(
+name="Greet",
+type="operation",
+actions=[
+Action(
+functionRef=FunctionRef(
+refName="greetingFunction",
+arguments={
+"name": "${ .person.name }"
 }
-],
-functions=[
-{
-"name": "greetingFunction",
-"operation": "file://myapis/greetingapis.json#greeting"
-}
-]
+),
+actionDataFilter=ActionDataFilter(
+results="${ .greeting }"
 )
+)
+],
+end=True
+)
+],
+functions=[
+Function(name="greetingFunction",
+operation="file://myapis/greetingapis.json#greeting")
+]
+)
 
 def test_workflow_to_json(self):
 expected = """{
@@ -106,6 +107,16 @@ def test_workflow_to_yaml(self):
 """
 self.assertEqual(expected, self.workflow.to_yaml())
 
+def test_programmatically_create_workflow(self):
+
+self.assertEqual("greeting", self.workflow.id)
+self.assertEqual("operation", self.workflow.states[0].type)
+self.assertTrue(isinstance(self.workflow.states[0], OperationState))
+self.assertEqual(True, self.workflow.states[0].end)
+self.assertTrue(isinstance(self.workflow.states[0].actions[0], Action))
+self.assertTrue(isinstance(self.workflow.states[0].actions[0].functionRef, FunctionRef))
+self.assertTrue(isinstance(self.workflow.functions[0], Function))
+
 def test_workflow_from_source_json(self):
 examples_dir = os.path.join(os.path.dirname(__file__), '../../examples')
 examples = listdir(examples_dir)