<s> import sys , os , unittest <EOL> from smisk . test . live import LiveTestCase , suite_if_possible <EOL> class KVSExample ( LiveTestCase ) : <EOL> def test_0_simple_response ( self ) : <EOL> client = self . connection ( ) <EOL> rsp = client . request ( '<STR_LIT:GET>' , '<STR_LIT:/>' , headers = [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT:utf-8>' ) , <EOL> ] ) <EOL> payload = eval ( rsp . body ) <EOL> self . assertTrue ( '<STR_LIT>' in payload ) <EOL> self . assertEquals ( len ( payload [ '<STR_LIT>' ] ) , <NUM_LIT:0> ) <EOL> client . disconnect ( ) <EOL> def test_1_put_get_delete_single_connn ( self ) : <EOL> def reconnect ( client ) : <EOL> return client <EOL> self . _test_put_get_delete ( reconnect ) <EOL> def test_2_put_get_delete_serial_conns ( self ) : <EOL> def reconnect ( client ) : <EOL> client . disconnect ( ) <EOL> return self . connection ( ) <EOL> def test_3_put_get_delete_parallel_conns ( self ) : <EOL> def reconnect ( client ) : <EOL> return self . connection ( ) <EOL> def _test_put_get_delete ( self , reconnect ) : <EOL> orig_payload = { '<STR_LIT:value>' : { '<STR_LIT:a>' : u'<STR_LIT>' , '<STR_LIT:b>' : u'<STR_LIT>' , '<STR_LIT>' : u'<STR_LIT>' } } <EOL> orig_payload_s = repr ( orig_payload ) <EOL> send_payload_headers = [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , str ( len ( orig_payload_s ) ) ) <EOL> ] <EOL> client = self . connection ( ) <EOL> rsp = client . request ( '<STR_LIT>' , '<STR_LIT>' , body = orig_payload_s , headers = send_payload_headers ) <EOL> rsp . read ( ) <EOL> try : <EOL> self . assertTrue ( rsp . is_ok , rsp . status ) <EOL> except : <EOL> print >> sys . stderr , rsp . body <EOL> raise <EOL> client = reconnect ( client ) <EOL> rsp = client . request ( '<STR_LIT:GET>' , '<STR_LIT>' ) <EOL> self . assertTrue ( rsp . is_ok ) <EOL> rsp_payload = eval ( rsp . body ) <EOL> print rsp_payload <EOL> self . assertTrue ( '<STR_LIT:value>' in rsp_payload ) <EOL> value = rsp_payload [ '<STR_LIT:value>' ] <EOL> self . assertEquals ( rsp_payload [ '<STR_LIT:value>' ] [ '<STR_LIT:a>' ] , orig_payload [ '<STR_LIT:value>' ] [ '<STR_LIT:a>' ] ) <EOL> self . assertEquals ( rsp_payload [ '<STR_LIT:value>' ] [ '<STR_LIT:b>' ] , orig_payload [ '<STR_LIT:value>' ] [ '<STR_LIT:b>' ] ) <EOL> self . assertEquals ( rsp_payload [ '<STR_LIT:value>' ] [ '<STR_LIT>' ] , orig_payload [ '<STR_LIT:value>' ] [ '<STR_LIT>' ] ) <EOL> client . disconnect ( ) <EOL> def unused ( ) : <EOL> rsp = self . client . request ( '<STR_LIT:GET>' , '<STR_LIT:/>' , headers = [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ] ) <EOL> self . assertTrue ( rsp . is_ok ) <EOL> self . assertResponseHeaderIsSet ( rsp , '<STR_LIT>' ) <EOL> self . assertResponseHeaderEquals ( rsp , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertResponseHeaderEquals ( rsp , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertResponseHeaderIsSet ( rsp , '<STR_LIT>' ) <EOL> for v in rsp . header ( '<STR_LIT>' ) : <EOL> self . assertTrue ( '<STR_LIT>' in v ) <EOL> self . assertTrue ( '<STR_LIT>' in v ) <EOL> self . assertNotEquals ( rsp . body , '<STR_LIT>' ) <EOL> def suite ( enable_experimental = False ) : <EOL> if enable_experimental : <EOL> return suite_if_possible ( KVSExample ) <EOL> else : <EOL> return unittest . TestSuite ( [ ] ) <EOL> def test ( enable_experimental = False ) : <EOL> return unittest . TextTestRunner ( ) . run ( suite ( enable_experimental ) ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> import logging <EOL> logging . basicConfig ( level = logging . INFO , format = "<STR_LIT>" ) <EOL> test ( True ) </s>
