Fix how action result is assigned to task 'state_info' field

Closes-Bug: #1802477

Change-Id: Ia8848b3bb0417f66422c4995b64be7a803dde1e7

mistral/engine/tasks.py

@@ -375,8 +375,13 @@ class RegularTask(Task):
         # TODO(rakhmerov): Here we can define more informative messages for
         # cases when action is successful and when it's not. For example,
         # in state_info we can specify the cause action.
-        state_info = (None if state == states.SUCCESS
-                      else action_ex.output.get('result'))
+
+        if state == states.SUCCESS:
+            state_info = None
+        else:
+            action_result = action_ex.output.get('result')
+
+            state_info = str(action_result) if action_result else None
 
         self.complete(state, state_info)
 

mistral/tests/unit/engine/test_direct_workflow.py

@@ -13,6 +13,8 @@
 #    See the License for the specific language governing permissions and
 #    limitations under the License.
 
+import six
+
 from oslo_config import cfg
 
 from mistral.db.v2 import api as db_api
@@ -1090,3 +1092,44 @@ class DirectWorkflowEngineTest(base.EngineTestCase):
 
         self._assert_single_item(task_execs, name='task0')
         self._assert_single_item(task_execs, name='task{}'.format(task_cnt))
+
+    def test_action_error_with_array_result(self):
+        wf_text = """---
+        version: '2.0'
+
+        wf:
+          tasks:
+            task1:
+              action: std.fail error_data=[1,2,3]
+        """
+
+        wf_service.create_workflows(wf_text)
+
+        # Start workflow.
+        wf_ex = self.engine.start_workflow('wf')
+
+        self.await_workflow_error(wf_ex.id)
+
+        with db_api.transaction():
+            # Note: We need to reread execution to access related tasks.
+            wf_ex = db_api.get_workflow_execution(wf_ex.id)
+
+            task_ex = self._assert_single_item(
+                wf_ex.task_executions,
+                name='task1',
+                state=states.ERROR
+            )
+
+            a_ex = self._assert_single_item(
+                task_ex.action_executions,
+                name='std.fail'
+            )
+
+            self.assertIsInstance(a_ex.output.get('result'), list)
+
+            # NOTE(rakhmerov): This was previously failing but only Python 2.7
+            # probably because SQLAlchemy works differently on different
+            # versions of Python. On Python 3 this field's value was always
+            # converted into a string no matter what we tried to assign. But
+            # that didn't happen on Python 2.7 which caused an SQL exception.
+            self.assertIsInstance(task_ex.state_info, six.string_types)

releasenotes/notes/fix_task_state_info_assignment-e25481ce8c3193ba.yaml

@@ -0,0 +1,9 @@
+---
+fixes:
+  - |
+    If an action execution fails but returns a result as a list (error=[])
+    the result of this action is assigned to the task execution 'state_info'
+    field which is a string according to the DB model. On Python 3 it this
+    list magically converts to a string. On Python 2.7 it doesn't. The reason
+    is probably in how SQLAlchemy works on different versions of Python. This
+    has now been fixed with an explicit type coercion.