Add .delete() to all Storages.
Reviewed in http://codereview.appspot.com/5608049/.
diff --git a/tests/test_oauth2client_appengine.py b/tests/test_oauth2client_appengine.py
index f2f470f..90ee335 100644
--- a/tests/test_oauth2client_appengine.py
+++ b/tests/test_oauth2client_appengine.py
@@ -203,6 +203,30 @@
q = parse_qs(response.headers['Location'].split('?', 1)[1])
self.assertEqual('http://localhost/oauth2callback', q['redirect_uri'][0])
+ def test_storage_delete(self):
+ # An initial request to an oauth_required decorated path should be a
+ # redirect to start the OAuth dance.
+ response = self.app.get('/foo_path')
+ self.assertTrue(response.status.startswith('302'))
+
+ # Now simulate the callback to /oauth2callback.
+ response = self.app.get('/oauth2callback', {
+ 'code': 'foo_access_code',
+ 'state': 'foo_path',
+ })
+ self.assertEqual('http://localhost/foo_path', response.headers['Location'])
+ self.assertEqual(None, self.decorator.credentials)
+
+ # Now requesting the decorated path should work.
+ response = self.app.get('/foo_path')
+
+ # Invalidate the stored Credentials.
+ self.decorator.credentials.store.delete()
+
+ # Invalid Credentials should start the OAuth dance again.
+ response = self.app.get('/foo_path')
+ self.assertTrue(response.status.startswith('302'))
+
def test_aware(self):
# An initial request to an oauth_aware decorated path should not redirect.
response = self.app.get('/bar_path')
diff --git a/tests/test_oauth2client_file.py b/tests/test_oauth2client_file.py
index 7ff1eeb..596a4b6 100644
--- a/tests/test_oauth2client_file.py
+++ b/tests/test_oauth2client_file.py
@@ -98,7 +98,6 @@
self.assertEquals(data['_module'], OAuth2Credentials.__module__)
def test_token_refresh(self):
- # Write a file with a pickled OAuth2Credentials.
access_token = 'foo'
client_id = 'some_client_id'
client_secret = 'cOuDdkfjxxnv+'
@@ -122,6 +121,27 @@
credentials._refresh(lambda x: x)
self.assertEquals(credentials.access_token, 'bar')
+ def test_credentials_delete(self):
+ access_token = 'foo'
+ client_id = 'some_client_id'
+ client_secret = 'cOuDdkfjxxnv+'
+ refresh_token = '1/0/a.df219fjls0'
+ token_expiry = datetime.datetime.utcnow()
+ token_uri = 'https://www.google.com/accounts/o8/oauth2/token'
+ user_agent = 'refresh_checker/1.0'
+
+ credentials = OAuth2Credentials(
+ access_token, client_id, client_secret,
+ refresh_token, token_expiry, token_uri,
+ user_agent)
+
+ s = Storage(FILENAME)
+ s.put(credentials)
+ credentials = s.get()
+ self.assertNotEquals(None, credentials)
+ s.delete()
+ credentials = s.get()
+ self.assertEquals(None, credentials)
def test_access_token_credentials(self):
access_token = 'foo'
@@ -205,6 +225,11 @@
self.assertNotEquals(None, credentials)
self.assertEquals('foo', credentials.access_token)
+ store.delete()
+ credentials = store.get()
+
+ self.assertEquals(None, credentials)
+
if os.name == 'posix':
self.assertEquals('0600', oct(stat.S_IMODE(os.stat(FILENAME).st_mode)))