<s> from __future__ import unicode_literals <EOL> from __future__ import print_function <EOL> import logging <EOL> import sys <EOL> from zerotest . fuzzy_matcher import FuzzyMatcher <EOL> LOG = logging . getLogger ( __name__ ) <EOL> _SERIALIZABLE_CONTENT_TYPE = { '<STR_LIT:application/json>' : '<STR_LIT>' } <EOL> class ResponseMatcher ( object ) : <EOL> def __init__ ( self , fuzzy_match = False , fuzzy_match_options = None , ignore_headers = None , ignore_all_headers = False , <EOL> ignore_fields = None ) : <EOL> """<STR_LIT>""" <EOL> ignore_headers = ignore_headers or [ ] <EOL> self . _fuzzy_match = fuzzy_match <EOL> self . _fuzzy_match_options = fuzzy_match_options or { } <EOL> self . _ignore_headers = set ( map ( lambda h : h . upper ( ) , ignore_headers ) ) <EOL> self . _ignore_all_headers = ignore_all_headers <EOL> self . _ignore_fields = ignore_fields <EOL> def _compare_status ( self , r1 , r2 ) : <EOL> assert r1 . status == r2 . status <EOL> def __remove_ignore_headers ( self , headers ) : <EOL> return { k . upper ( ) : headers [ k ] for k in headers if <EOL> k . upper ( ) not in self . _ignore_headers } <EOL> def _compare_headers ( self , expect , real ) : <EOL> if not self . _ignore_all_headers : <EOL> expect_headers = self . __remove_ignore_headers ( expect . headers ) <EOL> real_headers = self . __remove_ignore_headers ( real . headers ) <EOL> assert expect_headers == real_headers <EOL> def __delete_ignore_fields ( self , content ) : <EOL> from zerotest . utils . data_helper import delete_path_from_dict <EOL> for field_path in self . _ignore_fields or ( ) : <EOL> if field_path in content : <EOL> content . pop ( field_path , None ) <EOL> else : <EOL> delete_path_from_dict ( content , field_path ) <EOL> def _handle_content_type_json ( self , content ) : <EOL> import json <EOL> try : <EOL> content = json . loads ( content ) <EOL> except : <EOL> LOG . error ( "<STR_LIT>" ) <EOL> raise <EOL> self . __delete_ignore_fields ( content ) <EOL> return content <EOL> def _prepare_body_data ( self , r1 , r2 ) : <EOL> r1_content_type = r1 . get_header ( '<STR_LIT:Content-Type>' ) <EOL> r2_content_type = r2 . get_header ( '<STR_LIT:Content-Type>' ) <EOL> assert r1_content_type == r2_content_type <EOL> r1_content = r1 . body <EOL> r2_content = r2 . body <EOL> content_type = None <EOL> if r1_content_type : <EOL> values = r1_content_type . split ( '<STR_LIT:;>' ) <EOL> for v in values : <EOL> content_type = _SERIALIZABLE_CONTENT_TYPE . get ( v . lower ( ) ) <EOL> if content_type : <EOL> break <EOL> if content_type : <EOL> handler = getattr ( self , '<STR_LIT>' . format ( content_type ) ) <EOL> if handler : <EOL> r1_content = handler ( r1_content ) <EOL> r2_content = handler ( r2_content ) <EOL> return r1_content , r2_content <EOL> def _compare_body ( self , r1 , r2 ) : <EOL> r1_content , r2_content = self . _prepare_body_data ( r1 , r2 ) <EOL> is_formatted = all ( [ isinstance ( c , dict ) for c in ( r1_content , r2_content ) ] ) <EOL> if self . _fuzzy_match and is_formatted : <EOL> fuzzy_matcher = FuzzyMatcher ( ** self . _fuzzy_match_options ) <EOL> fuzzy_matcher . set_items ( r1_content , r2_content ) <EOL> try : <EOL> fuzzy_matcher . compare ( ) <EOL> except AssertionError : <EOL> print ( '<STR_LIT>' . format ( r1_content ) , file = sys . stderr ) <EOL> print ( '<STR_LIT>' . format ( r2_content ) , file = sys . stderr ) <EOL> raise <EOL> else : <EOL> assert r1_content == r2_content <EOL> def match_responses ( self , expect , real ) : <EOL> """<STR_LIT>""" <EOL> for attr in ( '<STR_LIT:status>' , '<STR_LIT>' , '<STR_LIT:body>' ) : <EOL> compare_func = '<STR_LIT>' . format ( attr ) <EOL> getattr ( self , compare_func ) ( expect , real ) </s>
