Fix empty with items list resulting in parent workflow stuck in running

CHANGELOG.rst

@@ -24,6 +24,8 @@ Fixed
   their state to the MongoDB database. (bug fix) #4932
 
   Contributed by Nick Maludy (@nmaludy Encore Technologies)
+* Fix a bug where passing an empty list to a with items task in a subworkflow causes
+  the parent workflow to be stuck in running status. (bug fix) #4954
 
 
 3.2.0 - April 27, 2020

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

@@ -0,0 +1,6 @@
+---
+name: orquesta-test-subworkflow-empty-with-items
+description: A workflow for testing passing an emtpy list to with items task in a subworkflow.
+runner_type: orquesta
+entry_point: workflows/tests/orquesta-test-subworkflow-empty-with-items.yaml
+enabled: true

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

@@ -0,0 +1,12 @@
+version: 1.0
+
+tasks:
+  task1:
+    action: examples.orquesta-test-with-items
+    input:
+      tempfiles: []
+    next:
+      - when: <% succeeded() %>
+        do: task2
+  task2:
+    action: core.noop

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

@@ -582,3 +582,31 @@ def test_with_items_concurrency_pause_and_resume(self):
         self.assertEqual(wf_ex_db.status, wf_statuses.SUCCEEDED)
         lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
         self.assertEqual(lv_ac_db.status, action_constants.LIVEACTION_STATUS_SUCCEEDED)
+
+    def test_subworkflow_with_items_empty_list(self):
+        wf_input = {'members': []}
+        wf_meta = base.get_wf_fixture_meta_data(TEST_PACK_PATH, 'with-items-empty-parent.yaml')
+        lv_ac_db = lv_db_models.LiveActionDB(action=wf_meta['name'], parameters=wf_input)
+        lv_ac_db, ac_ex_db = action_service.request(lv_ac_db)
+
+        # Identify the records for the main workflow.
+        wf_ex_db = wf_db_access.WorkflowExecution.query(action_execution=str(ac_ex_db.id))[0]
+        tk_ex_dbs = wf_db_access.TaskExecution.query(workflow_execution=str(wf_ex_db.id))
+        self.assertEqual(len(tk_ex_dbs), 1)
+
+        # Identify the records for the tasks.
+        t1_ac_ex_db = ex_db_access.ActionExecution.query(task_execution=str(tk_ex_dbs[0].id))[0]
+        t1_wf_ex_db = wf_db_access.WorkflowExecution.query(action_execution=str(t1_ac_ex_db.id))[0]
+        self.assertEqual(t1_ac_ex_db.status, action_constants.LIVEACTION_STATUS_SUCCEEDED)
+        self.assertEqual(t1_wf_ex_db.status, wf_statuses.SUCCEEDED)
+
+        # Manually processing completion of the subworkflow in task1.
+        workflows.get_engine().process(t1_ac_ex_db)
+        t1_ex_db = wf_db_access.TaskExecution.get_by_id(tk_ex_dbs[0].id)
+        self.assertEqual(t1_ex_db.status, wf_statuses.SUCCEEDED)
+
+        # Check that the workflow execution is completed.
+        wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(wf_ex_db.id)
+        self.assertEqual(wf_ex_db.status, wf_statuses.SUCCEEDED)
+        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
+        self.assertEqual(lv_ac_db.status, action_constants.LIVEACTION_STATUS_SUCCEEDED)

st2common/st2common/services/workflows.py

@@ -595,7 +595,7 @@ def request_task_execution(wf_ex_db, st2_ctx, task_ex_req):
 
             # Fast forward task execution to completion.
             update_task_execution(str(task_ex_db.id), statuses.SUCCEEDED)
-            update_task_state(str(task_ex_db.id), statuses.SUCCEEDED, publish=False)
+            update_task_state(str(task_ex_db.id), statuses.SUCCEEDED)
 
             # Refresh and return the task execution
             return wf_db_access.TaskExecution.get_by_id(str(task_ex_db.id))

st2tests/integration/orquesta/test_wiring_with_items.py

@@ -287,3 +287,10 @@ def test_with_items_concurrency_pause_and_resume(self):
 
         # Wait for completion.
         ex = self._wait_for_state(ex, ac_const.LIVEACTION_STATUS_SUCCEEDED)
+
+    def test_subworkflow_empty_with_items(self):
+        wf_name = 'examples.orquesta-test-subworkflow-empty-with-items'
+        ex = self._execute_workflow(wf_name)
+        ex = self._wait_for_completion(ex)
+
+        self.assertEqual(ex.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)

st2tests/st2tests/fixtures/packs/orquesta_tests/actions/with-items-empty-parent.yaml

@@ -0,0 +1,12 @@
+---
+name: with-items-empty-parent
+description: A workflow to test passing of empty array to subworkflow that has a with items task.
+pack: orquesta_tests
+runner_type: orquesta
+entry_point: workflows/with-items-empty-parent.yaml
+enabled: true
+parameters:
+  members:
+    required: true
+    type: array
+    default: []

st2tests/st2tests/fixtures/packs/orquesta_tests/actions/with-items-empty.yaml

@@ -0,0 +1,12 @@
+---
+name: with-items-empty
+description: A workflow to test passing of empty array to with items task.
+pack: orquesta_tests
+runner_type: orquesta
+entry_point: workflows/with-items-empty.yaml
+enabled: true
+parameters:
+  members:
+    required: true
+    type: array
+    default: []

st2tests/st2tests/fixtures/packs/orquesta_tests/actions/workflows/with-items-empty-parent.yaml

@@ -0,0 +1,10 @@
+version: 1.0
+
+description: A workflow to test passing of empty array to subworkflow that has a with items task.
+
+input:
+  - members
+
+tasks:
+  task1:
+    action: orquesta_tests.with-items-empty members=<% ctx().members %>

st2tests/st2tests/fixtures/packs/orquesta_tests/actions/workflows/with-items-empty.yaml

@@ -0,0 +1,11 @@
+version: 1.0
+
+description: A workflow to test passing of empty array to with items task.
+
+input:
+  - members
+
+tasks:
+  task1:
+    with: <% ctx(members) %>
+    action: core.echo message="<% item() %>, resistance is futile!"