<s> """<STR_LIT>""" <EOL> import json <EOL> import requests <EOL> import six <EOL> from six . moves . urllib import parse <EOL> from magnumclient . openstack . common . apiclient import client <EOL> def assert_has_keys ( dct , required = None , optional = None ) : <EOL> required = required or [ ] <EOL> optional = optional or [ ] <EOL> for k in required : <EOL> try : <EOL> assert k in dct <EOL> except AssertionError : <EOL> extra_keys = set ( dct . keys ( ) ) . difference ( set ( required + optional ) ) <EOL> raise AssertionError ( "<STR_LIT>" % <EOL> list ( extra_keys ) ) <EOL> class TestResponse ( requests . Response ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , data ) : <EOL> super ( TestResponse , self ) . __init__ ( ) <EOL> self . _content_consumed = True <EOL> if isinstance ( data , dict ) : <EOL> self . status_code = data . get ( '<STR_LIT>' , <NUM_LIT:200> ) <EOL> text = data . get ( '<STR_LIT:text>' , "<STR_LIT>" ) <EOL> if isinstance ( text , ( dict , list ) ) : <EOL> self . _content = json . dumps ( text ) <EOL> default_headers = { <EOL> "<STR_LIT:Content-Type>" : "<STR_LIT:application/json>" , <EOL> } <EOL> else : <EOL> self . _content = text <EOL> default_headers = { } <EOL> if six . PY3 and isinstance ( self . _content , six . string_types ) : <EOL> self . _content = self . _content . encode ( '<STR_LIT:utf-8>' , '<STR_LIT:strict>' ) <EOL> self . headers = data . get ( '<STR_LIT>' ) or default_headers <EOL> else : <EOL> self . status_code = data <EOL> def __eq__ ( self , other ) : <EOL> return ( self . status_code == other . status_code and <EOL> self . headers == other . headers and <EOL> self . _content == other . _content ) <EOL> class FakeHTTPClient ( client . HTTPClient ) : <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> self . callstack = [ ] <EOL> self . fixtures = kwargs . pop ( "<STR_LIT>" , None ) or { } <EOL> if not args and "<STR_LIT>" not in kwargs : <EOL> args = ( None , ) <EOL> super ( FakeHTTPClient , self ) . __init__ ( * args , ** kwargs ) <EOL> def assert_called ( self , method , url , body = None , pos = - <NUM_LIT:1> ) : <EOL> """<STR_LIT>""" <EOL> expected = ( method , url ) <EOL> called = self . callstack [ pos ] [ <NUM_LIT:0> : <NUM_LIT:2> ] <EOL> assert self . callstack , "<STR_LIT>" % expected <EOL> assert expected == called , '<STR_LIT>' % ( expected + called ) <EOL> if body is not None : <EOL> if self . callstack [ pos ] [ <NUM_LIT:3> ] != body : <EOL> raise AssertionError ( '<STR_LIT>' % <EOL> ( self . callstack [ pos ] [ <NUM_LIT:3> ] , body ) ) <EOL> def assert_called_anytime ( self , method , url , body = None ) : <EOL> """<STR_LIT>""" <EOL> expected = ( method , url ) <EOL> assert self . callstack , "<STR_LIT>" % expected <EOL> found = False <EOL> entry = None <EOL> for entry in self . callstack : <EOL> if expected == entry [ <NUM_LIT:0> : <NUM_LIT:2> ] : <EOL> found = True <EOL> break <EOL> assert found , '<STR_LIT>' % ( method , url , self . callstack ) <EOL> if body is not None : <EOL> assert entry [ <NUM_LIT:3> ] == body , "<STR_LIT>" % ( entry [ <NUM_LIT:3> ] , body ) <EOL> self . callstack = [ ] <EOL> def clear_callstack ( self ) : <EOL> self . callstack = [ ] <EOL> def authenticate ( self ) : <EOL> pass <EOL> def client_request ( self , client , method , url , ** kwargs ) : <EOL> if method in [ "<STR_LIT:GET>" , "<STR_LIT>" ] : <EOL> assert "<STR_LIT>" not in kwargs <EOL> self . callstack . append ( <EOL> ( method , <EOL> url , <EOL> kwargs . get ( "<STR_LIT>" ) or { } , <EOL> kwargs . get ( "<STR_LIT>" ) or kwargs . get ( "<STR_LIT:data>" ) ) ) <EOL> try : <EOL> fixture = self . fixtures [ url ] [ method ] <EOL> except KeyError : <EOL> pass <EOL> else : <EOL> return TestResponse ( { "<STR_LIT>" : fixture [ <NUM_LIT:0> ] , <EOL> "<STR_LIT:text>" : fixture [ <NUM_LIT:1> ] } ) <EOL> args = parse . parse_qsl ( parse . urlparse ( url ) [ <NUM_LIT:4> ] ) <EOL> kwargs . update ( args ) <EOL> munged_url = url . rsplit ( '<STR_LIT:?>' , <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> munged_url = munged_url . strip ( '<STR_LIT:/>' ) . replace ( '<STR_LIT:/>' , '<STR_LIT:_>' ) . replace ( '<STR_LIT:.>' , '<STR_LIT:_>' ) <EOL> munged_url = munged_url . replace ( '<STR_LIT:->' , '<STR_LIT:_>' ) <EOL> callback = "<STR_LIT>" % ( method . lower ( ) , munged_url ) <EOL> if not hasattr ( self , callback ) : <EOL> raise AssertionError ( '<STR_LIT>' <EOL> '<STR_LIT>' % <EOL> ( method , url , callback ) ) <EOL> resp = getattr ( self , callback ) ( ** kwargs ) <EOL> if len ( resp ) == <NUM_LIT:3> : <EOL> status , headers , body = resp <EOL> else : <EOL> status , body = resp <EOL> headers = { } <EOL> self . last_request_id = headers . get ( '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> return TestResponse ( { <EOL> "<STR_LIT>" : status , <EOL> "<STR_LIT:text>" : body , <EOL> "<STR_LIT>" : headers , <EOL> } ) </s>
