Stage 1 conversion to JSON for storing Credentials.
Reviewed in http://codereview.appspot.com/4972065/
diff --git a/tests/test_oauth2client.py b/tests/test_oauth2client.py
index de11810..14beb97 100644
--- a/tests/test_oauth2client.py
+++ b/tests/test_oauth2client.py
@@ -22,6 +22,7 @@
__author__ = 'jcgregorio@google.com (Joe Gregorio)'
+import datetime
import httplib2
import unittest
import urlparse
@@ -31,6 +32,16 @@
except ImportError:
from cgi import parse_qs
+try: # pragma: no cover
+ import simplejson
+except ImportError: # pragma: no cover
+ try:
+ # Try to import from django, should work on App Engine
+ from django.utils import simplejson
+ except ImportError:
+ # Should work for Python2.6 and higher.
+ import json as simplejson
+
from apiclient.http import HttpMockSequence
from oauth2client.client import AccessTokenCredentials
from oauth2client.client import AccessTokenCredentialsError
@@ -48,7 +59,7 @@
client_id = "some_client_id"
client_secret = "cOuDdkfjxxnv+"
refresh_token = "1/0/a.df219fjls0"
- token_expiry = "ignored"
+ token_expiry = datetime.datetime.utcnow()
token_uri = "https://www.google.com/accounts/o8/oauth2/token"
user_agent = "refresh_checker/1.0"
self.credentials = OAuth2Credentials(
@@ -86,6 +97,12 @@
resp, content = http.request("http://example.com")
self.assertEqual(400, resp.status)
+ def test_to_from_json(self):
+ json = self.credentials.to_json()
+ instance = OAuth2Credentials.from_json(json)
+ self.assertEquals(type(instance), OAuth2Credentials)
+ self.assertEquals(self.credentials.__dict__, instance.__dict__)
+
class AccessTokenCredentialsTests(unittest.TestCase):
diff --git a/tests/test_oauth2client_appengine.py b/tests/test_oauth2client_appengine.py
index f9b5094..9362220 100644
--- a/tests/test_oauth2client_appengine.py
+++ b/tests/test_oauth2client_appengine.py
@@ -173,7 +173,7 @@
self.assertEqual('code', q['response_type'][0])
self.assertEqual(False, self.decorator.has_credentials())
- # Now simulate the callback to /oauth2callback
+ # Now simulate the callback to /oauth2callback.
response = self.app.get('/oauth2callback', {
'code': 'foo_access_code',
'state': 'foo_path',
@@ -181,7 +181,7 @@
self.assertEqual('http://localhost/foo_path', response.headers['Location'])
self.assertEqual(None, self.decorator.credentials)
- # Now requesting the decorated path should work
+ # Now requesting the decorated path should work.
response = self.app.get('/foo_path')
self.assertEqual('200 OK', response.status)
self.assertEqual(True, self.decorator.has_credentials())
@@ -190,18 +190,18 @@
self.assertEqual('foo_access_token',
self.decorator.credentials.access_token)
- # Invalidate the stored Credentials
+ # Invalidate the stored Credentials.
self.decorator.credentials.invalid = True
self.decorator.credentials.store.put(self.decorator.credentials)
- # Invalid Credentials should start the OAuth dance again
+ # Invalid Credentials should start the OAuth dance again.
response = self.app.get('/foo_path')
self.assertTrue(response.status.startswith('302'))
q = parse_qs(response.headers['Location'].split('?', 1)[1])
self.assertEqual('http://localhost/oauth2callback', q['redirect_uri'][0])
def test_aware(self):
- # An initial request to an oauth_aware decorated path should not redirect
+ # An initial request to an oauth_aware decorated path should not redirect.
response = self.app.get('/bar_path')
self.assertEqual('Hello World!', response.body)
self.assertEqual('200 OK', response.status)
@@ -214,7 +214,7 @@
self.assertEqual('http://localhost/bar_path', q['state'][0])
self.assertEqual('code', q['response_type'][0])
- # Now simulate the callback to /oauth2callback
+ # Now simulate the callback to /oauth2callback.
url = self.decorator.authorize_url()
response = self.app.get('/oauth2callback', {
'code': 'foo_access_code',
@@ -223,7 +223,7 @@
self.assertEqual('http://localhost/bar_path', response.headers['Location'])
self.assertEqual(False, self.decorator.has_credentials())
- # Now requesting the decorated path will have credentials
+ # Now requesting the decorated path will have credentials.
response = self.app.get('/bar_path')
self.assertEqual('200 OK', response.status)
self.assertEqual('Hello World!', response.body)
diff --git a/tests/test_oauth2client_file.py b/tests/test_oauth2client_file.py
new file mode 100644
index 0000000..05deaa0
--- /dev/null
+++ b/tests/test_oauth2client_file.py
@@ -0,0 +1,157 @@
+#!/usr/bin/python2.4
+#
+# Copyright 2010 Google Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+"""Oauth2client.file tests
+
+Unit tests for oauth2client.file
+"""
+
+__author__ = 'jcgregorio@google.com (Joe Gregorio)'
+
+import os
+import pickle
+import unittest
+import datetime
+
+
+try: # pragma: no cover
+ import simplejson
+except ImportError: # pragma: no cover
+ try:
+ # Try to import from django, should work on App Engine
+ from django.utils import simplejson
+ except ImportError:
+ # Should work for Python2.6 and higher.
+ import json as simplejson
+
+
+from oauth2client.client import OAuth2Credentials
+from oauth2client.client import AccessTokenCredentials
+from oauth2client.client import AssertionCredentials
+from oauth2client.file import Storage
+from oauth2client import multistore_file
+
+
+FILENAME = os.path.join(os.path.dirname(__file__), 'test_file_storage.data')
+
+
+class OAuth2ClientFileTests(unittest.TestCase):
+
+ def tearDown(self):
+ try:
+ os.unlink(FILENAME)
+ except OSError:
+ pass
+
+ def setUp(self):
+ try:
+ os.unlink(FILENAME)
+ except OSError:
+ pass
+
+ def test_non_existent_file_storage(self):
+ s = Storage(FILENAME)
+ credentials = s.get()
+ self.assertEquals(None, credentials)
+
+ def test_pickle_and_json_interop(self):
+ # Write a file with a pickled OAuth2Credentials.
+ 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)
+
+ f = open(FILENAME, 'w')
+ pickle.dump(credentials, f)
+ f.close()
+
+ # Storage should be able to read that object.
+ # TODO(jcgregorio) This should fail once pickle support is removed.
+ s = Storage(FILENAME)
+ credentials = s.get()
+ self.assertNotEquals(None, credentials)
+ self.assertEquals('foo', credentials.access_token)
+
+ # Now write it back out and confirm it has been rewritten as JSON
+ s.put(credentials)
+ f = file(FILENAME)
+ data = simplejson.load(f)
+ f.close()
+
+ self.assertEquals(data['access_token'], 'foo')
+ self.assertEquals(data['_class'], 'OAuth2Credentials')
+ self.assertEquals(data['_module'], 'oauth2client.client')
+
+ def test_access_token_credentials(self):
+ access_token = 'foo'
+ user_agent = 'refresh_checker/1.0'
+
+ credentials = AccessTokenCredentials(access_token, user_agent)
+
+ s = Storage(FILENAME)
+ credentials = s.put(credentials)
+ credentials = s.get()
+
+ self.assertNotEquals(None, credentials)
+ self.assertEquals('foo', credentials.access_token)
+
+ def test_multistore_non_existent_file(self):
+ store = multistore_file.get_credential_storage(
+ FILENAME,
+ 'some_client_id',
+ 'user-agent/1.0',
+ 'some-scope')
+
+ credentials = store.get()
+ self.assertEquals(None, credentials)
+
+ def test_multistore_file(self):
+ access_token = 'foo'
+ 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'
+ client_id = 'some_client_id'
+
+ credentials = OAuth2Credentials(
+ access_token, client_id, client_secret,
+ refresh_token, token_expiry, token_uri,
+ user_agent)
+
+ store = multistore_file.get_credential_storage(
+ FILENAME,
+ credentials.client_id,
+ credentials.user_agent,
+ 'some-scope')
+
+ store.put(credentials)
+ credentials = store.get()
+
+ self.assertNotEquals(None, credentials)
+ self.assertEquals('foo', credentials.access_token)
+
+if __name__ == '__main__':
+ unittest.main()