<s> """<STR_LIT>""" <EOL> import datetime <EOL> import json <EOL> import unittest <EOL> from six . moves import http_client <EOL> import mock <EOL> import requests <EOL> from acme import challenges <EOL> from acme import errors <EOL> from acme import jose <EOL> from acme import jws as acme_jws <EOL> from acme import messages <EOL> from acme import messages_test <EOL> from acme import test_util <EOL> CERT_DER = test_util . load_vector ( '<STR_LIT>' ) <EOL> KEY = jose . JWKRSA . load ( test_util . load_vector ( '<STR_LIT>' ) ) <EOL> KEY2 = jose . JWKRSA . load ( test_util . load_vector ( '<STR_LIT>' ) ) <EOL> class ClientTest ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . response = mock . MagicMock ( <EOL> ok = True , status_code = http_client . OK , headers = { } , links = { } ) <EOL> self . net = mock . MagicMock ( ) <EOL> self . net . post . return_value = self . response <EOL> self . net . get . return_value = self . response <EOL> self . directory = messages . Directory ( { <EOL> messages . NewRegistration : <EOL> '<STR_LIT>' , <EOL> messages . Revocation : <EOL> '<STR_LIT>' , <EOL> messages . NewAuthorization : <EOL> '<STR_LIT>' , <EOL> } ) <EOL> from acme . client import Client <EOL> self . client = Client ( <EOL> directory = self . directory , key = KEY , alg = jose . RS256 , net = self . net ) <EOL> self . identifier = messages . Identifier ( <EOL> typ = messages . IDENTIFIER_FQDN , value = '<STR_LIT>' ) <EOL> self . contact = ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> reg = messages . Registration ( <EOL> contact = self . contact , key = KEY . public_key ( ) ) <EOL> self . new_reg = messages . NewRegistration ( ** dict ( reg ) ) <EOL> self . regr = messages . RegistrationResource ( <EOL> body = reg , uri = '<STR_LIT>' , <EOL> new_authzr_uri = '<STR_LIT>' , <EOL> terms_of_service = '<STR_LIT>' ) <EOL> authzr_uri = '<STR_LIT>' <EOL> challb = messages . ChallengeBody ( <EOL> uri = ( authzr_uri + '<STR_LIT>' ) , status = messages . STATUS_VALID , <EOL> chall = challenges . DNS ( token = jose . b64decode ( <EOL> '<STR_LIT>' ) ) ) <EOL> self . challr = messages . ChallengeResource ( <EOL> body = challb , authzr_uri = authzr_uri ) <EOL> self . authz = messages . Authorization ( <EOL> identifier = messages . Identifier ( <EOL> typ = messages . IDENTIFIER_FQDN , value = '<STR_LIT>' ) , <EOL> challenges = ( challb , ) , combinations = None ) <EOL> self . authzr = messages . AuthorizationResource ( <EOL> body = self . authz , uri = authzr_uri , <EOL> new_cert_uri = '<STR_LIT>' ) <EOL> self . certr = messages . CertificateResource ( <EOL> body = messages_test . CERT , authzrs = ( self . authzr , ) , <EOL> uri = '<STR_LIT>' , <EOL> cert_chain_uri = '<STR_LIT>' ) <EOL> def test_init_downloads_directory ( self ) : <EOL> uri = '<STR_LIT>' <EOL> from acme . client import Client <EOL> self . client = Client ( <EOL> directory = uri , key = KEY , alg = jose . RS256 , net = self . net ) <EOL> self . net . get . assert_called_once_with ( uri ) <EOL> def test_register ( self ) : <EOL> self . response . status_code = http_client . CREATED <EOL> self . response . json . return_value = self . regr . body . to_json ( ) <EOL> self . response . headers [ '<STR_LIT>' ] = self . regr . uri <EOL> self . response . links . update ( { <EOL> '<STR_LIT>' : { '<STR_LIT:url>' : self . regr . new_authzr_uri } , <EOL> '<STR_LIT>' : { '<STR_LIT:url>' : self . regr . terms_of_service } , <EOL> } ) <EOL> self . assertEqual ( self . regr , self . client . register ( self . new_reg ) ) <EOL> reg_wrong_key = self . regr . body . update ( key = KEY2 . public_key ( ) ) <EOL> self . response . json . return_value = reg_wrong_key . to_json ( ) <EOL> self . assertRaises ( <EOL> errors . UnexpectedUpdate , self . client . register , self . new_reg ) <EOL> def test_register_missing_next ( self ) : <EOL> self . response . status_code = http_client . CREATED <EOL> self . assertRaises ( <EOL> errors . ClientError , self . client . register , self . new_reg ) <EOL> def test_update_registration ( self ) : <EOL> self . response . headers [ '<STR_LIT>' ] = self . regr . uri <EOL> self . response . json . return_value = self . regr . body . to_json ( ) <EOL> self . assertEqual ( self . regr , self . client . update_registration ( self . regr ) ) <EOL> self . response . json . return_value = self . regr . body . update ( <EOL> contact = ( ) ) . to_json ( ) <EOL> self . assertRaises ( <EOL> errors . UnexpectedUpdate , self . client . update_registration , self . regr ) <EOL> def test_query_registration ( self ) : <EOL> self . response . json . return_value = self . regr . body . to_json ( ) <EOL> self . assertEqual ( self . regr , self . client . query_registration ( self . regr ) ) <EOL> def test_query_registration_updates_new_authzr_uri ( self ) : <EOL> self . response . json . return_value = self . regr . body . to_json ( ) <EOL> self . response . links = { '<STR_LIT>' : { '<STR_LIT:url>' : '<STR_LIT>' } } <EOL> self . assertEqual ( <EOL> '<STR_LIT>' , <EOL> self . client . query_registration ( self . regr ) . new_authzr_uri ) <EOL> def test_agree_to_tos ( self ) : <EOL> self . client . update_registration = mock . Mock ( ) <EOL> self . client . agree_to_tos ( self . regr ) <EOL> regr = self . client . update_registration . call_args [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> self . assertEqual ( self . regr . terms_of_service , regr . body . agreement ) <EOL> def _prepare_response_for_request_challenges ( self ) : <EOL> self . response . status_code = http_client . CREATED <EOL> self . response . headers [ '<STR_LIT>' ] = self . authzr . uri <EOL> self . response . json . return_value = self . authz . to_json ( ) <EOL> self . response . links = { <EOL> '<STR_LIT>' : { '<STR_LIT:url>' : self . authzr . new_cert_uri } , <EOL> } <EOL> def test_request_challenges ( self ) : <EOL> self . _prepare_response_for_request_challenges ( ) <EOL> self . client . request_challenges ( self . identifier ) <EOL> self . net . post . assert_called_once_with ( <EOL> self . directory . new_authz , <EOL> messages . NewAuthorization ( identifier = self . identifier ) ) <EOL> def test_requets_challenges_custom_uri ( self ) : <EOL> self . _prepare_response_for_request_challenges ( ) <EOL> self . client . request_challenges ( self . identifier , '<STR_LIT>' ) <EOL> self . net . post . assert_called_once_with ( '<STR_LIT>' , mock . ANY ) <EOL> def test_request_challenges_unexpected_update ( self ) : <EOL> self . _prepare_response_for_request_challenges ( ) <EOL> self . response . json . return_value = self . authz . update ( <EOL> identifier = self . identifier . update ( value = '<STR_LIT:foo>' ) ) . to_json ( ) <EOL> self . assertRaises ( <EOL> errors . UnexpectedUpdate , self . client . request_challenges , <EOL> self . identifier , self . authzr . uri ) <EOL> def test_request_challenges_missing_next ( self ) : <EOL> self . response . status_code = http_client . CREATED <EOL> self . assertRaises ( errors . ClientError , self . client . request_challenges , <EOL> self . identifier ) <EOL> def test_request_domain_challenges ( self ) : <EOL> self . client . request_challenges = mock . MagicMock ( ) <EOL> self . assertEqual ( <EOL> self . client . request_challenges ( self . identifier ) , <EOL> self . client . request_domain_challenges ( '<STR_LIT>' ) ) <EOL> def test_request_domain_challenges_custom_uri ( self ) : <EOL> self . client . request_challenges = mock . MagicMock ( ) <EOL> self . assertEqual ( <EOL> self . client . request_challenges ( self . identifier , '<STR_LIT>' ) , <EOL> self . client . request_domain_challenges ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> def test_answer_challenge ( self ) : <EOL> self . response . links [ '<STR_LIT>' ] = { '<STR_LIT:url>' : self . challr . authzr_uri } <EOL> self . response . json . return_value = self . challr . body . to_json ( ) <EOL> chall_response = challenges . DNSResponse ( validation = None ) <EOL> self . client . answer_challenge ( self . challr . body , chall_response ) <EOL> self . assertRaises ( errors . UnexpectedUpdate , self . client . answer_challenge , <EOL> self . challr . body . update ( uri = '<STR_LIT:foo>' ) , chall_response ) <EOL> def test_answer_challenge_missing_next ( self ) : <EOL> self . assertRaises ( <EOL> errors . ClientError , self . client . answer_challenge , <EOL> self . challr . body , challenges . DNSResponse ( validation = None ) ) <EOL> def test_retry_after_date ( self ) : <EOL> self . response . headers [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . assertEqual ( <EOL> datetime . datetime ( <NUM_LIT> , <NUM_LIT:12> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ) , <EOL> self . client . retry_after ( response = self . response , default = <NUM_LIT:10> ) ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_retry_after_invalid ( self , dt_mock ) : <EOL> dt_mock . datetime . now . return_value = datetime . datetime ( <NUM_LIT> , <NUM_LIT:3> , <NUM_LIT> ) <EOL> dt_mock . timedelta = datetime . timedelta <EOL> self . response . headers [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . assertEqual ( <EOL> datetime . datetime ( <NUM_LIT> , <NUM_LIT:3> , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:10> ) , <EOL> self . client . retry_after ( response = self . response , default = <NUM_LIT:10> ) ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_retry_after_overflow ( self , dt_mock ) : <EOL> dt_mock . datetime . now . return_value = datetime . datetime ( <NUM_LIT> , <NUM_LIT:3> , <NUM_LIT> ) <EOL> dt_mock . timedelta = datetime . timedelta <EOL> dt_mock . datetime . side_effect = datetime . datetime <EOL> self . response . headers [ '<STR_LIT>' ] = "<STR_LIT>" <EOL> self . assertEqual ( <EOL> datetime . datetime ( <NUM_LIT> , <NUM_LIT:3> , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:10> ) , <EOL> self . client . retry_after ( response = self . response , default = <NUM_LIT:10> ) ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_retry_after_seconds ( self , dt_mock ) : <EOL> dt_mock . datetime . now . return_value = datetime . datetime ( <NUM_LIT> , <NUM_LIT:3> , <NUM_LIT> ) <EOL> dt_mock . timedelta = datetime . timedelta <EOL> self . response . headers [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . assertEqual ( <EOL> datetime . datetime ( <NUM_LIT> , <NUM_LIT:3> , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:50> ) , <EOL> self . client . retry_after ( response = self . response , default = <NUM_LIT:10> ) ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_retry_after_missing ( self , dt_mock ) : <EOL> dt_mock . datetime . now . return_value = datetime . datetime ( <NUM_LIT> , <NUM_LIT:3> , <NUM_LIT> ) <EOL> dt_mock . timedelta = datetime . timedelta <EOL> self . assertEqual ( <EOL> datetime . datetime ( <NUM_LIT> , <NUM_LIT:3> , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:10> ) , <EOL> self . client . retry_after ( response = self . response , default = <NUM_LIT:10> ) ) <EOL> def test_poll ( self ) : <EOL> self . response . json . return_value = self . authzr . body . to_json ( ) <EOL> self . assertEqual ( ( self . authzr , self . response ) , <EOL> self . client . poll ( self . authzr ) ) <EOL> self . response . json . return_value = self . authz . update ( <EOL> identifier = self . identifier . update ( value = '<STR_LIT:foo>' ) ) . to_json ( ) <EOL> self . assertRaises ( <EOL> errors . UnexpectedUpdate , self . client . poll , self . authzr ) <EOL> def test_request_issuance ( self ) : <EOL> self . response . content = CERT_DER <EOL> self . response . headers [ '<STR_LIT>' ] = self . certr . uri <EOL> self . response . links [ '<STR_LIT>' ] = { '<STR_LIT:url>' : self . certr . cert_chain_uri } <EOL> self . assertEqual ( self . certr , self . client . request_issuance ( <EOL> messages_test . CSR , ( self . authzr , ) ) ) <EOL> def test_request_issuance_missing_up ( self ) : <EOL> self . response . content = CERT_DER <EOL> self . response . headers [ '<STR_LIT>' ] = self . certr . uri <EOL> self . assertEqual ( <EOL> self . certr . update ( cert_chain_uri = None ) , <EOL> self . client . request_issuance ( messages_test . CSR , ( self . authzr , ) ) ) <EOL> def test_request_issuance_missing_location ( self ) : <EOL> self . assertRaises ( <EOL> errors . ClientError , self . client . request_issuance , <EOL> messages_test . CSR , ( self . authzr , ) ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_poll_and_request_issuance ( self , time_mock , dt_mock ) : <EOL> clock = mock . MagicMock ( dt = datetime . datetime ( <NUM_LIT> , <NUM_LIT:3> , <NUM_LIT> ) ) <EOL> def sleep ( seconds ) : <EOL> """<STR_LIT>""" <EOL> clock . dt += datetime . timedelta ( seconds = seconds ) <EOL> time_mock . sleep . side_effect = sleep <EOL> def now ( ) : <EOL> """<STR_LIT>""" <EOL> return clock . dt <EOL> dt_mock . datetime . now . side_effect = now <EOL> dt_mock . timedelta = datetime . timedelta <EOL> def poll ( authzr ) : <EOL> authzr . times . append ( clock . dt ) <EOL> clock . dt += datetime . timedelta ( seconds = <NUM_LIT:2> ) <EOL> if len ( authzr . retries ) == <NUM_LIT:1> : <EOL> done = mock . MagicMock ( uri = authzr . uri , times = authzr . times ) <EOL> done . body . status = authzr . retries [ <NUM_LIT:0> ] <EOL> return done , [ ] <EOL> return ( mock . MagicMock ( retries = authzr . retries [ <NUM_LIT:1> : ] , <EOL> uri = authzr . uri + '<STR_LIT:.>' , times = authzr . times ) , <EOL> authzr . retries [ <NUM_LIT:0> ] ) <EOL> self . client . poll = mock . MagicMock ( side_effect = poll ) <EOL> mintime = <NUM_LIT:7> <EOL> def retry_after ( response , default ) : <EOL> self . assertEqual ( default , mintime ) <EOL> return clock . dt + datetime . timedelta ( seconds = response ) <EOL> self . client . retry_after = mock . MagicMock ( side_effect = retry_after ) <EOL> def request_issuance ( csr , authzrs ) : <EOL> return csr , authzrs <EOL> self . client . request_issuance = mock . MagicMock ( <EOL> side_effect = request_issuance ) <EOL> csr = mock . MagicMock ( ) <EOL> authzrs = ( <EOL> mock . MagicMock ( uri = '<STR_LIT:a>' , times = [ ] , retries = ( <EOL> <NUM_LIT:8> , <NUM_LIT:20> , <NUM_LIT:30> , messages . STATUS_VALID ) ) , <EOL> mock . MagicMock ( uri = '<STR_LIT:b>' , times = [ ] , retries = ( <EOL> <NUM_LIT:5> , messages . STATUS_VALID ) ) , <EOL> ) <EOL> cert , updated_authzrs = self . client . poll_and_request_issuance ( <EOL> csr , authzrs , mintime = mintime , <EOL> max_attempts = max ( len ( authzrs [ <NUM_LIT:0> ] . retries ) , len ( authzrs [ <NUM_LIT:1> ] . retries ) ) ) <EOL> self . assertTrue ( cert [ <NUM_LIT:0> ] is csr ) <EOL> self . assertTrue ( cert [ <NUM_LIT:1> ] is updated_authzrs ) <EOL> self . assertEqual ( updated_authzrs [ <NUM_LIT:0> ] . uri , '<STR_LIT>' ) <EOL> self . assertEqual ( updated_authzrs [ <NUM_LIT:1> ] . uri , '<STR_LIT>' ) <EOL> self . assertEqual ( updated_authzrs [ <NUM_LIT:0> ] . times , [ <EOL> datetime . datetime ( <NUM_LIT> , <NUM_LIT:3> , <NUM_LIT> ) , <EOL> datetime . datetime ( <NUM_LIT> , <NUM_LIT:3> , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:11> ) , <EOL> datetime . datetime ( <NUM_LIT> , <NUM_LIT:3> , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT> ) , <EOL> datetime . datetime ( <NUM_LIT> , <NUM_LIT:3> , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:5> ) , <EOL> ] ) <EOL> self . assertEqual ( updated_authzrs [ <NUM_LIT:1> ] . times , [ <EOL> datetime . datetime ( <NUM_LIT> , <NUM_LIT:3> , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:2> ) , <EOL> datetime . datetime ( <NUM_LIT> , <NUM_LIT:3> , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:9> ) , <EOL> ] ) <EOL> self . assertEqual ( clock . dt , datetime . datetime ( <NUM_LIT> , <NUM_LIT:3> , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:7> ) ) <EOL> invalid_authzr = mock . MagicMock ( <EOL> times = [ ] , retries = [ messages . STATUS_INVALID ] ) <EOL> self . assertRaises ( <EOL> errors . PollError , self . client . poll_and_request_issuance , <EOL> csr , authzrs = ( invalid_authzr , ) , mintime = mintime ) <EOL> self . assertRaises ( <EOL> errors . PollError , self . client . poll_and_request_issuance , <EOL> csr , authzrs , mintime = mintime , max_attempts = <NUM_LIT:2> ) <EOL> def test_check_cert ( self ) : <EOL> self . response . headers [ '<STR_LIT>' ] = self . certr . uri <EOL> self . response . content = CERT_DER <EOL> self . assertEqual ( self . certr . update ( body = messages_test . CERT ) , <EOL> self . client . check_cert ( self . certr ) ) <EOL> self . response . headers [ '<STR_LIT>' ] = '<STR_LIT:foo>' <EOL> self . assertRaises ( <EOL> errors . UnexpectedUpdate , self . client . check_cert , self . certr ) <EOL> def test_check_cert_missing_location ( self ) : <EOL> self . response . content = CERT_DER <EOL> self . assertRaises ( <EOL> errors . ClientError , self . client . check_cert , self . certr ) <EOL> def test_refresh ( self ) : <EOL> self . client . check_cert = mock . MagicMock ( ) <EOL> self . assertEqual ( <EOL> self . client . check_cert ( self . certr ) , self . client . refresh ( self . certr ) ) <EOL> def test_fetch_chain_no_up_link ( self ) : <EOL> self . assertEqual ( [ ] , self . client . fetch_chain ( self . certr . update ( <EOL> cert_chain_uri = None ) ) ) <EOL> def test_fetch_chain_single ( self ) : <EOL> self . client . _get_cert = mock . MagicMock ( ) <EOL> self . client . _get_cert . return_value = ( <EOL> mock . MagicMock ( links = { } ) , "<STR_LIT>" ) <EOL> self . assertEqual ( [ self . client . _get_cert ( self . certr . cert_chain_uri ) [ <NUM_LIT:1> ] ] , <EOL> self . client . fetch_chain ( self . certr ) ) <EOL> def test_fetch_chain_max ( self ) : <EOL> up_response = mock . MagicMock ( links = { '<STR_LIT>' : { '<STR_LIT:url>' : '<STR_LIT>' } } ) <EOL> noup_response = mock . MagicMock ( links = { } ) <EOL> self . client . _get_cert = mock . MagicMock ( ) <EOL> self . client . _get_cert . side_effect = [ <EOL> ( up_response , "<STR_LIT>" ) ] * <NUM_LIT:9> + [ ( noup_response , "<STR_LIT>" ) ] <EOL> chain = self . client . fetch_chain ( self . certr , max_length = <NUM_LIT:10> ) <EOL> self . assertEqual ( chain , [ "<STR_LIT>" ] * <NUM_LIT:9> + [ "<STR_LIT>" ] ) <EOL> def test_fetch_chain_too_many ( self ) : <EOL> response = mock . MagicMock ( links = { '<STR_LIT>' : { '<STR_LIT:url>' : '<STR_LIT>' } } ) <EOL> self . client . _get_cert = mock . MagicMock ( ) <EOL> self . client . _get_cert . return_value = ( response , "<STR_LIT>" ) <EOL> self . assertRaises ( errors . Error , self . client . fetch_chain , self . certr ) <EOL> def test_revoke ( self ) : <EOL> self . client . revoke ( self . certr . body ) <EOL> self . net . post . assert_called_once_with ( <EOL> self . directory [ messages . Revocation ] , mock . ANY , content_type = None ) <EOL> def test_revoke_bad_status_raises_error ( self ) : <EOL> self . response . status_code = http_client . METHOD_NOT_ALLOWED <EOL> self . assertRaises ( errors . ClientError , self . client . revoke , self . certr ) <EOL> class ClientNetworkTest ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . verify_ssl = mock . MagicMock ( ) <EOL> self . wrap_in_jws = mock . MagicMock ( return_value = mock . sentinel . wrapped ) <EOL> from acme . client import ClientNetwork <EOL> self . net = ClientNetwork ( <EOL> key = KEY , alg = jose . RS256 , verify_ssl = self . verify_ssl , <EOL> user_agent = '<STR_LIT>' ) <EOL> self . response = mock . MagicMock ( ok = True , status_code = http_client . OK ) <EOL> self . response . headers = { } <EOL> self . response . links = { } <EOL> def test_init ( self ) : <EOL> self . assertTrue ( self . net . verify_ssl is self . verify_ssl ) <EOL> def test_wrap_in_jws ( self ) : <EOL> class MockJSONDeSerializable ( jose . JSONDeSerializable ) : <EOL> def __init__ ( self , value ) : <EOL> self . value = value <EOL> def to_partial_json ( self ) : <EOL> return { '<STR_LIT:foo>' : self . value } <EOL> @ classmethod <EOL> def from_json ( cls , value ) : <EOL> pass <EOL> jws_dump = self . net . _wrap_in_jws ( <EOL> MockJSONDeSerializable ( '<STR_LIT:foo>' ) , nonce = b'<STR_LIT>' ) <EOL> jws = acme_jws . JWS . json_loads ( jws_dump ) <EOL> self . assertEqual ( json . loads ( jws . payload . decode ( ) ) , { '<STR_LIT:foo>' : '<STR_LIT:foo>' } ) <EOL> self . assertEqual ( jws . signature . combined . nonce , b'<STR_LIT>' ) <EOL> def test_check_response_not_ok_jobj_no_error ( self ) : <EOL> self . response . ok = False <EOL> self . response . json . return_value = { } <EOL> self . assertRaises ( <EOL> errors . ClientError , self . net . _check_response , self . response ) <EOL> def test_check_response_not_ok_jobj_error ( self ) : <EOL> self . response . ok = False <EOL> self . response . json . return_value = messages . Error ( <EOL> detail = '<STR_LIT:foo>' , typ = '<STR_LIT>' , title = '<STR_LIT>' ) . to_json ( ) <EOL> self . assertRaises ( <EOL> messages . Error , self . net . _check_response , self . response ) <EOL> def test_check_response_not_ok_no_jobj ( self ) : <EOL> self . response . ok = False <EOL> self . response . json . side_effect = ValueError <EOL> self . assertRaises ( <EOL> errors . ClientError , self . net . _check_response , self . response ) <EOL> def test_check_response_ok_no_jobj_ct_required ( self ) : <EOL> self . response . json . side_effect = ValueError <EOL> for response_ct in [ self . net . JSON_CONTENT_TYPE , '<STR_LIT:foo>' ] : <EOL> self . response . headers [ '<STR_LIT:Content-Type>' ] = response_ct <EOL> self . assertRaises ( <EOL> errors . ClientError , self . net . _check_response , self . response , <EOL> content_type = self . net . JSON_CONTENT_TYPE ) <EOL> def test_check_response_ok_no_jobj_no_ct ( self ) : <EOL> self . response . json . side_effect = ValueError <EOL> for response_ct in [ self . net . JSON_CONTENT_TYPE , '<STR_LIT:foo>' ] : <EOL> self . response . headers [ '<STR_LIT:Content-Type>' ] = response_ct <EOL> self . assertEqual ( <EOL> self . response , self . net . _check_response ( self . response ) ) <EOL> def test_check_response_jobj ( self ) : <EOL> self . response . json . return_value = { } <EOL> for response_ct in [ self . net . JSON_CONTENT_TYPE , '<STR_LIT:foo>' ] : <EOL> self . response . headers [ '<STR_LIT:Content-Type>' ] = response_ct <EOL> self . assertEqual ( <EOL> self . response , self . net . _check_response ( self . response ) ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_send_request ( self , mock_requests ) : <EOL> mock_requests . request . return_value = self . response <EOL> self . assertEqual ( self . response , self . net . _send_request ( <EOL> '<STR_LIT>' , '<STR_LIT:url>' , '<STR_LIT:foo>' , bar = '<STR_LIT>' ) ) <EOL> mock_requests . request . assert_called_once_with ( <EOL> '<STR_LIT>' , '<STR_LIT:url>' , '<STR_LIT:foo>' , verify = mock . ANY , bar = '<STR_LIT>' , headers = mock . ANY ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_send_request_verify_ssl ( self , mock_requests ) : <EOL> for verify in True , False : <EOL> mock_requests . request . reset_mock ( ) <EOL> mock_requests . request . return_value = self . response <EOL> self . net . verify_ssl = verify <EOL> self . assertEqual ( <EOL> self . response , self . net . _send_request ( '<STR_LIT:GET>' , '<STR_LIT:url>' ) ) <EOL> mock_requests . request . assert_called_once_with ( <EOL> '<STR_LIT:GET>' , '<STR_LIT:url>' , verify = verify , headers = mock . ANY ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_send_request_user_agent ( self , mock_requests ) : <EOL> mock_requests . request . return_value = self . response <EOL> self . net . _send_request ( '<STR_LIT:GET>' , '<STR_LIT:url>' , headers = { '<STR_LIT:bar>' : '<STR_LIT>' } ) <EOL> mock_requests . request . assert_called_once_with ( <EOL> '<STR_LIT:GET>' , '<STR_LIT:url>' , verify = mock . ANY , <EOL> headers = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:bar>' : '<STR_LIT>' } ) <EOL> self . net . _send_request ( '<STR_LIT:GET>' , '<STR_LIT:url>' , headers = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> mock_requests . request . assert_called_with ( <EOL> '<STR_LIT:GET>' , '<STR_LIT:url>' , verify = mock . ANY , headers = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_requests_error_passthrough ( self , mock_requests ) : <EOL> mock_requests . exceptions = requests . exceptions <EOL> mock_requests . request . side_effect = requests . exceptions . RequestException <EOL> self . assertRaises ( requests . exceptions . RequestException , <EOL> self . net . _send_request , '<STR_LIT:GET>' , '<STR_LIT>' ) <EOL> class ClientNetworkWithMockedResponseTest ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> from acme . client import ClientNetwork <EOL> self . net = ClientNetwork ( key = None , alg = None ) <EOL> self . response = mock . MagicMock ( ok = True , status_code = http_client . OK ) <EOL> self . response . headers = { } <EOL> self . response . links = { } <EOL> self . checked_response = mock . MagicMock ( ) <EOL> self . obj = mock . MagicMock ( ) <EOL> self . wrapped_obj = mock . MagicMock ( ) <EOL> self . content_type = mock . sentinel . content_type <EOL> self . all_nonces = [ jose . b64encode ( b'<STR_LIT>' ) , jose . b64encode ( b'<STR_LIT>' ) ] <EOL> self . available_nonces = self . all_nonces [ : ] <EOL> def send_request ( * args , ** kwargs ) : <EOL> if self . available_nonces : <EOL> self . response . headers = { <EOL> self . net . REPLAY_NONCE_HEADER : <EOL> self . available_nonces . pop ( ) . decode ( ) } <EOL> else : <EOL> self . response . headers = { } <EOL> return self . response <EOL> self . net . _send_request = self . send_request = mock . MagicMock ( <EOL> side_effect = send_request ) <EOL> self . net . _check_response = self . check_response <EOL> self . net . _wrap_in_jws = mock . MagicMock ( return_value = self . wrapped_obj ) <EOL> def check_response ( self , response , content_type ) : <EOL> self . assertEqual ( self . response , response ) <EOL> self . assertEqual ( self . content_type , content_type ) <EOL> return self . checked_response <EOL> def test_head ( self ) : <EOL> self . assertEqual ( self . response , self . net . head ( '<STR_LIT:url>' , '<STR_LIT:foo>' , bar = '<STR_LIT>' ) ) <EOL> self . send_request . assert_called_once_with ( <EOL> '<STR_LIT>' , '<STR_LIT:url>' , '<STR_LIT:foo>' , bar = '<STR_LIT>' ) <EOL> def test_get ( self ) : <EOL> self . assertEqual ( self . checked_response , self . net . get ( <EOL> '<STR_LIT:url>' , content_type = self . content_type , bar = '<STR_LIT>' ) ) <EOL> self . send_request . assert_called_once_with ( '<STR_LIT:GET>' , '<STR_LIT:url>' , bar = '<STR_LIT>' ) <EOL> def test_post ( self ) : <EOL> self . assertEqual ( self . checked_response , self . net . post ( <EOL> '<STR_LIT>' , self . obj , content_type = self . content_type ) ) <EOL> self . net . _wrap_in_jws . assert_called_once_with ( <EOL> self . obj , jose . b64decode ( self . all_nonces . pop ( ) ) ) <EOL> assert not self . available_nonces <EOL> self . assertRaises ( errors . MissingNonce , self . net . post , <EOL> '<STR_LIT>' , self . obj , content_type = self . content_type ) <EOL> self . net . _wrap_in_jws . assert_called_with ( <EOL> self . obj , jose . b64decode ( self . all_nonces . pop ( ) ) ) <EOL> def test_post_wrong_initial_nonce ( self ) : <EOL> self . available_nonces = [ b'<STR_LIT:f>' , jose . b64encode ( b'<STR_LIT>' ) ] <EOL> self . assertRaises ( errors . BadNonce , self . net . post , '<STR_LIT>' , <EOL> self . obj , content_type = self . content_type ) <EOL> def test_post_wrong_post_response_nonce ( self ) : <EOL> self . available_nonces = [ jose . b64encode ( b'<STR_LIT>' ) , b'<STR_LIT:f>' ] <EOL> self . assertRaises ( errors . BadNonce , self . net . post , '<STR_LIT>' , <EOL> self . obj , content_type = self . content_type ) <EOL> def test_head_get_post_error_passthrough ( self ) : <EOL> self . send_request . side_effect = requests . exceptions . RequestException <EOL> for method in self . net . head , self . net . get : <EOL> self . assertRaises ( <EOL> requests . exceptions . RequestException , method , '<STR_LIT:GET>' , '<STR_LIT>' ) <EOL> self . assertRaises ( requests . exceptions . RequestException , <EOL> self . net . post , '<STR_LIT>' , obj = self . obj ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
