parallel_sequence helper

Makefile

@@ -1,5 +1,5 @@
 lint:
-	flake8 --ignore=E131,E731,W503 --max-line-length=100 effect/
+	flake8 --ignore=E131,E301,E731,W503,E704 --max-line-length=100 effect/
 
 build-dist:
 	rm -rf dist

effect/test_testing.py

@@ -11,12 +11,15 @@
                                 raises)
 
 from . import (
+    ComposedDispatcher,
     Constant,
     Effect,
     base_dispatcher,
     parallel,
-    sync_perform)
+    sync_perform,
+    sync_performer)
 from .do import do, do_return
+from .fold import FoldError, sequence
 from .testing import (
     ESConstant,
     ESError,
@@ -25,6 +28,7 @@
     EQFDispatcher,
     SequenceDispatcher,
     fail_effect,
+    parallel_sequence,
     perform_sequence,
     resolve_effect,
     resolve_stubs)
@@ -403,3 +407,59 @@ def code_under_test():
     expected = ("sequence: MyIntent(val='a')\n"
                 "NOT FOUND: OtherIntent(val='b')")
     assert expected in str(exc.value)
+
+
+def test_parallel_sequence():
+    """
+    Ensures that all parallel effects are found in the given intents, in
+    order, and returns the results associated with those intents.
+    """
+    seq = [
+        parallel_sequence([
+            [(1, lambda i: "one!")],
+            [(2, lambda i: "two!")],
+            [(3, lambda i: "three!")],
+        ])
+    ]
+    p = parallel([Effect(1), Effect(2), Effect(3)])
+    assert perform_sequence(seq, p) == ['one!', 'two!', 'three!']
+
+
+def test_parallel_sequence_fallback():
+    """
+    Accepts a ``fallback`` dispatcher that will be used when the sequence
+    doesn't contain an intent.
+    """
+    def dispatch_2(intent):
+        if intent == 2:
+            return sync_performer(lambda d, i: "two!")
+    fallback = ComposedDispatcher([dispatch_2, base_dispatcher])
+    seq = [
+        parallel_sequence([
+            [(1, lambda i: 'one!')],
+            [],  # only implicit effects in this slot
+            [(3, lambda i: 'three!')],
+        ],
+            fallback_dispatcher=fallback),
+    ]
+    p = parallel([Effect(1), Effect(2), Effect(3)])
+    assert perform_sequence(seq, p) == ['one!', 'two!', 'three!']
+
+
+def test_parallel_sequence_must_be_parallel():
+    """
+    If the sequences aren't run in parallel, the parallel_sequence won't
+    match and a FoldError of NoPerformerFoundError will be raised.
+    """
+    seq = [
+        parallel_sequence([
+            [(1, lambda i: "one!")],
+            [(2, lambda i: "two!")],
+            [(3, lambda i: "three!")],
+        ])
+    ]
+    p = sequence([Effect(1), Effect(2), Effect(3)])
+    with pytest.raises(FoldError) as excinfo:
+        perform_sequence(seq, p)
+    print excinfo.value
+    assert excinfo.value.wrapped_exception[0] is AssertionError

effect/testing.py

@@ -107,6 +107,64 @@ def dispatcher(intent):
         return sync_perform(dispatcher, eff)
 
 
+@object.__new__
+class _ANY(object):
+    def __eq__(self, o): return True
+    def __ne__(self, o): return False
+
+
+def parallel_sequence(parallel_seqs, fallback_dispatcher=None):
+    """
+    Convenience for expecting a ParallelEffects in an expected intent sequence,
+    as required by :func:`perform_sequence` or :obj:`SequenceDispatcher`.
+
+    This lets you verify that intents are performed in parallel in the
+    context of :func:`perform_sequence`. It returns a two-tuple as expected by
+    that function, so you can use it like this::
+
+        @do
+        def code_under_test():
+            r = yield Effect(SerialIntent('serial'))
+            r2 = yield parallel([Effect(MyIntent('a')),
+                                 Effect(OtherIntent('b'))])
+            yield do_return((r, r2))
+
+        def test_code():
+            seq = [
+                (SerialIntent('serial'), lambda i: 'result1'),
+                nested_parallel([
+                    [(MyIntent('a'), lambda i: 'a result')],
+                    [(OtherIntent('b'), lambda i: 'b result')]
+                ]),
+            ]
+            eff = code_under_test()
+            assert perform_sequence(seq, eff) == ('result1', 'result2')
+
+
+    The argument is expected to be a list of intent sequences, one for each
+    parallel effect expected. Each sequence will be performed with
+    :func:`perform_sequence` and the respective effect that's being run in
+    parallel. The order of the sequences must match that of the order of
+    parallel effects.
+
+    :param parallel_seqs: list of lists of (intent, performer), like
+        what :func:`perform_sequence` accepts.
+    :param fallback_dispatcher: an optional dispatcher to compose onto the
+        sequence dispatcher.
+    """
+    perf = partial(perform_sequence, fallback_dispatcher=fallback_dispatcher)
+    def performer(intent):
+        if len(intent.effects) != len(parallel_seqs):
+            raise AssertionError(
+                "Need one list in parallel_seqs per parallel effect. "
+                "Got %s effects and %s seqs.\n"
+                "Effects: %s\n"
+                "parallel_seqs: %s" % (len(intent.effects), len(parallel_seqs),
+                                       intent.effects, parallel_seqs))
+        return map(perf, parallel_seqs, intent.effects)
+    return (ParallelEffects(effects=_ANY), performer)
+
+
 @attr.s
 class Stub(object):
     """