<s> import json <EOL> import time <EOL> import mock <EOL> import six <EOL> from six . moves import http_client <EOL> from ironicclient . common import filecache <EOL> from ironicclient . common import http <EOL> from ironicclient import exc <EOL> from ironicclient . tests . unit import utils <EOL> DEFAULT_TIMEOUT = <NUM_LIT> <EOL> DEFAULT_HOST = '<STR_LIT:localhost>' <EOL> DEFAULT_PORT = '<STR_LIT>' <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> def _session_client ( ** kwargs ) : <EOL> return http . SessionClient ( os_ironic_api_version = '<STR_LIT>' , <EOL> api_version_select_state = '<STR_LIT:default>' , <EOL> max_retries = <NUM_LIT:5> , <EOL> retry_interval = <NUM_LIT:2> , <EOL> auth = None , <EOL> interface = '<STR_LIT>' , <EOL> service_type = '<STR_LIT>' , <EOL> region_name = '<STR_LIT>' , <EOL> endpoint = '<STR_LIT>' % ( DEFAULT_HOST , <EOL> DEFAULT_PORT ) , <EOL> ** kwargs ) <EOL> class VersionNegotiationMixinTest ( utils . BaseTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( VersionNegotiationMixinTest , self ) . setUp ( ) <EOL> self . test_object = http . VersionNegotiationMixin ( ) <EOL> self . test_object . os_ironic_api_version = '<STR_LIT>' <EOL> self . test_object . api_version_select_state = '<STR_LIT:default>' <EOL> self . test_object . endpoint = "<STR_LIT>" <EOL> self . mock_mcu = mock . MagicMock ( ) <EOL> self . test_object . _make_connection_url = self . mock_mcu <EOL> self . response = utils . FakeResponse ( <EOL> { } , status = http_client . NOT_ACCEPTABLE ) <EOL> self . test_object . get_server = mock . MagicMock ( <EOL> return_value = ( '<STR_LIT:localhost>' , '<STR_LIT>' ) ) <EOL> def test__generic_parse_version_headers_has_headers ( self ) : <EOL> response = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> expected = ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> result = self . test_object . _generic_parse_version_headers ( response . get ) <EOL> self . assertEqual ( expected , result ) <EOL> def test__generic_parse_version_headers_missing_headers ( self ) : <EOL> response = { } <EOL> expected = ( None , None ) <EOL> result = self . test_object . _generic_parse_version_headers ( response . get ) <EOL> self . assertEqual ( expected , result ) <EOL> @ mock . patch . object ( filecache , '<STR_LIT>' , autospec = True ) <EOL> def test_negotiate_version_bad_state ( self , mock_save_data ) : <EOL> self . test_object . api_version_select_state = '<STR_LIT>' <EOL> self . assertRaises ( <EOL> RuntimeError , <EOL> self . test_object . negotiate_version , <EOL> None , None ) <EOL> self . assertEqual ( <NUM_LIT:0> , mock_save_data . call_count ) <EOL> @ mock . patch . object ( filecache , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( http . VersionNegotiationMixin , '<STR_LIT>' , <EOL> autospec = True ) <EOL> def test_negotiate_version_server_older ( self , mock_pvh , mock_save_data ) : <EOL> latest_ver = '<STR_LIT>' <EOL> mock_pvh . return_value = ( '<STR_LIT>' , latest_ver ) <EOL> mock_conn = mock . MagicMock ( ) <EOL> result = self . test_object . negotiate_version ( mock_conn , self . response ) <EOL> self . assertEqual ( latest_ver , result ) <EOL> self . assertEqual ( <NUM_LIT:1> , mock_pvh . call_count ) <EOL> host , port = http . get_server ( self . test_object . endpoint ) <EOL> mock_save_data . assert_called_once_with ( host = host , port = port , <EOL> data = latest_ver ) <EOL> @ mock . patch . object ( filecache , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( http . VersionNegotiationMixin , '<STR_LIT>' , <EOL> autospec = True ) <EOL> def test_negotiate_version_server_newer ( self , mock_pvh , mock_save_data ) : <EOL> mock_pvh . return_value = ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> mock_conn = mock . MagicMock ( ) <EOL> result = self . test_object . negotiate_version ( mock_conn , self . response ) <EOL> self . assertEqual ( '<STR_LIT>' , result ) <EOL> self . assertEqual ( <NUM_LIT:1> , mock_pvh . call_count ) <EOL> mock_save_data . assert_called_once_with ( host = DEFAULT_HOST , <EOL> port = DEFAULT_PORT , <EOL> data = '<STR_LIT>' ) <EOL> @ mock . patch . object ( filecache , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( http . VersionNegotiationMixin , '<STR_LIT>' , <EOL> autospec = True ) <EOL> @ mock . patch . object ( http . VersionNegotiationMixin , '<STR_LIT>' , <EOL> autospec = True ) <EOL> def test_negotiate_version_server_no_version_on_error ( <EOL> self , mock_pvh , mock_msr , mock_save_data ) : <EOL> mock_pvh . side_effect = iter ( [ ( None , None ) , ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> mock_conn = mock . MagicMock ( ) <EOL> result = self . test_object . negotiate_version ( mock_conn , self . response ) <EOL> self . assertEqual ( '<STR_LIT>' , result ) <EOL> self . assertTrue ( mock_msr . called ) <EOL> self . assertEqual ( <NUM_LIT:2> , mock_pvh . call_count ) <EOL> self . assertEqual ( <NUM_LIT:1> , mock_save_data . call_count ) <EOL> @ mock . patch . object ( filecache , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( http . VersionNegotiationMixin , '<STR_LIT>' , <EOL> autospec = True ) <EOL> def test_negotiate_version_server_explicit_too_high ( self , mock_pvh , <EOL> mock_save_data ) : <EOL> mock_pvh . return_value = ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> mock_conn = mock . MagicMock ( ) <EOL> self . test_object . api_version_select_state = '<STR_LIT:user>' <EOL> self . test_object . os_ironic_api_version = '<STR_LIT>' <EOL> self . assertRaises ( <EOL> exc . UnsupportedVersion , <EOL> self . test_object . negotiate_version , <EOL> mock_conn , self . response ) <EOL> self . assertEqual ( <NUM_LIT:1> , mock_pvh . call_count ) <EOL> self . assertEqual ( <NUM_LIT:0> , mock_save_data . call_count ) <EOL> @ mock . patch . object ( filecache , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( http . VersionNegotiationMixin , '<STR_LIT>' , <EOL> autospec = True ) <EOL> def test_negotiate_version_server_explicit_not_supported ( self , mock_pvh , <EOL> mock_save_data ) : <EOL> mock_pvh . return_value = ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> mock_conn = mock . MagicMock ( ) <EOL> self . test_object . api_version_select_state = '<STR_LIT>' <EOL> self . test_object . os_ironic_api_version = '<STR_LIT>' <EOL> self . assertRaises ( <EOL> exc . UnsupportedVersion , <EOL> self . test_object . negotiate_version , <EOL> mock_conn , self . response ) <EOL> self . assertEqual ( <NUM_LIT:1> , mock_pvh . call_count ) <EOL> self . assertEqual ( <NUM_LIT:0> , mock_save_data . call_count ) <EOL> def test_get_server ( self ) : <EOL> host = '<STR_LIT>' <EOL> port = '<STR_LIT>' <EOL> endpoint = '<STR_LIT>' % ( host , port ) <EOL> self . assertEqual ( ( host , port ) , http . get_server ( endpoint ) ) <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_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> client = http . HTTPClient ( '<STR_LIT>' ) <EOL> client . session = utils . FakeSession ( <EOL> { '<STR_LIT:Content-Type>' : '<STR_LIT:application/json>' } , <EOL> six . StringIO ( error_body ) , <EOL> version = <NUM_LIT:1> , <EOL> status_code = http_client . INTERNAL_SERVER_ERROR ) <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> client = http . HTTPClient ( '<STR_LIT>' ) <EOL> client . session = utils . FakeSession ( <EOL> { '<STR_LIT:Content-Type>' : '<STR_LIT:application/json>' } , <EOL> six . StringIO ( error_body ) , <EOL> version = <NUM_LIT:1> , <EOL> status_code = http_client . INTERNAL_SERVER_ERROR ) <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> client = http . HTTPClient ( '<STR_LIT>' ) <EOL> client . session = utils . FakeSession ( <EOL> { '<STR_LIT:Content-Type>' : '<STR_LIT:application/json>' } , <EOL> six . StringIO ( error_body ) , <EOL> version = <NUM_LIT:1> , <EOL> status_code = http_client . INTERNAL_SERVER_ERROR ) <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_https_request_ok ( self ) : <EOL> client = http . HTTPClient ( '<STR_LIT>' ) <EOL> client . session = utils . FakeSession ( <EOL> { '<STR_LIT:Content-Type>' : '<STR_LIT:application/json>' } , <EOL> six . StringIO ( "<STR_LIT>" ) , <EOL> version = <NUM_LIT:1> , <EOL> status_code = http_client . OK ) <EOL> client . json_request ( '<STR_LIT:GET>' , '<STR_LIT>' ) <EOL> def test_server_https_empty_body ( self ) : <EOL> error_body = _get_error_body ( ) <EOL> client = http . HTTPClient ( '<STR_LIT>' ) <EOL> client . session = utils . FakeSession ( <EOL> { '<STR_LIT:Content-Type>' : '<STR_LIT:application/json>' } , <EOL> six . StringIO ( error_body ) , <EOL> version = <NUM_LIT:1> , <EOL> status_code = http_client . INTERNAL_SERVER_ERROR ) <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_401_unauthorized_exception ( self ) : <EOL> error_body = _get_error_body ( ) <EOL> client = http . HTTPClient ( '<STR_LIT>' ) <EOL> client . session = utils . FakeSession ( <EOL> { '<STR_LIT:Content-Type>' : '<STR_LIT>' } , <EOL> six . StringIO ( error_body ) , <EOL> version = <NUM_LIT:1> , <EOL> status_code = http_client . UNAUTHORIZED ) <EOL> self . assertRaises ( exc . Unauthorized , client . json_request , <EOL> '<STR_LIT:GET>' , '<STR_LIT>' ) <EOL> def test_http_request_not_valid_request ( self ) : <EOL> client = http . HTTPClient ( '<STR_LIT>' ) <EOL> client . session . request = mock . Mock ( <EOL> side_effect = http . requests . exceptions . InvalidSchema ) <EOL> self . assertRaises ( exc . ValidationError , client . _http_request , <EOL> '<STR_LIT>' , '<STR_LIT:GET>' ) <EOL> def test__parse_version_headers ( self ) : <EOL> error_body = _get_error_body ( ) <EOL> expected_result = ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> client = http . HTTPClient ( '<STR_LIT>' ) <EOL> fake_resp = utils . FakeSessionResponse ( <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:Content-Type>' : '<STR_LIT>' , <EOL> } , <EOL> six . StringIO ( error_body ) , <EOL> version = <NUM_LIT:1> , <EOL> status_code = http_client . NOT_ACCEPTABLE ) <EOL> result = client . _parse_version_headers ( fake_resp ) <EOL> self . assertEqual ( expected_result , result ) <EOL> @ mock . patch . object ( filecache , '<STR_LIT>' , autospec = True ) <EOL> def test__http_request_client_fallback_fail ( self , mock_save_data ) : <EOL> host , port , latest_ver = '<STR_LIT:localhost>' , '<STR_LIT>' , '<STR_LIT>' <EOL> error_body = _get_error_body ( ) <EOL> client = http . HTTPClient ( '<STR_LIT>' % ( host , port ) ) <EOL> client . session = utils . FakeSession ( <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : latest_ver , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } , <EOL> six . StringIO ( error_body ) , <EOL> version = <NUM_LIT:1> , <EOL> status_code = http_client . NOT_ACCEPTABLE ) <EOL> self . assertRaises ( <EOL> exc . UnsupportedVersion , <EOL> client . _http_request , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT:GET>' ) <EOL> mock_save_data . assert_called_once_with ( host = host , data = latest_ver , <EOL> port = port ) <EOL> @ mock . patch . object ( http . VersionNegotiationMixin , '<STR_LIT>' , <EOL> autospec = False ) <EOL> def test__http_request_client_fallback_success ( self , mock_negotiate ) : <EOL> mock_negotiate . return_value = '<STR_LIT>' <EOL> error_body = _get_error_body ( ) <EOL> bad_resp = utils . FakeSessionResponse ( <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } , <EOL> six . StringIO ( error_body ) , <EOL> version = <NUM_LIT:1> , <EOL> status_code = http_client . NOT_ACCEPTABLE ) <EOL> good_resp = utils . FakeSessionResponse ( <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } , <EOL> six . StringIO ( "<STR_LIT>" ) , <EOL> version = <NUM_LIT:1> , <EOL> status_code = http_client . OK ) <EOL> client = http . HTTPClient ( '<STR_LIT>' ) <EOL> with mock . patch . object ( client , '<STR_LIT>' , <EOL> autospec = True ) as mock_session : <EOL> mock_session . request . side_effect = iter ( [ bad_resp , good_resp ] ) <EOL> response , body_iter = client . _http_request ( '<STR_LIT>' , '<STR_LIT:GET>' ) <EOL> self . assertEqual ( http_client . OK , response . status_code ) <EOL> self . assertEqual ( <NUM_LIT:1> , mock_negotiate . call_count ) <EOL> @ mock . patch . object ( http . LOG , '<STR_LIT>' , autospec = True ) <EOL> def test_log_curl_request_mask_password ( self , mock_log ) : <EOL> client = http . HTTPClient ( '<STR_LIT>' ) <EOL> kwargs = { '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } , <EOL> '<STR_LIT:body>' : '<STR_LIT>' } <EOL> client . log_curl_request ( '<STR_LIT:foo>' , '<STR_LIT>' , kwargs ) <EOL> expected_log = ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> mock_log . assert_called_once_with ( expected_log ) <EOL> @ mock . patch . object ( http . LOG , '<STR_LIT>' , autospec = True ) <EOL> def test_log_http_response_mask_password ( self , mock_log ) : <EOL> client = http . HTTPClient ( '<STR_LIT>' ) <EOL> fake_response = utils . FakeResponse ( { } , version = <NUM_LIT:1> , reason = '<STR_LIT:foo>' , <EOL> status = <NUM_LIT:200> ) <EOL> body = '<STR_LIT>' <EOL> client . log_http_response ( fake_response , body = body ) <EOL> expected_log = ( "<STR_LIT>" ) <EOL> mock_log . assert_called_once_with ( expected_log ) <EOL> def test__https_init_ssl_args_insecure ( self ) : <EOL> client = http . HTTPClient ( '<STR_LIT>' , insecure = True ) <EOL> self . assertEqual ( False , client . session . verify ) <EOL> def test__https_init_ssl_args_secure ( self ) : <EOL> client = http . HTTPClient ( '<STR_LIT>' , ca_file = '<STR_LIT>' , <EOL> key_file = '<STR_LIT>' , cert_file = '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , client . session . verify ) <EOL> self . assertEqual ( ( '<STR_LIT>' , '<STR_LIT>' ) , client . session . cert ) <EOL> @ mock . patch . object ( http . LOG , '<STR_LIT>' , autospec = True ) <EOL> def test_log_curl_request_with_body_and_header ( self , mock_log ) : <EOL> client = http . HTTPClient ( '<STR_LIT>' ) <EOL> headers = { '<STR_LIT>' : '<STR_LIT>' } <EOL> body = '<STR_LIT>' <EOL> client . log_curl_request ( '<STR_LIT:GET>' , '<STR_LIT>' , <EOL> { '<STR_LIT>' : headers , '<STR_LIT:body>' : body } ) <EOL> self . assertTrue ( mock_log . called ) <EOL> self . assertTrue ( mock_log . call_args [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> mock_log . call_args [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] ) <EOL> @ mock . patch . object ( http . LOG , '<STR_LIT>' , autospec = True ) <EOL> def test_log_curl_request_with_certs ( self , mock_log ) : <EOL> headers = { '<STR_LIT>' : '<STR_LIT>' } <EOL> client = http . HTTPClient ( '<STR_LIT>' , key_file = '<STR_LIT:key>' , <EOL> cert_file = '<STR_LIT>' , cacert = '<STR_LIT>' , <EOL> token = '<STR_LIT>' ) <EOL> client . log_curl_request ( '<STR_LIT:GET>' , '<STR_LIT>' , { '<STR_LIT>' : headers } ) <EOL> self . assertTrue ( mock_log . called ) <EOL> self . assertTrue ( mock_log . call_args [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> mock_log . call_args [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] ) <EOL> @ mock . patch . object ( http . LOG , '<STR_LIT>' , autospec = True ) <EOL> def test_log_curl_request_with_insecure_param ( self , mock_log ) : <EOL> headers = { '<STR_LIT>' : '<STR_LIT>' } <EOL> http_client_object = http . HTTPClient ( '<STR_LIT>' , insecure = True , <EOL> token = '<STR_LIT>' ) <EOL> http_client_object . log_curl_request ( '<STR_LIT:GET>' , '<STR_LIT>' , <EOL> { '<STR_LIT>' : headers } ) <EOL> self . assertTrue ( mock_log . called ) <EOL> self . assertTrue ( mock_log . call_args [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> mock_log . call_args [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] ) <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> http_client . INTERNAL_SERVER_ERROR ) <EOL> client = _session_client ( 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> http_client . INTERNAL_SERVER_ERROR ) <EOL> client = _session_client ( 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__parse_version_headers ( self ) : <EOL> fake_session = utils . FakeSession ( <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } , <EOL> None , <EOL> http_client . HTTP_VERSION_NOT_SUPPORTED ) <EOL> expected_result = ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> client = _session_client ( session = fake_session ) <EOL> result = client . _parse_version_headers ( fake_session ) <EOL> self . assertEqual ( expected_result , result ) <EOL> def _test_endpoint_override ( self , endpoint ) : <EOL> fake_session = utils . FakeSession ( { '<STR_LIT>' : '<STR_LIT:application/json>' } , <EOL> status_code = http_client . NO_CONTENT ) <EOL> request_mock = mock . Mock ( ) <EOL> fake_session . request = request_mock <EOL> request_mock . return_value = utils . FakeSessionResponse ( <EOL> headers = { '<STR_LIT>' : '<STR_LIT:application/json>' } , <EOL> status_code = http_client . NO_CONTENT ) <EOL> client = _session_client ( session = fake_session , <EOL> endpoint_override = endpoint ) <EOL> client . json_request ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> expected_args_dict = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:Content-Type>' : '<STR_LIT:application/json>' , <EOL> '<STR_LIT>' : '<STR_LIT:application/json>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } , <EOL> '<STR_LIT>' : None , '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> } <EOL> if isinstance ( endpoint , six . string_types ) : <EOL> trimmed = http . _trim_endpoint_api_version ( endpoint ) <EOL> expected_args_dict [ '<STR_LIT>' ] = trimmed <EOL> request_mock . assert_called_once_with ( <EOL> '<STR_LIT>' , '<STR_LIT>' , raise_exc = False , <EOL> ** expected_args_dict <EOL> ) <EOL> def test_endpoint_override ( self ) : <EOL> self . _test_endpoint_override ( '<STR_LIT>' ) <EOL> def test_endpoint_override_with_version ( self ) : <EOL> self . _test_endpoint_override ( '<STR_LIT>' ) <EOL> def test_endpoint_override_not_valid ( self ) : <EOL> self . _test_endpoint_override ( True ) <EOL> @ mock . patch . object ( time , '<STR_LIT>' , lambda * _ : None ) <EOL> class RetriesTestCase ( utils . BaseTestCase ) : <EOL> def test_http_no_retry ( self ) : <EOL> error_body = _get_error_body ( ) <EOL> bad_resp = utils . FakeSessionResponse ( <EOL> { '<STR_LIT:Content-Type>' : '<STR_LIT>' } , <EOL> six . StringIO ( error_body ) , <EOL> version = <NUM_LIT:1> , <EOL> status_code = http_client . CONFLICT ) <EOL> client = http . HTTPClient ( '<STR_LIT>' , max_retries = <NUM_LIT:0> ) <EOL> with mock . patch . object ( client . session , '<STR_LIT>' , autospec = True , <EOL> return_value = bad_resp ) as mock_request : <EOL> self . assertRaises ( exc . Conflict , client . _http_request , <EOL> '<STR_LIT>' , '<STR_LIT:GET>' ) <EOL> self . assertEqual ( <NUM_LIT:1> , mock_request . call_count ) <EOL> def test_http_retry ( self ) : <EOL> error_body = _get_error_body ( ) <EOL> bad_resp = utils . FakeSessionResponse ( <EOL> { '<STR_LIT:Content-Type>' : '<STR_LIT>' } , <EOL> six . StringIO ( error_body ) , <EOL> version = <NUM_LIT:1> , <EOL> status_code = http_client . CONFLICT ) <EOL> good_resp = utils . FakeSessionResponse ( <EOL> { '<STR_LIT:Content-Type>' : '<STR_LIT>' } , <EOL> six . StringIO ( "<STR_LIT>" ) , <EOL> version = <NUM_LIT:1> , <EOL> status_code = http_client . OK ) <EOL> client = http . HTTPClient ( '<STR_LIT>' ) <EOL> with mock . patch . object ( client , '<STR_LIT>' , <EOL> autospec = True ) as mock_session : <EOL> mock_session . request . side_effect = iter ( [ bad_resp , good_resp ] ) <EOL> response , body_iter = client . _http_request ( '<STR_LIT>' , '<STR_LIT:GET>' ) <EOL> self . assertEqual ( http_client . OK , response . status_code ) <EOL> self . assertEqual ( <NUM_LIT:2> , mock_session . request . call_count ) <EOL> def test_http_retry_503 ( self ) : <EOL> error_body = _get_error_body ( ) <EOL> bad_resp = utils . FakeSessionResponse ( <EOL> { '<STR_LIT:Content-Type>' : '<STR_LIT>' } , <EOL> six . StringIO ( error_body ) , <EOL> version = <NUM_LIT:1> , <EOL> status_code = http_client . SERVICE_UNAVAILABLE ) <EOL> good_resp = utils . FakeSessionResponse ( <EOL> { '<STR_LIT:Content-Type>' : '<STR_LIT>' } , <EOL> six . StringIO ( "<STR_LIT>" ) , <EOL> version = <NUM_LIT:1> , <EOL> status_code = http_client . OK ) <EOL> client = http . HTTPClient ( '<STR_LIT>' ) <EOL> with mock . patch . object ( client , '<STR_LIT>' , <EOL> autospec = True ) as mock_session : <EOL> mock_session . request . side_effect = iter ( [ bad_resp , good_resp ] ) <EOL> response , body_iter = client . _http_request ( '<STR_LIT>' , '<STR_LIT:GET>' ) <EOL> self . assertEqual ( http_client . OK , response . status_code ) <EOL> self . assertEqual ( <NUM_LIT:2> , mock_session . request . call_count ) <EOL> def test_http_retry_connection_refused ( self ) : <EOL> good_resp = utils . FakeSessionResponse ( <EOL> { '<STR_LIT>' : '<STR_LIT>' } , <EOL> six . StringIO ( "<STR_LIT>" ) , <EOL> version = <NUM_LIT:1> , <EOL> status_code = http_client . OK ) <EOL> client = http . HTTPClient ( '<STR_LIT>' ) <EOL> with mock . patch . object ( client , '<STR_LIT>' , <EOL> autospec = True ) as mock_session : <EOL> mock_session . request . side_effect = iter ( [ exc . ConnectionRefused ( ) , <EOL> good_resp ] ) <EOL> response , body_iter = client . _http_request ( '<STR_LIT>' , '<STR_LIT:GET>' ) <EOL> self . assertEqual ( http_client . OK , response . status_code ) <EOL> self . assertEqual ( <NUM_LIT:2> , mock_session . request . call_count ) <EOL> def test_http_failed_retry ( self ) : <EOL> error_body = _get_error_body ( ) <EOL> bad_resp = utils . FakeSessionResponse ( <EOL> { '<STR_LIT>' : '<STR_LIT>' } , <EOL> six . StringIO ( error_body ) , <EOL> version = <NUM_LIT:1> , <EOL> status_code = http_client . CONFLICT ) <EOL> client = http . HTTPClient ( '<STR_LIT>' ) <EOL> with mock . patch . object ( client , '<STR_LIT>' , <EOL> autospec = True ) as mock_session : <EOL> mock_session . request . return_value = bad_resp <EOL> self . assertRaises ( exc . Conflict , client . _http_request , <EOL> '<STR_LIT>' , '<STR_LIT:GET>' ) <EOL> self . assertEqual ( http . DEFAULT_MAX_RETRIES + <NUM_LIT:1> , <EOL> mock_session . request . call_count ) <EOL> def test_http_max_retries_none ( self ) : <EOL> error_body = _get_error_body ( ) <EOL> bad_resp = utils . FakeSessionResponse ( <EOL> { '<STR_LIT>' : '<STR_LIT>' } , <EOL> six . StringIO ( error_body ) , <EOL> version = <NUM_LIT:1> , <EOL> status_code = http_client . CONFLICT ) <EOL> client = http . HTTPClient ( '<STR_LIT>' , max_retries = None ) <EOL> with mock . patch . object ( client , '<STR_LIT>' , <EOL> autospec = True ) as mock_session : <EOL> mock_session . request . return_value = bad_resp <EOL> self . assertRaises ( exc . Conflict , client . _http_request , <EOL> '<STR_LIT>' , '<STR_LIT:GET>' ) <EOL> self . assertEqual ( http . DEFAULT_MAX_RETRIES + <NUM_LIT:1> , <EOL> mock_session . request . call_count ) <EOL> def test_http_change_max_retries ( self ) : <EOL> error_body = _get_error_body ( ) <EOL> bad_resp = utils . FakeSessionResponse ( <EOL> { '<STR_LIT>' : '<STR_LIT>' } , <EOL> six . StringIO ( error_body ) , <EOL> version = <NUM_LIT:1> , <EOL> status_code = http_client . CONFLICT ) <EOL> client = http . HTTPClient ( '<STR_LIT>' , <EOL> max_retries = http . DEFAULT_MAX_RETRIES + <NUM_LIT:1> ) <EOL> with mock . patch . object ( client , '<STR_LIT>' , <EOL> autospec = True ) as mock_session : <EOL> mock_session . request . return_value = bad_resp <EOL> self . assertRaises ( exc . Conflict , client . _http_request , <EOL> '<STR_LIT>' , '<STR_LIT:GET>' ) <EOL> self . assertEqual ( http . DEFAULT_MAX_RETRIES + <NUM_LIT:2> , <EOL> mock_session . request . call_count ) <EOL> def test_session_retry ( self ) : <EOL> error_body = _get_error_body ( ) <EOL> fake_resp = utils . FakeSessionResponse ( <EOL> { '<STR_LIT:Content-Type>' : '<STR_LIT:application/json>' } , <EOL> error_body , <EOL> http_client . CONFLICT ) <EOL> ok_resp = utils . FakeSessionResponse ( <EOL> { '<STR_LIT:Content-Type>' : '<STR_LIT:application/json>' } , <EOL> b"<STR_LIT:OK>" , <EOL> http_client . OK ) <EOL> fake_session = mock . Mock ( spec = utils . FakeSession ) <EOL> fake_session . request . side_effect = iter ( ( fake_resp , ok_resp ) ) <EOL> client = _session_client ( session = fake_session ) <EOL> client . json_request ( '<STR_LIT:GET>' , '<STR_LIT>' ) <EOL> self . assertEqual ( <NUM_LIT:2> , fake_session . request . call_count ) <EOL> def test_session_retry_503 ( self ) : <EOL> error_body = _get_error_body ( ) <EOL> fake_resp = utils . FakeSessionResponse ( <EOL> { '<STR_LIT:Content-Type>' : '<STR_LIT:application/json>' } , <EOL> error_body , <EOL> http_client . SERVICE_UNAVAILABLE ) <EOL> ok_resp = utils . FakeSessionResponse ( <EOL> { '<STR_LIT:Content-Type>' : '<STR_LIT:application/json>' } , <EOL> b"<STR_LIT:OK>" , <EOL> http_client . OK ) <EOL> fake_session = mock . Mock ( spec = utils . FakeSession ) <EOL> fake_session . request . side_effect = iter ( ( fake_resp , ok_resp ) ) <EOL> client = _session_client ( session = fake_session ) <EOL> client . json_request ( '<STR_LIT:GET>' , '<STR_LIT>' ) <EOL> self . assertEqual ( <NUM_LIT:2> , fake_session . request . call_count ) <EOL> def test_session_retry_connection_refused ( self ) : <EOL> ok_resp = utils . FakeSessionResponse ( <EOL> { '<STR_LIT:Content-Type>' : '<STR_LIT:application/json>' } , <EOL> b"<STR_LIT:OK>" , <EOL> http_client . OK ) <EOL> fake_session = mock . Mock ( spec = utils . FakeSession ) <EOL> fake_session . request . side_effect = iter ( ( exc . ConnectionRefused ( ) , <EOL> ok_resp ) ) <EOL> client = _session_client ( session = fake_session ) <EOL> client . json_request ( '<STR_LIT:GET>' , '<STR_LIT>' ) <EOL> self . assertEqual ( <NUM_LIT:2> , fake_session . request . call_count ) <EOL> def test_session_retry_fail ( self ) : <EOL> error_body = _get_error_body ( ) <EOL> fake_resp = utils . FakeSessionResponse ( <EOL> { '<STR_LIT:Content-Type>' : '<STR_LIT:application/json>' } , <EOL> error_body , <EOL> http_client . CONFLICT ) <EOL> fake_session = mock . Mock ( spec = utils . FakeSession ) <EOL> fake_session . request . return_value = fake_resp <EOL> client = _session_client ( session = fake_session ) <EOL> self . assertRaises ( exc . Conflict , client . json_request , <EOL> '<STR_LIT:GET>' , '<STR_LIT>' ) <EOL> self . assertEqual ( http . DEFAULT_MAX_RETRIES + <NUM_LIT:1> , <EOL> fake_session . request . call_count ) <EOL> def test_session_max_retries_none ( self ) : <EOL> error_body = _get_error_body ( ) <EOL> fake_resp = utils . FakeSessionResponse ( <EOL> { '<STR_LIT:Content-Type>' : '<STR_LIT:application/json>' } , <EOL> error_body , <EOL> http_client . CONFLICT ) <EOL> fake_session = mock . Mock ( spec = utils . FakeSession ) <EOL> fake_session . request . return_value = fake_resp <EOL> client = _session_client ( session = fake_session ) <EOL> client . conflict_max_retries = None <EOL> self . assertRaises ( exc . Conflict , client . json_request , <EOL> '<STR_LIT:GET>' , '<STR_LIT>' ) <EOL> self . assertEqual ( http . DEFAULT_MAX_RETRIES + <NUM_LIT:1> , <EOL> fake_session . request . call_count ) <EOL> def test_session_change_max_retries ( self ) : <EOL> error_body = _get_error_body ( ) <EOL> fake_resp = utils . FakeSessionResponse ( <EOL> { '<STR_LIT:Content-Type>' : '<STR_LIT:application/json>' } , <EOL> error_body , <EOL> http_client . CONFLICT ) <EOL> fake_session = mock . Mock ( spec = utils . FakeSession ) <EOL> fake_session . request . return_value = fake_resp <EOL> client = _session_client ( session = fake_session ) <EOL> client . conflict_max_retries = http . DEFAULT_MAX_RETRIES + <NUM_LIT:1> <EOL> self . assertRaises ( exc . Conflict , client . json_request , <EOL> '<STR_LIT:GET>' , '<STR_LIT>' ) <EOL> self . assertEqual ( http . DEFAULT_MAX_RETRIES + <NUM_LIT:2> , <EOL> fake_session . request . call_count ) </s>
