Added tests and updated placeholders documentation

Author: ca-nguyen

doc/placeholders.rst

@@ -1,16 +1,32 @@
 Placeholders
 =============
 
+
 Once defined, a workflow is static unless you update it explicitly. But, you can pass
 input to workflow executions. You can have dynamic values
 that you use in the **parameters** or **result_selector** fields of the steps in your workflow. For this,
 the AWS Step Functions Data Science SDK provides a way to define placeholders to pass around when you
-create your workflow. There are 3 mechanisms for passing dynamic values in a workflow.
+create your workflow.
+
+.. autoclass:: stepfunctions.inputs.Placeholder
+
+There are 3 mechanisms for passing dynamic values in a workflow:
+
+- `Execution Input <#execution-input>`__
 
+- `Step Input <#step-input>`__
+
+- `Step Result <#step-result>`__
+
+Execution Input
+---------------
 The first mechanism is a global input to the workflow execution. This input is
 accessible to all the steps in the workflow. The SDK provides :py:meth:`stepfunctions.inputs.ExecutionInput`
 to define the schema for this input, and to access the values in your workflow.
 
+.. autoclass:: stepfunctions.inputs.ExecutionInput
+                :inherited-members:
+
 .. code-block:: python
 
   # Create an instance of ExecutionInput class, and define a schema. Defining
@@ -50,13 +66,19 @@ to define the schema for this input, and to access the values in your workflow.
 
   workflow.execute(inputs={'myDynamicInput': "WorldHello"})
 
+
+Step Input
+----------
 The second mechanism is for passing dynamic values from one step to the next
 step. The output of one step becomes the input of the next step.
 The SDK provides the :py:meth:`stepfunctions.inputs.StepInput` class for this.
 
 By default, each step has an output method :py:meth:`stepfunctions.steps.states.State.output`
 that returns the placeholder output for that step.
 
+.. autoclass:: stepfunctions.inputs.StepInput
+                :inherited-members:
+
 .. code-block:: python
 
   lambda_state_first = LambdaStep(
@@ -81,6 +103,9 @@ that returns the placeholder output for that step.
 
   definition = Chain([lambda_state_first, lambda_state_second])
 
+
+Step Result
+-----------
 The third mechanism is a placeholder for a step's result. The result of a step can be modified
 with the **result_selector** field to replace the step's result. 
 
@@ -100,13 +125,8 @@ with the **result_selector** field to replace the step's result.
       }
   )
 
-.. autoclass:: stepfunctions.inputs.Placeholder
 
-.. autoclass:: stepfunctions.inputs.ExecutionInput
-                :inherited-members:
 
-.. autoclass:: stepfunctions.inputs.StepInput
-                :inherited-members:
 
 .. autoclass:: stepfunctions.inputs.StepResult
                 :inherited-members:

src/stepfunctions/steps/service.py

@@ -250,6 +250,7 @@ def __init__(self, state_id, wait_for_completion=True, **kwargs):
             heartbeat_seconds_path (str, optional): Path specifying the state's heartbeat value in seconds from the state input. When resolved, the path must select a field whose value is a positive integer.
             input_path (str, optional): Path applied to the state’s raw input to select some or all of it; that selection is used by the state. (default: '$')
             parameters (dict, optional): The value of this field becomes the effective input for the state.
+            result_selector (dict, optional): The value of this field becomes the effective result of the state.
             result_path (str, optional): Path specifying the raw input’s combination with or replacement by the state’s result. (default: '$')
             output_path (str, optional): Path applied to the state’s output after the application of `result_path`, producing the effective output which serves as the raw input for the next state. (default: '$')
             wait_for_completion (bool, optional): Boolean value set to `True` if the Task state should wait to complete before proceeding to the next step in the workflow. (default: True)
@@ -289,6 +290,7 @@ def __init__(self, state_id, wait_for_completion=True, **kwargs):
             heartbeat_seconds_path (str, optional): Path specifying the state's heartbeat value in seconds from the state input. When resolved, the path must select a field whose value is a positive integer.
             input_path (str, optional): Path applied to the state’s raw input to select some or all of it; that selection is used by the state. (default: '$')
             parameters (dict, optional): The value of this field becomes the effective input for the state.
