<s> '''<STR_LIT>''' <EOL> import functools <EOL> import itertools <EOL> from . compat import contextlib , mock <EOL> from . stubs import VCRHTTPConnection , VCRHTTPSConnection <EOL> from six . moves import http_client as httplib <EOL> _HTTPConnection = httplib . HTTPConnection <EOL> _HTTPSConnection = httplib . HTTPSConnection <EOL> try : <EOL> import requests . packages . urllib3 . connectionpool as cpool <EOL> except ImportError : <EOL> pass <EOL> else : <EOL> _VerifiedHTTPSConnection = cpool . VerifiedHTTPSConnection <EOL> _cpoolHTTPConnection = cpool . HTTPConnection <EOL> _cpoolHTTPSConnection = cpool . HTTPSConnection <EOL> try : <EOL> import urllib3 <EOL> except ImportError : <EOL> pass <EOL> else : <EOL> _VerifiedHTTPSConnection = urllib3 . connectionpool . VerifiedHTTPSConnection <EOL> try : <EOL> import httplib2 <EOL> except ImportError : <EOL> pass <EOL> else : <EOL> _HTTPConnectionWithTimeout = httplib2 . HTTPConnectionWithTimeout <EOL> _HTTPSConnectionWithTimeout = httplib2 . HTTPSConnectionWithTimeout <EOL> _SCHEME_TO_CONNECTION = httplib2 . SCHEME_TO_CONNECTION <EOL> try : <EOL> import boto . https_connection <EOL> except ImportError : <EOL> pass <EOL> else : <EOL> _CertValidatingHTTPSConnection = boto . https_connection . CertValidatingHTTPSConnection <EOL> try : <EOL> import tornado . simple_httpclient <EOL> except ImportError : <EOL> pass <EOL> else : <EOL> _SimpleAsyncHTTPClient_fetch_impl = tornado . simple_httpclient . SimpleAsyncHTTPClient . fetch_impl <EOL> try : <EOL> import tornado . curl_httpclient <EOL> except ImportError : <EOL> pass <EOL> else : <EOL> _CurlAsyncHTTPClient_fetch_impl = tornado . curl_httpclient . CurlAsyncHTTPClient . fetch_impl <EOL> class CassettePatcherBuilder ( object ) : <EOL> def _build_patchers_from_mock_triples_decorator ( function ) : <EOL> @ functools . wraps ( function ) <EOL> def wrapped ( self , * args , ** kwargs ) : <EOL> return self . _build_patchers_from_mock_triples ( <EOL> function ( self , * args , ** kwargs ) <EOL> ) <EOL> return wrapped <EOL> def __init__ ( self , cassette ) : <EOL> self . _cassette = cassette <EOL> self . _class_to_cassette_subclass = { } <EOL> def build ( self ) : <EOL> return itertools . chain ( <EOL> self . _httplib ( ) , self . _requests ( ) , self . _urllib3 ( ) , <EOL> self . _httplib2 ( ) , self . _boto ( ) , self . _tornado ( ) , <EOL> self . _build_patchers_from_mock_triples ( <EOL> self . _cassette . custom_patches <EOL> ) , <EOL> ) <EOL> def _build_patchers_from_mock_triples ( self , mock_triples ) : <EOL> for args in mock_triples : <EOL> patcher = self . _build_patcher ( * args ) <EOL> if patcher : <EOL> yield patcher <EOL> def _build_patcher ( self , obj , patched_attribute , replacement_class ) : <EOL> if not hasattr ( obj , patched_attribute ) : <EOL> return <EOL> return mock . patch . object ( obj , patched_attribute , <EOL> self . _recursively_apply_get_cassette_subclass ( <EOL> replacement_class ) ) <EOL> def _recursively_apply_get_cassette_subclass ( self , replacement_dict_or_obj ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( replacement_dict_or_obj , dict ) : <EOL> for key , replacement_obj in replacement_dict_or_obj . items ( ) : <EOL> replacement_obj = self . _recursively_apply_get_cassette_subclass ( <EOL> replacement_obj ) <EOL> replacement_dict_or_obj [ key ] = replacement_obj <EOL> return replacement_dict_or_obj <EOL> if hasattr ( replacement_dict_or_obj , '<STR_LIT>' ) : <EOL> replacement_dict_or_obj = self . _get_cassette_subclass ( <EOL> replacement_dict_or_obj ) <EOL> return replacement_dict_or_obj <EOL> def _get_cassette_subclass ( self , klass ) : <EOL> if klass . cassette is not None : <EOL> return klass <EOL> if klass not in self . _class_to_cassette_subclass : <EOL> subclass = self . _build_cassette_subclass ( klass ) <EOL> self . _class_to_cassette_subclass [ klass ] = subclass <EOL> return self . _class_to_cassette_subclass [ klass ] <EOL> def _build_cassette_subclass ( self , base_class ) : <EOL> bases = ( base_class , ) <EOL> if not issubclass ( base_class , object ) : <EOL> bases += ( object , ) <EOL> return type ( '<STR_LIT>' . format ( base_class . __name__ , self . _cassette . _path ) , <EOL> bases , dict ( cassette = self . _cassette ) ) <EOL> @ _build_patchers_from_mock_triples_decorator <EOL> def _httplib ( self ) : <EOL> yield httplib , '<STR_LIT>' , VCRHTTPConnection <EOL> yield httplib , '<STR_LIT>' , VCRHTTPSConnection <EOL> def _requests ( self ) : <EOL> try : <EOL> import requests . packages . urllib3 . connectionpool as cpool <EOL> except ImportError : <EOL> return ( ) <EOL> from . stubs import requests_stubs <EOL> return self . _urllib3_patchers ( cpool , requests_stubs ) <EOL> def _patched_get_conn ( self , connection_pool_class , connection_class_getter ) : <EOL> get_conn = connection_pool_class . _get_conn <EOL> @ functools . wraps ( get_conn ) <EOL> def patched_get_conn ( pool , timeout = None ) : <EOL> connection = get_conn ( pool , timeout ) <EOL> connection_class = ( <EOL> pool . ConnectionCls if hasattr ( pool , '<STR_LIT>' ) <EOL> else connection_class_getter ( ) ) <EOL> while not isinstance ( connection , connection_class ) : <EOL> connection = get_conn ( pool , timeout ) <EOL> return connection <EOL> return patched_get_conn <EOL> def _patched_new_conn ( self , connection_pool_class , connection_remover ) : <EOL> new_conn = connection_pool_class . _new_conn <EOL> @ functools . wraps ( new_conn ) <EOL> def patched_new_conn ( pool ) : <EOL> new_connection = new_conn ( pool ) <EOL> connection_remover . add_connection_to_pool_entry ( pool , new_connection ) <EOL> return new_connection <EOL> return patched_new_conn <EOL> def _urllib3 ( self ) : <EOL> try : <EOL> import urllib3 . connectionpool as cpool <EOL> except ImportError : <EOL> return ( ) <EOL> from . stubs import urllib3_stubs <EOL> return self . _urllib3_patchers ( cpool , urllib3_stubs ) <EOL> @ _build_patchers_from_mock_triples_decorator <EOL> def _httplib2 ( self ) : <EOL> try : <EOL> import httplib2 as cpool <EOL> except ImportError : <EOL> pass <EOL> else : <EOL> from . stubs . httplib2_stubs import VCRHTTPConnectionWithTimeout <EOL> from . stubs . httplib2_stubs import VCRHTTPSConnectionWithTimeout <EOL> yield cpool , '<STR_LIT>' , VCRHTTPConnectionWithTimeout <EOL> yield cpool , '<STR_LIT>' , VCRHTTPSConnectionWithTimeout <EOL> yield cpool , '<STR_LIT>' , { '<STR_LIT:http>' : VCRHTTPConnectionWithTimeout , <EOL> '<STR_LIT>' : VCRHTTPSConnectionWithTimeout } <EOL> @ _build_patchers_from_mock_triples_decorator <EOL> def _boto ( self ) : <EOL> try : <EOL> import boto . https_connection as cpool <EOL> except ImportError : <EOL> pass <EOL> else : <EOL> from . stubs . boto_stubs import VCRCertValidatingHTTPSConnection <EOL> yield cpool , '<STR_LIT>' , VCRCertValidatingHTTPSConnection <EOL> @ _build_patchers_from_mock_triples_decorator <EOL> def _tornado ( self ) : <EOL> try : <EOL> import tornado . simple_httpclient as simple <EOL> except ImportError : <EOL> pass <EOL> else : <EOL> from . stubs . tornado_stubs import vcr_fetch_impl <EOL> new_fetch_impl = vcr_fetch_impl ( <EOL> self . _cassette , _SimpleAsyncHTTPClient_fetch_impl <EOL> ) <EOL> yield simple . SimpleAsyncHTTPClient , '<STR_LIT>' , new_fetch_impl <EOL> try : <EOL> import tornado . curl_httpclient as curl <EOL> except ImportError : <EOL> pass <EOL> else : <EOL> from . stubs . tornado_stubs import vcr_fetch_impl <EOL> new_fetch_impl = vcr_fetch_impl ( <EOL> self . _cassette , _CurlAsyncHTTPClient_fetch_impl <EOL> ) <EOL> yield curl . CurlAsyncHTTPClient , '<STR_LIT>' , new_fetch_impl <EOL> def _urllib3_patchers ( self , cpool , stubs ) : <EOL> http_connection_remover = ConnectionRemover ( <EOL> self . _get_cassette_subclass ( stubs . VCRRequestsHTTPConnection ) <EOL> ) <EOL> https_connection_remover = ConnectionRemover ( <EOL> self . _get_cassette_subclass ( stubs . VCRRequestsHTTPSConnection ) <EOL> ) <EOL> mock_triples = ( <EOL> ( cpool , '<STR_LIT>' , stubs . VCRRequestsHTTPSConnection ) , <EOL> ( cpool , '<STR_LIT>' , stubs . VCRRequestsHTTPSConnection ) , <EOL> ( cpool , '<STR_LIT>' , stubs . VCRRequestsHTTPConnection ) , <EOL> ( cpool , '<STR_LIT>' , stubs . VCRRequestsHTTPSConnection ) , <EOL> ( cpool , '<STR_LIT>' , mock . Mock ( return_value = False ) ) , <EOL> ( cpool . HTTPConnectionPool , '<STR_LIT>' , stubs . VCRRequestsHTTPConnection ) , <EOL> ( cpool . HTTPSConnectionPool , '<STR_LIT>' , stubs . VCRRequestsHTTPSConnection ) , <EOL> ) <EOL> mock_triples += ( ( cpool . HTTPConnectionPool , '<STR_LIT>' , <EOL> self . _patched_get_conn ( cpool . HTTPConnectionPool , <EOL> lambda : cpool . HTTPConnection ) ) , <EOL> ( cpool . HTTPSConnectionPool , '<STR_LIT>' , <EOL> self . _patched_get_conn ( cpool . HTTPSConnectionPool , <EOL> lambda : cpool . HTTPSConnection ) ) , <EOL> ( cpool . HTTPConnectionPool , '<STR_LIT>' , <EOL> self . _patched_new_conn ( cpool . HTTPConnectionPool , <EOL> http_connection_remover ) ) , <EOL> ( cpool . HTTPSConnectionPool , '<STR_LIT>' , <EOL> self . _patched_new_conn ( cpool . HTTPSConnectionPool , <EOL> https_connection_remover ) ) ) <EOL> return itertools . chain ( self . _build_patchers_from_mock_triples ( mock_triples ) , <EOL> ( http_connection_remover , https_connection_remover ) ) <EOL> class ConnectionRemover ( object ) : <EOL> def __init__ ( self , connection_class ) : <EOL> self . _connection_class = connection_class <EOL> self . _connection_pool_to_connections = { } <EOL> def add_connection_to_pool_entry ( self , pool , connection ) : <EOL> if isinstance ( connection , self . _connection_class ) : <EOL> self . _connection_pool_to_connections . setdefault ( pool , set ( ) ) . add ( connection ) <EOL> def remove_connection_to_pool_entry ( self , pool , connection ) : <EOL> if isinstance ( connection , self . _connection_class ) : <EOL> self . _connection_pool_to_connections [ self . _connection_class ] . remove ( connection ) <EOL> def __enter__ ( self ) : <EOL> return self <EOL> def __exit__ ( self , * args ) : <EOL> for pool , connections in self . _connection_pool_to_connections . items ( ) : <EOL> readd_connections = [ ] <EOL> while pool . pool and not pool . pool . empty ( ) and connections : <EOL> connection = pool . pool . get ( ) <EOL> if isinstance ( connection , self . _connection_class ) : <EOL> connections . remove ( connection ) <EOL> else : <EOL> readd_connections . append ( connection ) <EOL> for connection in readd_connections : <EOL> pool . _put_conn ( connection ) <EOL> def reset_patchers ( ) : <EOL> yield mock . patch . object ( httplib , '<STR_LIT>' , _HTTPConnection ) <EOL> yield mock . patch . object ( httplib , '<STR_LIT>' , _HTTPSConnection ) <EOL> try : <EOL> import requests . packages . urllib3 . connectionpool as cpool <EOL> except ImportError : <EOL> pass <EOL> else : <EOL> yield mock . patch . object ( cpool , '<STR_LIT>' , _VerifiedHTTPSConnection ) <EOL> yield mock . patch . object ( cpool , '<STR_LIT>' , _cpoolHTTPConnection ) <EOL> if hasattr ( cpool . HTTPConnectionPool , '<STR_LIT>' ) : <EOL> yield mock . patch . object ( cpool . HTTPConnectionPool , '<STR_LIT>' , <EOL> _cpoolHTTPConnection ) <EOL> yield mock . patch . object ( cpool . HTTPSConnectionPool , '<STR_LIT>' , <EOL> _cpoolHTTPSConnection ) <EOL> if hasattr ( cpool , '<STR_LIT>' ) : <EOL> yield mock . patch . object ( cpool , '<STR_LIT>' , _cpoolHTTPSConnection ) <EOL> try : <EOL> import urllib3 . connectionpool as cpool <EOL> except ImportError : <EOL> pass <EOL> else : <EOL> yield mock . patch . object ( cpool , '<STR_LIT>' , _VerifiedHTTPSConnection ) <EOL> yield mock . patch . object ( cpool , '<STR_LIT>' , _HTTPConnection ) <EOL> yield mock . patch . object ( cpool , '<STR_LIT>' , _HTTPSConnection ) <EOL> if hasattr ( cpool . HTTPConnectionPool , '<STR_LIT>' ) : <EOL> yield mock . patch . object ( cpool . HTTPConnectionPool , '<STR_LIT>' , _HTTPConnection ) <EOL> yield mock . patch . object ( cpool . HTTPSConnectionPool , '<STR_LIT>' , _HTTPSConnection ) <EOL> try : <EOL> import httplib2 as cpool <EOL> except ImportError : <EOL> pass <EOL> else : <EOL> yield mock . patch . object ( cpool , '<STR_LIT>' , _HTTPConnectionWithTimeout ) <EOL> yield mock . patch . object ( cpool , '<STR_LIT>' , _HTTPSConnectionWithTimeout ) <EOL> yield mock . patch . object ( cpool , '<STR_LIT>' , _SCHEME_TO_CONNECTION ) <EOL> try : <EOL> import boto . https_connection as cpool <EOL> except ImportError : <EOL> pass <EOL> else : <EOL> yield mock . patch . object ( cpool , '<STR_LIT>' , <EOL> _CertValidatingHTTPSConnection ) <EOL> try : <EOL> import tornado . simple_httpclient as simple <EOL> except ImportError : <EOL> pass <EOL> else : <EOL> yield mock . patch . object ( <EOL> simple . SimpleAsyncHTTPClient , <EOL> '<STR_LIT>' , <EOL> _SimpleAsyncHTTPClient_fetch_impl , <EOL> ) <EOL> try : <EOL> import tornado . curl_httpclient as curl <EOL> except ImportError : <EOL> pass <EOL> else : <EOL> yield mock . patch . object ( <EOL> curl . CurlAsyncHTTPClient , <EOL> '<STR_LIT>' , <EOL> _CurlAsyncHTTPClient_fetch_impl , <EOL> ) <EOL> @ contextlib . contextmanager <EOL> def force_reset ( ) : <EOL> with contextlib . ExitStack ( ) as exit_stack : <EOL> for patcher in reset_patchers ( ) : <EOL> exit_stack . enter_context ( patcher ) <EOL> yield </s>
