text: # Copyright 2015 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.
  
  import json
  import unittest
  
  from django.conf.urls import include, url
  from django.core import exceptions
  from django import http
  from django import test
  import mock
  from oauth2client.client import FlowExchangeError, OAuth2WebServerFlow
  import django.conf
  from oauth2client.contrib import django_util
  from oauth2client.contrib.django_util import decorators
  from oauth2client.contrib.django_util import site
  from oauth2client.contrib.django_util import storage
  from oauth2client.contrib.django_util import views
  from six.moves import http_client
  from six.moves.urllib import parse
  
  urlpatterns = [
      url(r'^oauth2/', include(site.urls))
  ]
  
  urlpatterns += [url(r'^oauth2/', include(site.urls))]
  
  
  class OAuth2SetupTest(unittest.TestCase):
  
      @mock.patch("oauth2client.contrib.django_util.clientsecrets")
      def test_settings_initialize(self, clientsecrets):
          django.conf.settings.GOOGLE_OAUTH2_CLIENT_SECRETS_JSON = 'file.json'
          clientsecrets.loadfile.return_value = (
              clientsecrets.TYPE_WEB,
              {
                  'client_id': 'myid',
                  'client_secret': 'hunter2'
              }
          )
  
          oauth2_settings = django_util.OAuth2Settings(django.conf.settings)
          self.assertTrue(clientsecrets.loadfile.called)
          self.assertEqual(oauth2_settings.client_id, 'myid')
          self.assertEqual(oauth2_settings.client_secret, 'hunter2')
  
      @mock.patch("oauth2client.contrib.django_util.clientsecrets")
      def test_settings_initialize_invalid_type(self, clientsecrets):
          django.conf.settings.GOOGLE_OAUTH2_CLIENT_SECRETS_JSON = 'file.json'
          clientsecrets.loadfile.return_value = (
              "wrong_type",
              {
                  'client_id': 'myid',
                  'client_secret': 'hunter2'
              }
          )
  
          self.assertRaises(
              ValueError,
              django_util.OAuth2Settings.__init__,
              object.__new__(django_util.OAuth2Settings),
              django.conf.settings)
  
      @mock.patch("oauth2client.contrib.django_util.clientsecrets")
      def test_no_settings(self, clientsecrets):
          django.conf.settings.GOOGLE_OAUTH2_CLIENT_SECRETS_JSON = None
          django.conf.settings.GOOGLE_OAUTH2_CLIENT_SECRET = None
          django.conf.settings.GOOGLE_OAUTH2_CLIENT_ID = None
  
          self.assertRaises(
              exceptions.ImproperlyConfigured,
              django_util.OAuth2Settings.__init__,
              object.__new__(django_util.OAuth2Settings),
              django.conf.settings)
  
      @mock.patch("oauth2client.contrib.django_util.clientsecrets")
      def test_no_session_middleware(self, clientsecrets):
          old_classes = django.conf.settings.MIDDLEWARE_CLASSES
          django.conf.settings.MIDDLEWARE_CLASSES = ()
  
          self.assertRaises(
              exceptions.ImproperlyConfigured,
              django_util.OAuth2Settings.__init__,
              object.__new__(django_util.OAuth2Settings),
              django.conf.settings)
          django.conf.settings.MIDDLEWARE_CLASSES = old_classes
  
  
  class TestWithSession(test.TestCase):
      def setUp(self):
          self.factory = test.RequestFactory()
          from django.contrib.sessions.backends.file import SessionStore
  
          store = SessionStore()
          store.save()
          self.session = store
  
  
  class OAuth2EnabledDecoratorTest(TestWithSession):
      def test_no_credentials_without_credentials(self):
          request = self.factory.get('/test')
          request.session = self.session
  
          @decorators.oauth_enabled
          def test_view(request):
              return http.HttpResponse("test")  # pragma: NO COVER
  
          response = test_view(request)
          self.assertEquals(response.status_code, http_client.OK)
          self.assertIsNotNone(request.oauth)
          self.assertFalse(request.oauth.has_credentials())
          self.assertIsNone(request.oauth.http)
  
      @mock.patch('oauth2client.contrib.dictionary_storage.OAuth2Credentials')
      def test_has_credentials_in_storage(self, OAuth2Credentials):
          request = self.factory.get('/test')
          request.session = mock.MagicMock()
  
          credentials_mock = mock.Mock(
              scopes=set(django.conf.settings.GOOGLE_OAUTH2_SCOPES))
          credentials_mock.has_scopes.return_value = True
          credentials_mock.invalid = False
          OAuth2Credentials.from_json.return_value = credentials_mock
  
          @decorators.oauth_enabled
          def test_view(request):
              return http.HttpResponse("test")
  
          response = test_view(request)
          self.assertEquals(response.status_code, http_client.OK)
          self.assertEquals(response.content, b"test")
          self.assertTrue(request.oauth.has_credentials())
          self.assertIsNotNone(request.oauth.http)
  
      @mock.patch('oauth2client.contrib.dictionary_storage.OAuth2Credentials')
      def test_specified_scopes(self, OAuth2Credentials):
          request = self.factory.get('/test')
          request.session = mock.MagicMock()
  
          credentials_mock = mock.Mock(
              scopes=set(django.conf.settings.GOOGLE_OAUTH2_SCOPES))
          credentials_mock.has_scopes = True
          credentials_mock.is_valid = True
          OAuth2Credentials.from_json.return_value = credentials_mock
  
          @decorators.oauth_enabled(scopes=['additional-scope'])
          def test_view(request):
              return http.HttpResponse("hello world")  # pragma: NO COVER
  
          response = test_view(request)
          self.assertEquals(response.status_code, http_client.OK)
          self.assertIsNotNone(request.oauth)
          self.assertFalse(request.oauth.has_credentials())
  
  
  class OAuth2RequiredDecoratorTest(TestWithSession):
      def test_redirects_without_credentials(self):
          request = self.factory.get('/test')
          request.session = self.session
  
          @decorators.oauth_required
          def test_view(request):
              return http.HttpResponse("test")  # pragma: NO COVER
  
          response = test_view(request)
          self.assertTrue(isinstance(response, http.HttpResponseRedirect))
          self.assertEquals(parse.urlparse(response['Location']).path,
                            "/oauth2/oauth2authorize/")
          self.assertTrue(
              "return_url=%2Ftest" in parse.urlparse(response['Location']).query)
  
          self.assertEquals(response.status_code, 302)
  
      @mock.patch('oauth2client.contrib.django_util.UserOAuth2', autospec=True)
      def test_has_credentials_in_storage(self, UserOAuth2):
          request = self.factory.get('/test')
          request.session = mock.MagicMock()
  
          @decorators.oauth_required
          def test_view(request):
              return http.HttpResponse("test")
  
          my_user_oauth = mock.MagicMock()
  
          UserOAuth2.return_value = my_user_oauth
          my_user_oauth.has_credentials.return_value = True
  
          response = test_view(request)
          self.assertEquals(response.status_code, http_client.OK)
          self.assertEquals(response.content, b"test")
  
      @mock.patch('oauth2client.contrib.dictionary_storage.OAuth2Credentials')
      def test_has_credentials_in_storage_no_scopes(self, OAuth2Credentials):
          request = self.factory.get('/test')
  
          request.session = mock.MagicMock()
          credentials_mock = mock.Mock(
              scopes=set(django.conf.settings.GOOGLE_OAUTH2_SCOPES))
          credentials_mock.has_scopes.return_value = False
  
          OAuth2Credentials.from_json.return_value = credentials_mock
  
          @decorators.oauth_required
          def test_view(request):
              return http.HttpResponse("test")  # pragma: NO COVER
  
          response = test_view(request)
          self.assertEquals(response.status_code, 302)
  
      @mock.patch('oauth2client.contrib.dictionary_storage.OAuth2Credentials')
      def test_specified_scopes(self, OAuth2Credentials):
          request = self.factory.get('/test')
          request.session = mock.MagicMock()
  
          credentials_mock = mock.Mock(
              scopes=set(django.conf.settings.GOOGLE_OAUTH2_SCOPES))
          credentials_mock.has_scopes = False
          OAuth2Credentials.from_json.return_value = credentials_mock
  
          @decorators.oauth_required(scopes=['additional-scope'])
          def test_view(request):
              return http.HttpResponse("hello world")  # pragma: NO COVER
  
          response = test_view(request)
          self.assertEquals(response.status_code, 302)
  
  
  class Oauth2AuthorizeTest(TestWithSession):
  
      def test_authorize_works(self):
          request = self.factory.get('oauth2/oauth2authorize')
          request.session = self.session
          response = views.oauth2_authorize(request)
          self.assertTrue(isinstance(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
          response = views.oauth2_authorize(request)
          self.assertTrue(isinstance(response, http.HttpResponseRedirect))
  
  
  class Oauth2CallbackTest(TestWithSession):
  
      def setUp(self):
          global mycallback
          mycallback = mock.Mock()
  
          super(Oauth2CallbackTest, self).setUp()
          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
          }
  
      @mock.patch("oauth2client.contrib.django_util.views.pickle")
      def test_callback_works(self, pickle):
          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 = OAuth2WebServerFlow(
              client_id='clientid',
              client_secret='clientsecret',
              scope=['email'],
              state=json.dumps(self.fake_state),
              redirect_uri=request.build_absolute_uri("oauth2/oauth2callback"))
  
          self.session['google_oauth2_flow_{0}'.format(self.CSRF_TOKEN)] \
              = pickle.dumps(flow)
          flow.step2_exchange = mock.Mock()
          pickle.loads.return_value = flow
  
          request.session = self.session
          response = views.oauth2_callback(request)
          self.assertTrue(isinstance(response, http.HttpResponseRedirect))
          self.assertEquals(response.status_code, 302)
          self.assertEquals(response['Location'], self.RETURN_URL)
  
      @mock.patch("oauth2client.contrib.django_util.views.pickle")
      def test_callback_handles_bad_flow_exchange(self, pickle):
          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 = OAuth2WebServerFlow(
              client_id='clientid',
              client_secret='clientsecret',
              scope=['email'],
              state=json.dumps(self.fake_state),
              redirect_uri=request.build_absolute_uri("oauth2/oauth2callback"))
  
          self.session['google_oauth2_flow_{0}'.format(self.CSRF_TOKEN)]\
              = pickle.dumps(flow)
  
          def local_throws(code):
              raise FlowExchangeError("test")
  
          flow.step2_exchange = local_throws
          pickle.loads.return_value = flow
  
          request.session = self.session
          response = views.oauth2_callback(request)
          self.assertTrue(isinstance(response, http.HttpResponseBadRequest))
  
      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.assertTrue(isinstance(response, http.HttpResponseBadRequest))
          self.assertTrue(b"Authorization failed" in 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.assertTrue(isinstance(response, http.HttpResponseBadRequest))
          self.assertEquals(
              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.assertTrue(isinstance(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.assertTrue(isinstance(response, http.HttpResponseBadRequest))
          self.assertEquals(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.assertTrue(isinstance(response, http.HttpResponseBadRequest))
          self.assertEquals(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.assertTrue(isinstance(response, http.HttpResponseBadRequest))
          self.assertEquals(response.content, b'Missing Oauth2 flow.')
  
  
  class MockObjectWithSession(object):
      def __init__(self, session):
          self.session = session
  
  
  class StorageTest(TestWithSession):
  
      def test_session_delete(self):
          self.session[storage._CREDENTIALS_KEY] = "test_val"
          request = MockObjectWithSession(self.session)
          django_storage = storage.get_storage(request)
          django_storage.delete()
          self.assertIsNone(self.session.get(storage._CREDENTIALS_KEY))
  
      def test_session_delete_nothing(self):
          request = MockObjectWithSession(self.session)
          django_storage = storage.get_storage(request)
          django_storage.delete()
  