How to use the slackclient.exceptions.TokenRefreshError function in slackclient

To help you get started, we’ve selected a few slackclient examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github slackapi / python-slackclient / tests / test_slackclient.py View on Github external
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")
github slackapi / python-slackclient / tests / test_slackclient.py View on Github external
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",
        )
github slackapi / python-slackclient / slackclient / client.py View on Github external
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
github slackapi / python-slackclient / slackclient / client.py View on Github external
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)
github slackapi / python-slackclient / slackclient / client.py View on Github external
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")