forked from slackapi/python-slack-sdk
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_async_web_client.py
More file actions
158 lines (138 loc) · 5.98 KB
/
Copy pathtest_async_web_client.py
File metadata and controls
158 lines (138 loc) · 5.98 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
import io
import re
import unittest
import slack.errors as err
from slack import AsyncWebClient
from tests.helpers import async_test
from tests.web.mock_web_api_server import (
setup_mock_web_api_server,
cleanup_mock_web_api_server,
)
class TestAsyncWebClient(unittest.TestCase):
def setUp(self):
setup_mock_web_api_server(self)
self.client = AsyncWebClient(
token="xoxp-1234", base_url="http://localhost:8888",
)
def tearDown(self):
cleanup_mock_web_api_server(self)
pattern_for_language = re.compile("python/(\\S+)", re.IGNORECASE)
pattern_for_package_identifier = re.compile("slackclient/(\\S+)")
@async_test
async def test_api_calls_return_a_future(self):
self.client.token = "xoxb-api_test"
resp = await self.client.api_test()
self.assertEqual(200, resp.status_code)
self.assertTrue(resp["ok"])
@async_test
async def test_requests_can_be_paginated(self):
self.client.token = "xoxb-users_list_pagination"
users = []
async for page in await self.client.users_list(limit=2):
users = users + page["members"]
self.assertTrue(len(users) == 4)
@async_test
async def test_request_pagination_stops_when_next_cursor_is_missing(self):
self.client.token = "xoxb-users_list_pagination_1"
users = []
async for page in await self.client.users_list(limit=2):
users = users + page["members"]
self.assertTrue(len(users) == 2)
@async_test
async def test_json_can_only_be_sent_with_post_requests(self):
with self.assertRaises(err.SlackRequestError):
await self.client.api_call("fake.method", http_verb="GET", json={})
@async_test
async def test_slack_api_error_is_raised_on_unsuccessful_responses(self):
self.client.token = "xoxb-api_test_false"
with self.assertRaises(err.SlackApiError):
await self.client.api_test()
self.client.token = "xoxb-500"
with self.assertRaises(err.SlackApiError):
await self.client.api_test()
@async_test
async def test_slack_api_rate_limiting_exception_returns_retry_after(self):
self.client.token = "xoxb-rate_limited"
try:
await self.client.api_test()
except err.SlackApiError as slack_api_error:
self.assertFalse(slack_api_error.response["ok"])
self.assertEqual(429, slack_api_error.response.status_code)
self.assertEqual(30, int(slack_api_error.response.headers["Retry-After"]))
@async_test
async def test_the_api_call_files_argument_creates_the_expected_data(self):
self.client.token = "xoxb-users_setPhoto"
resp = await self.client.users_setPhoto(image="tests/data/slack_logo.png")
self.assertEqual(200, resp.status_code)
@async_test
async def test_issue_560_bool_in_params_sync(self):
self.client.token = "xoxb-conversations_list"
await self.client.conversations_list(exclude_archived=1) # ok
await self.client.conversations_list(exclude_archived="true") # ok
await self.client.conversations_list(exclude_archived=True) # ok
@async_test
async def test_issue_690_oauth_v2_access_async(self):
self.client.token = ""
resp = await self.client.oauth_v2_access(
client_id="111.222", client_secret="secret", code="codeeeeeeeeee",
)
self.assertIsNone(resp["error"])
with self.assertRaises(err.SlackApiError):
await self.client.oauth_v2_access(
client_id="999.999", client_secret="secret", code="codeeeeeeeeee",
)
@async_test
async def test_issue_690_oauth_access_async(self):
self.client.token = ""
resp = await self.client.oauth_access(
client_id="111.222", client_secret="secret", code="codeeeeeeeeee"
)
self.assertIsNone(resp["error"])
with self.assertRaises(err.SlackApiError):
await self.client.oauth_access(
client_id="999.999", client_secret="secret", code="codeeeeeeeeee"
)
@async_test
async def test_token_param_async(self):
with self.assertRaises(err.SlackApiError):
await self.client.users_list()
resp = await self.client.users_list(token="xoxb-users_list_pagination")
self.assertIsNone(resp["error"])
with self.assertRaises(err.SlackApiError):
await self.client.users_list()
@async_test
async def test_timeout_issue_712_async(self):
with self.assertRaises(Exception):
await self.client.users_list(token="xoxb-timeout")
@async_test
async def test_html_response_body_issue_718_async(self):
try:
await self.client.users_list(token="xoxb-html_response")
self.fail("SlackApiError expected here")
except err.SlackApiError as e:
self.assertEqual(
"The request to the Slack API failed.\n"
"The server responded with: {}",
str(e),
)
@async_test
async def test_user_agent_customization_issue_769_async(self):
client = AsyncWebClient(
token="xoxb-user-agent this_is test",
base_url="http://localhost:8888",
user_agent_prefix="this_is",
user_agent_suffix="test",
)
resp = await client.api_test()
self.assertTrue(resp["ok"])
@async_test
async def test_issue_809_filename_for_IOBase(self):
self.client.token = "xoxb-api_test"
file = io.BytesIO(b"here is my data but not sure what is wrong.......")
resp = await self.client.files_upload(file=file)
self.assertIsNone(resp["error"])
# if file:
# if "filename" not in kwargs:
# # use the local filename if filename is missing
# > kwargs["filename"] = file.split(os.path.sep)[-1]
# E AttributeError: '_io.BytesIO' object has no attribute 'split'