Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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="token", connect=True)
for call in rsps.calls:
assert call.request.url in [
"https://slack.com/api/rtm.start"
]
def test_server():
server = Server(token="valid_token", connect=False)
assert type(server) == Server
# The server eqs to a string, either the token or workspace domain
assert server.token == "valid_token"
def test_server():
server = Server(token="valid_token", connect=False)
assert type(server) == Server
# The server eqs to a string, either the token or workspace domain
assert server.token == "valid_token"
def test_output():
''' Test that sending a message behaves as expected '''
rtmbot = init_rtmbot()
# Mock the slack_client object with Server, Channel objects and needed methods
slackclient_mock = create_autospec(SlackClient)
server_mock = create_autospec(Server)
# Mock Server with channels method and correct return value
slackclient_mock.server = server_mock
searchlist_mock = create_autospec(SearchList)
server_mock.channels = searchlist_mock
channel_mock = create_autospec(Channel)
slackclient_mock.server.channels.find.return_value = channel_mock
rtmbot.slack_client = slackclient_mock
# mock the plugin object to return a sample response
plugin_mock = create_autospec(Plugin)
plugin_mock.do_output.return_value = [['C12345678', 'test message']]
rtmbot.bot_plugins.append(plugin_mock)
rtmbot.output()
import threading
from slackclient.server import Server
from slackclient.client import SlackClient
class ThreadedSlackClient(SlackClient):
def __init__(self, *args, **kwargs):
super(ThreadedSlackClient, self).__init__(*args, **kwargs)
self.server = ThreadedServer(self.token, False)
class ThreadedServer(Server):
def __init__(self, *args, **kwargs):
super(ThreadedServer, self).__init__(*args, **kwargs)
self.api_call_lock = threading.RLock()
def api_call(self, *args, **kwargs):
with self.api_call_lock:
return super(ThreadedServer, self).api_call(*args, **kwargs)
if callable(token_update_callback):
self.server = Server(
connect=False,
proxies=proxies,
refresh_token=refresh_token,
client_id=client_id,
client_secret=client_secret,
token_update_callback=token_update_callback,
)
else:
raise TokenRefreshError(
"Token refresh callback function is required when using refresh token."
)
else:
# Slack app configs
self.server = Server(token=token, connect=False, proxies=proxies)
def __init__(self, token: str, proxies: Optional[Dict[str,str]] = None) -> None:
self.token = token
self.server = Server(self.token, proxies)
def __init__(self, token, proxies=None):
self.token = token
self.server = Server(self.token, False, proxies)
client_id (str): Your app's Client ID
client_secret (srt): Your app's Client Secret (Used for OAuth requests)
refresh_callback (function): Your application's function for updating Slack
OAuth tokens inside your data store
"""
self.client_id = client_id
self.client_secret = client_secret
self.refresh_token = refresh_token
self.token_update_callback = token_update_callback
self.token = token
self.access_token_expires_at = 0
if refresh_token:
if callable(token_update_callback):
self.server = Server(
connect=False,
proxies=proxies,
refresh_token=refresh_token,
client_id=client_id,
client_secret=client_secret,
token_update_callback=token_update_callback,
)
else:
raise TokenRefreshError(
"Token refresh callback function is required when using refresh token."
)
else:
# Slack app configs
self.server = Server(token=token, connect=False, proxies=proxies)