306 lines
12 KiB
Python
306 lines
12 KiB
Python
# Copyright 2016 Google Inc. All Rights Reserved.
|
|
#
|
|
# 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 test for django_util views"""
|
|
|
|
import copy
|
|
import json
|
|
|
|
import django
|
|
from django import http
|
|
import django.conf
|
|
from django.contrib.auth import models as django_models
|
|
import mock
|
|
from six.moves import reload_module
|
|
|
|
from oauth2client import client
|
|
import oauth2client.contrib.django_util
|
|
from oauth2client.contrib.django_util import views
|
|
from tests.contrib import django_util as tests_django_util
|
|
from tests.contrib.django_util import models as tests_models
|
|
|
|
|
|
class OAuth2AuthorizeTest(tests_django_util.TestWithDjangoEnvironment):
|
|
|
|
def setUp(self):
|
|
super(OAuth2AuthorizeTest, self).setUp()
|
|
self.save_settings = copy.deepcopy(django.conf.settings)
|
|
reload_module(oauth2client.contrib.django_util)
|
|
self.user = django_models.User.objects.create_user(
|
|
username='bill', email='bill@example.com', password='hunter2')
|
|
|
|
def tearDown(self):
|
|
django.conf.settings = copy.deepcopy(self.save_settings)
|
|
|
|
def test_authorize_works(self):
|
|
request = self.factory.get('oauth2/oauth2authorize')
|
|
request.session = self.session
|
|
request.user = self.user
|
|
response = views.oauth2_authorize(request)
|
|
self.assertIsInstance(response, http.HttpResponseRedirect)
|
|
|
|
def test_authorize_anonymous_user(self):
|
|
request = self.factory.get('oauth2/oauth2authorize')
|
|
request.session = self.session
|
|
request.user = django_models.AnonymousUser()
|
|
response = views.oauth2_authorize(request)
|
|
self.assertIsInstance(response, http.HttpResponseRedirect)
|
|
|
|
def test_authorize_works_explicit_return_url(self):
|
|
request = self.factory.get('oauth2/oauth2authorize',
|
|
data={'return_url': '/return_endpoint'})
|
|
request.session = self.session
|
|
request.user = self.user
|
|
response = views.oauth2_authorize(request)
|
|
self.assertIsInstance(response, http.HttpResponseRedirect)
|
|
|
|
|
|
class Oauth2AuthorizeStorageModelTest(
|
|
tests_django_util.TestWithDjangoEnvironment):
|
|
|
|
def setUp(self):
|
|
super(Oauth2AuthorizeStorageModelTest, self).setUp()
|
|
self.save_settings = copy.deepcopy(django.conf.settings)
|
|
|
|
STORAGE_MODEL = {
|
|
'model': 'tests.contrib.django_util.models.CredentialsModel',
|
|
'user_property': 'user_id',
|
|
'credentials_property': 'credentials'
|
|
}
|
|
django.conf.settings.GOOGLE_OAUTH2_STORAGE_MODEL = STORAGE_MODEL
|
|
|
|
# OAuth2 Settings gets configured based on Django settings
|
|
# at import time, so in order for us to reload the settings
|
|
# we need to reload the module
|
|
reload_module(oauth2client.contrib.django_util)
|
|
self.user = django_models.User.objects.create_user(
|
|
username='bill', email='bill@example.com', password='hunter2')
|
|
|
|
def tearDown(self):
|
|
django.conf.settings = copy.deepcopy(self.save_settings)
|
|
|
|
def test_authorize_works(self):
|
|
request = self.factory.get('oauth2/oauth2authorize')
|
|
request.session = self.session
|
|
request.user = self.user
|
|
response = views.oauth2_authorize(request)
|
|
self.assertIsInstance(response, http.HttpResponseRedirect)
|
|
# redirects to Google oauth
|
|
self.assertIn('accounts.google.com', response.url)
|
|
|
|
def test_authorize_anonymous_user_redirects_login(self):
|
|
request = self.factory.get('oauth2/oauth2authorize')
|
|
request.session = self.session
|
|
request.user = django_models.AnonymousUser()
|
|
response = views.oauth2_authorize(request)
|
|
self.assertIsInstance(response, http.HttpResponseRedirect)
|
|
# redirects to Django login
|
|
self.assertIn(django.conf.settings.LOGIN_URL, response.url)
|
|
|
|
def test_authorize_works_explicit_return_url(self):
|
|
request = self.factory.get('oauth2/oauth2authorize',
|
|
data={'return_url': '/return_endpoint'})
|
|
request.session = self.session
|
|
request.user = self.user
|
|
response = views.oauth2_authorize(request)
|
|
self.assertIsInstance(response, http.HttpResponseRedirect)
|
|
|
|
def test_authorized_user_no_credentials_redirects(self):
|
|
request = self.factory.get('oauth2/oauth2authorize',
|
|
data={'return_url': '/return_endpoint'})
|
|
request.session = self.session
|
|
|
|
authorized_user = django_models.User.objects.create_user(
|
|
username='bill2', email='bill@example.com', password='hunter2')
|
|
|
|
tests_models.CredentialsModel.objects.create(
|
|
user_id=authorized_user,
|
|
credentials=None)
|
|
|
|
request.user = authorized_user
|
|
response = views.oauth2_authorize(request)
|
|
self.assertIsInstance(response, http.HttpResponseRedirect)
|
|
|
|
def test_already_authorized(self):
|
|
request = self.factory.get('oauth2/oauth2authorize',
|
|
data={'return_url': '/return_endpoint'})
|
|
request.session = self.session
|
|
|
|
authorized_user = django_models.User.objects.create_user(
|
|
username='bill2', email='bill@example.com', password='hunter2')
|
|
|
|
credentials = _Credentials()
|
|
tests_models.CredentialsModel.objects.create(
|
|
user_id=authorized_user,
|
|
credentials=credentials)
|
|
|
|
request.user = authorized_user
|
|
response = views.oauth2_authorize(request)
|
|
self.assertIsInstance(response, http.HttpResponseRedirect)
|
|
self.assertEqual(response.url, '/return_endpoint')
|
|
|
|
|
|
class _Credentials(object):
|
|
# Can't use mock when testing Django models
|
|
# https://code.djangoproject.com/ticket/25493
|
|
def __init__(self):
|
|
self.invalid = False
|
|
self.scopes = set()
|
|
|
|
def has_scopes(self, _):
|
|
return True
|
|
|
|
|
|
class Oauth2CallbackTest(tests_django_util.TestWithDjangoEnvironment):
|
|
|
|
def setUp(self):
|
|
super(Oauth2CallbackTest, self).setUp()
|
|
self.save_settings = copy.deepcopy(django.conf.settings)
|
|
reload_module(oauth2client.contrib.django_util)
|
|
|
|
self.CSRF_TOKEN = 'token'
|
|
self.RETURN_URL = 'http://return-url.com'
|
|
self.fake_state = {
|
|
'csrf_token': self.CSRF_TOKEN,
|
|
'return_url': self.RETURN_URL,
|
|
'scopes': django.conf.settings.GOOGLE_OAUTH2_SCOPES
|
|
}
|
|
self.user = django_models.User.objects.create_user(
|
|
username='bill', email='bill@example.com', password='hunter2')
|
|
|
|
@mock.patch('oauth2client.contrib.django_util.views.jsonpickle')
|
|
def test_callback_works(self, jsonpickle_mock):
|
|
request = self.factory.get('oauth2/oauth2callback', data={
|
|
'state': json.dumps(self.fake_state),
|
|
'code': 123
|
|
})
|
|
|
|
self.session['google_oauth2_csrf_token'] = self.CSRF_TOKEN
|
|
|
|
flow = client.OAuth2WebServerFlow(
|
|
client_id='clientid',
|
|
client_secret='clientsecret',
|
|
scope=['email'],
|
|
state=json.dumps(self.fake_state),
|
|
redirect_uri=request.build_absolute_uri("oauth2/oauth2callback"))
|
|
|
|
name = 'google_oauth2_flow_{0}'.format(self.CSRF_TOKEN)
|
|
pickled_flow = object()
|
|
self.session[name] = pickled_flow
|
|
flow.step2_exchange = mock.Mock()
|
|
jsonpickle_mock.decode.return_value = flow
|
|
|
|
request.session = self.session
|
|
request.user = self.user
|
|
response = views.oauth2_callback(request)
|
|
self.assertIsInstance(response, http.HttpResponseRedirect)
|
|
self.assertEqual(
|
|
response.status_code, django.http.HttpResponseRedirect.status_code)
|
|
self.assertEqual(response['Location'], self.RETURN_URL)
|
|
jsonpickle_mock.decode.assert_called_once_with(pickled_flow)
|
|
|
|
@mock.patch('oauth2client.contrib.django_util.views.jsonpickle')
|
|
def test_callback_handles_bad_flow_exchange(self, jsonpickle_mock):
|
|
request = self.factory.get('oauth2/oauth2callback', data={
|
|
"state": json.dumps(self.fake_state),
|
|
"code": 123
|
|
})
|
|
|
|
self.session['google_oauth2_csrf_token'] = self.CSRF_TOKEN
|
|
|
|
flow = client.OAuth2WebServerFlow(
|
|
client_id='clientid',
|
|
client_secret='clientsecret',
|
|
scope=['email'],
|
|
state=json.dumps(self.fake_state),
|
|
redirect_uri=request.build_absolute_uri('oauth2/oauth2callback'))
|
|
|
|
session_key = 'google_oauth2_flow_{0}'.format(self.CSRF_TOKEN)
|
|
pickled_flow = object()
|
|
self.session[session_key] = pickled_flow
|
|
|
|
def local_throws(code):
|
|
raise client.FlowExchangeError('test')
|
|
|
|
flow.step2_exchange = local_throws
|
|
jsonpickle_mock.decode.return_value = flow
|
|
|
|
request.session = self.session
|
|
response = views.oauth2_callback(request)
|
|
self.assertIsInstance(response, http.HttpResponseBadRequest)
|
|
jsonpickle_mock.decode.assert_called_once_with(pickled_flow)
|
|
|
|
def test_error_returns_bad_request(self):
|
|
request = self.factory.get('oauth2/oauth2callback', data={
|
|
'error': 'There was an error in your authorization.',
|
|
})
|
|
response = views.oauth2_callback(request)
|
|
self.assertIsInstance(response, http.HttpResponseBadRequest)
|
|
self.assertIn(b'Authorization failed', response.content)
|
|
|
|
def test_no_session(self):
|
|
request = self.factory.get('oauth2/oauth2callback', data={
|
|
'code': 123,
|
|
'state': json.dumps(self.fake_state)
|
|
})
|
|
|
|
request.session = self.session
|
|
response = views.oauth2_callback(request)
|
|
self.assertIsInstance(response, http.HttpResponseBadRequest)
|
|
self.assertEqual(
|
|
response.content, b'No existing session for this flow.')
|
|
|
|
def test_missing_state_returns_bad_request(self):
|
|
request = self.factory.get('oauth2/oauth2callback', data={
|
|
'code': 123
|
|
})
|
|
self.session['google_oauth2_csrf_token'] = "token"
|
|
request.session = self.session
|
|
response = views.oauth2_callback(request)
|
|
self.assertIsInstance(response, http.HttpResponseBadRequest)
|
|
|
|
def test_bad_state(self):
|
|
request = self.factory.get('oauth2/oauth2callback', data={
|
|
'code': 123,
|
|
'state': json.dumps({'wrong': 'state'})
|
|
})
|
|
self.session['google_oauth2_csrf_token'] = 'token'
|
|
request.session = self.session
|
|
response = views.oauth2_callback(request)
|
|
self.assertIsInstance(response, http.HttpResponseBadRequest)
|
|
self.assertEqual(response.content, b'Invalid state parameter.')
|
|
|
|
def test_bad_csrf(self):
|
|
request = self.factory.get('oauth2/oauth2callback', data={
|
|
"state": json.dumps(self.fake_state),
|
|
"code": 123
|
|
})
|
|
self.session['google_oauth2_csrf_token'] = 'WRONG TOKEN'
|
|
request.session = self.session
|
|
response = views.oauth2_callback(request)
|
|
self.assertIsInstance(response, http.HttpResponseBadRequest)
|
|
self.assertEqual(response.content, b'Invalid CSRF token.')
|
|
|
|
def test_no_saved_flow(self):
|
|
request = self.factory.get('oauth2/oauth2callback', data={
|
|
'state': json.dumps(self.fake_state),
|
|
'code': 123
|
|
})
|
|
self.session['google_oauth2_csrf_token'] = self.CSRF_TOKEN
|
|
self.session['google_oauth2_flow_{0}'.format(self.CSRF_TOKEN)] = None
|
|
request.session = self.session
|
|
response = views.oauth2_callback(request)
|
|
self.assertIsInstance(response, http.HttpResponseBadRequest)
|
|
self.assertEqual(response.content, b'Missing Oauth2 flow.')
|