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
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
|
import time
import datetime
import pydle
from pytest import mark
from .fixtures import with_client
from .mocks import Mock, MockEventLoop, MockConnection
class Passed:
def __init__(self):
self._passed = False
def __bool__(self):
return self._passed
def set(self):
self._passed = True
def reset(self):
self._passed = False
## Client.
@mark.meta
@with_client(connected=False)
def test_mock_client_connect(server, client):
assert not client.connected
client.on_connect = Mock()
client.connect('mock://local', 1337, eventloop=MockEventLoop())
assert client.connected
assert client.on_connect.called
client.disconnect()
assert not client.connected
@mark.meta
@with_client()
def test_mock_client_send(server, client):
client.raw('benis')
assert server.receiveddata('benis')
client.rawmsg('INSTALL', 'Gentoo')
assert server.received('INSTALL', 'Gentoo')
@mark.meta
@with_client(pydle.features.RFC1459Support)
def test_mock_client_receive(server, client):
client.on_raw = Mock()
server.send('PING', 'test')
assert client.on_raw.called
message = client.on_raw.call_args[0][0]
assert isinstance(message, pydle.protocol.Message)
assert message.source is None
assert message.command == 'PING'
assert message.params == ('test',)
## Connection.
@mark.meta
def test_mock_connection_connect():
serv = Mock()
conn = MockConnection('mock.local', port=1337, mock_server=serv)
conn.connect()
assert conn.connected
assert serv.connection is conn
@mark.meta
def test_mock_connection_disconnect():
serv = Mock()
conn = MockConnection('mock.local', port=1337, mock_server=serv)
conn.connect()
conn.disconnect()
assert not conn.connected
## Event loop.
@mark.meta
def test_mock_eventloop_schedule():
ev = MockEventLoop()
passed = Passed()
ev.schedule(lambda: passed.set())
assert passed
ev.stop()
@mark.meta
@mark.slow
def test_mock_eventloop_schedule_in():
ev = MockEventLoop()
passed = Passed()
ev.schedule_in(1, lambda: passed.set())
time.sleep(1.1)
assert passed
ev.stop()
@mark.meta
@mark.slow
def test_mock_eventloop_schedule_in_timedelta():
ev = MockEventLoop()
passed = Passed()
ev.schedule_in(datetime.timedelta(seconds=1), lambda: passed.set())
time.sleep(1.1)
assert passed
@mark.meta
@mark.slow
def test_mock_eventloop_schedule_periodically():
ev = MockEventLoop()
passed = Passed()
ev.schedule_periodically(1, lambda: passed.set())
time.sleep(1.1)
assert passed
passed.reset()
time.sleep(1)
assert passed
ev.stop()
@mark.meta
@mark.slow
def test_mock_eventloop_unschedule_in():
ev = MockEventLoop()
passed = Passed()
handle = ev.schedule_in(1, lambda: passed.set())
ev.unschedule(handle)
time.sleep(1.1)
assert not passed
@mark.meta
@mark.slow
def test_mock_eventloop_unschedule_periodically():
ev = MockEventLoop()
passed = Passed()
handle = ev.schedule_periodically(1, lambda: passed.set())
ev.unschedule(handle)
time.sleep(1.1)
assert not passed
@mark.meta
@mark.slow
def test_mock_eventloop_unschedule_periodically_after():
ev = MockEventLoop()
passed = Passed()
handle = ev.schedule_periodically(1, lambda: passed.set())
time.sleep(1.1)
assert passed
passed.reset()
ev.unschedule(handle)
time.sleep(1.0)
assert not passed
|
