Move 'fill_iter' to 'iter_utils.fill'

This is better placed in the iterator utility module
as it acts on iterables and provides its own iterator
that fills up to (and potentially beyond) a provided
iterator.

Also adds some nice tests to make sure it keeps on
working as expected; as well as tests for other parts
of iter_utils to ensure they keep on working as
expected as well.

Change-Id: Ica90816cbdedfd87f3861a111d7a852655c1fb74

3 files changed, 111 insertions, 25 deletions

diff --git a/taskflow/tests/unit/test_utils_iter_utils.py b/taskflow/tests/unit/test_utils_iter_utils.py
index 82d470f..8887981 100644
--- a/taskflow/tests/unit/test_utils_iter_utils.py
+++ b/taskflow/tests/unit/test_utils_iter_utils.py
@@ -30,6 +30,51 @@ def forever_it():
 
 
 class IterUtilsTest(test.TestCase):
+    def test_fill_empty(self):
+        self.assertEqual([], list(iter_utils.fill([1, 2, 3], 0)))
+
+    def test_bad_unique_seen(self):
+        iters = [
+            ['a', 'b'],
+            2,
+            None,
+        ]
+        self.assertRaises(ValueError,
+                          iter_utils.unique_seen, *iters)
+
+    def test_unique_seen(self):
+        iters = [
+            ['a', 'b'],
+            ['a', 'c', 'd'],
+            ['a', 'e', 'f'],
+            ['f', 'm', 'n'],
+        ]
+        self.assertEqual(['a', 'b', 'c', 'd', 'e', 'f', 'm', 'n'],
+                         list(iter_utils.unique_seen(*iters)))
+
+    def test_bad_fill(self):
+        self.assertRaises(ValueError, iter_utils.fill, 2, 2)
+
+    def test_fill_many_empty(self):
+        result = list(iter_utils.fill(compat_range(0, 50), 500))
+        self.assertEqual(450, sum(1 for x in result if x is None))
+        self.assertEqual(50, sum(1 for x in result if x is not None))
+
+    def test_fill_custom_filler(self):
+        self.assertEqual("abcd",
+                         "".join(iter_utils.fill("abc", 4, filler='d')))
+
+    def test_fill_less_needed(self):
+        self.assertEqual("ab", "".join(iter_utils.fill("abc", 2)))
+
+    def test_fill(self):
+        self.assertEqual([None, None], list(iter_utils.fill([], 2)))
+        self.assertEqual((None, None), tuple(iter_utils.fill([], 2)))
+
+    def test_bad_find_first_match(self):
+        self.assertRaises(ValueError,
+                          iter_utils.find_first_match, 2, lambda v: False)
+
     def test_find_first_match(self):
         it = forever_it()
         self.assertEqual(100, iter_utils.find_first_match(it,
@@ -40,6 +85,9 @@ class IterUtilsTest(test.TestCase):
         self.assertIsNone(iter_utils.find_first_match(it,
                                                       lambda v: v == ''))
 
+    def test_bad_count(self):
+        self.assertRaises(ValueError, iter_utils.count, 2)
+
     def test_count(self):
         self.assertEqual(0, iter_utils.count([]))
         self.assertEqual(1, iter_utils.count(['a']))
@@ -48,6 +96,9 @@ class IterUtilsTest(test.TestCase):
         self.assertEqual(0, iter_utils.count(compat_range(0)))
         self.assertEqual(0, iter_utils.count(compat_range(-1)))
 
+    def test_bad_while_is_not(self):
+        self.assertRaises(ValueError, iter_utils.while_is_not, 2, 'a')
+
     def test_while_is_not(self):
         it = iter(string.ascii_lowercase)
         self.assertEqual(['a'],
diff --git a/taskflow/types/failure.py b/taskflow/types/failure.py
index 34c3047..2663fb8 100644
--- a/taskflow/types/failure.py
+++ b/taskflow/types/failure.py
@@ -24,6 +24,7 @@ from oslo_utils import reflection
 import six
 
 from taskflow import exceptions as exc
+from taskflow.utils import iter_utils
 from taskflow.utils import mixins
 from taskflow.utils import schema_utils as su
 
@@ -40,23 +41,6 @@ def _copy_exc_info(exc_info):
     return (exc_type, copy.copy(exc_value), tb)
 
 
-def _fill_iter(it, desired_len, filler=None):
-    """Iterates over a provided iterator up to the desired length.
-
-    If the source iterator does not have enough values then the filler
-    value is yielded until the desired length is reached.
-    """
-    count = 0
-    for value in it:
-        if count >= desired_len:
-            return
-        yield value
-        count += 1
-    while count < desired_len:
-        yield filler
-        count += 1
-
-
 def _are_equal_exc_info_tuples(ei1, ei2):
     if ei1 == ei2:
         return True
@@ -444,7 +428,7 @@ class Failure(mixins.StrMixin):
             # what the twisted people have done, see for example
             # twisted-13.0.0/twisted/python/failure.py#L89 for how they
             # created a fake traceback object...
-            self._exc_info = tuple(_fill_iter(dct['exc_info'], 3))
+            self._exc_info = tuple(iter_utils.fill(dct['exc_info'], 3))
         else:
             self._exc_info = None
         causes = dct.get('causes')
diff --git a/taskflow/utils/iter_utils.py b/taskflow/utils/iter_utils.py
index 1a36684..5d0aff1 100644
--- a/taskflow/utils/iter_utils.py
+++ b/taskflow/utils/iter_utils.py
@@ -16,10 +16,45 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+import collections
 import itertools
+
+import six
 from six.moves import range as compat_range
 
 
+def _ensure_iterable(func):
+
+    @six.wraps(func)
+    def wrapper(it, *args, **kwargs):
+        if not isinstance(it, collections.Iterable):
+            raise ValueError("Iterable expected, but '%s' is not"
+                             " iterable" % it)
+        return func(it, *args, **kwargs)
+
+    return wrapper
+
+
+@_ensure_iterable
+def fill(it, desired_len, filler=None):
+    """Iterates over a provided iterator up to the desired length.
+
+    If the source iterator does not have enough values then the filler
+    value is yielded until the desired length is reached.
+    """
+    if desired_len > 0:
+        count = 0
+        for value in it:
+            yield value
+            count += 1
+            if count >= desired_len:
+                return
+        while count < desired_len:
+            yield filler
+            count += 1
+
+
+@_ensure_iterable
 def count(it):
     """Returns how many values in the iterator (depletes the iterator)."""
     return sum(1 for _value in it)
@@ -27,15 +62,30 @@ def count(it):
 
 def unique_seen(it, *its):
     """Yields unique values from iterator(s) (and retains order)."""
-    seen = set()
-    for value in itertools.chain(it, *its):
-        if value in seen:
-            continue
-        else:
-            yield value
-            seen.add(value)
+
+    def _gen_it(all_its):
+        # NOTE(harlowja): Generation is delayed so that validation
+        # can happen before generation/iteration... (instead of
+        # during generation/iteration)
+        seen = set()
+        while all_its:
+            it = all_its.popleft()
+            for value in it:
+                if value not in seen:
+                    yield value
+                    seen.add(value)
+
+    all_its = collections.deque([it])
+    if its:
+        all_its.extend(its)
+    for it in all_its:
+        if not isinstance(it, collections.Iterable):
+            raise ValueError("Iterable expected, but '%s' is"
+                             " not iterable" % it)
+    return _gen_it(all_its)
 
 
+@_ensure_iterable
 def find_first_match(it, matcher, not_found_value=None):
     """Searches iterator for first value that matcher callback returns true."""
     for value in it:
@@ -44,6 +94,7 @@ def find_first_match(it, matcher, not_found_value=None):
     return not_found_value
 
 
+@_ensure_iterable
 def while_is_not(it, stop_value):
     """Yields given values from iterator until stop value is passed.