Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions nokia/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -190,6 +190,19 @@ def set_token(self, token):
if self.refresh_cb:
self.refresh_cb(token)

def refresh_token(self, force=False):
current_time = ts()
expiration_time = int(self.credentials.token_expiry)

# Don't refresh unless the token will expire soon.
if force is False and expiration_time - current_time > 300:
return

# Refresh the token. No need to do anything with the return value
# as the client's token_updater will be called which means this
# classes' set_token method will be called (if successful).
self.client.refresh_token(self.client.auto_refresh_url)

def request(self, service, action, params=None, method='GET',
version=None):
params = params or {}
Expand Down
66 changes: 65 additions & 1 deletion tests/test_nokia_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
import unittest

from requests import Session
from requests_oauthlib import OAuth2Session
from nokia import (
NokiaActivity,
NokiaApi,
Expand Down Expand Up @@ -140,7 +141,70 @@ def test_set_token_refresh_cb(self):

self.assertEqual(api.token, token)
refresh_cb.assert_called_once_with(token)


def test_refresh_token_not_expired(self):
OAuth2Session.refresh_token = MagicMock()

timestamp = int((
datetime.datetime.utcnow() - datetime.datetime(1970, 1, 1)
).total_seconds())
creds = NokiaCredentials(token_expiry=(timestamp + 600))
refresh_cb = MagicMock()
api = NokiaApi(creds, refresh_cb=refresh_cb)

OAuth2Session.refresh_token.reset_mock()
api.refresh_token()

OAuth2Session.refresh_token.assert_not_called()

def test_refresh_token_not_expired_force(self):
OAuth2Session.refresh_token = MagicMock()

timestamp = int((
datetime.datetime.utcnow() - datetime.datetime(1970, 1, 1)
).total_seconds())
creds = NokiaCredentials(token_expiry=(timestamp + 600))
refresh_cb = MagicMock()
api = NokiaApi(creds, refresh_cb=refresh_cb)

api.refresh_token(force=True)

OAuth2Session.refresh_token.assert_called_once_with(
api.client.auto_refresh_url
)

def test_refresh_token_expired(self):
OAuth2Session.refresh_token = MagicMock()

timestamp = int((
datetime.datetime.utcnow() - datetime.datetime(1970, 1, 1)
).total_seconds())
creds = NokiaCredentials(token_expiry=(timestamp - 600))
refresh_cb = MagicMock()
api = NokiaApi(creds, refresh_cb=refresh_cb)

api.refresh_token()

OAuth2Session.refresh_token.assert_called_once_with(
api.client.auto_refresh_url
)

def test_refresh_token_expired_force(self):
OAuth2Session.refresh_token = MagicMock()

timestamp = int((
datetime.datetime.utcnow() - datetime.datetime(1970, 1, 1)
).total_seconds())
creds = NokiaCredentials(token_expiry=(timestamp - 600))
refresh_cb = MagicMock()
api = NokiaApi(creds, refresh_cb=refresh_cb)

api.refresh_token(force=True)

OAuth2Session.refresh_token.assert_called_once_with(
api.client.auto_refresh_url
)

def test_request(self):
"""
Make sure the request method builds the proper URI and returns the
Expand Down