Storage: restore injected data on resumption

In storage we now use single task for injector (we look it up by name)
and we restore result mapping for that task in constructor if there
is one in flow detail we are given. This has visible effect
of injected data staying in storage after resumption.

Closes-bug: #1240931
Change-Id: I2e8fb2c4b5cca769d36f578a934119db4a530f5c

1 files changed, 27 insertions, 0 deletions

diff --git a/taskflow/tests/unit/test_storage.py b/taskflow/tests/unit/test_storage.py
index c2019ab..ec62915 100644
--- a/taskflow/tests/unit/test_storage.py
+++ b/taskflow/tests/unit/test_storage.py
@@ -180,6 +180,33 @@ class StorageTest(test.TestCase):
             'spam': 'eggs',
         })
 
+    def test_inject_twice(self):
+        s = self._get_storage()
+        s.inject({'foo': 'bar'})
+        self.assertEquals(s.fetch_all(), {'foo': 'bar'})
+        s.inject({'spam': 'eggs'})
+        self.assertEquals(s.fetch_all(), {
+            'foo': 'bar',
+            'spam': 'eggs',
+        })
+
+    def test_inject_resumed(self):
+        s = self._get_storage()
+        s.inject({'foo': 'bar', 'spam': 'eggs'})
+        # verify it's there
+        self.assertEquals(s.fetch_all(), {
+            'foo': 'bar',
+            'spam': 'eggs',
+        })
+        # imagine we are resuming, so we need to make new
+        # storage from same flow details
+        s2 = storage.Storage(s._flowdetail, backend=self.backend)
+        # injected data should still be there:
+        self.assertEquals(s2.fetch_all(), {
+            'foo': 'bar',
+            'spam': 'eggs',
+        })
+
     def test_fetch_meapped_args(self):
         s = self._get_storage()
         s.inject({'foo': 'bar', 'spam': 'eggs'})