-
Notifications
You must be signed in to change notification settings - Fork 15
Expand file tree
/
Copy pathtest_parallel_performers.py
More file actions
83 lines (71 loc) · 2.6 KB
/
Copy pathtest_parallel_performers.py
File metadata and controls
83 lines (71 loc) · 2.6 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
from functools import partial
from characteristic import attributes
import six
from testtools.matchers import MatchesStructure, Equals
from . import Effect
from ._intents import Constant, Func, FirstError, parallel
from ._sync import sync_perform
from ._test_utils import MatchesReraisedExcInfo, get_exc_info
@attributes(['message'])
class EquitableException(Exception):
pass
class ParallelPerformerTestsMixin(object):
"""Common tests for any performer of :obj:`effect.ParallelEffects`."""
def test_empty(self):
"""
When given an empty list of effects, ``perform_parallel_async`` returns
an empty list synchronusly.
"""
result = sync_perform(
self.dispatcher,
parallel([]))
self.assertEqual(result, [])
def test_parallel(self):
"""
'parallel' results in a list of results of the given effects, in the
same order that they were passed to parallel.
"""
result = sync_perform(
self.dispatcher,
parallel([Effect(Constant('a')),
Effect(Constant('b'))]))
self.assertEqual(result, ['a', 'b'])
def test_error(self):
"""
When given an effect that results in a Error,
``perform_parallel_async`` result in ``FirstError``.
"""
expected_exc_info = get_exc_info(EquitableException(message='foo'))
reraise = partial(six.reraise, *expected_exc_info)
try:
sync_perform(
self.dispatcher,
parallel([Effect(Func(reraise))]))
except FirstError as fe:
self.assertThat(
fe,
MatchesStructure(
index=Equals(0),
exc_info=MatchesReraisedExcInfo(expected_exc_info)))
else:
self.fail("sync_perform should have raised FirstError.")
def test_error_index(self):
"""
The ``index`` of a :obj:`FirstError` is the index of the effect that
failed in the list.
"""
expected_exc_info = get_exc_info(EquitableException(message='foo'))
reraise = partial(six.reraise, *expected_exc_info)
try:
sync_perform(
self.dispatcher,
parallel([
Effect(Constant(1)),
Effect(Func(reraise)),
Effect(Constant(2))]))
except FirstError as fe:
self.assertThat(
fe,
MatchesStructure(
index=Equals(1),
exc_info=MatchesReraisedExcInfo(expected_exc_info)))