+            result_selector (dict, optional): The value of this field becomes the effective result of the state.
             result_path (str, optional): Path specifying the raw input’s combination with or replacement by the state’s result. (default: '$')
             output_path (str, optional): Path applied to the state’s output after the application of `result_path`, producing the effective output which serves as the raw input for the next state. (default: '$')
             wait_for_completion (bool, optional): Boolean value set to `True` if the Task state should wait to complete before proceeding to the next step in the workflow. (default: True)
@@ -328,6 +330,7 @@ def __init__(self, state_id, wait_for_completion=True, **kwargs):
             heartbeat_seconds_path (str, optional): Path specifying the state's heartbeat value in seconds from the state input. When resolved, the path must select a field whose value is a positive integer.
             input_path (str, optional): Path applied to the state’s raw input to select some or all of it; that selection is used by the state. (default: '$')
             parameters (dict, optional): The value of this field becomes the effective input for the state.
+            result_selector (dict, optional): The value of this field becomes the effective result of the state.
             result_path (str, optional): Path specifying the raw input’s combination with or replacement by the state’s result. (default: '$')
             output_path (str, optional): Path applied to the state’s output after the application of `result_path`, producing the effective output which serves as the raw input for the next state. (default: '$')
             wait_for_completion (bool, optional): Boolean value set to `True` if the Task state should wait to complete before proceeding to the next step in the workflow. (default: True)
@@ -367,6 +370,7 @@ def __init__(self, state_id, wait_for_completion=True, **kwargs):
             heartbeat_seconds_path (str, optional): Path specifying the state's heartbeat value in seconds from the state input. When resolved, the path must select a field whose value is a positive integer.
             input_path (str, optional): Path applied to the state’s raw input to select some or all of it; that selection is used by the state. (default: '$')
             parameters (dict, optional): The value of this field becomes the effective input for the state.
+            result_selector (dict, optional): The value of this field becomes the effective result of the state.
             result_path (str, optional): Path specifying the raw input’s combination with or replacement by the state’s result. (default: '$')
             output_path (str, optional): Path applied to the state’s output after the application of `result_path`, producing the effective output which serves as the raw input for the next state. (default: '$')
             wait_for_completion (bool, optional): Boolean value set to `True` if the Task state should wait to complete before proceeding to the next step in the workflow. (default: True)
@@ -406,6 +410,7 @@ def __init__(self, state_id, wait_for_completion=True, **kwargs):
             heartbeat_seconds_path (str, optional): Path specifying the state's heartbeat value in seconds from the state input. When resolved, the path must select a field whose value is a positive integer.
             input_path (str, optional): Path applied to the state’s raw input to select some or all of it; that selection is used by the state. (default: '$')
             parameters (dict, optional): The value of this field becomes the effective input for the state.
+            result_selector (dict, optional): The value of this field becomes the effective result of the state.
             result_path (str, optional): Path specifying the raw input’s combination with or replacement by the state’s result. (default: '$')
             output_path (str, optional): Path applied to the state’s output after the application of `result_path`, producing the effective output which serves as the raw input for the next state. (default: '$')
             wait_for_completion (bool, optional): Boolean value set to `True` if the Task state should wait to complete before proceeding to the next step in the workflow. (default: True)
@@ -445,6 +450,7 @@ def __init__(self, state_id, wait_for_completion=True, **kwargs):
             heartbeat_seconds_path (str, optional): Path specifying the state's heartbeat value in seconds from the state input. When resolved, the path must select a field whose value is a positive integer.
             input_path (str, optional): Path applied to the state’s raw input to select some or all of it; that selection is used by the state. (default: '$')
             parameters (dict, optional): The value of this field becomes the effective input for the state.
+            result_selector (dict, optional): The value of this field becomes the effective result of the state.
             result_path (str, optional): Path specifying the raw input’s combination with or replacement by the state’s result. (default: '$')
             output_path (str, optional): Path applied to the state’s output after the application of `result_path`, producing the effective output which serves as the raw input for the next state. (default: '$')
             wait_for_completion (bool, optional): Boolean value set to `True` if the Task state should wait to complete before proceeding to the next step in the workflow. (default: True)
@@ -484,6 +490,7 @@ def __init__(self, state_id, wait_for_completion=True, **kwargs):
             heartbeat_seconds_path (str, optional): Path specifying the state's heartbeat value in seconds from the state input. When resolved, the path must select a field whose value is a positive integer.
             input_path (str, optional): Path applied to the state’s raw input to select some or all of it; that selection is used by the state. (default: '$')
             parameters (dict, optional): The value of this field becomes the effective input for the state.
