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
|
"""Define tests for the Dispatch module."""
import asyncio
import functools
import pytest
from pysmartapp.dispatch import Dispatcher
class TestDispatcher:
"""Define tests for the dispatcher class."""
@staticmethod
@pytest.mark.asyncio
async def test_connect(handler):
"""Tests the connect function."""
# Arrange
dispatcher = Dispatcher()
# Act
dispatcher.connect('TEST', handler)
# Assert
assert handler in dispatcher.signals['TEST']
@staticmethod
@pytest.mark.asyncio
async def test_disconnect(handler):
"""Tests the disconnect function."""
# Arrange
dispatcher = Dispatcher()
disconnect = dispatcher.connect('TEST', handler)
# Act
disconnect()
# Assert
assert handler not in dispatcher.signals['TEST']
@staticmethod
@pytest.mark.asyncio
async def test_disconnect_all(handler):
"""Tests the disconnect all function."""
# Arrange
dispatcher = Dispatcher()
dispatcher.connect('TEST', handler)
dispatcher.connect('TEST', handler)
dispatcher.connect('TEST2', handler)
dispatcher.connect('TEST3', handler)
# Act
dispatcher.disconnect_all()
# Assert
assert handler not in dispatcher.signals['TEST']
assert handler not in dispatcher.signals['TEST2']
assert handler not in dispatcher.signals['TEST3']
@staticmethod
@pytest.mark.asyncio
async def test_already_disconnected(handler):
"""Tests that disconnect can be called more than once."""
# Arrange
dispatcher = Dispatcher()
disconnect = dispatcher.connect('TEST', handler)
disconnect()
# Act
disconnect()
# Assert
assert handler not in dispatcher.signals['TEST']
@staticmethod
@pytest.mark.asyncio
async def test_send_async_handler(async_handler):
"""Tests sending to async handlers."""
# Arrange
dispatcher = Dispatcher()
dispatcher.connect('TEST', async_handler)
# Act
await asyncio.gather(*dispatcher.send('TEST'))
# Assert
assert async_handler.fired
@staticmethod
@pytest.mark.asyncio
async def test_send_async_partial_handler(async_handler):
"""Tests sending to async handlers."""
# Arrange
partial = functools.partial(async_handler)
dispatcher = Dispatcher()
dispatcher.connect('TEST', partial)
# Act
await asyncio.gather(*dispatcher.send('TEST'))
# Assert
assert async_handler.fired
@staticmethod
@pytest.mark.asyncio
async def test_send(handler):
"""Tests sending to async handlers."""
# Arrange
dispatcher = Dispatcher()
dispatcher.connect('TEST', handler)
args = object()
# Act
await asyncio.gather(*dispatcher.send('TEST', args))
# Assert
assert handler.fired
assert handler.args[0] == args
@staticmethod
@pytest.mark.asyncio
async def test_custom_connect_and_send(handler):
"""Tests using the custom connect and send implementations."""
# Arrange
test_signal = 'PREFIX_TEST'
stored_target = None
def connect(signal, target):
assert signal == test_signal
nonlocal stored_target
stored_target = target
def disconnect():
nonlocal stored_target
stored_target = None
return disconnect
def send(signal, *args):
assert signal == test_signal
stored_target(*args) # pylint:disable=not-callable
dispatcher = Dispatcher(connect=connect, send=send,
signal_prefix='PREFIX_')
# Act
dispatcher.connect('TEST', handler)
dispatcher.send('TEST')
# Assert
assert handler.fired
|
