| # 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')) |
