forked from slackapi/python-slack-sdk
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_server.py
More file actions
219 lines (170 loc) · 6.72 KB
/
Copy pathtest_server.py
File metadata and controls
219 lines (170 loc) · 6.72 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
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
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
import json
import time
import pytest
import requests
import responses
from mock import patch
from slackclient.user import User
from slackclient.server import Server, SlackLoginError, SlackConnectionError
from slackclient.channel import Channel
from requests.packages.urllib3.exceptions import InsecureRequestWarning
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
@pytest.fixture
def rtm_start_fixture():
file_login_data = open('tests/data/rtm.start.json', 'r').read()
json_login_data = json.loads(file_login_data)
return json_login_data
def test_server():
server = Server("valid_token", connect=False)
assert type(server) == Server
# The server eqs to a string, either the token or workspace domain
assert server == "valid_token"
assert server != "invalid_token"
def test_server_connect(rtm_start_fixture):
with responses.RequestsMock() as rsps:
rsps.add(
responses.POST,
"https://slack.com/api/rtm.start",
status=200,
json=rtm_start_fixture
)
Server("token", connect=True)
for call in rsps.calls:
assert call.request.url in [
"https://slack.com/api/rtm.start"
]
def test_server_is_hashable(server):
server_map = {server: server.token}
assert server_map[server] == 'xoxp-1234123412341234-12341234-1234'
assert (server_map[server] == 'foo') is False
@patch('time.sleep', return_value=None)
def test_rate_limiting(patched_time_sleep, server):
# Testing for rate limit retry headers
with responses.RequestsMock() as rsps:
rsps.add(
responses.POST,
"https://slack.com/api/rtm.start",
status=429,
json={"ok": False},
headers={'Retry-After': "1"}
)
with pytest.raises(SlackConnectionError) as e:
server.rtm_connect()
for call in rsps.calls:
assert call.response.status_code == 429
assert e.message == "RTM connection attempt was rate limited 10 times."
def test_custom_agent(server):
server.append_user_agent("test agent", 1.0)
assert server.api_requester.custom_user_agent[0] == ['test agent', 1.0]
def test_server_parse_channel_data(server, rtm_start_fixture):
server.parse_channel_data(rtm_start_fixture["channels"])
assert type(server.channels.find('general')) == Channel
def test_server_parse_user_data(server, rtm_start_fixture):
server.parse_user_data(rtm_start_fixture["users"])
# Find user by Name
user_by_name = server.users.find('fakeuser')
assert type(user_by_name) == User
assert user_by_name == "fakeuser"
assert user_by_name != "someotheruser"
# Find user by ID
user_by_id = server.users.find('U10CX1234')
assert type(user_by_id) == User
assert user_by_id == "fakeuser"
# Don't find invalid user
user_by_id = server.users.find('invaliduser')
assert user_by_id is None
def test_server_cant_connect(server):
with responses.RequestsMock() as rsps:
rsps.add(
responses.POST,
"https://slack.com/api/rtm.start",
status=403,
json={"ok": False}
)
with pytest.raises(SlackConnectionError) as e:
server.rtm_connect()
def test_reconnect_flag(server, rtm_start_fixture):
with responses.RequestsMock() as rsps:
rsps.add(
responses.POST,
"https://slack.com/api/rtm.start",
status=200,
json=rtm_start_fixture
)
server.rtm_connect(auto_reconnect=True)
assert server.auto_reconnect is True
for call in rsps.calls:
assert call.request.url in [
"https://slack.com/api/rtm.start"
]
def test_rtm_reconnect(server, rtm_start_fixture):
with responses.RequestsMock() as rsps:
rsps.add(
responses.POST,
"https://slack.com/api/rtm.connect",
status=200,
json=rtm_start_fixture
)
server.rtm_connect(auto_reconnect=True, reconnect=True, use_rtm_start=False)
for call in rsps.calls:
assert call.request.url in [
"https://slack.com/api/rtm.connect"
]
@patch('time.sleep', return_value=None)
def test_rtm_max_reconnect_timeout(patched_time_sleep, server, rtm_start_fixture):
with responses.RequestsMock() as rsps:
rsps.add(
responses.POST,
"https://slack.com/api/rtm.connect",
status=200,
json=rtm_start_fixture
)
server.reconnect_count = 4
server.last_connected_at = time.time()
server.rtm_connect(auto_reconnect=True, reconnect=True, use_rtm_start=False)
assert server.reconnect_count == 5
def test_rtm_reconnect_timeout_recently_connected(server, rtm_start_fixture):
# If reconnected recently, server must wait to reconnect and increment the counter
with responses.RequestsMock() as rsps:
rsps.add(
responses.POST,
"https://slack.com/api/rtm.connect",
status=200,
json=rtm_start_fixture
)
server.reconnect_count = 0
server.last_connected_at = time.time()
server.rtm_connect(auto_reconnect=True, reconnect=True, use_rtm_start=False)
assert server.reconnect_count == 1
for call in rsps.calls:
assert call.request.url in [
"https://slack.com/api/rtm.connect"
]
def test_rtm_reconnect_timeout_not_recently_connected(server, rtm_start_fixture):
# If reconnecting after 3 minutes since last reconnect, reset counter and connect without wait
with responses.RequestsMock() as rsps:
rsps.add(
responses.POST,
"https://slack.com/api/rtm.connect",
status=200,
json=rtm_start_fixture
)
server.reconnect_count = 1
server.last_connected_at = time.time() - 180
server.rtm_connect(auto_reconnect=True, reconnect=True, use_rtm_start=False)
assert server.reconnect_count == 0
for call in rsps.calls:
assert call.request.url in [
"https://slack.com/api/rtm.connect"
]
def test_max_rtm_reconnects(server, monkeypatch):
monkeypatch.setattr("time.sleep", None)
with pytest.raises(SlackConnectionError) as e:
server.reconnect_count = 5
server.rtm_connect(auto_reconnect=True, reconnect=True, use_rtm_start=False)
assert e.message == "RTM connection failed, reached max reconnects."
@pytest.mark.xfail
def test_server_ping(server, monkeypatch):
monkeypatch.setattr("websocket.create_connection", lambda: True)
reply = server.ping()