100% coverage for oauth2client.tools

oauth2client/tools.py

@@ -52,7 +52,7 @@
 def _CreateArgumentParser():
     try:
         import argparse
-    except ImportError:
+    except ImportError:  # pragma: NO COVER
         return None
     parser = argparse.ArgumentParser(add_help=False)
     parser.add_argument('--auth_host_name', default='localhost',

tests/test_tools.py

@@ -1,12 +1,17 @@
 """Unit tests for oauth2client.tools."""
 
-import unittest
 from oauth2client import tools
+from oauth2client.client import OOB_CALLBACK_URN
+from oauth2client.client import FlowExchangeError
+import mock
 from six.moves.urllib import request
+import socket
+import sys
 import threading
+import unittest2
 
 
-class TestClientRedirectServer(unittest.TestCase):
+class TestClientRedirectServer(unittest2.TestCase):
     """Test the ClientRedirectServer and ClientRedirectHandler classes."""
 
     def test_ClientRedirectServer(self):
@@ -27,5 +32,114 @@ def test_ClientRedirectServer(self):
         self.assertEqual(httpd.query_params.get('code'), code)
 
 
+class _Namespace(object):
+    def __init__(self, **kwargs):
+        self.__dict__.update(kwargs)
+
+
+class TestRunFlow(unittest2.TestCase):
+    @mock.patch.object(sys, 'argv',
+        [sys.executable, '--noauth_local_webserver'])
+    @mock.patch('oauth2client.tools.logging')
+    @mock.patch('oauth2client.tools.input')
+    def test_run_flow_no_webserver(self, input_mock, logging_mock):
+        flow = mock.Mock()
+        storage = mock.Mock()
+        credentials = mock.Mock()
+
+        input_mock.return_value = 'auth_code'
+        flow.step1_get_authorize_url.return_value = 'http://example.com/auth'
+        flow.step2_exchange.return_value = credentials
+
+        # Successful exchange.
+        returned_credentials = tools.run_flow(flow, storage)
+
+        self.assertEqual(credentials, returned_credentials)
+        self.assertEqual(flow.redirect_uri, OOB_CALLBACK_URN)
+        flow.step2_exchange.assert_called_once_with('auth_code', http=None)
+        storage.put.assert_called_once_with(credentials)
+        credentials.set_store.assert_called_once_with(storage)
+
+        # Same thing, but with explicit flags
+        flags = _Namespace(noauth_local_webserver=True, logging_level='INFO')
+        returned_credentials = tools.run_flow(flow, storage, flags=flags)
+        self.assertEqual(credentials, returned_credentials)
+
+        # Error while exchanging.
+        with self.assertRaises(SystemExit):
+            flow.step2_exchange.side_effect = FlowExchangeError()
+            returned_credentials = tools.run_flow(flow, storage)
+
+
+    @mock.patch.object(sys, 'argv', [sys.executable,])
+    @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 = mock.Mock()
+        flow = mock.Mock()
+        storage = mock.Mock()
+        credentials = mock.Mock()
+
+        server_ctor_mock.return_value = server
+        server.query_params = {'code': 'auth_code'}
+        flow.step1_get_authorize_url.return_value = 'http://example.com/auth'
+        flow.step2_exchange.return_value = credentials
+
+        # Successful exchange.
+        returned_credentials = tools.run_flow(flow, storage)
+
+        self.assertEqual(credentials, returned_credentials)
+        self.assertEqual(flow.redirect_uri, 'http://localhost:8080/')
+        flow.step2_exchange.assert_called_once_with('auth_code', http=None)
+        storage.put.assert_called_once_with(credentials)
+        credentials.set_store.assert_called_once_with(storage)
+        webbrowser_open_mock.assert_called_once_with(
+            'http://example.com/auth', autoraise=True, new=1)
+        self.assertTrue(server.handle_request.called)
+
+        # Exchange returned an error code.
+        with self.assertRaises(SystemExit):
+            server.query_params = {'error': 'any error'}
+            returned_credentials = tools.run_flow(flow, storage)
+
+        # No code found in response
+        with self.assertRaises(SystemExit):
+            server.query_params = {}
+            returned_credentials = tools.run_flow(flow, storage)
+
+    @mock.patch.object(sys, 'argv', [sys.executable,])
+    @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):
+        flow = mock.Mock()
+        storage = mock.Mock()
+        credentials = mock.Mock()
+
+        server_ctor_mock.side_effect = socket.error()
+        input_mock.return_value = 'auth_code'
+        flow.step1_get_authorize_url.return_value = 'http://example.com/auth'
+        flow.step2_exchange.return_value = credentials
+
+        # It should catch the socket error and proceed as if
+        # noauth_local_webserver was specified.
+        returned_credentials = tools.run_flow(flow, storage)
+
+        self.assertEqual(credentials, returned_credentials)
+        self.assertEqual(flow.redirect_uri, OOB_CALLBACK_URN)
+        flow.step2_exchange.assert_called_once_with('auth_code', http=None)
+        storage.put.assert_called_once_with(credentials)
+        credentials.set_store.assert_called_once_with(storage)
+        self.assertTrue(server_ctor_mock.called)
+
+
+class TestMessageIfMissing(unittest2.TestCase):
+    def test_message_if_missing(self):
+        self.assertIn('somefile.txt', tools.message_if_missing('somefile.txt'))
+
+
 if __name__ == '__main__':  # pragma: NO COVER
     unittest.main()