Check whether a credential loaded from a store is expired before using it.

mattmoor · mattmoor · commit b930ddee0e66 · 2015-04-04T09:00:54.000-07:00
diff --git a/oauth2client/client.py b/oauth2client/client.py
@@ -756,7 +756,8 @@ def _refresh(self, http_request):
       try:
         new_cred = self.store.locked_get()
         if (new_cred and not new_cred.invalid and
-            new_cred.access_token != self.access_token):
+            new_cred.access_token != self.access_token and
+            not new_cred.access_token_expired):
           logger.info('Updated access_token read from Storage')
           self._updateFromCredential(new_cred)
         else:
diff --git a/tests/test_file.py b/tests/test_file.py
@@ -32,6 +32,7 @@
 import tempfile
 import unittest
 
+from http_mock import HttpMockSequence
 from oauth2client import GOOGLE_TOKEN_URI
 from oauth2client import file
 from oauth2client import locked_file
@@ -64,11 +65,12 @@ def setUp(self):
     except OSError:
       pass
 
-  def create_test_credentials(self, client_id='some_client_id'):
+  def create_test_credentials(self, client_id='some_client_id',
+                              expiration=None):
     access_token = 'foo'
     client_secret = 'cOuDdkfjxxnv+'
     refresh_token = '1/0/a.df219fjls0'
-    token_expiry = datetime.datetime.utcnow()
+    token_expiry = datetime.datetime.utcnow() if not expiration else expiration
     token_uri = 'https://www.google.com/accounts/o8/oauth2/token'
     user_agent = 'refresh_checker/1.0'
 
@@ -119,8 +121,29 @@ def test_pickle_and_json_interop(self):
     self.assertEquals(data['_class'], 'OAuth2Credentials')
     self.assertEquals(data['_module'], OAuth2Credentials.__module__)
 
-  def test_token_refresh(self):
-    credentials = self.create_test_credentials()
+  def test_token_refresh_store_expired(self):
+    expiration = datetime.datetime.utcnow() - datetime.timedelta(minutes=15)
+    credentials = self.create_test_credentials(expiration=expiration)
+
+    s = file.Storage(FILENAME)
+    s.put(credentials)
+    credentials = s.get()
+    new_cred = copy.copy(credentials)
+    new_cred.access_token = 'bar'
+    s.put(new_cred)
+
+    access_token = '1/3w'
+    token_response = {'access_token': access_token, 'expires_in': 3600}
+    http = HttpMockSequence([
+        ({'status': '200'}, json.dumps(token_response)),
+    ])
+
+    credentials._refresh(http.request)
+    self.assertEquals(credentials.access_token, access_token)
+
+  def test_token_refresh_good_store(self):
+    expiration = datetime.datetime.utcnow() + datetime.timedelta(minutes=15)
+    credentials = self.create_test_credentials(expiration=expiration)
 
     s = file.Storage(FILENAME)
     s.put(credentials)
