<s> import socket <EOL> import sys <EOL> import threading <EOL> import mock <EOL> from six . moves . urllib import request <EOL> import unittest2 <EOL> from oauth2client . client import FlowExchangeError <EOL> from oauth2client . client import OOB_CALLBACK_URN <EOL> from oauth2client import tools <EOL> try : <EOL> import argparse <EOL> except ImportError : <EOL> raise unittest2 . SkipTest ( '<STR_LIT>' ) <EOL> class TestClientRedirectServer ( unittest2 . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_ClientRedirectServer ( self ) : <EOL> httpd = tools . ClientRedirectServer ( ( '<STR_LIT:localhost>' , <NUM_LIT:0> ) , <EOL> tools . ClientRedirectHandler ) <EOL> code = '<STR_LIT:foo>' <EOL> url = '<STR_LIT>' % ( httpd . server_address [ <NUM_LIT:1> ] , code ) <EOL> t = threading . Thread ( target = httpd . handle_request ) <EOL> t . setDaemon ( True ) <EOL> t . start ( ) <EOL> f = request . urlopen ( url ) <EOL> self . assertTrue ( f . read ( ) ) <EOL> t . join ( ) <EOL> httpd . server_close ( ) <EOL> self . assertEqual ( httpd . query_params . get ( '<STR_LIT:code>' ) , code ) <EOL> class TestRunFlow ( unittest2 . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . server = mock . Mock ( ) <EOL> self . flow = mock . Mock ( ) <EOL> self . storage = mock . Mock ( ) <EOL> self . credentials = mock . Mock ( ) <EOL> self . flow . step1_get_authorize_url . return_value = ( <EOL> '<STR_LIT>' ) <EOL> self . flow . step2_exchange . return_value = self . credentials <EOL> self . flags = argparse . Namespace ( <EOL> noauth_local_webserver = True , logging_level = '<STR_LIT>' ) <EOL> self . server_flags = argparse . Namespace ( <EOL> noauth_local_webserver = False , <EOL> logging_level = '<STR_LIT>' , <EOL> auth_host_port = [ <NUM_LIT> , ] , <EOL> auth_host_name = '<STR_LIT:localhost>' ) <EOL> @ mock . patch . object ( sys , '<STR_LIT>' , <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_run_flow_no_webserver ( self , input_mock , logging_mock ) : <EOL> input_mock . return_value = '<STR_LIT>' <EOL> returned_credentials = tools . run_flow ( self . flow , self . storage ) <EOL> self . assertEqual ( self . credentials , returned_credentials ) <EOL> self . assertEqual ( self . flow . redirect_uri , OOB_CALLBACK_URN ) <EOL> self . flow . step2_exchange . assert_called_once_with ( <EOL> '<STR_LIT>' , http = None ) <EOL> self . storage . put . assert_called_once_with ( self . credentials ) <EOL> self . credentials . set_store . assert_called_once_with ( self . storage ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_run_flow_no_webserver_explicit_flags ( <EOL> self , input_mock , logging_mock ) : <EOL> input_mock . return_value = '<STR_LIT>' <EOL> returned_credentials = tools . run_flow ( <EOL> self . flow , self . storage , flags = self . flags ) <EOL> self . assertEqual ( self . credentials , returned_credentials ) <EOL> self . assertEqual ( self . flow . redirect_uri , OOB_CALLBACK_URN ) <EOL> self . flow . step2_exchange . assert_called_once_with ( <EOL> '<STR_LIT>' , http = None ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_run_flow_no_webserver_exchange_error ( <EOL> self , input_mock , logging_mock ) : <EOL> input_mock . return_value = '<STR_LIT>' <EOL> self . flow . step2_exchange . side_effect = FlowExchangeError ( ) <EOL> with self . assertRaises ( SystemExit ) : <EOL> tools . run_flow ( self . flow , self . storage , flags = self . flags ) <EOL> self . flow . step2_exchange . assert_called_once_with ( <EOL> '<STR_LIT>' , http = None ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_run_flow_webserver ( <EOL> self , webbrowser_open_mock , server_ctor_mock , logging_mock ) : <EOL> server_ctor_mock . return_value = self . server <EOL> self . server . query_params = { '<STR_LIT:code>' : '<STR_LIT>' } <EOL> returned_credentials = tools . run_flow ( <EOL> self . flow , self . storage , flags = self . server_flags ) <EOL> self . assertEqual ( self . credentials , returned_credentials ) <EOL> self . assertEqual ( self . flow . redirect_uri , '<STR_LIT>' ) <EOL> self . flow . step2_exchange . assert_called_once_with ( <EOL> '<STR_LIT>' , http = None ) <EOL> self . storage . put . assert_called_once_with ( self . credentials ) <EOL> self . credentials . set_store . assert_called_once_with ( self . storage ) <EOL> self . assertTrue ( self . server . handle_request . called ) <EOL> webbrowser_open_mock . assert_called_once_with ( <EOL> '<STR_LIT>' , autoraise = True , new = <NUM_LIT:1> ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_run_flow_webserver_exchange_error ( <EOL> self , webbrowser_open_mock , server_ctor_mock , logging_mock ) : <EOL> server_ctor_mock . return_value = self . server <EOL> self . server . query_params = { '<STR_LIT:error>' : '<STR_LIT>' } <EOL> with self . assertRaises ( SystemExit ) : <EOL> returned_credentials = tools . run_flow ( <EOL> self . flow , self . storage , flags = self . server_flags ) <EOL> self . assertTrue ( self . server . handle_request . called ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_run_flow_webserver_no_code ( <EOL> self , webbrowser_open_mock , server_ctor_mock , logging_mock ) : <EOL> server_ctor_mock . return_value = self . server <EOL> self . server . query_params = { } <EOL> with self . assertRaises ( SystemExit ) : <EOL> returned_credentials = tools . run_flow ( <EOL> self . flow , self . storage , flags = self . server_flags ) <EOL> self . assertTrue ( self . server . handle_request . called ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_run_flow_webserver_fallback ( <EOL> self , input_mock , server_ctor_mock , logging_mock ) : <EOL> server_ctor_mock . side_effect = socket . error ( ) <EOL> input_mock . return_value = '<STR_LIT>' <EOL> returned_credentials = tools . run_flow ( <EOL> self . flow , self . storage , flags = self . server_flags ) <EOL> self . assertEqual ( self . credentials , returned_credentials ) <EOL> self . assertEqual ( self . flow . redirect_uri , OOB_CALLBACK_URN ) <EOL> self . flow . step2_exchange . assert_called_once_with ( <EOL> '<STR_LIT>' , http = None ) <EOL> self . assertTrue ( server_ctor_mock . called ) <EOL> self . assertFalse ( self . server . handle_request . called ) <EOL> class TestMessageIfMissing ( unittest2 . TestCase ) : <EOL> def test_message_if_missing ( self ) : <EOL> self . assertIn ( '<STR_LIT>' , tools . message_if_missing ( '<STR_LIT>' ) ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
