Allow injected atom args to be persisted

Instead of only storing injected atom arguments in memory
allow for specifying those to be persisted; so that users
who desire this feature can persist them (it defaults to
being transient to retain the old API behavior).

This also reworks the validating of engine dependencies
to be more correct. It removes the validation of these
dependencies from the prepare() method and moves them to a
new engine validate() method; this allows users to prepare()
the engine, then inject there atom non-transient arguments
and then validate(); the validation would fail prior to this
at preparation time since no injected arguments would
exist and the user would not have the ability to inject any
that target a specific atom, since the atom detail would
not have been created yet (since that is populated in the
prepartion method).

Change-Id: I2846d0334db32a115592f850d85b206d9e6a3f07

1 files changed, 25 insertions, 0 deletions

diff --git a/taskflow/tests/unit/test_storage.py b/taskflow/tests/unit/test_storage.py
index a27a811..7401282 100644
--- a/taskflow/tests/unit/test_storage.py
+++ b/taskflow/tests/unit/test_storage.py
@@ -533,6 +533,31 @@ class StorageTestMixin(object):
         intention = s.get_atom_intention('my retry')
         self.assertEqual(intention, states.RETRY)
 
+    def test_inject_persistent_missing(self):
+        t = test_utils.ProgressingTask('my retry', requires=['x'])
+        s = self._get_storage()
+        s.ensure_atom(t)
+        missing = s.fetch_unsatisfied_args(t.name, t.rebind)
+        self.assertEqual(set(['x']), missing)
+        s.inject_atom_args(t.name, {'x': 2}, transient=False)
+        missing = s.fetch_unsatisfied_args(t.name, t.rebind)
+        self.assertEqual(set(), missing)
+        args = s.fetch_mapped_args(t.rebind, atom_name=t.name)
+        self.assertEqual(2, args['x'])
+
+    def test_inject_persistent_and_transient_missing(self):
+        t = test_utils.ProgressingTask('my retry', requires=['x'])
+        s = self._get_storage()
+        s.ensure_atom(t)
+        missing = s.fetch_unsatisfied_args(t.name, t.rebind)
+        self.assertEqual(set(['x']), missing)
+        s.inject_atom_args(t.name, {'x': 2}, transient=False)
+        s.inject_atom_args(t.name, {'x': 3}, transient=True)
+        missing = s.fetch_unsatisfied_args(t.name, t.rebind)
+        self.assertEqual(set(), missing)
+        args = s.fetch_mapped_args(t.rebind, atom_name=t.name)
+        self.assertEqual(3, args['x'])
+
 
 class StorageMemoryTest(StorageTestMixin, test.TestCase):
     def setUp(self):