FazBrowse GitHub Viewer
|
Trending
|
URL:
|
Home
Tools:
[Download Repo ZIP]
[View Raw Code]
[Original HTTPS Page]
effect/effect/test_sync.py at master · python-effect/effect · GitHub
Uh oh!
There was an error while loading.
Please reload this page
.
python-effect
/
effect
Public
Notifications
You must be signed in to change notification settings
Fork
15
Star
386
Code
Issues
10
Pull requests
1
Discussions
Actions
Projects
Wiki
Security and quality
0
Insights
Additional navigation options
Code
Issues
Pull requests
Discussions
Actions
Projects
Wiki
Security and quality
Insights
Expand file tree
Breadcrumbs
effect
/
effect
/
test_sync.py
Copy path
More file actions
More file actions
Latest commit
History
History
History
138 lines (105 loc) · 3.92 KB
Breadcrumbs
effect
/
effect
/
test_sync.py
Copy path
File metadata and controls
138 lines (105 loc) · 3.92 KB
Raw
Copy raw file
Download raw file
Open symbols panel
Edit and raw actions
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
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
from
functools
import
partial
from
testtools
import
TestCase
from
testtools
.
matchers
import
raises
from
.
_base
import
Effect
from
.
_sync
import
NotSynchronousError
,
sync_perform
,
sync_performer
from
.
_test_utils
import
MatchesException
def
func_dispatcher
(
intent
):
def
performer
(
dispatcher
,
intent
,
box
):
intent
(
box
)
return
performer
class
SyncPerformTests
(
TestCase
):
"""Tests for :func:`sync_perform`."""
def
test_sync_perform_effect_function_dispatch
(
self
):
"""
sync_perform returns the result of the effect.
"""
intent
=
lambda
box
:
box
.
succeed
(
"foo"
)
self
.
assertEqual
(
sync_perform
(
func_dispatcher
,
Effect
(
intent
)),
"foo"
)
def
test_sync_perform_async_effect
(
self
):
"""If an effect is asynchronous, sync_effect raises an error."""
intent
=
lambda
box
:
None
self
.
assertRaises
(
NotSynchronousError
,
lambda
:
sync_perform
(
func_dispatcher
,
Effect
(
intent
))
)
def
test_error_bubbles_up
(
self
):
"""
When effect performance fails, the exception is raised up through
sync_perform.
"""
def
fail
(
box
):
box
.
fail
(
ValueError
(
"oh dear"
))
self
.
assertThat
(
lambda
:
sync_perform
(
func_dispatcher
,
Effect
(
fail
)),
raises
(
ValueError
(
"oh dear"
)),
)
class
SyncPerformerTests
(
TestCase
):
"""
Tests for :func:`sync_performer`.
"""
def
test_success
(
self
):
"""Return value of the performer becomes the result of the Effect."""
@
sync_performer
def
succeed
(
dispatcher
,
intent
):
return
intent
dispatcher
=
lambda
_
:
succeed
result
=
sync_perform
(
dispatcher
,
Effect
(
"foo"
))
self
.
assertEqual
(
result
,
"foo"
)
def
test_failure
(
self
):
"""
Errors are caught and cause the effect to fail with the exception info.
"""
@
sync_performer
def
fail
(
dispatcher
,
intent
):
raise
intent
dispatcher
=
lambda
_
:
fail
self
.
assertThat
(
sync_perform
(
dispatcher
,
Effect
(
ValueError
(
"oh dear"
)).
on
(
error
=
lambda
e
:
e
)
),
MatchesException
(
ValueError
(
"oh dear"
)),
)
def
test_instance_method_performer
(
self
):
"""The decorator works on instance methods."""
eff
=
Effect
(
"meaningless"
)
class
PerformerContainer
(
object
):
@
sync_performer
def
performer
(
self
,
dispatcher
,
intent
):
return
(
self
,
dispatcher
,
intent
)
container
=
PerformerContainer
()
dispatcher
=
lambda
i
:
container
.
performer
result
=
sync_perform
(
dispatcher
,
eff
)
self
.
assertEqual
(
result
, (
container
,
dispatcher
,
"meaningless"
))
def
test_promote_metadata
(
self
):
"""
The decorator copies metadata from the wrapped function onto the
wrapper.
"""
def
original
(
dispatcher
,
intent
):
"""Original!"""
pass
original
.
attr
=
1
wrapped
=
sync_performer
(
original
)
self
.
assertEqual
(
wrapped
.
__name__
,
"original"
)
self
.
assertEqual
(
wrapped
.
attr
,
1
)
self
.
assertEqual
(
wrapped
.
__doc__
,
"Original!"
)
def
test_ignore_lack_of_metadata
(
self
):
"""
When the original callable is not a function, a new function is still
returned.
"""
def
original
(
something
,
dispatcher
,
intent
):
"""Original!"""
pass
new_func
=
partial
(
original
,
"something"
)
original
.
attr
=
1
wrapped
=
sync_performer
(
new_func
)
self
.
assertEqual
(
wrapped
.
__name__
,
"sync_wrapper"
)
def
test_kwargs
(
self
):
"""Additional kwargs are passed through."""
@
sync_performer
def
p
(
dispatcher
,
intent
,
extra
):
return
extra
dispatcher
=
lambda
_
:
partial
(
p
,
extra
=
"extra val"
)
result
=
sync_perform
(
dispatcher
,
Effect
(
"foo"
))
self
.
assertEqual
(
result
,
"extra val"
)
Back
|
FazBrowse Home
|
New Git URL