Fix the use of both adhoc actions and "with-items" in workflows

The WithItemsTask previously skipped the validate_input call.
Adding this resolves the issues with input and with-items.

Closes-Bug: 1597640
Change-Id: Id6a1dad09d6f21e24861bd4292df7152843bd9f3
Co-authored-by: Dawid Deja <dawid.deja@intel.com>

mistral/engine/tasks.py

@@ -406,6 +406,8 @@ class WithItemsTask(RegularTask):
 
             action = self._build_action()
 
+            action.validate_input(input_dict)
+
             action.schedule(
                 input_dict,
                 target,

mistral/tests/unit/engine/test_with_items.py

@@ -1241,3 +1241,62 @@ class WithItemsEngineTest(base.EngineTestCase):
         result = [item['result'] for item in task_result]
 
         self.assertListEqual(sorted(result), sorted(names))
+
+    def test_with_items_and_adhoc_action(self):
+        wb_text = """---
+        version: "2.0"
+
+        name: test
+
+        actions:
+          http:
+            input:
+              - url: http://www.example.com
+              - method: GET
+              - timeout: 10
+
+            output: <% $.content %>
+
+            base: std.http
+            base-input:
+              url: <% $.url %>
+              method: <% $.method %>
+              timeout: <% $.timeout %>
+
+        workflows:
+          with_items_default_bug:
+            description: Re-create the with-items bug with default values
+            type: direct
+
+            tasks:
+              get_pages:
+                with-items: page in <% range(0, 1) %>
+                action: test.http
+                input:
+                  url: http://www.example.com
+                  method: GET
+                on-success:
+                  - well_done
+
+              well_done:
+                action: std.echo output="Well done"
+        """
+
+        wb_service.create_workbook_v2(wb_text)
+
+        # Start workflow.
+        wf_ex = self.engine.start_workflow('test.with_items_default_bug', {})
+
+        self.await_workflow_success(wf_ex.id)
+
+        # Note: We need to reread execution to access related tasks.
+        wf_ex = db_api.get_workflow_execution(wf_ex.id)
+
+        task_execs = wf_ex.task_executions
+
+        task1_ex = self._assert_single_item(task_execs, name='get_pages')
+        task2_ex = self._assert_single_item(task_execs, name='well_done')
+
+        self.assertEqual(2, len(task_execs))
+        self.assertEqual(states.SUCCESS, task1_ex.state)
+        self.assertEqual(states.SUCCESS, task2_ex.state)