From 9550c8fd86314a15d693a3389b68468b17813156 Mon Sep 17 00:00:00 2001 From: Tony Jackson Date: Mon, 21 Dec 2020 02:31:21 -0800 Subject: [PATCH] Create CacheHandler to abstract caching tokens (#625) * Refactor functions into static methods of AuthBase Functions is_token_expired was a loose function that was added to SpotifyClientCredentials, SpotifyPKCE, and SpotifyImplicitGrant classes through a method on each class that passed the call to the loose is_token_expired function. Function _is_scope_subset was duplicated on SpotifyClientCredentials, SpotifyPKCE, and SpotifyImplicitGrant classes. Refactoring is_token_expired and _is_scope_subset to be static methods on SpotifyAuthBase means both are available for all derived classes and require less boilerplate. * Create CacheHandler to abstract caching tokens Previous code only supported caching to and from json files in a given directory. In addition, the get_cached_token method mixed getting and getting the token in the same method. This change creates a CacheHandler class to abstract out the caching implementation and allow the user to cache tokens in any way they see fit. For example, the user could create a MongoCache class to store and retrieve tokens from a Mongo database and specify that cache_handler=MongoCache in creating an auth_manager object. To implement the CacheHandler abstraction, the following changes are implemented: The validation code in each get_cached_token method in SpotifyOAuth, SpotifyPKCE, and SpotifyImplicitGrant is moved into a validate_token method in each class. The CacheHandler class is created with get_cached_token and save_token_to_cache methods. Previous instances of self.get_cached_token() are now replaced with self.validate_token(self.cache_handler.get_cached_token()) to preserve the getting and validation behaviour. cache_handler is added as an argument to SpotifyOAuth, SpotifyPKCE, and SpotifyImplicitGrant. Specifying a cache_handler now overrides any specification of cache_path and/or username. To preserve backwards compatibility in handling cache files, a CacheFileHandler class extending CacheHandler is created. If no cache_handler is specified, the cache_path and username arguments are used to create an instance of CacheFileHandler. It may be worth deprecating the cache_path and username fields in favour of using CacheFileHandler. Tests are also modified and extended to cover the new functionality. A sample MemoryCache CacheHandler is created to test getting and saving to a custom CacheHandler. * Fix cache_handler subclass check for Python 2 * Split assert message to fix line over max length * Split cache handlers into cache_handler.py * flake8 and autopep fixes * Fix init to allow importing CacheHandler When spotipy is installed as a package, CacheHandler is not accessible from a `from spotipy import CacheHandler` statement because the import is not specified in the __init__.py file. This commit adds CacheHandler and CacheFileHandler to the init file so the user can import them. * flake8 fix --- CHANGELOG.md | 3 +- spotipy/__init__.py | 1 + spotipy/cache_handler.py | 84 ++++++++++++ spotipy/oauth2.py | 281 +++++++++++++++------------------------ spotipy/util.py | 2 +- tests/unit/test_oauth.py | 71 ++++++---- 6 files changed, 240 insertions(+), 202 deletions(-) create mode 100644 spotipy/cache_handler.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 89845c8..cda4df2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,7 +19,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added - Added log messages for when the access and refresh tokens are retrieved and when they are refreshed -- Support `market` optional parameter in `track` +- Support `market` optional parameter in `track` +- Added CacheHandler abstraction to allow users to cache tokens in any way they see fit ## [2.16.1] - 2020-10-24 diff --git a/spotipy/__init__.py b/spotipy/__init__.py index 17d1f0f..7f3d859 100644 --- a/spotipy/__init__.py +++ b/spotipy/__init__.py @@ -1,3 +1,4 @@ +from .cache_handler import * # noqa from .client import * # noqa from .exceptions import * # noqa from .oauth2 import * # noqa diff --git a/spotipy/cache_handler.py b/spotipy/cache_handler.py new file mode 100644 index 0000000..3ba3987 --- /dev/null +++ b/spotipy/cache_handler.py @@ -0,0 +1,84 @@ +__all__ = ['CacheHandler', 'CacheFileHandler'] + +import errno +import json +import logging + +logger = logging.getLogger(__name__) + + +class CacheHandler(): + """ + An abstraction layer for handling the caching and retrieval of + authorization tokens. + + Custom extensions of this class must implement get_cached_token + and save_token_to_cache methods with the same input and output + structure as the CacheHandler class. + """ + + def get_cached_token(self): + """ + Get and return a token_info dictionary object. + """ + # return token_info + raise NotImplementedError() + + def save_token_to_cache(self, token_info): + """ + Save a token_info dictionary object to the cache and return None. + """ + raise NotImplementedError() + return None + + +class CacheFileHandler(CacheHandler): + """ + Handles reading and writing cached Spotify authorization tokens + as json files on disk. + """ + + def __init__(self, + cache_path=None, + username=None): + """ + Parameters: + * cache_path: May be supplied, will otherwise be generated + (takes precedence over `username`) + * username: May be supplied or set as environment variable + (will set `cache_path` to `.cache-{username}`) + """ + + if cache_path: + self.cache_path = cache_path + else: + cache_path = ".cache" + if username: + cache_path += "-" + str(username) + self.cache_path = cache_path + + def get_cached_token(self): + token_info = None + + try: + f = open(self.cache_path) + token_info_string = f.read() + f.close() + token_info = json.loads(token_info_string) + + except IOError as error: + if error.errno == errno.ENOENT: + logger.debug("cache does not exist at: %s", self.cache_path) + else: + logger.warning("Couldn't read cache at: %s", self.cache_path) + + return token_info + + def save_token_to_cache(self, token_info): + try: + f = open(self.cache_path, "w") + f.write(json.dumps(token_info)) + f.close() + except IOError: + logger.warning('Couldn\'t write token to cache at: %s', + self.cache_path) diff --git a/spotipy/oauth2.py b/spotipy/oauth2.py index c0b41ce..4adb76a 100644 --- a/spotipy/oauth2.py +++ b/spotipy/oauth2.py @@ -1,7 +1,6 @@ # -*- coding: utf-8 -*- __all__ = [ - "is_token_expired", "SpotifyClientCredentials", "SpotifyOAuth", "SpotifyOauthError", @@ -11,8 +10,6 @@ __all__ = [ ] import base64 -import errno -import json import logging import os import time @@ -26,6 +23,7 @@ import six.moves.urllib.parse as urllibparse from six.moves.BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer from six.moves.urllib_parse import parse_qsl, urlparse +from spotipy.cache_handler import CacheFileHandler, CacheHandler from spotipy.exceptions import SpotifyException from spotipy.util import CLIENT_CREDS_ENV_VARS, get_host_port @@ -62,11 +60,6 @@ def _make_authorization_headers(client_id, client_secret): return {"Authorization": "Basic %s" % auth_header.decode("ascii")} -def is_token_expired(token_info): - now = int(time.time()) - return token_info["expires_at"] - now < 60 - - def _ensure_value(value, env_key): env_val = CLIENT_CREDS_ENV_VARS[env_key] _val = value or os.getenv(env_val) @@ -79,17 +72,6 @@ def _ensure_value(value, env_key): return _val -def _get_cache_path(cache_path, username): - if cache_path: - return cache_path - - cache_path = ".cache" - if username: - cache_path += "-" + str(username) - - return cache_path - - class SpotifyAuthBase(object): def __init__(self, requests_session): if isinstance(requests_session, requests.Session): @@ -132,6 +114,19 @@ class SpotifyAuthBase(object): except NameError: return input(prompt) + @staticmethod + def is_token_expired(token_info): + now = int(time.time()) + return token_info["expires_at"] - now < 60 + + @staticmethod + def _is_scope_subset(needle_scope, haystack_scope): + needle_scope = set(needle_scope.split()) if needle_scope else set() + haystack_scope = ( + set(haystack_scope.split()) if haystack_scope else set() + ) + return needle_scope <= haystack_scope + def __del__(self): """Make sure the connection (pool) gets closed""" if isinstance(self._session, requests.Session): @@ -220,9 +215,6 @@ class SpotifyClientCredentials(SpotifyAuthBase): token_info = response.json() return token_info - def is_token_expired(self, token_info): - return is_token_expired(token_info) - def _add_custom_values_to_token_info(self, token_info): """ Store some values that aren't directly provided by a Web API @@ -252,7 +244,8 @@ class SpotifyOAuth(SpotifyAuthBase): show_dialog=False, requests_session=True, requests_timeout=None, - open_browser=True + open_browser=True, + cache_handler=None ): """ Creates a SpotifyOAuth object @@ -263,6 +256,10 @@ class SpotifyOAuth(SpotifyAuthBase): * redirect_uri: Must be supplied or set as environment variable * state: May be supplied, no verification is performed * scope: May be supplied, intuitively converted to proper format + * cache_handler: An instance of the `CacheHandler` class to handle + getting and saving cached authorization tokens. + May be supplied, will otherwise use `CacheFileHandler`. + (takes precedence over `cache_path` and `username`) * cache_path: May be supplied, will otherwise be generated (takes precedence over `username`) * username: May be supplied or set as environment variable @@ -280,65 +277,39 @@ class SpotifyOAuth(SpotifyAuthBase): self.client_secret = client_secret self.redirect_uri = redirect_uri self.state = state - self.username = username or os.getenv( - CLIENT_CREDS_ENV_VARS["client_username"] - ) - self.cache_path = _get_cache_path(cache_path, self.username) self.scope = self._normalize_scope(scope) + if cache_handler: + assert issubclass(cache_handler.__class__, CacheHandler), \ + "cache_handler must be a subclass of CacheHandler: " + str(type(cache_handler)) \ + + " != " + str(CacheHandler) + self.cache_handler = cache_handler + else: + self.cache_handler = CacheFileHandler( + username=(username or os.getenv(CLIENT_CREDS_ENV_VARS["client_username"])), + cache_path=cache_path + ) self.proxies = proxies self.requests_timeout = requests_timeout self.show_dialog = show_dialog self.open_browser = open_browser - def get_cached_token(self): - """ Gets a cached auth token - """ - token_info = None + def validate_token(self, token_info): + if token_info is None: + return None - try: - f = open(self.cache_path) - token_info_string = f.read() - f.close() - token_info = json.loads(token_info_string) + # if scopes don't match, then bail + if "scope" not in token_info or not self._is_scope_subset( + self.scope, token_info["scope"] + ): + return None - # if scopes don't match, then bail - if "scope" not in token_info or not self._is_scope_subset( - self.scope, token_info["scope"] - ): - return None - - if self.is_token_expired(token_info): - token_info = self.refresh_access_token( - token_info["refresh_token"] - ) - except IOError as error: - if error.errno == errno.ENOENT: - logger.debug("cache does not exist at: %s", self.cache_path) - else: - logger.warning("Couldn't read cache at: %s", self.cache_path) + if self.is_token_expired(token_info): + token_info = self.refresh_access_token( + token_info["refresh_token"] + ) return token_info - def _save_token_info(self, token_info): - if self.cache_path: - try: - f = open(self.cache_path, "w") - f.write(json.dumps(token_info)) - f.close() - except IOError: - logger.warning('Couldn\'t write token to cache at: %s', - self.cache_path) - - def _is_scope_subset(self, needle_scope, haystack_scope): - needle_scope = set(needle_scope.split()) if needle_scope else set() - haystack_scope = ( - set(haystack_scope.split()) if haystack_scope else set() - ) - return needle_scope <= haystack_scope - - def is_token_expired(self, token_info): - return is_token_expired(token_info) - def get_authorize_url(self, state=None): """ Gets the URL to use to authorize this app """ @@ -479,9 +450,9 @@ class SpotifyOAuth(SpotifyAuthBase): stacklevel=2, ) if check_cache: - token_info = self.get_cached_token() + token_info = self.validate_token(self.cache_handler.get_cached_token()) if token_info is not None: - if is_token_expired(token_info): + if self.is_token_expired(token_info): token_info = self.refresh_access_token( token_info["refresh_token"] ) @@ -521,7 +492,7 @@ class SpotifyOAuth(SpotifyAuthBase): error_description=error_payload['error_description']) token_info = response.json() token_info = self._add_custom_values_to_token_info(token_info) - self._save_token_info(token_info) + self.cache_handler.save_token_to_cache(token_info) return token_info if as_dict else token_info["access_token"] def _normalize_scope(self, scope): @@ -571,7 +542,7 @@ class SpotifyOAuth(SpotifyAuthBase): token_info = self._add_custom_values_to_token_info(token_info) if "refresh_token" not in token_info: token_info["refresh_token"] = refresh_token - self._save_token_info(token_info) + self.cache_handler.save_token_to_cache(token_info) return token_info def _add_custom_values_to_token_info(self, token_info): @@ -609,7 +580,8 @@ class SpotifyPKCE(SpotifyAuthBase): proxies=None, requests_timeout=None, requests_session=True, - open_browser=True): + open_browser=True, + cache_handler=None): """ Creates Auth Manager with the PKCE Auth flow. @@ -619,6 +591,10 @@ class SpotifyPKCE(SpotifyAuthBase): * redirect_uri: Must be supplied or set as environment variable * state: May be supplied, no verification is performed * scope: May be supplied, intuitively converted to proper format + * cache_handler: An instance of the `CacheHandler` class to handle + getting and saving cached authorization tokens. + May be supplied, will otherwise use `CacheFileHandler`. + (takes precedence over `cache_path` and `username`) * cache_path: May be supplied, will otherwise be generated (takes precedence over `username`) * username: May be supplied or set as environment variable @@ -635,10 +611,15 @@ class SpotifyPKCE(SpotifyAuthBase): self.redirect_uri = redirect_uri self.state = state self.scope = self._normalize_scope(scope) - self.username = username or os.getenv( - CLIENT_CREDS_ENV_VARS["client_username"] - ) - self.cache_path = _get_cache_path(cache_path, self.username) + if cache_handler: + assert issubclass(type(cache_handler), CacheHandler), \ + "type(cache_handler): " + str(type(cache_handler)) + " != " + str(CacheHandler) + self.cache_handler = cache_handler + else: + self.cache_handler = CacheFileHandler( + username=(username or os.getenv(CLIENT_CREDS_ENV_VARS["client_username"])), + cache_path=cache_path + ) self.proxies = proxies self.requests_timeout = requests_timeout @@ -781,55 +762,23 @@ class SpotifyPKCE(SpotifyAuthBase): return self.parse_response_code(response) return self._get_auth_response() - def get_cached_token(self): - """ Gets a cached auth token - """ - token_info = None + def validate_token(self, token_info): + if token_info is None: + return None - try: - f = open(self.cache_path) - token_info_string = f.read() - f.close() - token_info = json.loads(token_info_string) + # if scopes don't match, then bail + if "scope" not in token_info or not self._is_scope_subset( + self.scope, token_info["scope"] + ): + return None - # if scopes don't match, then bail - if "scope" not in token_info or not self._is_scope_subset( - self.scope, token_info["scope"] - ): - return None - - if self.is_token_expired(token_info): - token_info = self.refresh_access_token( - token_info["refresh_token"] - ) - except IOError as error: - if error.errno == errno.ENOENT: - logger.debug("cache does not exist at: %s", self.cache_path) - else: - logger.warning("Couldn't read cache at: %s", self.cache_path) + if self.is_token_expired(token_info): + token_info = self.refresh_access_token( + token_info["refresh_token"] + ) return token_info - def _is_scope_subset(self, needle_scope, haystack_scope): - needle_scope = set(needle_scope.split()) if needle_scope else set() - haystack_scope = ( - set(haystack_scope.split()) if haystack_scope else set() - ) - return needle_scope <= haystack_scope - - def is_token_expired(self, token_info): - return is_token_expired(token_info) - - def _save_token_info(self, token_info): - if self.cache_path: - try: - f = open(self.cache_path, "w") - f.write(json.dumps(token_info)) - f.close() - except IOError: - logger.warning('Couldn\'t write token to cache at: %s', - self.cache_path) - def _add_custom_values_to_token_info(self, token_info): """ Store some values that aren't directly provided by a Web API @@ -856,9 +805,9 @@ class SpotifyPKCE(SpotifyAuthBase): """ if check_cache: - token_info = self.get_cached_token() + token_info = self.validate_token(self.cache_handler.get_cached_token()) if token_info is not None: - if is_token_expired(token_info): + if self.is_token_expired(token_info): token_info = self.refresh_access_token( token_info["refresh_token"] ) @@ -900,7 +849,7 @@ class SpotifyPKCE(SpotifyAuthBase): error_description=error_payload['error_description']) token_info = response.json() token_info = self._add_custom_values_to_token_info(token_info) - self._save_token_info(token_info) + self.cache_handler.save_token_to_cache(token_info) return token_info["access_token"] def refresh_access_token(self, refresh_token): @@ -944,7 +893,7 @@ class SpotifyPKCE(SpotifyAuthBase): token_info = self._add_custom_values_to_token_info(token_info) if "refresh_token" not in token_info: token_info["refresh_token"] = refresh_token - self._save_token_info(token_info) + self.cache_handler.save_token_to_cache(token_info) return token_info def parse_response_code(self, url): @@ -1006,7 +955,8 @@ class SpotifyImplicitGrant(SpotifyAuthBase): scope=None, cache_path=None, username=None, - show_dialog=False): + show_dialog=False, + cache_handler=None): """ Creates Auth Manager using the Implicit Grant flow **See help(SpotifyImplictGrant) for full Security Warning** @@ -1017,6 +967,10 @@ class SpotifyImplicitGrant(SpotifyAuthBase): * redirect_uri: Must be supplied or set as environment variable * state: May be supplied, no verification is performed * scope: May be supplied, intuitively converted to proper format + * cache_handler: An instance of the `CacheHandler` class to handle + getting and saving cached authorization tokens. + May be supplied, will otherwise use `CacheFileHandler`. + (takes precedence over `cache_path` and `username`) * cache_path: May be supplied, will otherwise be generated (takes precedence over `username`) * username: May be supplied or set as environment variable @@ -1032,59 +986,34 @@ class SpotifyImplicitGrant(SpotifyAuthBase): self.client_id = client_id self.redirect_uri = redirect_uri self.state = state - self.username = username or os.getenv( - CLIENT_CREDS_ENV_VARS["client_username"] - ) - self.cache_path = _get_cache_path(cache_path, self.username) + if cache_handler: + assert issubclass(type(cache_handler), CacheHandler), \ + "type(cache_handler): " + str(type(cache_handler)) + " != " + str(CacheHandler) + self.cache_handler = cache_handler + else: + self.cache_handler = CacheFileHandler( + username=(username or os.getenv(CLIENT_CREDS_ENV_VARS["client_username"])), + cache_path=cache_path + ) self.scope = self._normalize_scope(scope) self.show_dialog = show_dialog self._session = None # As to not break inherited __del__ - def get_cached_token(self): - """ Gets a cached auth token - """ - token_info = None + def validate_token(self, token_info): + if token_info is None: + return None - try: - f = open(self.cache_path) - token_info_string = f.read() - f.close() - token_info = json.loads(token_info_string) + # if scopes don't match, then bail + if "scope" not in token_info or not self._is_scope_subset( + self.scope, token_info["scope"] + ): + return None - # if scopes don't match, then bail - if "scope" not in token_info or not self._is_scope_subset( - self.scope, token_info["scope"] - ): - return None - - if self.is_token_expired(token_info): - return None - except IOError as error: - if error.errno == errno.ENOENT: - logger.debug("cache does not exist at: %s", self.cache_path) - else: - logger.warning("Couldn't read cache at: %s", self.cache_path) + if self.is_token_expired(token_info): + return None return token_info - def _save_token_info(self, token_info): - try: - f = open(self.cache_path, "w") - f.write(json.dumps(token_info)) - f.close() - except IOError: - logger.warning("Couldn't write token to cache at: %s", self.cache_path) - - def _is_scope_subset(self, needle_scope, haystack_scope): - needle_scope = set(needle_scope.split()) if needle_scope else set() - haystack_scope = ( - set(haystack_scope.split()) if haystack_scope else set() - ) - return needle_scope <= haystack_scope - - def is_token_expired(self, token_info): - return is_token_expired(token_info) - def get_access_token(self, state=None, response=None, @@ -1098,8 +1027,8 @@ class SpotifyImplicitGrant(SpotifyAuthBase): * check_cache: Interpreted as boolean """ if check_cache: - token_info = self.get_cached_token() - if not (token_info is None or is_token_expired(token_info)): + token_info = self.validate_token(self.cache_handler.get_cached_token()) + if not (token_info is None or self.is_token_expired(token_info)): return token_info["access_token"] if response: @@ -1107,7 +1036,7 @@ class SpotifyImplicitGrant(SpotifyAuthBase): else: token_info = self.get_auth_response(state) token_info = self._add_custom_values_to_token_info(token_info) - self._save_token_info(token_info) + self.cache_handler.save_token_to_cache(token_info) return token_info["access_token"] diff --git a/spotipy/util.py b/spotipy/util.py index 0345ac3..8961d33 100644 --- a/spotipy/util.py +++ b/spotipy/util.py @@ -93,7 +93,7 @@ def prompt_for_user_token( # if not in the cache, the create a new (this will send # the user to a web page where they can authorize this app) - token_info = sp_oauth.get_cached_token() + token_info = sp_oauth.validate_token(sp_oauth.cache_handler.get_cached_token()) if not token_info: code = sp_oauth.get_auth_response() diff --git a/tests/unit/test_oauth.py b/tests/unit/test_oauth.py index 816f1a2..4fea792 100644 --- a/tests/unit/test_oauth.py +++ b/tests/unit/test_oauth.py @@ -6,6 +6,7 @@ import unittest import six.moves.urllib.parse as urllibparse from spotipy import SpotifyOAuth, SpotifyImplicitGrant, SpotifyPKCE +from spotipy.cache_handler import CacheHandler from spotipy.oauth2 import SpotifyClientCredentials, SpotifyOauthError from spotipy.oauth2 import SpotifyStateError @@ -50,11 +51,23 @@ def _make_pkceauth(*args, **kwargs): return SpotifyPKCE("CLID", "REDIR", "STATE", *args, **kwargs) +class MemoryCache(CacheHandler): + def __init__(self, token_info=None): + self.token_info = token_info + + def get_cached_token(self): + return self.token_info + + def save_token_to_cache(self, token_info): + self.token_info = token_info + return None + + class OAuthCacheTest(unittest.TestCase): @patch.multiple(SpotifyOAuth, is_token_expired=DEFAULT, refresh_access_token=DEFAULT) - @patch('spotipy.oauth2.open', create=True) + @patch('spotipy.cache_handler.open', create=True) def test_gets_from_cache_path(self, opener, is_token_expired, refresh_access_token): scope = "playlist-modify-private" @@ -65,7 +78,7 @@ class OAuthCacheTest(unittest.TestCase): is_token_expired.return_value = False spot = _make_oauth(scope, path) - cached_tok = spot.get_cached_token() + cached_tok = spot.validate_token(spot.cache_handler.get_cached_token()) opener.assert_called_with(path) self.assertIsNotNone(cached_tok) @@ -73,7 +86,7 @@ class OAuthCacheTest(unittest.TestCase): @patch.multiple(SpotifyOAuth, is_token_expired=DEFAULT, refresh_access_token=DEFAULT) - @patch('spotipy.oauth2.open', create=True) + @patch('spotipy.cache_handler.open', create=True) def test_expired_token_refreshes(self, opener, is_token_expired, refresh_access_token): scope = "playlist-modify-private" @@ -86,7 +99,7 @@ class OAuthCacheTest(unittest.TestCase): refresh_access_token.return_value = fresh_tok spot = _make_oauth(scope, path) - spot.get_cached_token() + spot.validate_token(spot.cache_handler.get_cached_token()) is_token_expired.assert_called_with(expired_tok) refresh_access_token.assert_called_with(expired_tok['refresh_token']) @@ -94,7 +107,7 @@ class OAuthCacheTest(unittest.TestCase): @patch.multiple(SpotifyOAuth, is_token_expired=DEFAULT, refresh_access_token=DEFAULT) - @patch('spotipy.oauth2.open', create=True) + @patch('spotipy.cache_handler.open', create=True) def test_badly_scoped_token_bails(self, opener, is_token_expired, refresh_access_token): token_scope = "playlist-modify-public" @@ -106,13 +119,13 @@ class OAuthCacheTest(unittest.TestCase): is_token_expired.return_value = False spot = _make_oauth(requested_scope, path) - cached_tok = spot.get_cached_token() + cached_tok = spot.validate_token(spot.cache_handler.get_cached_token()) opener.assert_called_with(path) self.assertIsNone(cached_tok) self.assertEqual(refresh_access_token.call_count, 0) - @patch('spotipy.oauth2.open', create=True) + @patch('spotipy.cache_handler.open', create=True) def test_saves_to_cache_path(self, opener): scope = "playlist-modify-private" path = ".cache-username" @@ -122,11 +135,21 @@ class OAuthCacheTest(unittest.TestCase): opener.return_value = fi spot = SpotifyOAuth("CLID", "CLISEC", "REDIR", "STATE", scope, path) - spot._save_token_info(tok) + spot.cache_handler.save_token_to_cache(tok) opener.assert_called_with(path, 'w') self.assertTrue(fi.write.called) + def test_cache_handler(self): + scope = "playlist-modify-private" + tok = _make_fake_token(1, 1, scope) + + spot = _make_oauth(scope, cache_handler=MemoryCache()) + spot.cache_handler.save_token_to_cache(tok) + cached_tok = spot.cache_handler.get_cached_token() + + self.assertEqual(tok, cached_tok) + class TestSpotifyOAuthGetAuthorizeUrl(unittest.TestCase): @@ -224,7 +247,7 @@ class TestSpotifyClientCredentials(unittest.TestCase): class ImplicitGrantCacheTest(unittest.TestCase): @patch.object(SpotifyImplicitGrant, "is_token_expired", DEFAULT) - @patch('spotipy.oauth2.open', create=True) + @patch('spotipy.cache_handler.open', create=True) def test_gets_from_cache_path(self, opener, is_token_expired): scope = "playlist-modify-private" path = ".cache-username" @@ -234,13 +257,13 @@ class ImplicitGrantCacheTest(unittest.TestCase): is_token_expired.return_value = False spot = _make_implicitgrantauth(scope, path) - cached_tok = spot.get_cached_token() + cached_tok = spot.cache_handler.get_cached_token() opener.assert_called_with(path) self.assertIsNotNone(cached_tok) @patch.object(SpotifyImplicitGrant, "is_token_expired", DEFAULT) - @patch('spotipy.oauth2.open', create=True) + @patch('spotipy.cache_handler.open', create=True) def test_expired_token_returns_none(self, opener, is_token_expired): scope = "playlist-modify-private" path = ".cache-username" @@ -250,14 +273,14 @@ class ImplicitGrantCacheTest(unittest.TestCase): opener.return_value = token_file spot = _make_implicitgrantauth(scope, path) - cached_tok = spot.get_cached_token() + cached_tok = spot.validate_token(spot.cache_handler.get_cached_token()) is_token_expired.assert_called_with(expired_tok) opener.assert_any_call(path) self.assertIsNone(cached_tok) @patch.object(SpotifyImplicitGrant, "is_token_expired", DEFAULT) - @patch('spotipy.oauth2.open', create=True) + @patch('spotipy.cache_handler.open', create=True) def test_badly_scoped_token_bails(self, opener, is_token_expired): token_scope = "playlist-modify-public" requested_scope = "playlist-modify-private" @@ -268,12 +291,12 @@ class ImplicitGrantCacheTest(unittest.TestCase): is_token_expired.return_value = False spot = _make_implicitgrantauth(requested_scope, path) - cached_tok = spot.get_cached_token() + cached_tok = spot.validate_token(spot.cache_handler.get_cached_token()) opener.assert_called_with(path) self.assertIsNone(cached_tok) - @patch('spotipy.oauth2.open', create=True) + @patch('spotipy.cache_handler.open', create=True) def test_saves_to_cache_path(self, opener): scope = "playlist-modify-private" path = ".cache-username" @@ -283,7 +306,7 @@ class ImplicitGrantCacheTest(unittest.TestCase): opener.return_value = fi spot = SpotifyImplicitGrant("CLID", "REDIR", "STATE", scope, path) - spot._save_token_info(tok) + spot.cache_handler.save_token_to_cache(tok) opener.assert_called_with(path, 'w') self.assertTrue(fi.write.called) @@ -343,7 +366,7 @@ class SpotifyPKCECacheTest(unittest.TestCase): @patch.multiple(SpotifyPKCE, is_token_expired=DEFAULT, refresh_access_token=DEFAULT) - @patch('spotipy.oauth2.open', create=True) + @patch('spotipy.cache_handler.open', create=True) def test_gets_from_cache_path(self, opener, is_token_expired, refresh_access_token): scope = "playlist-modify-private" @@ -354,7 +377,7 @@ class SpotifyPKCECacheTest(unittest.TestCase): is_token_expired.return_value = False spot = _make_pkceauth(scope, path) - cached_tok = spot.get_cached_token() + cached_tok = spot.cache_handler.get_cached_token() opener.assert_called_with(path) self.assertIsNotNone(cached_tok) @@ -362,7 +385,7 @@ class SpotifyPKCECacheTest(unittest.TestCase): @patch.multiple(SpotifyPKCE, is_token_expired=DEFAULT, refresh_access_token=DEFAULT) - @patch('spotipy.oauth2.open', create=True) + @patch('spotipy.cache_handler.open', create=True) def test_expired_token_refreshes(self, opener, is_token_expired, refresh_access_token): scope = "playlist-modify-private" @@ -375,7 +398,7 @@ class SpotifyPKCECacheTest(unittest.TestCase): refresh_access_token.return_value = fresh_tok spot = _make_pkceauth(scope, path) - spot.get_cached_token() + spot.validate_token(spot.cache_handler.get_cached_token()) is_token_expired.assert_called_with(expired_tok) refresh_access_token.assert_called_with(expired_tok['refresh_token']) @@ -383,7 +406,7 @@ class SpotifyPKCECacheTest(unittest.TestCase): @patch.multiple(SpotifyPKCE, is_token_expired=DEFAULT, refresh_access_token=DEFAULT) - @patch('spotipy.oauth2.open', create=True) + @patch('spotipy.cache_handler.open', create=True) def test_badly_scoped_token_bails(self, opener, is_token_expired, refresh_access_token): token_scope = "playlist-modify-public" @@ -395,13 +418,13 @@ class SpotifyPKCECacheTest(unittest.TestCase): is_token_expired.return_value = False spot = _make_pkceauth(requested_scope, path) - cached_tok = spot.get_cached_token() + cached_tok = spot.validate_token(spot.cache_handler.get_cached_token()) opener.assert_called_with(path) self.assertIsNone(cached_tok) self.assertEqual(refresh_access_token.call_count, 0) - @patch('spotipy.oauth2.open', create=True) + @patch('spotipy.cache_handler.open', create=True) def test_saves_to_cache_path(self, opener): scope = "playlist-modify-private" path = ".cache-username" @@ -411,7 +434,7 @@ class SpotifyPKCECacheTest(unittest.TestCase): opener.return_value = fi spot = SpotifyPKCE("CLID", "REDIR", "STATE", scope, path) - spot._save_token_info(tok) + spot.cache_handler.save_token_to_cache(tok) opener.assert_called_with(path, 'w') self.assertTrue(fi.write.called)