need to copy staged so it isn't mutated for with items

orquesta/conducting.py

@@ -68,7 +68,7 @@ def deserialize(cls, data):
         instance.contexts = data.get("contexts", list())
         instance.routes = data.get("routes", list())
         instance.sequence = data.get("sequence", list())
-        instance.staged = data.get("staged", list())
+        instance.staged = json_util.deepcopy(data.get("staged", list()))
         instance.status = data.get("status", statuses.UNSET)
         instance.tasks = data.get("tasks", dict())
         instance.reruns = data.get("reruns", list())
@@ -481,8 +481,7 @@ def get_workflow_terminal_context(self):
         for idx, task in other_term_tasks:
             # Remove the initial context since the first task processed above already
             # inclulded that and we only want to apply the differences.
-            in_ctx_idxs = [i for index, i in enumerate(task["ctxs"]["in"]) if
-                           index != 0]
+            in_ctx_idxs = [i for index, i in enumerate(task["ctxs"]["in"]) if index != 0]
 
             wf_term_ctx = dict_util.merge_dicts(
                 wf_term_ctx, self.get_task_context(in_ctx_idxs), overwrite=True

orquesta/machines.py

@@ -527,8 +527,11 @@ def add_context_to_task_item_event(cls, workflow_state, task_id, task_route, ac_
             # Make a copy of the items and remove current item under evaluation.
             staged_task = workflow_state.get_staged_task(task_id, task_route)
             items = staged_task["items"]
-            items_status = [item.get("status", statuses.UNSET) for index, item in
-                            enumerate(items) if index != ac_ex_event.item_id]
+            items_status = [
+                item.get("status", statuses.UNSET)
+                for index, item in enumerate(items)
+                if index != ac_ex_event.item_id
+            ]
 
             # Assess various situations.
             # todo(aj) loop over list one time and add to each list

orquesta/tests/unit/base.py

@@ -209,20 +209,11 @@ def assert_task_list(self, conductor, actual, expected):
         expected_copy = copy.deepcopy(expected)
 
         for task in actual_copy:
-            for staged_task in task["ctx"]["__state"]["staged"]:
-                if "items" in staged_task:
-                    del staged_task["items"]
-
             task["spec"] = task["spec"].serialize()
 
         for task in expected_copy:
             task["ctx"]["__current_task"] = {"id": task["id"], "route": task["route"]}
-            task["ctx"]["__state"] = copy.deepcopy(conductor.workflow_state.serialize())
-
-            for staged_task in task["ctx"]["__state"]["staged"]:
-                if "items" in staged_task:
-                    del staged_task["items"]
-
+            task["ctx"]["__state"] = conductor.workflow_state.serialize()
             task["spec"] = task["spec"].serialize()
 
         self.assertListEqual(actual_copy, expected_copy)