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_token_refresh_failed():
# Client should raise TokenRefreshError is token refresh returns error
def token_update_callback(update_data):
return update_data
sc = SlackClient(
client_id="12345",
client_secret="12345",
refresh_token="refresh_token",
token_update_callback=token_update_callback,
)
with pytest.raises(TokenRefreshError):
with responses.RequestsMock() as rsps:
rsps.add(
responses.POST,
"https://slack.com/api/channels.list",
status=200,
json={"ok": False, "error": "invalid_auth"},
)
rsps.add(
responses.POST,
"https://slack.com/api/oauth.access",
status=200,
json={"ok": False, "error": "invalid_auth"},
)
sc.api_call("channels.list")
def test_noncallable_refresh_callback():
with pytest.raises(TokenRefreshError):
SlackClient(
client_id="12345",
client_secret="12345",
refresh_token="refresh_token",
token_update_callback="THIS IS A STRING, NOT A CALLABLE METHOD",
)
def rtm_connect(self, with_team_state=True, **kwargs):
"""
Connects to the RTM Websocket
:Args:
with_team_state (bool): Connect via `rtm.start` to pull workspace state information.
`False` connects via `rtm.connect`, which is lighter weight and better for very large
teams.
:Returns:
False on exceptions
"""
if self.refresh_token:
raise TokenRefreshError(
"Workspace tokens may not be used to connect to the RTM API."
)
try:
self.server.rtm_connect(use_rtm_start=with_team_state, **kwargs)
return self.server.connected
except Exception:
LOG.warn("Failed RTM connect", exc_info=True)
return False
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)
self.server.token = response_json["access_token"]
# Update the token expiration timestamp
current_ts = int(time.time())
expires_at = int(current_ts + response_json["expires_in"])
self.access_token_expires_at = expires_at
# Call the developer's token update callback
update_args = {
"enterprise_id": response_json["enterprise_id"],
"team_id": response_json["team_id"],
"access_token": response_json["access_token"],
"expires_in": response_json["expires_in"],
}
self.token_update_callback(update_args)
else:
raise TokenRefreshError("Token refresh failed")