Merge pull request #70 from python-effect/parallel-seq

parallel_sequence helper

Author: radix

Makefile

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

README.rst

@@ -45,7 +45,7 @@ A very quick example of using Effects:
 .. code:: python
 
     from __future__ import print_function
-    from effect import perform, sync_performer, Effect, TypeDispatcher
+    from effect import sync_perform, sync_performer, Effect, TypeDispatcher
 
     class ReadLine(object):
         def __init__(self, prompt):
@@ -65,14 +65,14 @@ A very quick example of using Effects:
             error=lambda e: print("sorry, there was an error. {}".format(e)))
 
         dispatcher = TypeDispatcher({ReadLine: perform_read_line})
-        perform(dispatcher, effect)
+        sync_perform(dispatcher, effect)
 
     if __name__ == '__main__':
         main()
 
 
 ``Effect`` takes what we call an ``intent``, which is any object. The
-``dispatcher`` argument to ``perform`` must have a ``performer`` function
+``dispatcher`` argument to ``sync_perform`` must have a ``performer`` function
 for your intent.
 
 This has a number of advantages. First, your unit tests for ``get_user_name``

docs/source/intro.rst

@@ -69,17 +69,19 @@ A (sometimes) nicer syntax is provided for adding callbacks, with the
             yield Effect(Print("Hello,", name))
 
 Finally, to actually perform these effects, they can be passed to
-:func:`effect.perform`, along with a dispatcher which looks up the performer
-based on the intent.
+:func:`effect.sync_perform`, along with a dispatcher which looks up the
+performer based on the intent.
 
 .. code:: python
 
+    from effect import sync_perform
+
     def main():
         eff = greet()
         dispatcher = ComposedDispatcher([
             TypeDispatcher({ReadLine: perform_read_line}),
             base_dispatcher])
-        perform(dispatcher, eff)
+        sync_perform(dispatcher, eff)
 
 This has a number of advantages. First, your unit tests for ``get_user_name``
 become simpler. You don't need to mock out or parameterize the ``raw_input``
@@ -115,7 +117,10 @@ A quick tour, with definitions
 - Box: An object that has ``succeed`` and ``fail`` methods for providing the
   result of an effect (potentially asynchronously). Usually you don't need
   to care about this, if you define your performers with
-  :func:`effect.sync_performer` or :func:`effect.twisted.deferred_performer`.
+  :func:`effect.sync_performer` or ``txeffect.deferred_performer`` from the
+  `txeffect`_ package.
+
+.. _`txeffect`: https://pypi.python.org/pypi/txeffect
 
 There's a few main things you need to do to use Effect.
 
@@ -126,11 +131,12 @@ There's a few main things you need to do to use Effect.
   ``Effect(HTTPRequest(...))`` and attach callbacks to them with
   :func:`Effect.on`.
 - As close as possible to the top-level of your application, perform your
-  effect(s) with :func:`effect.perform`.
-- You will need to pass a dispatcher to :func:`effect.perform`. You should create one
-  by creating a :class:`effect.TypeDispatcher` with your own performers (e.g. for
-  ``HTTPRequest``), and composing it with :obj:`effect.base_dispatcher` (which
-  has performers for built-in effects) using :class:`effect.ComposedDispatcher`.
+  effect(s) with :func:`effect.sync_perform`.
+- You will need to pass a dispatcher to :func:`effect.sync_perform`. You should
+  create one by creating a :class:`effect.TypeDispatcher` with your own
+  performers (e.g. for ``HTTPRequest``), and composing it with
+  :obj:`effect.base_dispatcher` (which has performers for built-in effects)
+  using :class:`effect.ComposedDispatcher`.
 
 
 Callback chains

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,58 @@ 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)
+    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 list(map(perf, parallel_seqs, intent.effects))
+    return (ParallelEffects(effects=_ANY), performer)
+
+
 @attr.s
 class Stub(object):
     """