forked from slackapi/python-slack-sdk
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_rtm_client.py
More file actions
88 lines (66 loc) · 2.83 KB
/
Copy pathtest_rtm_client.py
File metadata and controls
88 lines (66 loc) · 2.83 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
# Standard Imports
import collections
import unittest
from unittest import mock
# Internal Imports
import slack
import slack.errors as e
from tests.helpers import mock_rtm_response
class TestRTMClient(unittest.TestCase):
def setUp(self):
self.client = slack.RTMClient(token="xoxp-1234", auto_reconnect=False)
def tearDown(self):
slack.RTMClient._callbacks = collections.defaultdict(list)
def test_run_on_annotation_sets_callbacks(self):
@slack.RTMClient.run_on(event="message")
def say_run_on(**payload):
pass
self.assertTrue(self.client._callbacks["message"][0].__name__ == "say_run_on")
def test_on_sets_callbacks(self):
def say_on(**payload):
pass
self.client.on(event="message", callback=say_on)
self.assertTrue(self.client._callbacks["message"][0].__name__ == "say_on")
def test_on_accepts_a_list_of_callbacks(self):
def say_on(**payload):
pass
def say_off(**payload):
pass
self.client.on(event="message", callback=[say_on, say_off])
self.assertEqual(len(self.client._callbacks["message"]), 2)
def test_on_raises_when_not_callable(self):
invalid_callback = "a"
with self.assertRaises(e.SlackClientError) as context:
self.client.on(event="message", callback=invalid_callback)
expected_error = "The specified callback 'a' is not callable."
error = str(context.exception)
self.assertIn(expected_error, error)
def test_on_raises_when_kwargs_not_accepted(self):
def invalid_cb():
pass
with self.assertRaises(e.SlackClientError) as context:
self.client.on(event="message", callback=invalid_cb)
expected_error = (
"The callback 'invalid_cb' must accept keyword arguments (**kwargs)."
)
error = str(context.exception)
self.assertIn(expected_error, error)
def test_send_over_websocket_raises_when_not_connected(self):
with self.assertRaises(e.SlackClientError) as context:
self.client.send_over_websocket(payload={})
expected_error = "Websocket connection is closed."
error = str(context.exception)
self.assertIn(expected_error, error)
@mock.patch("slack.WebClient._send", new_callable=mock_rtm_response)
def test_start_raises_an_error_if_rtm_ws_url_is_not_returned(
self, mock_rtm_response
):
mock_rtm_response.coro.return_value = {
"data": {"ok": True},
"headers": {},
"status_code": 200,
}
with self.assertRaises(e.SlackApiError) as context:
slack.RTMClient(token="xoxp-1234", auto_reconnect=False).start()
expected_error = "Unable to retreive RTM URL from Slack"
self.assertIn(expected_error, str(context.exception))