v811_spc009/external/python/oauth2client/tests/test_tools.py

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