Increased coverage of parallel helper tests and type checking more strict in arrows.

4 files changed, 27 insertions, 16 deletions

diff --git a/src/pypeline/core/arrows/function_arrow.py b/src/pypeline/core/arrows/function_arrow.py
index 15b880e..cfec34c 100644
--- a/src/pypeline/core/arrows/function_arrow.py
+++ b/src/pypeline/core/arrows/function_arrow.py
@@ -58,8 +58,8 @@ class FunctionArrow(Arrow):
     # b --->+--- f ---+--- c ---->+--- g ---+---> d
     #       +---------+           +---------+
     def __rshift__(self, other):
-        if not isinstance(other, Arrow):
-            raise ValueError("Must be an arrow")
+        if not isinstance(other, FunctionArrow):
+            raise ValueError("Must be a FunctionArrow")
 
         return FunctionArrow(lambda b: other._func(self._func(b)))
 
@@ -72,8 +72,8 @@ class FunctionArrow(Arrow):
     # d --->+---------+---> d --->+--- g ---+---> e
     #       +---------+           +---------+
     def __pow__(self, other):
-        if not isinstance(other, Arrow):
-            raise ValueError("Must be an arrow")
+        if not isinstance(other, FunctionArrow):
+            raise ValueError("Must be a FunctionArrow")
 
         return self.first() >> other.second()
 
@@ -86,8 +86,8 @@ class FunctionArrow(Arrow):
     #       |    \----+---> b --->+---------+---> b --->+--- g ---+---> d
     #       +---------+           +---------+           +---------+
     def __and__(self, other):
-        if not isinstance(other, Arrow):
-            raise ValueError("Must be an arrow")
+        if not isinstance(other, FunctionArrow):
+            raise ValueError("Must be a FunctionArrow")
 
         return split() >> self.first() >> other.second()
 
diff --git a/src/pypeline/core/arrows/function_arrow_choice.py b/src/pypeline/core/arrows/function_arrow_choice.py
index 2b04201..c5ef6a3 100644
--- a/src/pypeline/core/arrows/function_arrow_choice.py
+++ b/src/pypeline/core/arrows/function_arrow_choice.py
@@ -73,8 +73,8 @@ class FunctionArrowChoice(ArrowChoice, FunctionArrow):
     #                 |   \-------/   |                   |   \-- g --/   |
     #                 +---------------+                   +---------------+
     def __add__(self, other):
-        if not isinstance(other, Arrow):
-            raise ValueError("Must be an arrow")
+        if not isinstance(other, FunctionArrowChoice):
+            raise ValueError("Must be a FunctionArrow")
 
         return self.left() >> other.right()
 
@@ -87,8 +87,8 @@ class FunctionArrowChoice(ArrowChoice, FunctionArrow):
     #                 |   \-------/   |                   |   \-- g --/   |                   |   \-- Right? -/   |
     #                 +---------------+                   +---------------+                   +-------------------+
     def __or__(self, other):
-        if not isinstance(other, Arrow):
-            raise ValueError("Must be an arrow")
+        if not isinstance(other, FunctionArrowChoice):
+            raise ValueError("Must be a FunctionArrow")
 
         def merge(either):
             if not isinstance(either, Either):
@@ -104,8 +104,8 @@ class FunctionArrowChoice(ArrowChoice, FunctionArrow):
 # test :: a b c -> a b (Either b b)
 #
 def test(arrow):
-    if not isinstance(arrow, Arrow):
-        raise ValueError("Must be an arrow")
+    if not isinstance(arrow, FunctionArrow):
+        raise ValueError("Must be a FunctionArrow")
 
     # (a &&& arr id) >>> arr (\(b, x) = if b then Left x else Right x)
     return (arrow & FunctionArrow(lambda x: x)) >> FunctionArrow(lambda t: Left(t[1]) if t[0] else Right(t[1])) 
diff --git a/src/pypeline/core/arrows/kleisli_arrow.py b/src/pypeline/core/arrows/kleisli_arrow.py
index 344641a..179d209 100644
--- a/src/pypeline/core/arrows/kleisli_arrow.py
+++ b/src/pypeline/core/arrows/kleisli_arrow.py
@@ -44,7 +44,7 @@ class KleisliArrow(Arrow):
         self._patcher = patcher
         self._func = f
 
-    # arr f = K(\b ->return(f b))
+    # arr f = K(\b -> return(f b))
     def arr(self, f):
         return KleisliArrow(self._patcher, lambda b: self._patcher(f(b)))
 
diff --git a/src/pypeline/helpers/tests/parallel_helper_pipeline_tests.py b/src/pypeline/helpers/tests/parallel_helper_pipeline_tests.py
index ed4f743..b9ceaed 100644
--- a/src/pypeline/helpers/tests/parallel_helper_pipeline_tests.py
+++ b/src/pypeline/helpers/tests/parallel_helper_pipeline_tests.py
@@ -72,21 +72,32 @@ class ParallelPypelineHelperUnitTest(unittest.TestCase):
      def test_parallel_pypeline_with_split_and_unsplit_wires(self):
           rev_msg_one = "reverse(top)"
           rev_msg_two = "reverse(bottom)"
+          top_msg = "top"
+          bottom_msg = "bottom"
 
           reverse_func = lambda a, s: a[::-1]
+          top_func = lambda a, s: " ".join([a, top_msg])
+          bottom_func = lambda a, s: " ".join([a, bottom_msg])
 
           comp_rev_top = cons_function_component(reverse_func,
                                                  state_mutator = lambda s: s.append(rev_msg_one) or s)
           comp_rev_bottom = cons_function_component(reverse_func,
                                                     state_mutator = lambda s: s.append(rev_msg_two) or s)
+          comp_para_top = cons_function_component(top_func,
+                                                  state_mutator = lambda s: s.append(top_msg) or s)
+          comp_para_bottom = cons_function_component(bottom_func,
+                                                     state_mutator = lambda s: s.append(bottom_msg) or s)
 
           unsplit_func = lambda t, b: {'top': t, 'bottom': b}
 
-          pipeline = (comp_rev_top & comp_rev_bottom) >> cons_unsplit_wire(unsplit_func)
+          pipeline = (comp_rev_top & comp_rev_bottom) >> \
+                     (comp_para_top ** comp_para_bottom) >> \
+                     cons_unsplit_wire(unsplit_func)
 
           value = "hello world"
-          target = (unsplit_func(reverse_func(value, None), reverse_func(value, None)),
-                    [rev_msg_one, rev_msg_two])
+          target = (unsplit_func(top_func(reverse_func(value, None), None),
+                                 bottom_func(reverse_func(value, None), None)),
+                    [rev_msg_one, rev_msg_two, top_msg, bottom_msg])
 
           result = ParallelPypelineHelperUnitTest.test(2, pipeline, "hello world", list())