<s> import copy <EOL> import inspect <EOL> import logging <EOL> import os <EOL> import zlib <EOL> from six . moves import urllib <EOL> import simplejson <EOL> import vcr <EOL> import wrapt <EOL> from okcupyd import settings <EOL> from okcupyd import util <EOL> log = logging . getLogger ( __name__ ) <EOL> TESTING_USERNAME = '<STR_LIT:username>' <EOL> TESTING_PASSWORD = '<STR_LIT:password>' <EOL> WBITS = <NUM_LIT:16> + zlib . MAX_WBITS <EOL> SHOULD_SCRUB = False <EOL> REPLACEMENTS = [ ] <EOL> REMOVE_OLD_CASSETTES = False <EOL> @ wrapt . decorator <EOL> def check_should_scrub ( function , instance , args , kwargs ) : <EOL> if SHOULD_SCRUB : <EOL> return function ( * args ) <EOL> else : <EOL> return args [ <NUM_LIT:0> ] <EOL> @ util . curry <EOL> def remove_headers ( request , headers_to_remove = ( ) ) : <EOL> headers = copy . copy ( request . headers ) <EOL> headers_to_remove = [ h . lower ( ) for h in headers_to_remove ] <EOL> keys = [ k for k in headers if k . lower ( ) in headers_to_remove ] <EOL> if keys : <EOL> for k in keys : <EOL> headers . pop ( k ) <EOL> request . headers = headers <EOL> return request <EOL> def scrub_request_body ( request ) : <EOL> if urllib . parse . urlsplit ( request . uri ) . path == '<STR_LIT>' : <EOL> request . body = scrub_query_string ( request . body ) <EOL> request . uri = scrub_uri ( request . uri ) <EOL> return request <EOL> def scrub_uri ( uri ) : <EOL> replaced = util . replace_all_case_insensitive ( uri , settings . USERNAME , <EOL> TESTING_USERNAME ) <EOL> return util . replace_all_case_insensitive ( replaced , settings . PASSWORD , <EOL> TESTING_PASSWORD ) <EOL> def scrub_query_string ( query_string ) : <EOL> request_dict = urllib . parse . parse_qs ( query_string ) <EOL> if '<STR_LIT:password>' not in request_dict : <EOL> return query_string <EOL> for key in request_dict : <EOL> request_dict [ key ] = request_dict [ key ] [ <NUM_LIT:0> ] <EOL> request_dict [ '<STR_LIT:username>' ] = TESTING_USERNAME <EOL> request_dict [ '<STR_LIT:password>' ] = TESTING_PASSWORD <EOL> return urllib . parse . urlencode ( request_dict ) <EOL> def gzip_string ( incoming ) : <EOL> if isinstance ( incoming , str ) and bytes is not str : <EOL> incoming = bytes ( incoming , '<STR_LIT:utf8>' ) <EOL> else : <EOL> incoming = incoming . encode ( '<STR_LIT:utf8>' ) <EOL> compress_object = zlib . compressobj ( <NUM_LIT:6> , zlib . DEFLATED , WBITS ) <EOL> start = compress_object . compress ( incoming ) <EOL> end = compress_object . flush ( ) <EOL> if not isinstance ( start , str ) : <EOL> return start + end <EOL> return '<STR_LIT>' . join ( [ start , end ] ) <EOL> def scrub_response_headers ( response ) : <EOL> for item in ( '<STR_LIT:location>' , '<STR_LIT>' ) : <EOL> if item in response [ '<STR_LIT>' ] : <EOL> response [ '<STR_LIT>' ] [ item ] = [ scrub_uri ( uri ) <EOL> for uri in response [ '<STR_LIT>' ] [ item ] ] <EOL> return response <EOL> def replace_json_fields ( body ) : <EOL> try : <EOL> response_dict = simplejson . loads ( body ) <EOL> except : <EOL> return body <EOL> if '<STR_LIT>' not in response_dict : <EOL> return body <EOL> if response_dict [ '<STR_LIT>' ] is not None : <EOL> response_dict [ '<STR_LIT>' ] = TESTING_USERNAME <EOL> response_dict [ '<STR_LIT>' ] = <NUM_LIT:1> <EOL> response_dict [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> return simplejson . dumps ( response_dict ) <EOL> def scrub_response ( response ) : <EOL> if not SHOULD_SCRUB : <EOL> return response <EOL> response = response . copy ( ) <EOL> response = scrub_response_headers ( response ) <EOL> body = response [ '<STR_LIT:body>' ] [ '<STR_LIT:string>' ] <EOL> try : <EOL> body = zlib . decompress ( response [ '<STR_LIT:body>' ] [ '<STR_LIT:string>' ] , WBITS ) . decode ( '<STR_LIT:utf8>' ) <EOL> except : <EOL> should_recompress = False <EOL> else : <EOL> should_recompress = True <EOL> body = replace_json_fields ( body ) <EOL> body = util . replace_all_case_insensitive ( body , settings . USERNAME , <EOL> TESTING_USERNAME ) <EOL> if should_recompress : <EOL> body = gzip_string ( body ) <EOL> response [ '<STR_LIT:body>' ] [ '<STR_LIT:string>' ] = body <EOL> return response <EOL> before_record = check_should_scrub ( util . compose ( <EOL> scrub_request_body , remove_headers ( headers_to_remove = ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' <EOL> ) ) <EOL> ) ) <EOL> def _maybe_decode ( maybe_bytes ) : <EOL> try : <EOL> return maybe_bytes . decode ( '<STR_LIT:utf-8>' ) <EOL> except ( AttributeError , UnicodeDecodeError ) : <EOL> return maybe_bytes <EOL> def _match_search_query ( left , right ) : <EOL> left_filter = set ( [ value for param_name , value in left <EOL> if '<STR_LIT>' in _maybe_decode ( param_name ) ] ) <EOL> right_filter = set ( [ value for param_name , value in right <EOL> if '<STR_LIT>' in _maybe_decode ( param_name ) ] ) <EOL> left_rest = set ( [ ( param_name , value ) for param_name , value in left <EOL> if '<STR_LIT>' not in _maybe_decode ( param_name ) ] ) <EOL> right_rest = set ( [ ( param_name , value ) for param_name , value in right <EOL> if '<STR_LIT>' not in _maybe_decode ( param_name ) ] ) <EOL> try : <EOL> log . info ( simplejson . dumps ( <EOL> { <EOL> '<STR_LIT>' : list ( <EOL> left_filter . symmetric_difference ( right_filter ) <EOL> ) , <EOL> '<STR_LIT>' : list ( <EOL> left_rest . symmetric_difference ( right_rest ) <EOL> ) , <EOL> } , encoding = '<STR_LIT:utf-8>' <EOL> ) ) <EOL> except Exception as e : <EOL> log . warning ( e ) <EOL> return left_filter == right_filter and left_rest == right_rest <EOL> def match_search_query ( left , right ) : <EOL> return _match_search_query ( left . query , right . query ) <EOL> def body_as_query_string ( left , right ) : <EOL> if left . path == right . path and '<STR_LIT>' in left . path : <EOL> return True <EOL> try : <EOL> left_qs_items = list ( urllib . parse . parse_qs ( left . body ) . items ( ) ) <EOL> right_qs_items = list ( urllib . parse . parse_qs ( right . body ) . items ( ) ) <EOL> except Exception as exc : <EOL> log . debug ( exc ) <EOL> return left . body == right . body <EOL> else : <EOL> left_qs_items = [ ( k , tuple ( v ) ) for k , v in left_qs_items ] <EOL> right_qs_items = [ ( k , tuple ( v ) ) for k , v in right_qs_items ] <EOL> return _match_search_query ( left_qs_items , right_qs_items ) <EOL> cassette_library_directory = os . path . join ( os . path . dirname ( os . path . dirname ( __file__ ) ) , <EOL> '<STR_LIT>' , '<STR_LIT>' ) <EOL> okcupyd_vcr = vcr . VCR ( match_on = ( '<STR_LIT:path>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ) , <EOL> before_record = ( before_record , ) , <EOL> before_record_response = scrub_response , <EOL> cassette_library_dir = cassette_library_directory , <EOL> path_transformer = vcr . VCR . ensure_suffix ( '<STR_LIT>' ) ) <EOL> okcupyd_vcr . register_matcher ( '<STR_LIT>' , body_as_query_string ) <EOL> okcupyd_vcr . register_matcher ( '<STR_LIT>' , match_search_query ) <EOL> match_on_no_body = list ( filter ( lambda x : '<STR_LIT:body>' not in x , okcupyd_vcr . match_on ) ) <EOL> @ wrapt . adapter_factory <EOL> def add_request_to_signature ( function ) : <EOL> argspec = inspect . getargspec ( function ) <EOL> return inspect . ArgSpec ( argspec . args + [ '<STR_LIT>' ] , argspec . varargs , argspec . keywords , argspec . defaults ) <EOL> @ wrapt . decorator ( adapter = add_request_to_signature ) <EOL> def skip_if_live ( function , instance , args , kwargs ) : <EOL> request = kwargs . pop ( '<STR_LIT>' ) <EOL> if request . config . getoption ( '<STR_LIT>' ) : <EOL> log . debug ( "<STR_LIT>" . format ( <EOL> function . __name__ <EOL> ) ) <EOL> else : <EOL> return function ( * args , ** kwargs ) <EOL> use_cassette = okcupyd_vcr . use_cassette </s>
