<s> import json <EOL> import mock <EOL> import six <EOL> from magnumclient . common . apiclient . exceptions import GatewayTimeout <EOL> from magnumclient . common import httpclient as http <EOL> from magnumclient import exceptions as exc <EOL> from magnumclient . tests import utils <EOL> def _get_error_body ( faultstring = None , debuginfo = None ) : <EOL> error_body = { <EOL> '<STR_LIT>' : faultstring , <EOL> '<STR_LIT>' : debuginfo <EOL> } <EOL> raw_error_body = json . dumps ( error_body ) <EOL> body = { '<STR_LIT>' : raw_error_body } <EOL> raw_body = json . dumps ( body ) <EOL> return raw_body <EOL> HTTP_CLASS = six . moves . http_client . HTTPConnection <EOL> HTTPS_CLASS = http . VerifiedHTTPSConnection <EOL> DEFAULT_TIMEOUT = <NUM_LIT> <EOL> class HttpClientTest ( utils . BaseTestCase ) : <EOL> def test_url_generation_trailing_slash_in_base ( self ) : <EOL> client = http . HTTPClient ( '<STR_LIT>' ) <EOL> url = client . _make_connection_url ( '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , url ) <EOL> def test_url_generation_without_trailing_slash_in_base ( self ) : <EOL> client = http . HTTPClient ( '<STR_LIT>' ) <EOL> url = client . _make_connection_url ( '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , url ) <EOL> def test_url_generation_prefix_slash_in_path ( self ) : <EOL> client = http . HTTPClient ( '<STR_LIT>' ) <EOL> url = client . _make_connection_url ( '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , url ) <EOL> def test_url_generation_without_prefix_slash_in_path ( self ) : <EOL> client = http . HTTPClient ( '<STR_LIT>' ) <EOL> url = client . _make_connection_url ( '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , url ) <EOL> def test_server_exception_empty_body ( self ) : <EOL> error_body = _get_error_body ( ) <EOL> fake_resp = utils . FakeResponse ( { '<STR_LIT>' : '<STR_LIT:application/json>' } , <EOL> six . StringIO ( error_body ) , <EOL> version = <NUM_LIT:1> , <EOL> status = <NUM_LIT> ) <EOL> client = http . HTTPClient ( '<STR_LIT>' ) <EOL> client . get_connection = ( <EOL> lambda * a , ** kw : utils . FakeConnection ( fake_resp ) ) <EOL> error = self . assertRaises ( exc . InternalServerError , <EOL> client . json_request , <EOL> '<STR_LIT:GET>' , '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , str ( error ) ) <EOL> def test_server_exception_msg_only ( self ) : <EOL> error_msg = '<STR_LIT>' <EOL> error_body = _get_error_body ( error_msg ) <EOL> fake_resp = utils . FakeResponse ( { '<STR_LIT>' : '<STR_LIT:application/json>' } , <EOL> six . StringIO ( error_body ) , <EOL> version = <NUM_LIT:1> , <EOL> status = <NUM_LIT> ) <EOL> client = http . HTTPClient ( '<STR_LIT>' ) <EOL> client . get_connection = ( <EOL> lambda * a , ** kw : utils . FakeConnection ( fake_resp ) ) <EOL> error = self . assertRaises ( exc . InternalServerError , <EOL> client . json_request , <EOL> '<STR_LIT:GET>' , '<STR_LIT>' ) <EOL> self . assertEqual ( error_msg + '<STR_LIT>' , str ( error ) ) <EOL> def test_server_exception_msg_and_traceback ( self ) : <EOL> error_msg = '<STR_LIT>' <EOL> error_trace = ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> error_body = _get_error_body ( error_msg , error_trace ) <EOL> fake_resp = utils . FakeResponse ( { '<STR_LIT>' : '<STR_LIT:application/json>' } , <EOL> six . StringIO ( error_body ) , <EOL> version = <NUM_LIT:1> , <EOL> status = <NUM_LIT> ) <EOL> client = http . HTTPClient ( '<STR_LIT>' ) <EOL> client . get_connection = ( <EOL> lambda * a , ** kw : utils . FakeConnection ( fake_resp ) ) <EOL> error = self . assertRaises ( exc . InternalServerError , <EOL> client . json_request , <EOL> '<STR_LIT:GET>' , '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> '<STR_LIT>' % { '<STR_LIT:error>' : error_msg , <EOL> '<STR_LIT>' : error_trace } , <EOL> "<STR_LIT>" % { '<STR_LIT:error>' : str ( error ) , <EOL> '<STR_LIT>' : str ( error . details ) } ) <EOL> def test_get_connection_params ( self ) : <EOL> endpoint = '<STR_LIT>' <EOL> expected = ( HTTP_CLASS , <EOL> ( '<STR_LIT>' , <NUM_LIT> , '<STR_LIT>' ) , <EOL> { '<STR_LIT>' : DEFAULT_TIMEOUT } ) <EOL> params = http . HTTPClient . get_connection_params ( endpoint ) <EOL> self . assertEqual ( expected , params ) <EOL> def test_get_connection_params_with_trailing_slash ( self ) : <EOL> endpoint = '<STR_LIT>' <EOL> expected = ( HTTP_CLASS , <EOL> ( '<STR_LIT>' , <NUM_LIT> , '<STR_LIT>' ) , <EOL> { '<STR_LIT>' : DEFAULT_TIMEOUT } ) <EOL> params = http . HTTPClient . get_connection_params ( endpoint ) <EOL> self . assertEqual ( expected , params ) <EOL> def test_get_connection_params_with_ssl ( self ) : <EOL> endpoint = '<STR_LIT>' <EOL> expected = ( HTTPS_CLASS , <EOL> ( '<STR_LIT>' , <NUM_LIT> , '<STR_LIT>' ) , <EOL> { <EOL> '<STR_LIT>' : DEFAULT_TIMEOUT , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : False , <EOL> } ) <EOL> params = http . HTTPClient . get_connection_params ( endpoint ) <EOL> self . assertEqual ( expected , params ) <EOL> def test_get_connection_params_with_ssl_params ( self ) : <EOL> endpoint = '<STR_LIT>' <EOL> ssl_args = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : True , <EOL> } <EOL> expected_kwargs = { '<STR_LIT>' : DEFAULT_TIMEOUT } <EOL> expected_kwargs . update ( ssl_args ) <EOL> expected = ( HTTPS_CLASS , <EOL> ( '<STR_LIT>' , <NUM_LIT> , '<STR_LIT>' ) , <EOL> expected_kwargs ) <EOL> params = http . HTTPClient . get_connection_params ( endpoint , ** ssl_args ) <EOL> self . assertEqual ( expected , params ) <EOL> def test_get_connection_params_with_timeout ( self ) : <EOL> endpoint = '<STR_LIT>' <EOL> expected = ( HTTP_CLASS , <EOL> ( '<STR_LIT>' , <NUM_LIT> , '<STR_LIT>' ) , <EOL> { '<STR_LIT>' : <NUM_LIT> } ) <EOL> params = http . HTTPClient . get_connection_params ( endpoint , timeout = <NUM_LIT> ) <EOL> self . assertEqual ( expected , params ) <EOL> def test_get_connection_params_with_version ( self ) : <EOL> endpoint = '<STR_LIT>' <EOL> expected = ( HTTP_CLASS , <EOL> ( '<STR_LIT>' , <NUM_LIT> , '<STR_LIT>' ) , <EOL> { '<STR_LIT>' : DEFAULT_TIMEOUT } ) <EOL> params = http . HTTPClient . get_connection_params ( endpoint ) <EOL> self . assertEqual ( expected , params ) <EOL> def test_get_connection_params_with_version_trailing_slash ( self ) : <EOL> endpoint = '<STR_LIT>' <EOL> expected = ( HTTP_CLASS , <EOL> ( '<STR_LIT>' , <NUM_LIT> , '<STR_LIT>' ) , <EOL> { '<STR_LIT>' : DEFAULT_TIMEOUT } ) <EOL> params = http . HTTPClient . get_connection_params ( endpoint ) <EOL> self . assertEqual ( expected , params ) <EOL> def test_get_connection_params_with_subpath ( self ) : <EOL> endpoint = '<STR_LIT>' <EOL> expected = ( HTTP_CLASS , <EOL> ( '<STR_LIT>' , <NUM_LIT> , '<STR_LIT>' ) , <EOL> { '<STR_LIT>' : DEFAULT_TIMEOUT } ) <EOL> params = http . HTTPClient . get_connection_params ( endpoint ) <EOL> self . assertEqual ( expected , params ) <EOL> def test_get_connection_params_with_subpath_trailing_slash ( self ) : <EOL> endpoint = '<STR_LIT>' <EOL> expected = ( HTTP_CLASS , <EOL> ( '<STR_LIT>' , <NUM_LIT> , '<STR_LIT>' ) , <EOL> { '<STR_LIT>' : DEFAULT_TIMEOUT } ) <EOL> params = http . HTTPClient . get_connection_params ( endpoint ) <EOL> self . assertEqual ( expected , params ) <EOL> def test_get_connection_params_with_subpath_version ( self ) : <EOL> endpoint = '<STR_LIT>' <EOL> expected = ( HTTP_CLASS , <EOL> ( '<STR_LIT>' , <NUM_LIT> , '<STR_LIT>' ) , <EOL> { '<STR_LIT>' : DEFAULT_TIMEOUT } ) <EOL> params = http . HTTPClient . get_connection_params ( endpoint ) <EOL> self . assertEqual ( expected , params ) <EOL> def test_get_connection_params_with_subpath_version_trailing_slash ( self ) : <EOL> endpoint = '<STR_LIT>' <EOL> expected = ( HTTP_CLASS , <EOL> ( '<STR_LIT>' , <NUM_LIT> , '<STR_LIT>' ) , <EOL> { '<STR_LIT>' : DEFAULT_TIMEOUT } ) <EOL> params = http . HTTPClient . get_connection_params ( endpoint ) <EOL> self . assertEqual ( expected , params ) <EOL> def test_401_unauthorized_exception ( self ) : <EOL> error_body = _get_error_body ( ) <EOL> fake_resp = utils . FakeResponse ( { '<STR_LIT>' : '<STR_LIT>' } , <EOL> six . StringIO ( error_body ) , <EOL> version = <NUM_LIT:1> , <EOL> status = <NUM_LIT> ) <EOL> client = http . HTTPClient ( '<STR_LIT>' ) <EOL> client . get_connection = ( lambda * a , <EOL> ** kw : utils . FakeConnection ( fake_resp ) ) <EOL> self . assertRaises ( exc . Unauthorized , client . json_request , <EOL> '<STR_LIT:GET>' , '<STR_LIT>' ) <EOL> class SessionClientTest ( utils . BaseTestCase ) : <EOL> def test_server_exception_msg_and_traceback ( self ) : <EOL> error_msg = '<STR_LIT>' <EOL> error_trace = ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> error_body = _get_error_body ( error_msg , error_trace ) <EOL> fake_session = utils . FakeSession ( { '<STR_LIT:Content-Type>' : '<STR_LIT:application/json>' } , <EOL> error_body , <EOL> <NUM_LIT> ) <EOL> client = http . SessionClient ( session = fake_session ) <EOL> error = self . assertRaises ( exc . InternalServerError , <EOL> client . json_request , <EOL> '<STR_LIT:GET>' , '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> '<STR_LIT>' % { '<STR_LIT:error>' : error_msg , <EOL> '<STR_LIT>' : error_trace } , <EOL> "<STR_LIT>" % { '<STR_LIT:error>' : str ( error ) , <EOL> '<STR_LIT>' : str ( error . details ) } ) <EOL> def test_server_exception_empty_body ( self ) : <EOL> error_body = _get_error_body ( ) <EOL> fake_session = utils . FakeSession ( { '<STR_LIT:Content-Type>' : '<STR_LIT:application/json>' } , <EOL> error_body , <EOL> <NUM_LIT> ) <EOL> client = http . SessionClient ( session = fake_session ) <EOL> error = self . assertRaises ( exc . InternalServerError , <EOL> client . json_request , <EOL> '<STR_LIT:GET>' , '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , str ( error ) ) <EOL> def test_bypass_url ( self ) : <EOL> fake_response = utils . FakeSessionResponse ( <EOL> { } , content = "<STR_LIT>" , status_code = <NUM_LIT> ) <EOL> fake_session = mock . MagicMock ( ) <EOL> fake_session . request . side_effect = [ fake_response ] <EOL> client = http . SessionClient ( <EOL> session = fake_session , endpoint_override = '<STR_LIT>' ) <EOL> client . json_request ( '<STR_LIT:GET>' , '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> fake_session . request . call_args [ <NUM_LIT:1> ] [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' <EOL> ) <EOL> def test_exception ( self ) : <EOL> fake_response = utils . FakeSessionResponse ( <EOL> { } , content = "<STR_LIT>" , status_code = <NUM_LIT> ) <EOL> fake_session = mock . MagicMock ( ) <EOL> fake_session . request . side_effect = [ fake_response ] <EOL> client = http . SessionClient ( <EOL> session = fake_session , endpoint_override = '<STR_LIT>' ) <EOL> self . assertRaises ( GatewayTimeout , <EOL> client . json_request , <EOL> '<STR_LIT:GET>' , '<STR_LIT>' ) </s>
