Merge pull request #4400 from StackStorm/orquesta-with-items

Implement with items for Orquesta workflows

Author: m4dcoder

contrib/examples/actions/orquesta-test-with-items.yaml

@@ -0,0 +1,12 @@
+---
+name: orquesta-test-with-items
+description: A workflow for testing with items and concurrency.
+runner_type: orquesta
+entry_point: workflows/tests/orquesta-test-with-items.yaml
+enabled: true
+parameters:
+  tempfiles:
+    required: true
+    type: array
+  concurrency:
+    type: integer

contrib/examples/actions/orquesta-with-items-concurrency.yaml

@@ -0,0 +1,14 @@
+---
+name: orquesta-with-items-concurrency
+description: A workflow demonstrating with items and concurrent processing.
+runner_type: orquesta
+entry_point: workflows/orquesta-with-items-concurrency.yaml
+enabled: true
+parameters:
+  members:
+    required: true
+    type: array
+    default:
+      - Lakshmi
+      - Lindsay
+      - Tomaz

contrib/examples/actions/orquesta-with-items.yaml

@@ -0,0 +1,14 @@
+---
+name: orquesta-with-items
+description: A workflow demonstrating with items.
+runner_type: orquesta
+entry_point: workflows/orquesta-with-items.yaml
+enabled: true
+parameters:
+  members:
+    required: true
+    type: array
+    default:
+      - Lakshmi
+      - Lindsay
+      - Tomaz

contrib/examples/actions/workflows/orquesta-with-items-concurrency.yaml

@@ -0,0 +1,16 @@
+version: 1.0
+  
+description: A workflow demonstrating with items and concurrent processing.
+
+input:
+  - members
+
+tasks:
+  task1:
+    with:
+      items: <% ctx(members) %>
+      concurrency: 2
+    action: core.echo message="<% item() %>, resistance is futile!"
+
+output:
+  - items: <% task(task1).result.items.select($.result.stdout) %>

contrib/examples/actions/workflows/orquesta-with-items.yaml

@@ -0,0 +1,14 @@
+version: 1.0
+  
+description: A workflow demonstrating with items.
+
+input:
+  - members
+
+tasks:
+  task1:
+    with: <% ctx(members) %>
+    action: core.echo message="<% item() %>, resistance is futile!"
+
+output:
+  - items: <% task(task1).result.items.select($.result.stdout) %>

contrib/examples/actions/workflows/tests/orquesta-test-with-items.yaml

@@ -0,0 +1,17 @@
+version: 1.0
+  
+description: A workflow for testing with items and concurrency.
+
+input:
+  - tempfiles
+  - concurrency: <% len(ctx(tempfiles)) %>
+
+tasks:
+  task1:
+    with:
+      items: tempfile in <% ctx(tempfiles) %>
+      concurrency: <% ctx(concurrency) %>
+    action: core.local
+    input:
+      cmd: "while [ -e '<% item(tempfile) %>' ]; do sleep 0.1; done"
+      timeout: 300

contrib/runners/orquesta_runner/in-requirements.txt

@@ -1 +1 @@
-git+https://github.com/StackStorm/orquesta.git#egg=orquesta
+git+https://github.com/StackStorm/orquesta.git@142a19ddabbeea6a8d9d10341ba70585114dc3b7#egg=orquesta

contrib/runners/orquesta_runner/orquesta_runner/orquesta_runner.py

@@ -134,7 +134,7 @@ def pause(self):
 
     def resume(self):
         # Resume the target workflow.
-        wf_svc.request_resume(self.execution)
+        wf_ex_db = wf_svc.request_resume(self.execution)
 
         # Request resume of tasks that are workflows and still running.
         for child_ex_id in self.execution.children:
@@ -147,7 +147,7 @@ def resume(self):
                 )
 
         return (
-            ac_const.LIVEACTION_STATUS_RUNNING,
+            wf_ex_db.status if wf_ex_db else ac_const.LIVEACTION_STATUS_RUNNING,
             self.liveaction.result,
             self.liveaction.context
         )

contrib/runners/orquesta_runner/requirements.txt

@@ -1,2 +1,2 @@
 # Don't edit this file. It's generated automatically!
-git+https://github.com/StackStorm/orquesta.git#egg=orquesta
+git+https://github.com/StackStorm/orquesta.git@142a19ddabbeea6a8d9d10341ba70585114dc3b7#egg=orquesta

contrib/runners/orquesta_runner/tests/unit/test_error_handling.py

@@ -138,8 +138,11 @@ def test_fail_inspection(self):
             },
             {
                 'type': 'syntax',
-                'message': '[{\'cmd\': \'echo <% ctx().macro %>\'}] is not of type \'object\'',
-                'schema_path': 'properties.tasks.patternProperties.^\w+$.properties.input.type',
+                'message': (
+                    '[{\'cmd\': \'echo <% ctx().macro %>\'}] is '
+                    'not valid under any of the given schemas'
+                ),
+                'schema_path': 'properties.tasks.patternProperties.^\w+$.properties.input.oneOf',
                 'spec_path': 'tasks.task2.input'
             }
         ]