3.2.7. effect.testing module

Various functions and dispatchers for testing effects.

Usually the best way to test effects is by using perform_sequence().

effect.testing.perform_sequence(seq, eff, fallback_dispatcher=None)

Perform an Effect by looking up performers for intents in an ordered “plan”.

First, an example:

@do
def code_under_test():
    r = yield Effect(MyIntent('a'))
    r2 = yield Effect(OtherIntent('b'))
    yield do_return((r, r2))

def test_code():
    seq = [
        (MyIntent('a'), lambda i: 'result1'),
        (OtherIntent('b'), lambda i: 'result2')
    ]
    eff = code_under_test()
    assert perform_sequence(seq, eff) == ('result1', 'result2')

Every time an intent is to be performed, it is checked against the next item in the sequence, and the associated function is used to calculate its result. Note that the objects used for intents must provide a meaningful __eq__ implementation, since they will be checked for equality. Using something like attrs or pyrsistent‘s PClass is recommended for your intents, since they will auto-generate __eq__ and many other methods useful for immutable objects.

If an intent can’t be found in the sequence or the fallback dispatcher, an AssertionError is raised with a log of all intents that were performed so far. Each item in the log starts with one of three prefixes:

  • sequence: this intent was found in the sequence
  • fallback: a performer for this intent was provided by the fallback dispatcher
  • NOT FOUND: no performer for this intent was found.
  • NEXT EXPECTED: the next item in the sequence, if there is one. This will appear immediately after a NOT FOUND.
Parameters:
  • sequence (list) – List of (intent, fn) tuples, where fn is a function that should accept an intent and return a result.
  • eff (Effect) – The Effect to perform.
  • fallback_dispatcher – A dispatcher to use for intents that aren’t found in the sequence. if None is provided, base_dispatcher is used.
Returns:

Result of performed sequence

effect.testing.parallel_sequence(parallel_seqs, fallback_dispatcher=None)

Convenience for expecting a ParallelEffects in an expected intent sequence, as required by perform_sequence() or SequenceDispatcher.

This lets you verify that intents are performed in parallel in the context of 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 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.

Parameters:
  • parallel_seqs – list of lists of (intent, performer), like what perform_sequence() accepts.
  • fallback_dispatcher – an optional dispatcher to compose onto the sequence dispatcher.
Returns:

(intent, performer) tuple as expected by perform_sequence() where intent is ParallelEffects object

effect.testing.nested_sequence(seq, get_effect=<operator.attrgetter object>, fallback_dispatcher=TypeDispatcher(mapping={<class 'effect._intents.Constant'>: <function perform_constant at 0x7f622de1dc80>, <class 'effect._intents.Error'>: <function perform_error at 0x7f622de23320>, <class 'effect._intents.Func'>: <function perform_func at 0x7f622de23a28>}))

Return a function of Intent -> a that performs an effect retrieved from the intent (by accessing its effect attribute, by default) with the given intent-sequence.

A demonstration is best:

SequenceDispatcher([
    (BoundFields(effect=mock.ANY, fields={...}),
     nested_sequence([(SomeIntent(), perform_some_intent)]))
])

The point is that sometimes you have an intent that wraps another effect, and you want to ensure that the nested effects follow some sequence in the context of that wrapper intent.

get_effect defaults to attrgetter('effect'), so you can override it if your intent stores its nested effect in a different attribute. Or, more interestingly, if it’s something other than a single effect, e.g. for ParallelEffects see the parallel_sequence() function.

Parameters:
  • seq (list) – sequence of intents like SequenceDispatcher takes
  • get_effect – callable to get the inner effect from the wrapper intent.
  • fallback_dispatcher – an optional dispatcher to compose onto the sequence dispatcher.
Returns:

callable that can be used as performer of a wrapped intent

class effect.testing.SequenceDispatcher(sequence)

Bases: object

A dispatcher which steps through a sequence of (intent, func) tuples and runs func to perform intents in strict sequence.

This is the dispatcher used by perform_sequence(). In general that function should be used directly, instead of this dispatcher.

It’s important to use with sequence.consume(): to ensure that all of the intents are performed. Otherwise, if your code has a bug that causes it to return before all effects are performed, your test may not fail.

None is returned if the next intent in the sequence is not equal to the intent being performed, or if there are no more items left in the sequence (this is standard behavior for dispatchers that don’t handle an intent). This lets this dispatcher be composed easily with others.

Parameters:sequence (list) – Sequence of (intent, fn).
consume(*args, **kwds)

Return a context manager that can be used with the with syntax to ensure that all steps are performed by the end.

consumed()

Return True if all of the steps were performed.

sequence = Attribute(name='sequence', default=NOTHING, validator=None, repr=True, cmp=True, hash=True, init=True, convert=None, metadata=mappingproxy({}))
effect.testing.noop(intent)

Return None. This is just a handy way to make your intent sequences (as used by perform_sequence()) more concise when the effects you’re expecting in a test don’t return a result (and are instead only performed for their side-effects):

seq = [
    (Prompt('Enter your name: '), lambda i: 'Chris')
    (Greet('Chris'), noop),
]
effect.testing.const(value)

Return function that takes an argument but always return given value. Useful when creating sequence used by perform_sequence(). For example,

>>> dt = datetime(1970, 1, 1)
>>> seq = [(Func(datetime.now), const(dt))]
Parameters:value – This will be returned when called by returned function
Returns:callable that takes an arg and always returns value
effect.testing.conste(excp)

Like const() but takes and exception and returns function that raises the exception

Parameters:excp – Exception that will be raised
Type:Exception
Returns:callable that will raise given exception
effect.testing.intent_func(fname)

Return function that returns Effect of tuple of fname and its args. Useful in writing tests that expect intent based on args. For example, if you are testing following function:

@do
def code_under_test(arg1, arg2, eff_returning_func=eff_returning_func):
    r = yield Effect(MyIntent('a'))
    r2 = yield eff_returning_func(arg1, arg2)
    yield do_return((r, r2))

you will need to know the intents which eff_returning_func generates to test this using perform_sequence(). You can avoid that by doing:

def test_code():
    test_eff_func = intent_func("erf")
    seq = [
        (MyIntent('a'), const('result1')),
        (("erf", 'a1', 'a2'), const('result2'))
    ]
    eff = code_under_test('a1', 'a2', eff_returning_func=test_eff_func)
    assert perform_sequence(seq, eff) == ('result1', 'result2')

Here, the seq ensures that eff_returning_func is called with arguments a1 and a2.

Parameters:fname (str) – First member of intent tuple returned
Returns:callable with multiple positional arguments
effect.testing.resolve_effect(effect, result, is_error=False)

Supply a result for an effect, allowing its callbacks to run.

Note that is a pretty low-level testing utility; it’s much better to use a higher-level tool like perform_sequence() in your tests.

The return value of the last callback is returned, unless any callback returns another Effect, in which case an Effect representing that operation plus the remaining callbacks will be returned.

This allows you to test your code in a somewhat “channel”-oriented way:

eff = do_thing() next_eff = resolve_effect(eff, first_result) next_eff = resolve_effect(next_eff, second_result) result = resolve_effect(next_eff, third_result)

Equivalently, if you don’t care about intermediate results:

result = resolve_effect(
    resolve_effect(
        resolve_effect(
            do_thing(),
            first_result),
        second_result),
    third_result)

NOTE: parallel effects have no special support. They can be resolved with a sequence, and if they’re returned from another effect’s callback they will be returned just like any other effect.

Parameters:
  • is_error (bool) – Indicate whether the result should be treated as an exception or a regular result.
  • result – If is_error is False, this can be any object and will be treated as the result of the effect. If is_error is True, this must be a three-tuple in the style of sys.exc_info.
effect.testing.fail_effect(effect, exception)

Resolve an effect with an exception, so its error handler will be run.

class effect.testing.EQDispatcher(mapping)

Bases: object

An equality-based (constant) dispatcher.

This dispatcher looks up intents by equality and performs them by returning an associated constant value.

This is sometimes useful, but perform_sequence() should be preferred, since it constrains the order of effects, which is usually important.

Users provide a mapping of intents to results, where the intents are matched against the intents being performed with a simple equality check (not a type check!).

The mapping must be provided as a sequence of two-tuples. We don’t use a dict because we don’t want to require that the intents be hashable (in practice a lot of them aren’t, and it’s a pain to require it). If you want to construct your mapping as a dict, you can, just pass in the result of d.items().

e.g.:

>>> sync_perform(EQDispatcher([(MyIntent(1, 2), 'the-result')]),
...              Effect(MyIntent(1, 2)))
'the-result'

assuming MyIntent supports __eq__ by value.

Parameters:mapping (list) – A sequence of tuples of (intent, result).
mapping = Attribute(name='mapping', default=NOTHING, validator=None, repr=True, cmp=True, hash=True, init=True, convert=None, metadata=mappingproxy({}))
class effect.testing.EQFDispatcher(mapping)

Bases: object

An Equality-based function dispatcher.

This dispatcher looks up intents by equality and performs them by invoking an associated function.

This is sometimes useful, but perform_sequence() should be preferred, since it constrains the order of effects, which is usually important.

Users provide a mapping of intents to functions, where the intents are matched against the intents being performed with a simple equality check (not a type check!). The functions in the mapping will be passed only the intent and are expected to return the result or raise an exception.

The mapping must be provided as a sequence of two-tuples. We don’t use a dict because we don’t want to require that the intents be hashable (in practice a lot of them aren’t, and it’s a pain to require it). If you want to construct your mapping as a dict, you can, just pass in the result of d.items().

e.g.:

>>> sync_perform(
...     EQFDispatcher([(
...         MyIntent(1, 2), lambda i: 'the-result')]),
...     Effect(MyIntent(1, 2)))
'the-result'

assuming MyIntent supports __eq__ by value.

Parameters:mapping (list) – A sequence of two-tuples of (intent, function).
mapping = Attribute(name='mapping', default=NOTHING, validator=None, repr=True, cmp=True, hash=True, init=True, convert=None, metadata=mappingproxy({}))
class effect.testing.Stub(intent)

Bases: object

DEPRECATED in favor of using perform_sequence().

An intent which wraps another intent, to flag that the intent should be automatically resolved by resolve_stub().

Stub is intentionally not performable by any default mechanism.

intent = Attribute(name='intent', default=NOTHING, validator=None, repr=True, cmp=True, hash=True, init=True, convert=None, metadata=mappingproxy({}))
effect.testing.ESConstant(x)

DEPRECATED. Return Effect(Stub(Constant(x)))

effect.testing.ESError(x)

DEPRECATED. Return Effect(Stub(Error(x)))

effect.testing.ESFunc(x)

DEPRECATED. Return Effect(Stub(Func(x)))

effect.testing.resolve_stubs(dispatcher, effect)

DEPRECATED in favor of using perform_sequence().

Successively performs effects with resolve_stub until a non-Effect value, or an Effect with a non-stub intent is returned, and return that value.

Parallel effects are supported by recursively invoking resolve_stubs on the child effects, if all of their children are stubs.

effect.testing.resolve_stub(dispatcher, effect)

DEPRECATED in favor of perform_sequence().

Automatically perform an effect, if its intent is a Stub.

Note that resolve_stubs is preferred to this function, since it handles chains of stub effects.