Remove duplicated oauth2client tests.
diff --git a/tests/test_oauth2client.py b/tests/test_oauth2client.py
deleted file mode 100644
index 738bfbe..0000000
--- a/tests/test_oauth2client.py
+++ /dev/null
@@ -1,624 +0,0 @@
-#!/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 tests
-
-Unit tests for oauth2client.
-"""
-
-__author__ = 'jcgregorio@google.com (Joe Gregorio)'
-
-import base64
-import datetime
-import httplib2
-import os
-import unittest
-import urlparse
-
-from googleapiclient.http import HttpMock
-from googleapiclient.http import HttpMockSequence
-from oauth2client import GOOGLE_REVOKE_URI
-from oauth2client import GOOGLE_TOKEN_URI
-from oauth2client.anyjson import simplejson
-from oauth2client.client import AccessTokenCredentials
-from oauth2client.client import AccessTokenCredentialsError
-from oauth2client.client import AccessTokenRefreshError
-from oauth2client.client import AssertionCredentials
-from oauth2client.client import Credentials
-from oauth2client.client import FlowExchangeError
-from oauth2client.client import MemoryCache
-from oauth2client.client import NonAsciiHeaderError
-from oauth2client.client import OAuth2Credentials
-from oauth2client.client import OAuth2WebServerFlow
-from oauth2client.client import OOB_CALLBACK_URN
-from oauth2client.client import REFRESH_STATUS_CODES
-from oauth2client.client import Storage
-from oauth2client.client import TokenRevokeError
-from oauth2client.client import VerifyJwtTokenError
-from oauth2client.client import _extract_id_token
-from oauth2client.client import _update_query_params
-from oauth2client.client import credentials_from_clientsecrets_and_code
-from oauth2client.client import credentials_from_code
-from oauth2client.client import flow_from_clientsecrets
-from oauth2client.clientsecrets import _loadfile
-from test_discovery import assertUrisEqual
-
-
-DATA_DIR = os.path.join(os.path.dirname(__file__), 'data')
-
-
-def datafile(filename):
- return os.path.join(DATA_DIR, filename)
-
-
-def load_and_cache(existing_file, fakename, cache_mock):
- client_type, client_info = _loadfile(datafile(existing_file))
- cache_mock.cache[fakename] = {client_type: client_info}
-
-
-class CacheMock(object):
- def __init__(self):
- self.cache = {}
-
- def get(self, key, namespace=''):
- # ignoring namespace for easier testing
- return self.cache.get(key, None)
-
- def set(self, key, value, namespace=''):
- # ignoring namespace for easier testing
- self.cache[key] = value
-
-
-class CredentialsTests(unittest.TestCase):
-
- def test_to_from_json(self):
- credentials = Credentials()
- json = credentials.to_json()
- restored = Credentials.new_from_json(json)
-
-
-class DummyDeleteStorage(Storage):
- delete_called = False
-
- def locked_delete(self):
- self.delete_called = True
-
-
-def _token_revoke_test_helper(testcase, status, revoke_raise,
- valid_bool_value, token_attr):
- current_store = getattr(testcase.credentials, 'store', None)
-
- dummy_store = DummyDeleteStorage()
- testcase.credentials.set_store(dummy_store)
-
- actual_do_revoke = testcase.credentials._do_revoke
- testcase.token_from_revoke = None
- def do_revoke_stub(http_request, token):
- testcase.token_from_revoke = token
- return actual_do_revoke(http_request, token)
- testcase.credentials._do_revoke = do_revoke_stub
-
- http = HttpMock(headers={'status': status})
- if revoke_raise:
- testcase.assertRaises(TokenRevokeError, testcase.credentials.revoke, http)
- else:
- testcase.credentials.revoke(http)
-
- testcase.assertEqual(getattr(testcase.credentials, token_attr),
- testcase.token_from_revoke)
- testcase.assertEqual(valid_bool_value, testcase.credentials.invalid)
- testcase.assertEqual(valid_bool_value, dummy_store.delete_called)
-
- testcase.credentials.set_store(current_store)
-
-
-class BasicCredentialsTests(unittest.TestCase):
-
- def setUp(self):
- access_token = 'foo'
- client_id = 'some_client_id'
- client_secret = 'cOuDdkfjxxnv+'
- refresh_token = '1/0/a.df219fjls0'
- token_expiry = datetime.datetime.utcnow()
- user_agent = 'refresh_checker/1.0'
- self.credentials = OAuth2Credentials(
- access_token, client_id, client_secret,
- refresh_token, token_expiry, GOOGLE_TOKEN_URI,
- user_agent, revoke_uri=GOOGLE_REVOKE_URI)
-
- def test_token_refresh_success(self):
- for status_code in REFRESH_STATUS_CODES:
- token_response = {'access_token': '1/3w', 'expires_in': 3600}
- http = HttpMockSequence([
- ({'status': status_code}, ''),
- ({'status': '200'}, simplejson.dumps(token_response)),
- ({'status': '200'}, 'echo_request_headers'),
- ])
- http = self.credentials.authorize(http)
- resp, content = http.request('http://example.com')
- self.assertEqual('Bearer 1/3w', content['Authorization'])
- self.assertFalse(self.credentials.access_token_expired)
- self.assertEqual(token_response, self.credentials.token_response)
-
- def test_token_refresh_failure(self):
- for status_code in REFRESH_STATUS_CODES:
- http = HttpMockSequence([
- ({'status': status_code}, ''),
- ({'status': '400'}, '{"error":"access_denied"}'),
- ])
- http = self.credentials.authorize(http)
- try:
- http.request('http://example.com')
- self.fail('should raise AccessTokenRefreshError exception')
- except AccessTokenRefreshError:
- pass
- self.assertTrue(self.credentials.access_token_expired)
- self.assertEqual(None, self.credentials.token_response)
-
- def test_token_revoke_success(self):
- _token_revoke_test_helper(
- self, '200', revoke_raise=False,
- valid_bool_value=True, token_attr='refresh_token')
-
- def test_token_revoke_failure(self):
- _token_revoke_test_helper(
- self, '400', revoke_raise=True,
- valid_bool_value=False, token_attr='refresh_token')
-
- def test_non_401_error_response(self):
- http = HttpMockSequence([
- ({'status': '400'}, ''),
- ])
- http = self.credentials.authorize(http)
- resp, content = http.request('http://example.com')
- self.assertEqual(400, resp.status)
- self.assertEqual(None, self.credentials.token_response)
-
- def test_to_from_json(self):
- json = self.credentials.to_json()
- instance = OAuth2Credentials.from_json(json)
- self.assertEqual(OAuth2Credentials, type(instance))
- instance.token_expiry = None
- self.credentials.token_expiry = None
-
- self.assertEqual(instance.__dict__, self.credentials.__dict__)
-
- def test_no_unicode_in_request_params(self):
- access_token = u'foo'
- client_id = u'some_client_id'
- client_secret = u'cOuDdkfjxxnv+'
- refresh_token = u'1/0/a.df219fjls0'
- token_expiry = unicode(datetime.datetime.utcnow())
- token_uri = unicode(GOOGLE_TOKEN_URI)
- revoke_uri = unicode(GOOGLE_REVOKE_URI)
- user_agent = u'refresh_checker/1.0'
- credentials = OAuth2Credentials(access_token, client_id, client_secret,
- refresh_token, token_expiry, token_uri,
- user_agent, revoke_uri=revoke_uri)
-
- http = HttpMock(headers={'status': '200'})
- http = credentials.authorize(http)
- http.request(u'http://example.com', method=u'GET', headers={u'foo': u'bar'})
- for k, v in http.headers.iteritems():
- self.assertEqual(str, type(k))
- self.assertEqual(str, type(v))
-
- # Test again with unicode strings that can't simple be converted to ASCII.
- try:
- http.request(
- u'http://example.com', method=u'GET', headers={u'foo': u'\N{COMET}'})
- self.fail('Expected exception to be raised.')
- except NonAsciiHeaderError:
- pass
-
- self.credentials.token_response = 'foobar'
- instance = OAuth2Credentials.from_json(self.credentials.to_json())
- self.assertEqual('foobar', instance.token_response)
-
-
-class AccessTokenCredentialsTests(unittest.TestCase):
-
- def setUp(self):
- access_token = 'foo'
- user_agent = 'refresh_checker/1.0'
- self.credentials = AccessTokenCredentials(access_token, user_agent,
- revoke_uri=GOOGLE_REVOKE_URI)
-
- def test_token_refresh_success(self):
- for status_code in REFRESH_STATUS_CODES:
- http = HttpMockSequence([
- ({'status': status_code}, ''),
- ])
- http = self.credentials.authorize(http)
- try:
- resp, content = http.request('http://example.com')
- self.fail('should throw exception if token expires')
- except AccessTokenCredentialsError:
- pass
- except Exception:
- self.fail('should only throw AccessTokenCredentialsError')
-
- def test_token_revoke_success(self):
- _token_revoke_test_helper(
- self, '200', revoke_raise=False,
- valid_bool_value=True, token_attr='access_token')
-
- def test_token_revoke_failure(self):
- _token_revoke_test_helper(
- self, '400', revoke_raise=True,
- valid_bool_value=False, token_attr='access_token')
-
- def test_non_401_error_response(self):
- http = HttpMockSequence([
- ({'status': '400'}, ''),
- ])
- http = self.credentials.authorize(http)
- resp, content = http.request('http://example.com')
- self.assertEqual(400, resp.status)
-
- def test_auth_header_sent(self):
- http = HttpMockSequence([
- ({'status': '200'}, 'echo_request_headers'),
- ])
- http = self.credentials.authorize(http)
- resp, content = http.request('http://example.com')
- self.assertEqual('Bearer foo', content['Authorization'])
-
-
-class TestAssertionCredentials(unittest.TestCase):
- assertion_text = 'This is the assertion'
- assertion_type = 'http://www.google.com/assertionType'
-
- class AssertionCredentialsTestImpl(AssertionCredentials):
-
- def _generate_assertion(self):
- return TestAssertionCredentials.assertion_text
-
- def setUp(self):
- user_agent = 'fun/2.0'
- self.credentials = self.AssertionCredentialsTestImpl(self.assertion_type,
- user_agent=user_agent)
-
- def test_assertion_body(self):
- body = urlparse.parse_qs(self.credentials._generate_refresh_request_body())
- self.assertEqual(self.assertion_text, body['assertion'][0])
- self.assertEqual('urn:ietf:params:oauth:grant-type:jwt-bearer',
- body['grant_type'][0])
-
- def test_assertion_refresh(self):
- http = HttpMockSequence([
- ({'status': '200'}, '{"access_token":"1/3w"}'),
- ({'status': '200'}, 'echo_request_headers'),
- ])
- http = self.credentials.authorize(http)
- resp, content = http.request('http://example.com')
- self.assertEqual('Bearer 1/3w', content['Authorization'])
-
- def test_token_revoke_success(self):
- _token_revoke_test_helper(
- self, '200', revoke_raise=False,
- valid_bool_value=True, token_attr='access_token')
-
- def test_token_revoke_failure(self):
- _token_revoke_test_helper(
- self, '400', revoke_raise=True,
- valid_bool_value=False, token_attr='access_token')
-
-
-class UpdateQueryParamsTest(unittest.TestCase):
- def test_update_query_params_no_params(self):
- uri = 'http://www.google.com'
- updated = _update_query_params(uri, {'a': 'b'})
- self.assertEqual(updated, uri + '?a=b')
-
- def test_update_query_params_existing_params(self):
- uri = 'http://www.google.com?x=y'
- updated = _update_query_params(uri, {'a': 'b', 'c': 'd&'})
- hardcoded_update = uri + '&a=b&c=d%26'
- assertUrisEqual(self, updated, hardcoded_update)
-
-
-class ExtractIdTokenTest(unittest.TestCase):
- """Tests _extract_id_token()."""
-
- def test_extract_success(self):
- body = {'foo': 'bar'}
- payload = base64.urlsafe_b64encode(simplejson.dumps(body)).strip('=')
- jwt = 'stuff.' + payload + '.signature'
-
- extracted = _extract_id_token(jwt)
- self.assertEqual(extracted, body)
-
- def test_extract_failure(self):
- body = {'foo': 'bar'}
- payload = base64.urlsafe_b64encode(simplejson.dumps(body)).strip('=')
- jwt = 'stuff.' + payload
-
- self.assertRaises(VerifyJwtTokenError, _extract_id_token, jwt)
-
-
-class OAuth2WebServerFlowTest(unittest.TestCase):
-
- def setUp(self):
- self.flow = OAuth2WebServerFlow(
- client_id='client_id+1',
- client_secret='secret+1',
- scope='foo',
- redirect_uri=OOB_CALLBACK_URN,
- user_agent='unittest-sample/1.0',
- revoke_uri='dummy_revoke_uri',
- )
-
- def test_construct_authorize_url(self):
- authorize_url = self.flow.step1_get_authorize_url()
-
- parsed = urlparse.urlparse(authorize_url)
- q = urlparse.parse_qs(parsed[4])
- self.assertEqual('client_id+1', q['client_id'][0])
- self.assertEqual('code', q['response_type'][0])
- self.assertEqual('foo', q['scope'][0])
- self.assertEqual(OOB_CALLBACK_URN, q['redirect_uri'][0])
- self.assertEqual('offline', q['access_type'][0])
-
- def test_override_flow_via_kwargs(self):
- """Passing kwargs to override defaults."""
- flow = OAuth2WebServerFlow(
- client_id='client_id+1',
- client_secret='secret+1',
- scope='foo',
- redirect_uri=OOB_CALLBACK_URN,
- user_agent='unittest-sample/1.0',
- access_type='online',
- response_type='token'
- )
- authorize_url = flow.step1_get_authorize_url()
-
- parsed = urlparse.urlparse(authorize_url)
- q = urlparse.parse_qs(parsed[4])
- self.assertEqual('client_id+1', q['client_id'][0])
- self.assertEqual('token', q['response_type'][0])
- self.assertEqual('foo', q['scope'][0])
- self.assertEqual(OOB_CALLBACK_URN, q['redirect_uri'][0])
- self.assertEqual('online', q['access_type'][0])
-
- def test_exchange_failure(self):
- http = HttpMockSequence([
- ({'status': '400'}, '{"error":"invalid_request"}'),
- ])
-
- try:
- credentials = self.flow.step2_exchange('some random code', http=http)
- self.fail('should raise exception if exchange doesn\'t get 200')
- except FlowExchangeError:
- pass
-
- def test_urlencoded_exchange_failure(self):
- http = HttpMockSequence([
- ({'status': '400'}, 'error=invalid_request'),
- ])
-
- try:
- credentials = self.flow.step2_exchange('some random code', http=http)
- self.fail('should raise exception if exchange doesn\'t get 200')
- except FlowExchangeError, e:
- self.assertEquals('invalid_request', str(e))
-
- def test_exchange_failure_with_json_error(self):
- # Some providers have 'error' attribute as a JSON object
- # in place of regular string.
- # This test makes sure no strange object-to-string coversion
- # exceptions are being raised instead of FlowExchangeError.
- http = HttpMockSequence([
- ({'status': '400'},
- """ {"error": {
- "type": "OAuthException",
- "message": "Error validating verification code."} }"""),
- ])
-
- try:
- credentials = self.flow.step2_exchange('some random code', http=http)
- self.fail('should raise exception if exchange doesn\'t get 200')
- except FlowExchangeError, e:
- pass
-
- def test_exchange_success(self):
- http = HttpMockSequence([
- ({'status': '200'},
- """{ "access_token":"SlAV32hkKG",
- "expires_in":3600,
- "refresh_token":"8xLOxBtZp8" }"""),
- ])
-
- credentials = self.flow.step2_exchange('some random code', http=http)
- self.assertEqual('SlAV32hkKG', credentials.access_token)
- self.assertNotEqual(None, credentials.token_expiry)
- self.assertEqual('8xLOxBtZp8', credentials.refresh_token)
- self.assertEqual('dummy_revoke_uri', credentials.revoke_uri)
-
- def test_urlencoded_exchange_success(self):
- http = HttpMockSequence([
- ({'status': '200'}, 'access_token=SlAV32hkKG&expires_in=3600'),
- ])
-
- credentials = self.flow.step2_exchange('some random code', http=http)
- self.assertEqual('SlAV32hkKG', credentials.access_token)
- self.assertNotEqual(None, credentials.token_expiry)
-
- def test_urlencoded_expires_param(self):
- http = HttpMockSequence([
- # Note the 'expires=3600' where you'd normally
- # have if named 'expires_in'
- ({'status': '200'}, 'access_token=SlAV32hkKG&expires=3600'),
- ])
-
- credentials = self.flow.step2_exchange('some random code', http=http)
- self.assertNotEqual(None, credentials.token_expiry)
-
- def test_exchange_no_expires_in(self):
- http = HttpMockSequence([
- ({'status': '200'}, """{ "access_token":"SlAV32hkKG",
- "refresh_token":"8xLOxBtZp8" }"""),
- ])
-
- credentials = self.flow.step2_exchange('some random code', http=http)
- self.assertEqual(None, credentials.token_expiry)
-
- def test_urlencoded_exchange_no_expires_in(self):
- http = HttpMockSequence([
- # This might be redundant but just to make sure
- # urlencoded access_token gets parsed correctly
- ({'status': '200'}, 'access_token=SlAV32hkKG'),
- ])
-
- credentials = self.flow.step2_exchange('some random code', http=http)
- self.assertEqual(None, credentials.token_expiry)
-
- def test_exchange_fails_if_no_code(self):
- http = HttpMockSequence([
- ({'status': '200'}, """{ "access_token":"SlAV32hkKG",
- "refresh_token":"8xLOxBtZp8" }"""),
- ])
-
- code = {'error': 'thou shall not pass'}
- try:
- credentials = self.flow.step2_exchange(code, http=http)
- self.fail('should raise exception if no code in dictionary.')
- except FlowExchangeError, e:
- self.assertTrue('shall not pass' in str(e))
-
- def test_exchange_id_token_fail(self):
- http = HttpMockSequence([
- ({'status': '200'}, """{ "access_token":"SlAV32hkKG",
- "refresh_token":"8xLOxBtZp8",
- "id_token": "stuff.payload"}"""),
- ])
-
- self.assertRaises(VerifyJwtTokenError, self.flow.step2_exchange,
- 'some random code', http=http)
-
- def test_exchange_id_token_fail(self):
- body = {'foo': 'bar'}
- payload = base64.urlsafe_b64encode(simplejson.dumps(body)).strip('=')
- jwt = (base64.urlsafe_b64encode('stuff')+ '.' + payload + '.' +
- base64.urlsafe_b64encode('signature'))
-
- http = HttpMockSequence([
- ({'status': '200'}, """{ "access_token":"SlAV32hkKG",
- "refresh_token":"8xLOxBtZp8",
- "id_token": "%s"}""" % jwt),
- ])
-
- credentials = self.flow.step2_exchange('some random code', http=http)
- self.assertEqual(credentials.id_token, body)
-
-
-class FlowFromCachedClientsecrets(unittest.TestCase):
-
- def test_flow_from_clientsecrets_cached(self):
- cache_mock = CacheMock()
- load_and_cache('client_secrets.json', 'some_secrets', cache_mock)
-
- flow = flow_from_clientsecrets(
- 'some_secrets', '', redirect_uri='oob', cache=cache_mock)
- self.assertEquals('foo_client_secret', flow.client_secret)
-
-
-class CredentialsFromCodeTests(unittest.TestCase):
- def setUp(self):
- self.client_id = 'client_id_abc'
- self.client_secret = 'secret_use_code'
- self.scope = 'foo'
- self.code = '12345abcde'
- self.redirect_uri = 'postmessage'
-
- def test_exchange_code_for_token(self):
- token = 'asdfghjkl'
- payload =simplejson.dumps({'access_token': token, 'expires_in': 3600})
- http = HttpMockSequence([
- ({'status': '200'}, payload),
- ])
- credentials = credentials_from_code(self.client_id, self.client_secret,
- self.scope, self.code, redirect_uri=self.redirect_uri,
- http=http)
- self.assertEquals(credentials.access_token, token)
- self.assertNotEqual(None, credentials.token_expiry)
-
- def test_exchange_code_for_token_fail(self):
- http = HttpMockSequence([
- ({'status': '400'}, '{"error":"invalid_request"}'),
- ])
-
- try:
- credentials = credentials_from_code(self.client_id, self.client_secret,
- self.scope, self.code, redirect_uri=self.redirect_uri,
- http=http)
- self.fail('should raise exception if exchange doesn\'t get 200')
- except FlowExchangeError:
- pass
-
- def test_exchange_code_and_file_for_token(self):
- http = HttpMockSequence([
- ({'status': '200'},
- """{ "access_token":"asdfghjkl",
- "expires_in":3600 }"""),
- ])
- credentials = credentials_from_clientsecrets_and_code(
- datafile('client_secrets.json'), self.scope,
- self.code, http=http)
- self.assertEquals(credentials.access_token, 'asdfghjkl')
- self.assertNotEqual(None, credentials.token_expiry)
-
- def test_exchange_code_and_cached_file_for_token(self):
- http = HttpMockSequence([
- ({'status': '200'}, '{ "access_token":"asdfghjkl"}'),
- ])
- cache_mock = CacheMock()
- load_and_cache('client_secrets.json', 'some_secrets', cache_mock)
-
- credentials = credentials_from_clientsecrets_and_code(
- 'some_secrets', self.scope,
- self.code, http=http, cache=cache_mock)
- self.assertEquals(credentials.access_token, 'asdfghjkl')
-
- def test_exchange_code_and_file_for_token_fail(self):
- http = HttpMockSequence([
- ({'status': '400'}, '{"error":"invalid_request"}'),
- ])
-
- try:
- credentials = credentials_from_clientsecrets_and_code(
- datafile('client_secrets.json'), self.scope,
- self.code, http=http)
- self.fail('should raise exception if exchange doesn\'t get 200')
- except FlowExchangeError:
- pass
-
-
-class MemoryCacheTests(unittest.TestCase):
-
- def test_get_set_delete(self):
- m = MemoryCache()
- self.assertEqual(None, m.get('foo'))
- self.assertEqual(None, m.delete('foo'))
- m.set('foo', 'bar')
- self.assertEqual('bar', m.get('foo'))
- m.delete('foo')
- self.assertEqual(None, m.get('foo'))
-
-
-if __name__ == '__main__':
- unittest.main()
diff --git a/tests/test_oauth2client_appengine.py b/tests/test_oauth2client_appengine.py
deleted file mode 100644
index 2c3ce75..0000000
--- a/tests/test_oauth2client_appengine.py
+++ /dev/null
@@ -1,866 +0,0 @@
-#!/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.
-
-
-"""Discovery document tests
-
-Unit tests for objects created from discovery documents.
-"""
-
-__author__ = 'jcgregorio@google.com (Joe Gregorio)'
-
-import base64
-import datetime
-import httplib2
-import mox
-import os
-import time
-import unittest
-import urllib
-
-try:
- from urlparse import parse_qs
-except ImportError:
- from cgi import parse_qs
-
-import dev_appserver
-dev_appserver.fix_sys_path()
-import webapp2
-
-from googleapiclient.http import HttpMockSequence
-from google.appengine.api import apiproxy_stub
-from google.appengine.api import apiproxy_stub_map
-from google.appengine.api import app_identity
-from google.appengine.api import memcache
-from google.appengine.api import users
-from google.appengine.api.memcache import memcache_stub
-from google.appengine.ext import db
-from google.appengine.ext import ndb
-from google.appengine.ext import testbed
-from google.appengine.runtime import apiproxy_errors
-from oauth2client import appengine
-from oauth2client import GOOGLE_TOKEN_URI
-from oauth2client.anyjson import simplejson
-from oauth2client.clientsecrets import _loadfile
-from oauth2client.clientsecrets import InvalidClientSecretsError
-from oauth2client.appengine import AppAssertionCredentials
-from oauth2client.appengine import CredentialsModel
-from oauth2client.appengine import CredentialsNDBModel
-from oauth2client.appengine import FlowNDBProperty
-from oauth2client.appengine import FlowProperty
-from oauth2client.appengine import OAuth2Decorator
-from oauth2client.appengine import StorageByKeyName
-from oauth2client.appengine import oauth2decorator_from_clientsecrets
-from oauth2client.client import AccessTokenRefreshError
-from oauth2client.client import Credentials
-from oauth2client.client import FlowExchangeError
-from oauth2client.client import OAuth2Credentials
-from oauth2client.client import flow_from_clientsecrets
-from webtest import TestApp
-
-
-DATA_DIR = os.path.join(os.path.dirname(__file__), 'data')
-
-
-def datafile(filename):
- return os.path.join(DATA_DIR, filename)
-
-
-def load_and_cache(existing_file, fakename, cache_mock):
- client_type, client_info = _loadfile(datafile(existing_file))
- cache_mock.cache[fakename] = {client_type: client_info}
-
-
-class CacheMock(object):
- def __init__(self):
- self.cache = {}
-
- def get(self, key, namespace=''):
- # ignoring namespace for easier testing
- return self.cache.get(key, None)
-
- def set(self, key, value, namespace=''):
- # ignoring namespace for easier testing
- self.cache[key] = value
-
-
-class UserMock(object):
- """Mock the app engine user service"""
-
- def __call__(self):
- return self
-
- def user_id(self):
- return 'foo_user'
-
-
-class UserNotLoggedInMock(object):
- """Mock the app engine user service"""
-
- def __call__(self):
- return None
-
-
-class Http2Mock(object):
- """Mock httplib2.Http"""
- status = 200
- content = {
- 'access_token': 'foo_access_token',
- 'refresh_token': 'foo_refresh_token',
- 'expires_in': 3600,
- 'extra': 'value',
- }
-
- def request(self, token_uri, method, body, headers, *args, **kwargs):
- self.body = body
- self.headers = headers
- return (self, simplejson.dumps(self.content))
-
-
-class TestAppAssertionCredentials(unittest.TestCase):
- account_name = "service_account_name@appspot.com"
- signature = "signature"
-
-
- class AppIdentityStubImpl(apiproxy_stub.APIProxyStub):
-
- def __init__(self):
- super(TestAppAssertionCredentials.AppIdentityStubImpl, self).__init__(
- 'app_identity_service')
-
- def _Dynamic_GetAccessToken(self, request, response):
- response.set_access_token('a_token_123')
- response.set_expiration_time(time.time() + 1800)
-
-
- class ErroringAppIdentityStubImpl(apiproxy_stub.APIProxyStub):
-
- def __init__(self):
- super(TestAppAssertionCredentials.ErroringAppIdentityStubImpl, self).__init__(
- 'app_identity_service')
-
- def _Dynamic_GetAccessToken(self, request, response):
- raise app_identity.BackendDeadlineExceeded()
-
- def test_raise_correct_type_of_exception(self):
- app_identity_stub = self.ErroringAppIdentityStubImpl()
- apiproxy_stub_map.apiproxy = apiproxy_stub_map.APIProxyStubMap()
- apiproxy_stub_map.apiproxy.RegisterStub('app_identity_service',
- app_identity_stub)
- apiproxy_stub_map.apiproxy.RegisterStub(
- 'memcache', memcache_stub.MemcacheServiceStub())
-
- scope = 'http://www.googleapis.com/scope'
- try:
- credentials = AppAssertionCredentials(scope)
- http = httplib2.Http()
- credentials.refresh(http)
- self.fail('Should have raised an AccessTokenRefreshError')
- except AccessTokenRefreshError:
- pass
-
- def test_get_access_token_on_refresh(self):
- app_identity_stub = self.AppIdentityStubImpl()
- apiproxy_stub_map.apiproxy = apiproxy_stub_map.APIProxyStubMap()
- apiproxy_stub_map.apiproxy.RegisterStub("app_identity_service",
- app_identity_stub)
- apiproxy_stub_map.apiproxy.RegisterStub(
- 'memcache', memcache_stub.MemcacheServiceStub())
-
- scope = [
- "http://www.googleapis.com/scope",
- "http://www.googleapis.com/scope2"]
- credentials = AppAssertionCredentials(scope)
- http = httplib2.Http()
- credentials.refresh(http)
- self.assertEqual('a_token_123', credentials.access_token)
-
- json = credentials.to_json()
- credentials = Credentials.new_from_json(json)
- self.assertEqual(
- 'http://www.googleapis.com/scope http://www.googleapis.com/scope2',
- credentials.scope)
-
- scope = "http://www.googleapis.com/scope http://www.googleapis.com/scope2"
- credentials = AppAssertionCredentials(scope)
- http = httplib2.Http()
- credentials.refresh(http)
- self.assertEqual('a_token_123', credentials.access_token)
- self.assertEqual(
- 'http://www.googleapis.com/scope http://www.googleapis.com/scope2',
- credentials.scope)
-
- def test_custom_service_account(self):
- scope = "http://www.googleapis.com/scope"
- account_id = "service_account_name_2@appspot.com"
- m = mox.Mox()
- m.StubOutWithMock(app_identity, 'get_access_token')
- app_identity.get_access_token(
- [scope], service_account_id=account_id).AndReturn(('a_token_456', None))
- m.ReplayAll()
-
- credentials = AppAssertionCredentials(scope, service_account_id=account_id)
- http = httplib2.Http()
- credentials.refresh(http)
- m.VerifyAll()
- m.UnsetStubs()
- self.assertEqual('a_token_456', credentials.access_token)
- self.assertEqual(scope, credentials.scope)
-
-
-class TestFlowModel(db.Model):
- flow = FlowProperty()
-
-
-class FlowPropertyTest(unittest.TestCase):
-
- def setUp(self):
- self.testbed = testbed.Testbed()
- self.testbed.activate()
- self.testbed.init_datastore_v3_stub()
-
- def tearDown(self):
- self.testbed.deactivate()
-
- def test_flow_get_put(self):
- instance = TestFlowModel(
- flow=flow_from_clientsecrets(datafile('client_secrets.json'), 'foo',
- redirect_uri='oob'),
- key_name='foo'
- )
- instance.put()
- retrieved = TestFlowModel.get_by_key_name('foo')
-
- self.assertEqual('foo_client_id', retrieved.flow.client_id)
-
-
-class TestFlowNDBModel(ndb.Model):
- flow = FlowNDBProperty()
-
-
-class FlowNDBPropertyTest(unittest.TestCase):
-
- def setUp(self):
- self.testbed = testbed.Testbed()
- self.testbed.activate()
- self.testbed.init_datastore_v3_stub()
- self.testbed.init_memcache_stub()
-
- def tearDown(self):
- self.testbed.deactivate()
-
- def test_flow_get_put(self):
- instance = TestFlowNDBModel(
- flow=flow_from_clientsecrets(datafile('client_secrets.json'), 'foo',
- redirect_uri='oob'),
- id='foo'
- )
- instance.put()
- retrieved = TestFlowNDBModel.get_by_id('foo')
-
- self.assertEqual('foo_client_id', retrieved.flow.client_id)
-
-
-def _http_request(*args, **kwargs):
- resp = httplib2.Response({'status': '200'})
- content = simplejson.dumps({'access_token': 'bar'})
-
- return resp, content
-
-
-class StorageByKeyNameTest(unittest.TestCase):
-
- def setUp(self):
- self.testbed = testbed.Testbed()
- self.testbed.activate()
- self.testbed.init_datastore_v3_stub()
- self.testbed.init_memcache_stub()
- self.testbed.init_user_stub()
-
- access_token = 'foo'
- client_id = 'some_client_id'
- client_secret = 'cOuDdkfjxxnv+'
- refresh_token = '1/0/a.df219fjls0'
- token_expiry = datetime.datetime.utcnow()
- user_agent = 'refresh_checker/1.0'
- self.credentials = OAuth2Credentials(
- access_token, client_id, client_secret,
- refresh_token, token_expiry, GOOGLE_TOKEN_URI,
- user_agent)
-
- def tearDown(self):
- self.testbed.deactivate()
-
- def test_get_and_put_simple(self):
- storage = StorageByKeyName(
- CredentialsModel, 'foo', 'credentials')
-
- self.assertEqual(None, storage.get())
- self.credentials.set_store(storage)
-
- self.credentials._refresh(_http_request)
- credmodel = CredentialsModel.get_by_key_name('foo')
- self.assertEqual('bar', credmodel.credentials.access_token)
-
- def test_get_and_put_cached(self):
- storage = StorageByKeyName(
- CredentialsModel, 'foo', 'credentials', cache=memcache)
-
- self.assertEqual(None, storage.get())
- self.credentials.set_store(storage)
-
- self.credentials._refresh(_http_request)
- credmodel = CredentialsModel.get_by_key_name('foo')
- self.assertEqual('bar', credmodel.credentials.access_token)
-
- # Now remove the item from the cache.
- memcache.delete('foo')
-
- # Check that getting refreshes the cache.
- credentials = storage.get()
- self.assertEqual('bar', credentials.access_token)
- self.assertNotEqual(None, memcache.get('foo'))
-
- # Deleting should clear the cache.
- storage.delete()
- credentials = storage.get()
- self.assertEqual(None, credentials)
- self.assertEqual(None, memcache.get('foo'))
-
- def test_get_and_put_set_store_on_cache_retrieval(self):
- storage = StorageByKeyName(
- CredentialsModel, 'foo', 'credentials', cache=memcache)
-
- self.assertEqual(None, storage.get())
- self.credentials.set_store(storage)
- storage.put(self.credentials)
- # Pre-bug 292 old_creds wouldn't have storage, and the _refresh wouldn't
- # be able to store the updated cred back into the storage.
- old_creds = storage.get()
- self.assertEqual(old_creds.access_token, 'foo')
- old_creds.invalid = True
- old_creds._refresh(_http_request)
- new_creds = storage.get()
- self.assertEqual(new_creds.access_token, 'bar')
-
- def test_get_and_put_ndb(self):
- # Start empty
- storage = StorageByKeyName(
- CredentialsNDBModel, 'foo', 'credentials')
- self.assertEqual(None, storage.get())
-
- # Refresh storage and retrieve without using storage
- self.credentials.set_store(storage)
- self.credentials._refresh(_http_request)
- credmodel = CredentialsNDBModel.get_by_id('foo')
- self.assertEqual('bar', credmodel.credentials.access_token)
- self.assertEqual(credmodel.credentials.to_json(),
- self.credentials.to_json())
-
- def test_delete_ndb(self):
- # Start empty
- storage = StorageByKeyName(
- CredentialsNDBModel, 'foo', 'credentials')
- self.assertEqual(None, storage.get())
-
- # Add credentials to model with storage, and check equivalent w/o storage
- storage.put(self.credentials)
- credmodel = CredentialsNDBModel.get_by_id('foo')
- self.assertEqual(credmodel.credentials.to_json(),
- self.credentials.to_json())
-
- # Delete and make sure empty
- storage.delete()
- self.assertEqual(None, storage.get())
-
- def test_get_and_put_mixed_ndb_storage_db_get(self):
- # Start empty
- storage = StorageByKeyName(
- CredentialsNDBModel, 'foo', 'credentials')
- self.assertEqual(None, storage.get())
-
- # Set NDB store and refresh to add to storage
- self.credentials.set_store(storage)
- self.credentials._refresh(_http_request)
-
- # Retrieve same key from DB model to confirm mixing works
- credmodel = CredentialsModel.get_by_key_name('foo')
- self.assertEqual('bar', credmodel.credentials.access_token)
- self.assertEqual(self.credentials.to_json(),
- credmodel.credentials.to_json())
-
- def test_get_and_put_mixed_db_storage_ndb_get(self):
- # Start empty
- storage = StorageByKeyName(
- CredentialsModel, 'foo', 'credentials')
- self.assertEqual(None, storage.get())
-
- # Set DB store and refresh to add to storage
- self.credentials.set_store(storage)
- self.credentials._refresh(_http_request)
-
- # Retrieve same key from NDB model to confirm mixing works
- credmodel = CredentialsNDBModel.get_by_id('foo')
- self.assertEqual('bar', credmodel.credentials.access_token)
- self.assertEqual(self.credentials.to_json(),
- credmodel.credentials.to_json())
-
- def test_delete_db_ndb_mixed(self):
- # Start empty
- storage_ndb = StorageByKeyName(
- CredentialsNDBModel, 'foo', 'credentials')
- storage = StorageByKeyName(
- CredentialsModel, 'foo', 'credentials')
-
- # First DB, then NDB
- self.assertEqual(None, storage.get())
- storage.put(self.credentials)
- self.assertNotEqual(None, storage.get())
-
- storage_ndb.delete()
- self.assertEqual(None, storage.get())
-
- # First NDB, then DB
- self.assertEqual(None, storage_ndb.get())
- storage_ndb.put(self.credentials)
-
- storage.delete()
- self.assertNotEqual(None, storage_ndb.get())
- # NDB uses memcache and an instance cache (Context)
- ndb.get_context().clear_cache()
- memcache.flush_all()
- self.assertEqual(None, storage_ndb.get())
-
-
-class MockRequest(object):
- url = 'https://example.org'
-
- def relative_url(self, rel):
- return self.url + rel
-
-
-class MockRequestHandler(object):
- request = MockRequest()
-
-
-class DecoratorTests(unittest.TestCase):
-
- def setUp(self):
- self.testbed = testbed.Testbed()
- self.testbed.activate()
- self.testbed.init_datastore_v3_stub()
- self.testbed.init_memcache_stub()
- self.testbed.init_user_stub()
-
- decorator = OAuth2Decorator(client_id='foo_client_id',
- client_secret='foo_client_secret',
- scope=['foo_scope', 'bar_scope'],
- user_agent='foo')
-
- self._finish_setup(decorator, user_mock=UserMock)
-
- def _finish_setup(self, decorator, user_mock):
- self.decorator = decorator
- self.had_credentials = False
- self.found_credentials = None
- self.should_raise = False
- parent = self
-
- class TestRequiredHandler(webapp2.RequestHandler):
- @decorator.oauth_required
- def get(self):
- if decorator.has_credentials():
- parent.had_credentials = True
- parent.found_credentials = decorator.credentials
- if parent.should_raise:
- raise Exception('')
-
- class TestAwareHandler(webapp2.RequestHandler):
- @decorator.oauth_aware
- def get(self, *args, **kwargs):
- self.response.out.write('Hello World!')
- assert(kwargs['year'] == '2012')
- assert(kwargs['month'] == '01')
- if decorator.has_credentials():
- parent.had_credentials = True
- parent.found_credentials = decorator.credentials
- if parent.should_raise:
- raise Exception('')
-
-
- application = webapp2.WSGIApplication([
- ('/oauth2callback', self.decorator.callback_handler()),
- ('/foo_path', TestRequiredHandler),
- webapp2.Route(r'/bar_path/<year:\d{4}>/<month:\d{2}>',
- handler=TestAwareHandler, name='bar')],
- debug=True)
- self.app = TestApp(application, extra_environ={
- 'wsgi.url_scheme': 'http',
- 'HTTP_HOST': 'localhost',
- })
- users.get_current_user = user_mock()
- self.httplib2_orig = httplib2.Http
- httplib2.Http = Http2Mock
-
- def tearDown(self):
- self.testbed.deactivate()
- httplib2.Http = self.httplib2_orig
-
- def test_required(self):
- # An initial request to an oauth_required decorated path should be a
- # redirect to start the OAuth dance.
- self.assertEqual(self.decorator.flow, None)
- self.assertEqual(self.decorator.credentials, None)
- response = self.app.get('http://localhost/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])
- self.assertEqual('foo_client_id', q['client_id'][0])
- self.assertEqual('foo_scope bar_scope', q['scope'][0])
- self.assertEqual('http://localhost/foo_path',
- q['state'][0].rsplit(':', 1)[0])
- self.assertEqual('code', q['response_type'][0])
- self.assertEqual(False, self.decorator.has_credentials())
-
- m = mox.Mox()
- m.StubOutWithMock(appengine, '_parse_state_value')
- appengine._parse_state_value('foo_path:xsrfkey123',
- mox.IgnoreArg()).AndReturn('foo_path')
- m.ReplayAll()
-
- # Now simulate the callback to /oauth2callback.
- response = self.app.get('/oauth2callback', {
- 'code': 'foo_access_code',
- 'state': 'foo_path:xsrfkey123',
- })
- parts = response.headers['Location'].split('?', 1)
- self.assertEqual('http://localhost/foo_path', parts[0])
- self.assertEqual(None, self.decorator.credentials)
- if self.decorator._token_response_param:
- response = parse_qs(parts[1])[self.decorator._token_response_param][0]
- self.assertEqual(Http2Mock.content,
- simplejson.loads(urllib.unquote(response)))
- self.assertEqual(self.decorator.flow, self.decorator._tls.flow)
- self.assertEqual(self.decorator.credentials,
- self.decorator._tls.credentials)
-
- m.UnsetStubs()
- m.VerifyAll()
-
- # Now requesting the decorated path should work.
- response = self.app.get('/foo_path')
- self.assertEqual('200 OK', response.status)
- self.assertEqual(True, self.had_credentials)
- self.assertEqual('foo_refresh_token',
- self.found_credentials.refresh_token)
- self.assertEqual('foo_access_token',
- self.found_credentials.access_token)
- self.assertEqual(None, self.decorator.credentials)
-
- # Raising an exception still clears the Credentials.
- self.should_raise = True
- try:
- response = self.app.get('/foo_path')
- self.fail('Should have raised an exception.')
- except Exception:
- pass
- self.assertEqual(None, self.decorator.credentials)
- self.should_raise = False
-
- # Invalidate the stored Credentials.
- self.found_credentials.invalid = True
- self.found_credentials.store.put(self.found_credentials)
-
- # 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_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'))
-
- m = mox.Mox()
- m.StubOutWithMock(appengine, '_parse_state_value')
- appengine._parse_state_value('foo_path:xsrfkey123',
- mox.IgnoreArg()).AndReturn('foo_path')
- m.ReplayAll()
-
- # Now simulate the callback to /oauth2callback.
- response = self.app.get('/oauth2callback', {
- 'code': 'foo_access_code',
- 'state': 'foo_path:xsrfkey123',
- })
- 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')
-
- self.assertTrue(self.had_credentials)
-
- # Credentials should be cleared after each call.
- self.assertEqual(None, self.decorator.credentials)
-
- # Invalidate the stored Credentials.
- self.found_credentials.store.delete()
-
- # Invalid Credentials should start the OAuth dance again.
- response = self.app.get('/foo_path')
- self.assertTrue(response.status.startswith('302'))
-
- m.UnsetStubs()
- m.VerifyAll()
-
- def test_aware(self):
- # An initial request to an oauth_aware decorated path should not redirect.
- response = self.app.get('http://localhost/bar_path/2012/01')
- self.assertEqual('Hello World!', response.body)
- self.assertEqual('200 OK', response.status)
- self.assertEqual(False, self.decorator.has_credentials())
- url = self.decorator.authorize_url()
- q = parse_qs(url.split('?', 1)[1])
- self.assertEqual('http://localhost/oauth2callback', q['redirect_uri'][0])
- self.assertEqual('foo_client_id', q['client_id'][0])
- self.assertEqual('foo_scope bar_scope', q['scope'][0])
- self.assertEqual('http://localhost/bar_path/2012/01',
- q['state'][0].rsplit(':', 1)[0])
- self.assertEqual('code', q['response_type'][0])
-
- m = mox.Mox()
- m.StubOutWithMock(appengine, '_parse_state_value')
- appengine._parse_state_value('bar_path:xsrfkey456',
- mox.IgnoreArg()).AndReturn('bar_path')
- m.ReplayAll()
-
- # Now simulate the callback to /oauth2callback.
- url = self.decorator.authorize_url()
- response = self.app.get('/oauth2callback', {
- 'code': 'foo_access_code',
- 'state': 'bar_path:xsrfkey456',
- })
- self.assertEqual('http://localhost/bar_path', response.headers['Location'])
- self.assertEqual(False, self.decorator.has_credentials())
-
- m.UnsetStubs()
- m.VerifyAll()
-
- # Now requesting the decorated path will have credentials.
- response = self.app.get('/bar_path/2012/01')
- self.assertEqual('200 OK', response.status)
- self.assertEqual('Hello World!', response.body)
- self.assertEqual(True, self.had_credentials)
- self.assertEqual('foo_refresh_token',
- self.found_credentials.refresh_token)
- self.assertEqual('foo_access_token',
- self.found_credentials.access_token)
-
- # Credentials should be cleared after each call.
- self.assertEqual(None, self.decorator.credentials)
-
- # Raising an exception still clears the Credentials.
- self.should_raise = True
- try:
- response = self.app.get('/bar_path/2012/01')
- self.fail('Should have raised an exception.')
- except Exception:
- pass
- self.assertEqual(None, self.decorator.credentials)
- self.should_raise = False
-
-
- def test_error_in_step2(self):
- # An initial request to an oauth_aware decorated path should not redirect.
- response = self.app.get('/bar_path/2012/01')
- url = self.decorator.authorize_url()
- response = self.app.get('/oauth2callback', {
- 'error': 'Bad<Stuff>Happened\''
- })
- self.assertEqual('200 OK', response.status)
- self.assertTrue('Bad<Stuff>Happened'' in response.body)
-
- def test_kwargs_are_passed_to_underlying_flow(self):
- decorator = OAuth2Decorator(client_id='foo_client_id',
- client_secret='foo_client_secret',
- user_agent='foo_user_agent',
- scope=['foo_scope', 'bar_scope'],
- access_type='offline',
- approval_prompt='force',
- revoke_uri='dummy_revoke_uri')
- request_handler = MockRequestHandler()
- decorator._create_flow(request_handler)
-
- self.assertEqual('https://example.org/oauth2callback',
- decorator.flow.redirect_uri)
- self.assertEqual('offline', decorator.flow.params['access_type'])
- self.assertEqual('force', decorator.flow.params['approval_prompt'])
- self.assertEqual('foo_user_agent', decorator.flow.user_agent)
- self.assertEqual('dummy_revoke_uri', decorator.flow.revoke_uri)
- self.assertEqual(None, decorator.flow.params.get('user_agent', None))
- self.assertEqual(decorator.flow, decorator._tls.flow)
-
- def test_token_response_param(self):
- self.decorator._token_response_param = 'foobar'
- self.test_required()
-
- def test_decorator_from_client_secrets(self):
- decorator = oauth2decorator_from_clientsecrets(
- datafile('client_secrets.json'),
- scope=['foo_scope', 'bar_scope'])
- self._finish_setup(decorator, user_mock=UserMock)
-
- self.assertFalse(decorator._in_error)
- self.decorator = decorator
- self.test_required()
- http = self.decorator.http()
- self.assertEquals('foo_access_token', http.request.credentials.access_token)
-
- # revoke_uri is not required
- self.assertEqual(self.decorator._revoke_uri,
- 'https://accounts.google.com/o/oauth2/revoke')
- self.assertEqual(self.decorator._revoke_uri,
- self.decorator.credentials.revoke_uri)
-
- def test_decorator_from_cached_client_secrets(self):
- cache_mock = CacheMock()
- load_and_cache('client_secrets.json', 'secret', cache_mock)
- decorator = oauth2decorator_from_clientsecrets(
- # filename, scope, message=None, cache=None
- 'secret', '', cache=cache_mock)
- self.assertFalse(decorator._in_error)
-
- def test_decorator_from_client_secrets_not_logged_in_required(self):
- decorator = oauth2decorator_from_clientsecrets(
- datafile('client_secrets.json'),
- scope=['foo_scope', 'bar_scope'], message='NotLoggedInMessage')
- self.decorator = decorator
- self._finish_setup(decorator, user_mock=UserNotLoggedInMock)
-
- self.assertFalse(decorator._in_error)
-
- # An initial request to an oauth_required decorated path should be a
- # redirect to login.
- response = self.app.get('/foo_path')
- self.assertTrue(response.status.startswith('302'))
- self.assertTrue('Login' in str(response))
-
- def test_decorator_from_client_secrets_not_logged_in_aware(self):
- decorator = oauth2decorator_from_clientsecrets(
- datafile('client_secrets.json'),
- scope=['foo_scope', 'bar_scope'], message='NotLoggedInMessage')
- self.decorator = decorator
- self._finish_setup(decorator, user_mock=UserNotLoggedInMock)
-
- # An initial request to an oauth_aware decorated path should be a
- # redirect to login.
- response = self.app.get('/bar_path/2012/03')
- self.assertTrue(response.status.startswith('302'))
- self.assertTrue('Login' in str(response))
-
- def test_decorator_from_unfilled_client_secrets_required(self):
- MESSAGE = 'File is missing'
- try:
- decorator = oauth2decorator_from_clientsecrets(
- datafile('unfilled_client_secrets.json'),
- scope=['foo_scope', 'bar_scope'], message=MESSAGE)
- except InvalidClientSecretsError:
- pass
-
- def test_decorator_from_unfilled_client_secrets_aware(self):
- MESSAGE = 'File is missing'
- try:
- decorator = oauth2decorator_from_clientsecrets(
- datafile('unfilled_client_secrets.json'),
- scope=['foo_scope', 'bar_scope'], message=MESSAGE)
- except InvalidClientSecretsError:
- pass
-
-
-class DecoratorXsrfSecretTests(unittest.TestCase):
- """Test xsrf_secret_key."""
-
- def setUp(self):
- self.testbed = testbed.Testbed()
- self.testbed.activate()
- self.testbed.init_datastore_v3_stub()
- self.testbed.init_memcache_stub()
-
- def tearDown(self):
- self.testbed.deactivate()
-
- def test_build_and_parse_state(self):
- secret = appengine.xsrf_secret_key()
-
- # Secret shouldn't change from call to call.
- secret2 = appengine.xsrf_secret_key()
- self.assertEqual(secret, secret2)
-
- # Secret shouldn't change if memcache goes away.
- memcache.delete(appengine.XSRF_MEMCACHE_ID,
- namespace=appengine.OAUTH2CLIENT_NAMESPACE)
- secret3 = appengine.xsrf_secret_key()
- self.assertEqual(secret2, secret3)
-
- # Secret should change if both memcache and the model goes away.
- memcache.delete(appengine.XSRF_MEMCACHE_ID,
- namespace=appengine.OAUTH2CLIENT_NAMESPACE)
- model = appengine.SiteXsrfSecretKey.get_or_insert('site')
- model.delete()
-
- secret4 = appengine.xsrf_secret_key()
- self.assertNotEqual(secret3, secret4)
-
- def test_ndb_insert_db_get(self):
- secret = appengine._generate_new_xsrf_secret_key()
- appengine.SiteXsrfSecretKeyNDB(id='site', secret=secret).put()
-
- site_key = appengine.SiteXsrfSecretKey.get_by_key_name('site')
- self.assertEqual(site_key.secret, secret)
-
- def test_db_insert_ndb_get(self):
- secret = appengine._generate_new_xsrf_secret_key()
- appengine.SiteXsrfSecretKey(key_name='site', secret=secret).put()
-
- site_key = appengine.SiteXsrfSecretKeyNDB.get_by_id('site')
- self.assertEqual(site_key.secret, secret)
-
-
-class DecoratorXsrfProtectionTests(unittest.TestCase):
- """Test _build_state_value and _parse_state_value."""
-
- def setUp(self):
- self.testbed = testbed.Testbed()
- self.testbed.activate()
- self.testbed.init_datastore_v3_stub()
- self.testbed.init_memcache_stub()
-
- def tearDown(self):
- self.testbed.deactivate()
-
- def test_build_and_parse_state(self):
- state = appengine._build_state_value(MockRequestHandler(), UserMock())
- self.assertEqual(
- 'https://example.org',
- appengine._parse_state_value(state, UserMock()))
- self.assertRaises(appengine.InvalidXsrfTokenError,
- appengine._parse_state_value, state[1:], UserMock())
-
-
-if __name__ == '__main__':
- unittest.main()
diff --git a/tests/test_oauth2client_clientsecrets.py b/tests/test_oauth2client_clientsecrets.py
deleted file mode 100644
index f69fb36..0000000
--- a/tests/test_oauth2client_clientsecrets.py
+++ /dev/null
@@ -1,145 +0,0 @@
-# Copyright 2011 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.
-
-"""Unit tests for oauth2client.clientsecrets."""
-
-__author__ = 'jcgregorio@google.com (Joe Gregorio)'
-
-
-import os
-import unittest
-import StringIO
-
-import httplib2
-
-from oauth2client import clientsecrets
-
-
-DATA_DIR = os.path.join(os.path.dirname(__file__), 'data')
-VALID_FILE = os.path.join(DATA_DIR, 'client_secrets.json')
-INVALID_FILE = os.path.join(DATA_DIR, 'unfilled_client_secrets.json')
-NONEXISTENT_FILE = os.path.join(__file__, '..', 'afilethatisntthere.json')
-
-class OAuth2CredentialsTests(unittest.TestCase):
-
- def setUp(self):
- pass
-
- def tearDown(self):
- pass
-
- def test_validate_error(self):
- ERRORS = [
- ('{}', 'Invalid'),
- ('{"foo": {}}', 'Unknown'),
- ('{"web": {}}', 'Missing'),
- ('{"web": {"client_id": "dkkd"}}', 'Missing'),
- ("""{
- "web": {
- "client_id": "[[CLIENT ID REQUIRED]]",
- "client_secret": "[[CLIENT SECRET REQUIRED]]",
- "redirect_uris": ["http://localhost:8080/oauth2callback"],
- "auth_uri": "",
- "token_uri": ""
- }
- }
- """, 'Property'),
- ]
- for src, match in ERRORS:
- # Test load(s)
- try:
- clientsecrets.loads(src)
- self.fail(src + ' should not be a valid client_secrets file.')
- except clientsecrets.InvalidClientSecretsError, e:
- self.assertTrue(str(e).startswith(match))
-
- # Test loads(fp)
- try:
- fp = StringIO.StringIO(src)
- clientsecrets.load(fp)
- self.fail(src + ' should not be a valid client_secrets file.')
- except clientsecrets.InvalidClientSecretsError, e:
- self.assertTrue(str(e).startswith(match))
-
- def test_load_by_filename(self):
- try:
- clientsecrets._loadfile(NONEXISTENT_FILE)
- self.fail('should fail to load a missing client_secrets file.')
- except clientsecrets.InvalidClientSecretsError, e:
- self.assertTrue(str(e).startswith('File'))
-
-
-class CachedClientsecretsTests(unittest.TestCase):
-
- class CacheMock(object):
- def __init__(self):
- self.cache = {}
- self.last_get_ns = None
- self.last_set_ns = None
-
- def get(self, key, namespace=''):
- # ignoring namespace for easier testing
- self.last_get_ns = namespace
- return self.cache.get(key, None)
-
- def set(self, key, value, namespace=''):
- # ignoring namespace for easier testing
- self.last_set_ns = namespace
- self.cache[key] = value
-
- def setUp(self):
- self.cache_mock = self.CacheMock()
-
- def test_cache_miss(self):
- client_type, client_info = clientsecrets.loadfile(
- VALID_FILE, cache=self.cache_mock)
- self.assertEquals('web', client_type)
- self.assertEquals('foo_client_secret', client_info['client_secret'])
-
- cached = self.cache_mock.cache[VALID_FILE]
- self.assertEquals({client_type: client_info}, cached)
-
- # make sure we're using non-empty namespace
- ns = self.cache_mock.last_set_ns
- self.assertTrue(bool(ns))
- # make sure they're equal
- self.assertEquals(ns, self.cache_mock.last_get_ns)
-
- def test_cache_hit(self):
- self.cache_mock.cache[NONEXISTENT_FILE] = { 'web': 'secret info' }
-
- client_type, client_info = clientsecrets.loadfile(
- NONEXISTENT_FILE, cache=self.cache_mock)
- self.assertEquals('web', client_type)
- self.assertEquals('secret info', client_info)
- # make sure we didn't do any set() RPCs
- self.assertEqual(None, self.cache_mock.last_set_ns)
-
- def test_validation(self):
- try:
- clientsecrets.loadfile(INVALID_FILE, cache=self.cache_mock)
- self.fail('Expected InvalidClientSecretsError to be raised '
- 'while loading %s' % INVALID_FILE)
- except clientsecrets.InvalidClientSecretsError:
- pass
-
- def test_without_cache(self):
- # this also ensures loadfile() is backward compatible
- client_type, client_info = clientsecrets.loadfile(VALID_FILE)
- self.assertEquals('web', client_type)
- self.assertEquals('foo_client_secret', client_info['client_secret'])
-
-
-if __name__ == '__main__':
- unittest.main()
diff --git a/tests/test_oauth2client_django_orm.py b/tests/test_oauth2client_django_orm.py
deleted file mode 100644
index c80ff1d..0000000
--- a/tests/test_oauth2client_django_orm.py
+++ /dev/null
@@ -1,86 +0,0 @@
-#!/usr/bin/python2.4
-#
-# Copyright 2011 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.
-
-
-"""Discovery document tests
-
-Unit tests for objects created from discovery documents.
-"""
-
-__author__ = 'conleyo@google.com (Conley Owens)'
-
-import base64
-import imp
-import os
-import pickle
-import sys
-import unittest
-
-# Ensure that if app engine is available, we use the correct django from it
-try:
- from google.appengine.dist import use_library
- use_library('django', '1.2')
-except ImportError:
- pass
-
-from oauth2client.client import Credentials
-from oauth2client.client import Flow
-
-# Mock a Django environment
-os.environ['DJANGO_SETTINGS_MODULE'] = 'django_settings'
-sys.modules['django_settings'] = imp.new_module('django_settings')
-
-from oauth2client.django_orm import CredentialsField
-from oauth2client.django_orm import FlowField
-
-
-class TestCredentialsField(unittest.TestCase):
- def setUp(self):
- self.field = CredentialsField()
- self.credentials = Credentials()
- self.pickle = base64.b64encode(pickle.dumps(self.credentials))
-
- def test_field_is_text(self):
- self.assertEquals(self.field.get_internal_type(), 'TextField')
-
- def test_field_unpickled(self):
- self.assertTrue(isinstance(self.field.to_python(self.pickle), Credentials))
-
- def test_field_pickled(self):
- prep_value = self.field.get_db_prep_value(self.credentials,
- connection=None)
- self.assertEqual(prep_value, self.pickle)
-
-
-class TestFlowField(unittest.TestCase):
- def setUp(self):
- self.field = FlowField()
- self.flow = Flow()
- self.pickle = base64.b64encode(pickle.dumps(self.flow))
-
- def test_field_is_text(self):
- self.assertEquals(self.field.get_internal_type(), 'TextField')
-
- def test_field_unpickled(self):
- self.assertTrue(isinstance(self.field.to_python(self.pickle), Flow))
-
- def test_field_pickled(self):
- prep_value = self.field.get_db_prep_value(self.flow, connection=None)
- self.assertEqual(prep_value, self.pickle)
-
-
-if __name__ == '__main__':
- unittest.main()
diff --git a/tests/test_oauth2client_file.py b/tests/test_oauth2client_file.py
deleted file mode 100644
index 910466a..0000000
--- a/tests/test_oauth2client_file.py
+++ /dev/null
@@ -1,323 +0,0 @@
-#!/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 copy
-import datetime
-import httplib2
-import os
-import pickle
-import stat
-import tempfile
-import unittest
-
-from googleapiclient.http import HttpMockSequence
-from oauth2client import GOOGLE_TOKEN_URI
-from oauth2client import file
-from oauth2client import locked_file
-from oauth2client import multistore_file
-from oauth2client import util
-from oauth2client.anyjson import simplejson
-from oauth2client.client import AccessTokenCredentials
-from oauth2client.client import AssertionCredentials
-from oauth2client.client import OAuth2Credentials
-
-
-FILENAME = tempfile.mktemp('oauth2client_test.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 create_test_credentials(self, client_id='some_client_id'):
- 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'
-
- credentials = OAuth2Credentials(
- access_token, client_id, client_secret,
- refresh_token, token_expiry, token_uri,
- user_agent)
- return credentials
-
- def test_non_existent_file_storage(self):
- s = file.Storage(FILENAME)
- credentials = s.get()
- self.assertEquals(None, credentials)
-
- def test_no_sym_link_credentials(self):
- if hasattr(os, 'symlink'):
- SYMFILENAME = FILENAME + '.sym'
- os.symlink(FILENAME, SYMFILENAME)
- s = file.Storage(SYMFILENAME)
- try:
- s.get()
- self.fail('Should have raised an exception.')
- except file.CredentialsFileSymbolicLinkError:
- pass
- finally:
- os.unlink(SYMFILENAME)
-
- def test_pickle_and_json_interop(self):
- # Write a file with a pickled OAuth2Credentials.
- credentials = self.create_test_credentials()
-
- f = open(FILENAME, 'w')
- pickle.dump(credentials, f)
- f.close()
-
- # Storage should be not be able to read that object, as the capability to
- # read and write credentials as pickled objects has been removed.
- s = file.Storage(FILENAME)
- read_credentials = s.get()
- self.assertEquals(None, read_credentials)
-
- # Now write it back out and confirm it has been rewritten as JSON
- s.put(credentials)
- f = open(FILENAME)
- data = simplejson.load(f)
- f.close()
-
- self.assertEquals(data['access_token'], 'foo')
- self.assertEquals(data['_class'], 'OAuth2Credentials')
- self.assertEquals(data['_module'], OAuth2Credentials.__module__)
-
- def test_token_refresh(self):
- credentials = self.create_test_credentials()
-
- s = file.Storage(FILENAME)
- s.put(credentials)
- credentials = s.get()
- new_cred = copy.copy(credentials)
- new_cred.access_token = 'bar'
- s.put(new_cred)
-
- credentials._refresh(lambda x: x)
- self.assertEquals(credentials.access_token, 'bar')
-
- def test_credentials_delete(self):
- credentials = self.create_test_credentials()
-
- s = file.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'
- user_agent = 'refresh_checker/1.0'
-
- credentials = AccessTokenCredentials(access_token, user_agent)
-
- s = file.Storage(FILENAME)
- credentials = s.put(credentials)
- credentials = s.get()
-
- self.assertNotEquals(None, credentials)
- self.assertEquals('foo', credentials.access_token)
- mode = os.stat(FILENAME).st_mode
-
- if os.name == 'posix':
- self.assertEquals('0600', oct(stat.S_IMODE(os.stat(FILENAME).st_mode)))
-
- def test_read_only_file_fail_lock(self):
- credentials = self.create_test_credentials()
-
- open(FILENAME, 'a+b').close()
- os.chmod(FILENAME, 0400)
-
- store = multistore_file.get_credential_storage(
- FILENAME,
- credentials.client_id,
- credentials.user_agent,
- ['some-scope', 'some-other-scope'])
-
- store.put(credentials)
- if os.name == 'posix':
- self.assertTrue(store._multistore._read_only)
- os.chmod(FILENAME, 0600)
-
- def test_multistore_no_symbolic_link_files(self):
- if hasattr(os, 'symlink'):
- SYMFILENAME = FILENAME + 'sym'
- os.symlink(FILENAME, SYMFILENAME)
- store = multistore_file.get_credential_storage(
- SYMFILENAME,
- 'some_client_id',
- 'user-agent/1.0',
- ['some-scope', 'some-other-scope'])
- try:
- store.get()
- self.fail('Should have raised an exception.')
- except locked_file.CredentialsFileSymbolicLinkError:
- pass
- finally:
- os.unlink(SYMFILENAME)
-
- def test_multistore_non_existent_file(self):
- store = multistore_file.get_credential_storage(
- FILENAME,
- 'some_client_id',
- 'user-agent/1.0',
- ['some-scope', 'some-other-scope'])
-
- credentials = store.get()
- self.assertEquals(None, credentials)
-
- def test_multistore_file(self):
- credentials = self.create_test_credentials()
-
- store = multistore_file.get_credential_storage(
- FILENAME,
- credentials.client_id,
- credentials.user_agent,
- ['some-scope', 'some-other-scope'])
-
- store.put(credentials)
- credentials = store.get()
-
- 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)))
-
- def test_multistore_file_custom_key(self):
- credentials = self.create_test_credentials()
-
- custom_key = {'myapp': 'testing', 'clientid': 'some client'}
- store = multistore_file.get_credential_storage_custom_key(
- FILENAME, custom_key)
-
- store.put(credentials)
- stored_credentials = store.get()
-
- self.assertNotEquals(None, stored_credentials)
- self.assertEqual(credentials.access_token, stored_credentials.access_token)
-
- store.delete()
- stored_credentials = store.get()
-
- self.assertEquals(None, stored_credentials)
-
- def test_multistore_file_custom_string_key(self):
- credentials = self.create_test_credentials()
-
- # store with string key
- store = multistore_file.get_credential_storage_custom_string_key(
- FILENAME, 'mykey')
-
- store.put(credentials)
- stored_credentials = store.get()
-
- self.assertNotEquals(None, stored_credentials)
- self.assertEqual(credentials.access_token, stored_credentials.access_token)
-
- # try retrieving with a dictionary
- store_dict = multistore_file.get_credential_storage_custom_string_key(
- FILENAME, {'key': 'mykey'})
- stored_credentials = store.get()
- self.assertNotEquals(None, stored_credentials)
- self.assertEqual(credentials.access_token, stored_credentials.access_token)
-
- store.delete()
- stored_credentials = store.get()
-
- self.assertEquals(None, stored_credentials)
-
- def test_multistore_file_backwards_compatibility(self):
- credentials = self.create_test_credentials()
- scopes = ['scope1', 'scope2']
-
- # store the credentials using the legacy key method
- store = multistore_file.get_credential_storage(
- FILENAME, 'client_id', 'user_agent', scopes)
- store.put(credentials)
-
- # retrieve the credentials using a custom key that matches the legacy key
- key = {'clientId': 'client_id', 'userAgent': 'user_agent',
- 'scope': util.scopes_to_string(scopes)}
- store = multistore_file.get_credential_storage_custom_key(FILENAME, key)
- stored_credentials = store.get()
-
- self.assertEqual(credentials.access_token, stored_credentials.access_token)
-
-
- def test_multistore_file_get_all_keys(self):
- # start with no keys
- keys = multistore_file.get_all_credential_keys(FILENAME)
- self.assertEquals([], keys)
-
- # store credentials
- credentials = self.create_test_credentials(client_id='client1')
- custom_key = {'myapp': 'testing', 'clientid': 'client1'}
- store1 = multistore_file.get_credential_storage_custom_key(
- FILENAME, custom_key)
- store1.put(credentials)
-
- keys = multistore_file.get_all_credential_keys(FILENAME)
- self.assertEquals([custom_key], keys)
-
- # store more credentials
- credentials = self.create_test_credentials(client_id='client2')
- string_key = 'string_key'
- store2 = multistore_file.get_credential_storage_custom_string_key(
- FILENAME, string_key)
- store2.put(credentials)
-
- keys = multistore_file.get_all_credential_keys(FILENAME)
- self.assertEquals(2, len(keys))
- self.assertTrue(custom_key in keys)
- self.assertTrue({'key': string_key} in keys)
-
- # back to no keys
- store1.delete()
- store2.delete()
- keys = multistore_file.get_all_credential_keys(FILENAME)
- self.assertEquals([], keys)
-
-
-if __name__ == '__main__':
- unittest.main()
diff --git a/tests/test_oauth2client_gce.py b/tests/test_oauth2client_gce.py
deleted file mode 100644
index 15e45c4..0000000
--- a/tests/test_oauth2client_gce.py
+++ /dev/null
@@ -1,92 +0,0 @@
-# Copyright 2012 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.
-
-
-"""Tests for oauth2client.gce.
-
-Unit tests for oauth2client.gce.
-"""
-
-__author__ = 'jcgregorio@google.com (Joe Gregorio)'
-
-import unittest
-import mox
-
-from oauth2client.client import AccessTokenRefreshError
-from oauth2client.client import Credentials
-from oauth2client.gce import AppAssertionCredentials
-
-
-class AssertionCredentialsTests(unittest.TestCase):
-
- def test_good_refresh(self):
- m = mox.Mox()
-
- httplib2_response = m.CreateMock(object)
- httplib2_response.status = 200
-
- httplib2_request = m.CreateMock(object)
- httplib2_request.__call__(
- ('http://metadata.google.internal/0.1/meta-data/service-accounts/'
- 'default/acquire'
- '?scope=http%3A%2F%2Fexample.com%2Fa%20http%3A%2F%2Fexample.com%2Fb'
- )).AndReturn((httplib2_response, '{"accessToken": "this-is-a-token"}'))
-
- m.ReplayAll()
-
- c = AppAssertionCredentials(scope=['http://example.com/a',
- 'http://example.com/b'])
-
- c._refresh(httplib2_request)
-
- self.assertEquals('this-is-a-token', c.access_token)
-
- m.UnsetStubs()
- m.VerifyAll()
-
-
- def test_fail_refresh(self):
- m = mox.Mox()
-
- httplib2_response = m.CreateMock(object)
- httplib2_response.status = 400
-
- httplib2_request = m.CreateMock(object)
- httplib2_request.__call__(
- ('http://metadata.google.internal/0.1/meta-data/service-accounts/'
- 'default/acquire'
- '?scope=http%3A%2F%2Fexample.com%2Fa%20http%3A%2F%2Fexample.com%2Fb'
- )).AndReturn((httplib2_response, '{"accessToken": "this-is-a-token"}'))
-
- m.ReplayAll()
-
- c = AppAssertionCredentials(scope=['http://example.com/a',
- 'http://example.com/b'])
-
- try:
- c._refresh(httplib2_request)
- self.fail('Should have raised exception on 400')
- except AccessTokenRefreshError:
- pass
-
- m.UnsetStubs()
- m.VerifyAll()
-
- def test_to_from_json(self):
- c = AppAssertionCredentials(scope=['http://example.com/a',
- 'http://example.com/b'])
- json = c.to_json()
- c2 = Credentials.new_from_json(json)
-
- self.assertEqual(c.access_token, c2.access_token)
diff --git a/tests/test_oauth2client_jwt.py b/tests/test_oauth2client_jwt.py
deleted file mode 100644
index f8cf00b..0000000
--- a/tests/test_oauth2client_jwt.py
+++ /dev/null
@@ -1,329 +0,0 @@
-#!/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 tests
-
-Unit tests for oauth2client.
-"""
-
-__author__ = 'jcgregorio@google.com (Joe Gregorio)'
-
-import httplib2
-import os
-import sys
-import tempfile
-import time
-import unittest
-import urlparse
-
-try:
- from urlparse import parse_qs
-except ImportError:
- from cgi import parse_qs
-
-from googleapiclient.http import HttpMockSequence
-from oauth2client import crypt
-from oauth2client.anyjson import simplejson
-from oauth2client.client import Credentials
-from oauth2client.client import SignedJwtAssertionCredentials
-from oauth2client.client import VerifyJwtTokenError
-from oauth2client.client import verify_id_token
-from oauth2client.client import HAS_OPENSSL
-from oauth2client.client import HAS_CRYPTO
-from oauth2client.file import Storage
-
-
-def datafile(filename):
- f = open(os.path.join(os.path.dirname(__file__), 'data', filename), 'r')
- data = f.read()
- f.close()
- return data
-
-
-class CryptTests(unittest.TestCase):
- def setUp(self):
- self.format = 'p12'
- self.signer = crypt.OpenSSLSigner
- self.verifier = crypt.OpenSSLVerifier
-
- def test_sign_and_verify(self):
- self._check_sign_and_verify('privatekey.%s' % self.format)
-
- def test_sign_and_verify_from_converted_pkcs12(self):
- """Tests that following instructions to convert from PKCS12 to PEM works."""
- if self.format == 'pem':
- self._check_sign_and_verify('pem_from_pkcs12.pem')
-
- def _check_sign_and_verify(self, private_key_file):
- private_key = datafile(private_key_file)
- public_key = datafile('publickey.pem')
-
- signer = self.signer.from_string(private_key)
- signature = signer.sign('foo')
-
- verifier = self.verifier.from_string(public_key, True)
-
- self.assertTrue(verifier.verify('foo', signature))
-
- self.assertFalse(verifier.verify('bar', signature))
- self.assertFalse(verifier.verify('foo', 'bad signagure'))
-
- def _check_jwt_failure(self, jwt, expected_error):
- try:
- public_key = datafile('publickey.pem')
- certs = {'foo': public_key}
- audience = 'https://www.googleapis.com/auth/id?client_id=' + \
- 'external_public_key@testing.gserviceaccount.com'
- contents = crypt.verify_signed_jwt_with_certs(jwt, certs, audience)
- self.fail('Should have thrown for %s' % jwt)
- except:
- e = sys.exc_info()[1]
- msg = e.args[0]
- self.assertTrue(expected_error in msg)
-
- def _create_signed_jwt(self):
- private_key = datafile('privatekey.%s' % self.format)
- signer = self.signer.from_string(private_key)
- audience = 'some_audience_address@testing.gserviceaccount.com'
- now = long(time.time())
-
- return crypt.make_signed_jwt(
- signer,
- {
- 'aud': audience,
- 'iat': now,
- 'exp': now + 300,
- 'user': 'billy bob',
- 'metadata': {'meta': 'data'},
- })
-
- def test_verify_id_token(self):
- jwt = self._create_signed_jwt()
- public_key = datafile('publickey.pem')
- certs = {'foo': public_key }
- audience = 'some_audience_address@testing.gserviceaccount.com'
- contents = crypt.verify_signed_jwt_with_certs(jwt, certs, audience)
- self.assertEqual('billy bob', contents['user'])
- self.assertEqual('data', contents['metadata']['meta'])
-
- def test_verify_id_token_with_certs_uri(self):
- jwt = self._create_signed_jwt()
-
- http = HttpMockSequence([
- ({'status': '200'}, datafile('certs.json')),
- ])
-
- contents = verify_id_token(jwt,
- 'some_audience_address@testing.gserviceaccount.com', http=http)
- self.assertEqual('billy bob', contents['user'])
- self.assertEqual('data', contents['metadata']['meta'])
-
- def test_verify_id_token_with_certs_uri_fails(self):
- jwt = self._create_signed_jwt()
-
- http = HttpMockSequence([
- ({'status': '404'}, datafile('certs.json')),
- ])
-
- self.assertRaises(VerifyJwtTokenError, verify_id_token, jwt,
- 'some_audience_address@testing.gserviceaccount.com', http=http)
-
- def test_verify_id_token_bad_tokens(self):
- private_key = datafile('privatekey.%s' % self.format)
-
- # Wrong number of segments
- self._check_jwt_failure('foo', 'Wrong number of segments')
-
- # Not json
- self._check_jwt_failure('foo.bar.baz',
- 'Can\'t parse token')
-
- # Bad signature
- jwt = 'foo.%s.baz' % crypt._urlsafe_b64encode('{"a":"b"}')
- self._check_jwt_failure(jwt, 'Invalid token signature')
-
- # No expiration
- signer = self.signer.from_string(private_key)
- audience = 'https:#www.googleapis.com/auth/id?client_id=' + \
- 'external_public_key@testing.gserviceaccount.com'
- jwt = crypt.make_signed_jwt(signer, {
- 'aud': 'audience',
- 'iat': time.time(),
- }
- )
- self._check_jwt_failure(jwt, 'No exp field in token')
-
- # No issued at
- jwt = crypt.make_signed_jwt(signer, {
- 'aud': 'audience',
- 'exp': time.time() + 400,
- }
- )
- self._check_jwt_failure(jwt, 'No iat field in token')
-
- # Too early
- jwt = crypt.make_signed_jwt(signer, {
- 'aud': 'audience',
- 'iat': time.time() + 301,
- 'exp': time.time() + 400,
- })
- self._check_jwt_failure(jwt, 'Token used too early')
-
- # Too late
- jwt = crypt.make_signed_jwt(signer, {
- 'aud': 'audience',
- 'iat': time.time() - 500,
- 'exp': time.time() - 301,
- })
- self._check_jwt_failure(jwt, 'Token used too late')
-
- # Wrong target
- jwt = crypt.make_signed_jwt(signer, {
- 'aud': 'somebody else',
- 'iat': time.time(),
- 'exp': time.time() + 300,
- })
- self._check_jwt_failure(jwt, 'Wrong recipient')
-
-
-class PEMCryptTestsPyCrypto(CryptTests):
- def setUp(self):
- self.format = 'pem'
- self.signer = crypt.PyCryptoSigner
- self.verifier = crypt.OpenSSLVerifier
-
-
-class PEMCryptTestsOpenSSL(CryptTests):
- def setUp(self):
- self.format = 'pem'
- self.signer = crypt.OpenSSLSigner
- self.verifier = crypt.OpenSSLVerifier
-
-
-class SignedJwtAssertionCredentialsTests(unittest.TestCase):
- def setUp(self):
- self.format = 'p12'
- crypt.Signer = crypt.OpenSSLSigner
-
- def test_credentials_good(self):
- private_key = datafile('privatekey.%s' % self.format)
- credentials = SignedJwtAssertionCredentials(
- 'some_account@example.com',
- private_key,
- scope='read+write',
- sub='joe@example.org')
- http = HttpMockSequence([
- ({'status': '200'}, '{"access_token":"1/3w","expires_in":3600}'),
- ({'status': '200'}, 'echo_request_headers'),
- ])
- http = credentials.authorize(http)
- resp, content = http.request('http://example.org')
- self.assertEqual('Bearer 1/3w', content['Authorization'])
-
- def test_credentials_to_from_json(self):
- private_key = datafile('privatekey.%s' % self.format)
- credentials = SignedJwtAssertionCredentials(
- 'some_account@example.com',
- private_key,
- scope='read+write',
- sub='joe@example.org')
- json = credentials.to_json()
- restored = Credentials.new_from_json(json)
- self.assertEqual(credentials.private_key, restored.private_key)
- self.assertEqual(credentials.private_key_password,
- restored.private_key_password)
- self.assertEqual(credentials.kwargs, restored.kwargs)
-
- def _credentials_refresh(self, credentials):
- http = HttpMockSequence([
- ({'status': '200'}, '{"access_token":"1/3w","expires_in":3600}'),
- ({'status': '401'}, ''),
- ({'status': '200'}, '{"access_token":"3/3w","expires_in":3600}'),
- ({'status': '200'}, 'echo_request_headers'),
- ])
- http = credentials.authorize(http)
- resp, content = http.request('http://example.org')
- return content
-
- def test_credentials_refresh_without_storage(self):
- private_key = datafile('privatekey.%s' % self.format)
- credentials = SignedJwtAssertionCredentials(
- 'some_account@example.com',
- private_key,
- scope='read+write',
- sub='joe@example.org')
-
- content = self._credentials_refresh(credentials)
-
- self.assertEqual('Bearer 3/3w', content['Authorization'])
-
- def test_credentials_refresh_with_storage(self):
- private_key = datafile('privatekey.%s' % self.format)
- credentials = SignedJwtAssertionCredentials(
- 'some_account@example.com',
- private_key,
- scope='read+write',
- sub='joe@example.org')
-
- (filehandle, filename) = tempfile.mkstemp()
- os.close(filehandle)
- store = Storage(filename)
- store.put(credentials)
- credentials.set_store(store)
-
- content = self._credentials_refresh(credentials)
-
- self.assertEqual('Bearer 3/3w', content['Authorization'])
- os.unlink(filename)
-
-
-class PEMSignedJwtAssertionCredentialsOpenSSLTests(
- SignedJwtAssertionCredentialsTests):
- def setUp(self):
- self.format = 'pem'
- crypt.Signer = crypt.OpenSSLSigner
-
-
-class PEMSignedJwtAssertionCredentialsPyCryptoTests(
- SignedJwtAssertionCredentialsTests):
- def setUp(self):
- self.format = 'pem'
- crypt.Signer = crypt.PyCryptoSigner
-
-
-class PKCSSignedJwtAssertionCredentialsPyCryptoTests(unittest.TestCase):
- def test_for_failure(self):
- crypt.Signer = crypt.PyCryptoSigner
- private_key = datafile('privatekey.p12')
- credentials = SignedJwtAssertionCredentials(
- 'some_account@example.com',
- private_key,
- scope='read+write',
- sub='joe@example.org')
- try:
- credentials._generate_assertion()
- self.fail()
- except NotImplementedError:
- pass
-
-class TestHasOpenSSLFlag(unittest.TestCase):
- def test_true(self):
- self.assertEqual(True, HAS_OPENSSL)
- self.assertEqual(True, HAS_CRYPTO)
-
-if __name__ == '__main__':
- unittest.main()
diff --git a/tests/test_oauth2client_keyring.py b/tests/test_oauth2client_keyring.py
deleted file mode 100644
index e5b9971..0000000
--- a/tests/test_oauth2client_keyring.py
+++ /dev/null
@@ -1,94 +0,0 @@
-# Copyright 2012 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.
-
-
-"""Tests for oauth2client.keyring_storage tests.
-
-Unit tests for oauth2client.keyring_storage.
-"""
-
-__author__ = 'jcgregorio@google.com (Joe Gregorio)'
-
-import datetime
-import keyring
-import unittest
-import mox
-
-from oauth2client import GOOGLE_TOKEN_URI
-from oauth2client.client import OAuth2Credentials
-from oauth2client.keyring_storage import Storage
-
-
-class OAuth2ClientKeyringTests(unittest.TestCase):
-
- def test_non_existent_credentials_storage(self):
- m = mox.Mox()
- m.StubOutWithMock(keyring, 'get_password')
- m.StubOutWithMock(keyring, 'set_password')
- keyring.get_password('my_unit_test', 'me').AndReturn(None)
- m.ReplayAll()
-
- s = Storage('my_unit_test', 'me')
- credentials = s.get()
- self.assertEquals(None, credentials)
-
- m.UnsetStubs()
- m.VerifyAll()
-
- def test_malformed_credentials_in_storage(self):
- m = mox.Mox()
- m.StubOutWithMock(keyring, 'get_password')
- m.StubOutWithMock(keyring, 'set_password')
- keyring.get_password('my_unit_test', 'me').AndReturn('{')
- m.ReplayAll()
-
- s = Storage('my_unit_test', 'me')
- credentials = s.get()
- self.assertEquals(None, credentials)
-
- m.UnsetStubs()
- m.VerifyAll()
-
- def test_json_credentials_storage(self):
- access_token = 'foo'
- client_id = 'some_client_id'
- client_secret = 'cOuDdkfjxxnv+'
- refresh_token = '1/0/a.df219fjls0'
- token_expiry = datetime.datetime.utcnow()
- user_agent = 'refresh_checker/1.0'
-
- credentials = OAuth2Credentials(
- access_token, client_id, client_secret,
- refresh_token, token_expiry, GOOGLE_TOKEN_URI,
- user_agent)
-
- m = mox.Mox()
- m.StubOutWithMock(keyring, 'get_password')
- m.StubOutWithMock(keyring, 'set_password')
- keyring.get_password('my_unit_test', 'me').AndReturn(None)
- keyring.set_password('my_unit_test', 'me', credentials.to_json())
- keyring.get_password('my_unit_test', 'me').AndReturn(credentials.to_json())
- m.ReplayAll()
-
- s = Storage('my_unit_test', 'me')
- self.assertEquals(None, s.get())
-
- s.put(credentials)
-
- restored = s.get()
- self.assertEqual('foo', restored.access_token)
- self.assertEqual('some_client_id', restored.client_id)
-
- m.UnsetStubs()
- m.VerifyAll()
diff --git a/tests/test_oauth2client_util.py b/tests/test_oauth2client_util.py
deleted file mode 100644
index 2d67316..0000000
--- a/tests/test_oauth2client_util.py
+++ /dev/null
@@ -1,44 +0,0 @@
-"""Unit tests for oauth2client.util."""
-
-__author__ = 'jcgregorio@google.com (Joe Gregorio)'
-
-import unittest
-
-from oauth2client import util
-
-
-class ScopeToStringTests(unittest.TestCase):
-
- def test_iterables(self):
- cases = [
- ('', ''),
- ('', ()),
- ('', []),
- ('', ('', )),
- ('', ['', ]),
- ('a', ('a', )),
- ('b', ['b', ]),
- ('a b', ['a', 'b']),
- ('a b', ('a', 'b')),
- ('a b', 'a b'),
- ('a b', (s for s in ['a', 'b'])),
- ]
- for expected, case in cases:
- self.assertEqual(expected, util.scopes_to_string(case))
-
-
-class KeyConversionTests(unittest.TestCase):
-
- def test_key_conversions(self):
- d = {'somekey': 'some value', 'another': 'something else', 'onemore': 'foo'}
- tuple_key = util.dict_to_tuple_key(d)
-
- # the resulting key should be naturally sorted
- self.assertEqual(
- (('another', 'something else'),
- ('onemore', 'foo'),
- ('somekey', 'some value')),
- tuple_key)
-
- # check we get the original dictionary back
- self.assertEqual(d, dict(tuple_key))
diff --git a/tests/test_oauth2client_xsrfutil.py b/tests/test_oauth2client_xsrfutil.py
deleted file mode 100644
index a86a15b..0000000
--- a/tests/test_oauth2client_xsrfutil.py
+++ /dev/null
@@ -1,111 +0,0 @@
-# Copyright 2012 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.
-"""Tests for oauth2client.xsrfutil.
-
-Unit tests for oauth2client.xsrfutil.
-"""
-
-__author__ = 'jcgregorio@google.com (Joe Gregorio)'
-
-import unittest
-
-from oauth2client import xsrfutil
-
-# Jan 17 2008, 5:40PM
-TEST_KEY = 'test key'
-TEST_TIME = 1200609642081230
-TEST_USER_ID_1 = 123832983
-TEST_USER_ID_2 = 938297432
-TEST_ACTION_ID_1 = 'some_action'
-TEST_ACTION_ID_2 = 'some_other_action'
-TEST_EXTRA_INFO_1 = 'extra_info_1'
-TEST_EXTRA_INFO_2 = 'more_extra_info'
-
-
-class XsrfUtilTests(unittest.TestCase):
- """Test xsrfutil functions."""
-
- def testGenerateAndValidateToken(self):
- """Test generating and validating a token."""
- token = xsrfutil.generate_token(TEST_KEY,
- TEST_USER_ID_1,
- action_id=TEST_ACTION_ID_1,
- when=TEST_TIME)
-
- # Check that the token is considered valid when it should be.
- self.assertTrue(xsrfutil.validate_token(TEST_KEY,
- token,
- TEST_USER_ID_1,
- action_id=TEST_ACTION_ID_1,
- current_time=TEST_TIME))
-
- # Should still be valid 15 minutes later.
- later15mins = TEST_TIME + 15*60
- self.assertTrue(xsrfutil.validate_token(TEST_KEY,
- token,
- TEST_USER_ID_1,
- action_id=TEST_ACTION_ID_1,
- current_time=later15mins))
-
- # But not if beyond the timeout.
- later2hours = TEST_TIME + 2*60*60
- self.assertFalse(xsrfutil.validate_token(TEST_KEY,
- token,
- TEST_USER_ID_1,
- action_id=TEST_ACTION_ID_1,
- current_time=later2hours))
-
- # Or if the key is different.
- self.assertFalse(xsrfutil.validate_token('another key',
- token,
- TEST_USER_ID_1,
- action_id=TEST_ACTION_ID_1,
- current_time=later15mins))
-
- # Or the user ID....
- self.assertFalse(xsrfutil.validate_token(TEST_KEY,
- token,
- TEST_USER_ID_2,
- action_id=TEST_ACTION_ID_1,
- current_time=later15mins))
-
- # Or the action ID...
- self.assertFalse(xsrfutil.validate_token(TEST_KEY,
- token,
- TEST_USER_ID_1,
- action_id=TEST_ACTION_ID_2,
- current_time=later15mins))
-
- # Invalid when truncated
- self.assertFalse(xsrfutil.validate_token(TEST_KEY,
- token[:-1],
- TEST_USER_ID_1,
- action_id=TEST_ACTION_ID_1,
- current_time=later15mins))
-
- # Invalid with extra garbage
- self.assertFalse(xsrfutil.validate_token(TEST_KEY,
- token + 'x',
- TEST_USER_ID_1,
- action_id=TEST_ACTION_ID_1,
- current_time=later15mins))
-
- # Invalid with token of None
- self.assertFalse(xsrfutil.validate_token(TEST_KEY,
- None,
- TEST_USER_ID_1,
- action_id=TEST_ACTION_ID_1))
-
-if __name__ == '__main__':
- unittest.main()