| # 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. | |
| import socket | |
| import sys | |
| import threading | |
| import mock | |
| from six.moves.urllib import request | |
| import unittest2 | |
| from oauth2client import client | |
| from oauth2client import tools | |
| try: | |
| import argparse | |
| except ImportError: # pragma: NO COVER | |
| raise unittest2.SkipTest('argparase unavailable.') | |
| class TestClientRedirectServer(unittest2.TestCase): | |
| """Test the ClientRedirectServer and ClientRedirectHandler classes.""" | |
| def test_ClientRedirectServer(self): | |
| # create a ClientRedirectServer and run it in a thread to listen | |
| # for a mock GET request with the access token | |
| # the server should return a 200 message and store the token | |
| httpd = tools.ClientRedirectServer(('localhost', 0), | |
| tools.ClientRedirectHandler) | |
| code = 'foo' | |
| url = 'http://localhost:{0}?code={1}'.format( | |
| httpd.server_address[1], code) | |
| t = threading.Thread(target=httpd.handle_request) | |
| t.setDaemon(True) | |
| t.start() | |
| f = request.urlopen(url) | |
| self.assertTrue(f.read()) | |
| t.join() | |
| httpd.server_close() | |
| self.assertEqual(httpd.query_params.get('code'), code) | |
| class TestRunFlow(unittest2.TestCase): | |
| def setUp(self): | |
| self.server = mock.Mock() | |
| self.flow = mock.Mock() | |
| self.storage = mock.Mock() | |
| self.credentials = mock.Mock() | |
| self.flow.step1_get_authorize_url.return_value = ( | |
| 'http://example.com/auth') | |
| self.flow.step2_exchange.return_value = self.credentials | |
| self.flags = argparse.Namespace( | |
| noauth_local_webserver=True, logging_level='INFO') | |
| self.server_flags = argparse.Namespace( | |
| noauth_local_webserver=False, | |
| logging_level='INFO', | |
| auth_host_port=[8080, ], | |
| auth_host_name='localhost') | |
| @mock.patch.object(sys, 'argv', ['ignored', '--noauth_local_webserver']) | |
| @mock.patch('oauth2client.tools.logging') | |
| @mock.patch('oauth2client.tools.input') | |
| def test_run_flow_no_webserver(self, input_mock, logging_mock): | |
| input_mock.return_value = 'auth_code' | |
| # Successful exchange. | |
| returned_credentials = tools.run_flow(self.flow, self.storage) | |
| self.assertEqual(self.credentials, returned_credentials) | |
| self.assertEqual(self.flow.redirect_uri, client.OOB_CALLBACK_URN) | |
| self.flow.step2_exchange.assert_called_once_with( | |
| 'auth_code', http=None) | |
| self.storage.put.assert_called_once_with(self.credentials) | |
| self.credentials.set_store.assert_called_once_with(self.storage) | |
| @mock.patch('oauth2client.tools.logging') | |
| @mock.patch('oauth2client.tools.input') | |
| def test_run_flow_no_webserver_explicit_flags( | |
| self, input_mock, logging_mock): | |
| input_mock.return_value = 'auth_code' | |
| # Successful exchange. | |
| returned_credentials = tools.run_flow( | |
| self.flow, self.storage, flags=self.flags) | |
| self.assertEqual(self.credentials, returned_credentials) | |
| self.assertEqual(self.flow.redirect_uri, client.OOB_CALLBACK_URN) | |
| self.flow.step2_exchange.assert_called_once_with( | |
| 'auth_code', http=None) | |
| @mock.patch('oauth2client.tools.logging') | |
| @mock.patch('oauth2client.tools.input') | |
| def test_run_flow_no_webserver_exchange_error( | |
| self, input_mock, logging_mock): | |
| input_mock.return_value = 'auth_code' | |
| self.flow.step2_exchange.side_effect = client.FlowExchangeError() | |
| # Error while exchanging. | |
| with self.assertRaises(SystemExit): | |
| tools.run_flow(self.flow, self.storage, flags=self.flags) | |
| self.flow.step2_exchange.assert_called_once_with( | |
| 'auth_code', http=None) | |
| @mock.patch('oauth2client.tools.logging') | |
| @mock.patch('oauth2client.tools.ClientRedirectServer') | |
| @mock.patch('webbrowser.open') | |
| def test_run_flow_webserver( | |
| self, webbrowser_open_mock, server_ctor_mock, logging_mock): | |
| server_ctor_mock.return_value = self.server | |
| self.server.query_params = {'code': 'auth_code'} | |
| # Successful exchange. | |
| returned_credentials = tools.run_flow( | |
| self.flow, self.storage, flags=self.server_flags) | |
| self.assertEqual(self.credentials, returned_credentials) | |
| self.assertEqual(self.flow.redirect_uri, 'http://localhost:8080/') | |
| self.flow.step2_exchange.assert_called_once_with( | |
| 'auth_code', http=None) | |
| self.storage.put.assert_called_once_with(self.credentials) | |
| self.credentials.set_store.assert_called_once_with(self.storage) | |
| self.assertTrue(self.server.handle_request.called) | |
| webbrowser_open_mock.assert_called_once_with( | |
| 'http://example.com/auth', autoraise=True, new=1) | |
| @mock.patch('oauth2client.tools.logging') | |
| @mock.patch('oauth2client.tools.ClientRedirectServer') | |
| @mock.patch('webbrowser.open') | |
| def test_run_flow_webserver_exchange_error( | |
| self, webbrowser_open_mock, server_ctor_mock, logging_mock): | |
| server_ctor_mock.return_value = self.server | |
| self.server.query_params = {'error': 'any error'} | |
| # Exchange returned an error code. | |
| with self.assertRaises(SystemExit): | |
| tools.run_flow(self.flow, self.storage, flags=self.server_flags) | |
| self.assertTrue(self.server.handle_request.called) | |
| @mock.patch('oauth2client.tools.logging') | |
| @mock.patch('oauth2client.tools.ClientRedirectServer') | |
| @mock.patch('webbrowser.open') | |
| def test_run_flow_webserver_no_code( | |
| self, webbrowser_open_mock, server_ctor_mock, logging_mock): | |
| server_ctor_mock.return_value = self.server | |
| self.server.query_params = {} | |
| # No code found in response | |
| with self.assertRaises(SystemExit): | |
| tools.run_flow(self.flow, self.storage, flags=self.server_flags) | |
| self.assertTrue(self.server.handle_request.called) | |
| @mock.patch('oauth2client.tools.logging') | |
| @mock.patch('oauth2client.tools.ClientRedirectServer') | |
| @mock.patch('oauth2client.tools.input') | |
| def test_run_flow_webserver_fallback( | |
| self, input_mock, server_ctor_mock, logging_mock): | |
| server_ctor_mock.side_effect = socket.error() | |
| input_mock.return_value = 'auth_code' | |
| # It should catch the socket error and proceed as if | |
| # noauth_local_webserver was specified. | |
| returned_credentials = tools.run_flow( | |
| self.flow, self.storage, flags=self.server_flags) | |
| self.assertEqual(self.credentials, returned_credentials) | |
| self.assertEqual(self.flow.redirect_uri, client.OOB_CALLBACK_URN) | |
| self.flow.step2_exchange.assert_called_once_with( | |
| 'auth_code', http=None) | |
| self.assertTrue(server_ctor_mock.called) | |
| self.assertFalse(self.server.handle_request.called) | |
| class TestMessageIfMissing(unittest2.TestCase): | |
| def test_message_if_missing(self): | |
| self.assertIn('somefile.txt', tools.message_if_missing('somefile.txt')) |