Fix malformed state_info when json is set to it

Fix bug where setting a dict to state_info did not convert to
JSON format properly.

Change-Id: Ib78d65483dd628e03de2484d3176ab7f275bc9d6
Closes-bug: #1662350

mistral/engine/tasks.py

@@ -16,6 +16,7 @@
 
 import abc
 import copy
+import json
 from oslo_config import cfg
 from oslo_log import log as logging
 from osprofiler import profiler
@@ -166,7 +167,8 @@ class Task(object):
                 return False
 
             self.task_ex = task_ex
-            self.task_ex.state_info = state_info
+            self.task_ex.state_info = json.dumps(state_info) \
+                if isinstance(state_info, dict) else state_info
             self.state_changed = True
 
             if processed is not None:
@@ -179,7 +181,7 @@ class Task(object):
                  self.task_ex.id,
                  cur_state,
                  state,
-                 state_info)
+                 self.task_ex.state_info)
             )
 
         return True

mistral/engine/workflows.py

@@ -15,6 +15,7 @@
 #    limitations under the License.
 
 import abc
+import json
 from oslo_config import cfg
 from oslo_log import log as logging
 from osprofiler import profiler
@@ -341,12 +342,16 @@ class Workflow(object):
                 return
 
             self.wf_ex = wf_ex
-            self.wf_ex.state_info = state_info
+            self.wf_ex.state_info = json.dumps(state_info) \
+                if isinstance(state_info, dict) else state_info
 
             wf_trace.info(
                 self.wf_ex,
-                "Workflow '%s' [%s -> %s, msg=%s]"
-                % (self.wf_ex.workflow_name, cur_state, state, state_info)
+                "Workflow '%s' [%s -> %s, msg=%s]" %
+                (self.wf_ex.workflow_name,
+                 cur_state,
+                 state,
+                 self.wf_ex.state_info)
             )
         else:
             msg = ("Can't change workflow execution state from %s to %s. "

mistral/tests/unit/engine/test_state_info.py

@@ -169,3 +169,30 @@ class ExecutionStateInfoTest(base.EngineTestCase):
 
         for action_ex in success_actions:
             self.assertNotIn(action_ex.id, wf_ex.state_info)
+
+    def test_state_info_with_json(self):
+        workflow = """---
+        version: "2.0"
+        wf_state_info:
+          type: direct
+          tasks:
+            main_task:
+              action: std.test_dict
+              input:
+                size: 1
+                key_prefix: "abc"
+                val: "pqr"
+              on-success:
+                - fail msg="<% task().result %>"
+        """
+        wf_service.create_workflows(workflow)
+
+        # Start workflow.
+        wf_ex = self.engine.start_workflow('wf_state_info')
+
+        self.await_workflow_error(wf_ex.id)
+
+        # Note: We need to reread execution to access related tasks.
+        wf_ex = db_api.get_workflow_execution(wf_ex.id)
+
+        self.assertIn('{"abc0": "pqr"}', wf_ex.state_info)