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
|
import time
import pydle
from pytest import raises, mark
from .fixtures import with_client
from .mocks import Mock, MockEventLoop
pydle.client.PING_TIMEOUT = 10
## Initialization.
@with_client(invalid_kwarg=False)
def test_client_superfluous_arguments(server, client):
assert client.logger.warning.called
## Connection.
@with_client()
def test_client_reconnect(server, client):
client.disconnect(expected=True)
assert not client.connected
client.connect(reconnect=True)
assert client.connected
@mark.slow
@with_client()
def test_client_unexpected_disconnect_reconnect(server, client):
client._reconnect_delay = Mock(return_value=0)
client.disconnect(expected=False)
assert client._reconnect_delay.called
time.sleep(0.1)
assert client.connected
@with_client()
def test_client_unexpected_reconnect_give_up(server, client):
client.RECONNECT_ON_ERROR = False
client.disconnect(expected=False)
assert not client.connected
@mark.slow
@with_client()
def test_client_unexpected_disconnect_reconnect_delay(server, client):
client._reconnect_delay = Mock(return_value=1)
client.disconnect(expected=False)
assert not client.connected
time.sleep(1.1)
assert client.connected
@with_client()
def test_client_reconnect_delay_calculation(server, client):
client.RECONNECT_DELAYED = False
assert client._reconnect_delay() == 0
client.RECONNECT_DELAYED = True
for expected_delay in client.RECONNECT_DELAYS:
delay = client._reconnect_delay()
assert delay == expected_delay
client._reconnect_attempts += 1
assert client._reconnect_delay() == client.RECONNECT_DELAYS[-1]
@with_client()
def test_client_disconnect_on_connect(server, client):
client.disconnect = Mock()
client.connect('mock://local', 1337)
assert client.connected
assert client.disconnect.called
@with_client(connected=False)
def test_client_connect_invalid_params(server, client):
with raises(ValueError):
client.connect()
with raises(ValueError):
client.connect(port=1337)
@mark.slow
@with_client()
def test_client_timeout(server, client):
client.on_data_error = Mock()
time.sleep(pydle.client.PING_TIMEOUT + 1)
assert client.on_data_error.called
assert isinstance(client.on_data_error.call_args[0][0], TimeoutError)
@with_client(connected=False)
def test_client_server_tag(server, client):
ev = MockEventLoop()
assert client.server_tag is None
client.connect('Mock.local', 1337, eventloop=ev)
assert client.server_tag == 'mock'
client.disconnect()
client.connect('irc.mock.local', 1337, eventloop=ev)
assert client.server_tag == 'mock'
client.disconnect()
client.connect('mock', 1337, eventloop=ev)
assert client.server_tag == 'mock'
client.disconnect()
client.connect('127.0.0.1', 1337, eventloop=ev)
assert client.server_tag == '127.0.0.1'
client.network = 'MockNet'
assert client.server_tag == 'mocknet'
client.disconnect()
## Messages.
@with_client()
def test_client_message(server, client):
client.on_raw_install = Mock()
server.send('INSTALL', 'gentoo')
assert client.on_raw_install.called
message = client.on_raw_install.call_args[0][0]
assert isinstance(message, pydle.protocol.Message)
assert message.command == 'INSTALL'
assert message.params == ('gentoo',)
@with_client()
def test_client_unknown(server, client):
client.on_unknown = Mock()
server.send('INSTALL', 'gentoo')
assert client.on_unknown.called
|