+            result_selector (dict, optional): The value of this field becomes the effective result of the state.
             result_path (str, optional): Path specifying the raw input’s combination with or replacement by the state’s result. (default: '$')
             output_path (str, optional): Path applied to the state’s output after the application of `result_path`, producing the effective output which serves as the raw input for the next state. (default: '$')
             wait_for_completion (bool, optional): Boolean value set to `True` if the Task state should wait to complete before proceeding to the next step in the workflow. (default: True)
@@ -523,6 +530,7 @@ def __init__(self, state_id, **kwargs):
             heartbeat_seconds_path (str, optional): Path specifying the state's heartbeat value in seconds from the state input. When resolved, the path must select a field whose value is a positive integer.
             input_path (str, optional): Path applied to the state’s raw input to select some or all of it; that selection is used by the state. (default: '$')
             parameters (dict, optional): The value of this field becomes the effective input for the state.
+            result_selector (dict, optional): The value of this field becomes the effective result of the state.
             result_path (str, optional): Path specifying the raw input’s combination with or replacement by the state’s result. (default: '$')
             output_path (str, optional): Path applied to the state’s output after the application of `result_path`, producing the effective output which serves as the raw input for the next state. (default: '$')
         """

tests/unit/test_compute_steps.py

@@ -16,16 +16,24 @@
 import boto3
 
 from unittest.mock import patch
+
+from stepfunctions.inputs import StepResult
 from stepfunctions.steps.compute import LambdaStep, GlueStartJobRunStep, BatchSubmitJobStep, EcsRunTaskStep
 
 
+STEP_RESULT = StepResult()
+RESULT_SELECTOR = {
+    "OutputA":  STEP_RESULT['A']
+}
+
 @patch.object(boto3.session.Session, 'region_name', 'us-east-1')
 def test_lambda_step_creation():
-    step = LambdaStep('Echo')
+    step = LambdaStep('Echo', result_selector=RESULT_SELECTOR)
 
     assert step.to_dict() == {
         'Type': 'Task',
         'Resource': 'arn:aws:states:::lambda:invoke',
+        'ResultSelector': {'OutputA.$': "$['A']"},
         'End': True
     }
 
@@ -51,11 +59,12 @@ def test_lambda_step_creation():
 
 @patch.object(boto3.session.Session, 'region_name', 'us-east-1')
 def test_glue_start_job_run_step_creation():
-    step = GlueStartJobRunStep('Glue Job', wait_for_completion=False)
+    step = GlueStartJobRunStep('Glue Job', wait_for_completion=False, result_selector=RESULT_SELECTOR)
 
     assert step.to_dict() == {
         'Type': 'Task',
         'Resource': 'arn:aws:states:::glue:startJobRun',
+        'ResultSelector': {'OutputA.$': "$['A']"},
         'End': True
     }
 
@@ -75,11 +84,12 @@ def test_glue_start_job_run_step_creation():
 
 @patch.object(boto3.session.Session, 'region_name', 'us-east-1')
 def test_batch_submit_job_step_creation():
-    step = BatchSubmitJobStep('Batch Job', wait_for_completion=False)
+    step = BatchSubmitJobStep('Batch Job', wait_for_completion=False, result_selector=RESULT_SELECTOR)
 
     assert step.to_dict() == {
         'Type': 'Task',
         'Resource': 'arn:aws:states:::batch:submitJob',
+        'ResultSelector': {'OutputA.$': "$['A']"},
         'End': True
     }
 
@@ -101,11 +111,12 @@ def test_batch_submit_job_step_creation():
 
 @patch.object(boto3.session.Session, 'region_name', 'us-east-1')
 def test_ecs_run_task_step_creation():
-    step = EcsRunTaskStep('Ecs Job', wait_for_completion=False)
+    step = EcsRunTaskStep('Ecs Job', wait_for_completion=False, result_selector=RESULT_SELECTOR)
 
     assert step.to_dict() == {
         'Type': 'Task',
         'Resource': 'arn:aws:states:::ecs:runTask',
+        'ResultSelector': {'OutputA.$': "$['A']"},
         'End': True
     }