<s> import json <EOL> import weakref <EOL> import requests <EOL> from requests . adapters import BaseAdapter <EOL> import six <EOL> from six . moves . urllib import parse as urlparse <EOL> from requests_mock import exceptions <EOL> from requests_mock import response <EOL> ANY = object ( ) <EOL> class _RequestObjectProxy ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , request , ** kwargs ) : <EOL> self . _request = request <EOL> self . _matcher = None <EOL> self . _url_parts_ = None <EOL> self . _qs = None <EOL> self . _timeout = kwargs . pop ( '<STR_LIT>' , None ) <EOL> self . _allow_redirects = kwargs . pop ( '<STR_LIT>' , None ) <EOL> def __getattr__ ( self , name ) : <EOL> return getattr ( self . _request , name ) <EOL> @ property <EOL> def _url_parts ( self ) : <EOL> if self . _url_parts_ is None : <EOL> self . _url_parts_ = urlparse . urlparse ( self . _request . url . lower ( ) ) <EOL> return self . _url_parts_ <EOL> @ property <EOL> def scheme ( self ) : <EOL> return self . _url_parts . scheme <EOL> @ property <EOL> def netloc ( self ) : <EOL> return self . _url_parts . netloc <EOL> @ property <EOL> def path ( self ) : <EOL> return self . _url_parts . path <EOL> @ property <EOL> def query ( self ) : <EOL> return self . _url_parts . query <EOL> @ property <EOL> def qs ( self ) : <EOL> if self . _qs is None : <EOL> self . _qs = urlparse . parse_qs ( self . query ) <EOL> return self . _qs <EOL> @ property <EOL> def timeout ( self ) : <EOL> return self . _timeout <EOL> @ property <EOL> def allow_redirects ( self ) : <EOL> return self . _allow_redirects <EOL> @ classmethod <EOL> def _create ( cls , * args , ** kwargs ) : <EOL> return cls ( requests . Request ( * args , ** kwargs ) . prepare ( ) ) <EOL> @ property <EOL> def text ( self ) : <EOL> body = self . body <EOL> if isinstance ( body , six . binary_type ) : <EOL> body = body . decode ( '<STR_LIT:utf-8>' ) <EOL> return body <EOL> def json ( self , ** kwargs ) : <EOL> return json . loads ( self . text , ** kwargs ) <EOL> @ property <EOL> def matcher ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _matcher ( ) <EOL> class _RequestHistoryTracker ( object ) : <EOL> def __init__ ( self ) : <EOL> self . request_history = [ ] <EOL> def _add_to_history ( self , request ) : <EOL> self . request_history . append ( request ) <EOL> @ property <EOL> def last_request ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return self . request_history [ - <NUM_LIT:1> ] <EOL> except IndexError : <EOL> return None <EOL> @ property <EOL> def called ( self ) : <EOL> return self . call_count > <NUM_LIT:0> <EOL> @ property <EOL> def call_count ( self ) : <EOL> return len ( self . request_history ) <EOL> class _Matcher ( _RequestHistoryTracker ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , method , url , responses , complete_qs , request_headers ) : <EOL> """<STR_LIT>""" <EOL> super ( _Matcher , self ) . __init__ ( ) <EOL> self . _method = method <EOL> self . _url = url <EOL> try : <EOL> self . _url_parts = urlparse . urlparse ( url . lower ( ) ) <EOL> except : <EOL> self . _url_parts = None <EOL> self . _responses = responses <EOL> self . _complete_qs = complete_qs <EOL> self . _request_headers = request_headers <EOL> def _match_method ( self , request ) : <EOL> if self . _method is ANY : <EOL> return True <EOL> if request . method . lower ( ) == self . _method . lower ( ) : <EOL> return True <EOL> return False <EOL> def _match_url ( self , request ) : <EOL> if self . _url is ANY : <EOL> return True <EOL> if hasattr ( self . _url , '<STR_LIT>' ) : <EOL> return self . _url . search ( request . url ) is not None <EOL> if self . _url_parts . scheme and request . scheme != self . _url_parts . scheme : <EOL> return False <EOL> if self . _url_parts . netloc and request . netloc != self . _url_parts . netloc : <EOL> return False <EOL> if ( request . path or '<STR_LIT:/>' ) != ( self . _url_parts . path or '<STR_LIT:/>' ) : <EOL> return False <EOL> request_qs = urlparse . parse_qs ( request . query ) <EOL> matcher_qs = urlparse . parse_qs ( self . _url_parts . query ) <EOL> for k , vals in six . iteritems ( matcher_qs ) : <EOL> for v in vals : <EOL> try : <EOL> request_qs . get ( k , [ ] ) . remove ( v ) <EOL> except ValueError : <EOL> return False <EOL> if self . _complete_qs : <EOL> for v in six . itervalues ( request_qs ) : <EOL> if v : <EOL> return False <EOL> return True <EOL> def _match_headers ( self , request ) : <EOL> for k , vals in six . iteritems ( self . _request_headers ) : <EOL> try : <EOL> header = request . headers [ k ] <EOL> except KeyError : <EOL> if not isinstance ( k , six . text_type ) : <EOL> return False <EOL> try : <EOL> header = request . headers [ k . encode ( '<STR_LIT:utf-8>' ) ] <EOL> except KeyError : <EOL> return False <EOL> if header != vals : <EOL> return False <EOL> return True <EOL> def _match ( self , request ) : <EOL> return ( self . _match_method ( request ) and <EOL> self . _match_url ( request ) and <EOL> self . _match_headers ( request ) ) <EOL> def __call__ ( self , request ) : <EOL> if not self . _match ( request ) : <EOL> return None <EOL> if len ( self . _responses ) > <NUM_LIT:1> : <EOL> response_matcher = self . _responses . pop ( <NUM_LIT:0> ) <EOL> else : <EOL> response_matcher = self . _responses [ <NUM_LIT:0> ] <EOL> self . _add_to_history ( request ) <EOL> return response_matcher . get_response ( request ) <EOL> class Adapter ( BaseAdapter , _RequestHistoryTracker ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> super ( Adapter , self ) . __init__ ( ) <EOL> self . _matchers = [ ] <EOL> def send ( self , request , ** kwargs ) : <EOL> request = _RequestObjectProxy ( request , ** kwargs ) <EOL> self . _add_to_history ( request ) <EOL> for matcher in reversed ( self . _matchers ) : <EOL> try : <EOL> resp = matcher ( request ) <EOL> except Exception : <EOL> request . _matcher = weakref . ref ( matcher ) <EOL> raise <EOL> if resp is not None : <EOL> request . _matcher = weakref . ref ( matcher ) <EOL> resp . connection = self <EOL> return resp <EOL> raise exceptions . NoMockAddress ( request ) <EOL> def close ( self ) : <EOL> pass <EOL> def register_uri ( self , method , url , response_list = None , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> complete_qs = kwargs . pop ( '<STR_LIT>' , False ) <EOL> request_headers = kwargs . pop ( '<STR_LIT>' , { } ) <EOL> if response_list and kwargs : <EOL> raise RuntimeError ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> elif not response_list : <EOL> response_list = [ kwargs ] <EOL> responses = [ response . _MatcherResponse ( ** k ) for k in response_list ] <EOL> matcher = _Matcher ( method , <EOL> url , <EOL> responses , <EOL> complete_qs = complete_qs , <EOL> request_headers = request_headers ) <EOL> self . add_matcher ( matcher ) <EOL> return matcher <EOL> def add_matcher ( self , matcher ) : <EOL> """<STR_LIT>""" <EOL> self . _matchers . append ( matcher ) <EOL> __all__ = [ '<STR_LIT>' ] </s>
