<s> import os , unittest , subprocess , re , signal , time <EOL> import nrepl <EOL> from collections import OrderedDict <EOL> from nrepl . bencode import encode , decode <EOL> import sys <EOL> PY2 = sys . version_info [ <NUM_LIT:0> ] == <NUM_LIT:2> <EOL> if not PY2 : <EOL> port_type = bytes <EOL> else : <EOL> port_type = ( ) <EOL> class BencodeTest ( unittest . TestCase ) : <EOL> def test_encoding ( self ) : <EOL> test_values = OrderedDict ( ( ( "<STR_LIT:a>" , <NUM_LIT:1> ) , ( "<STR_LIT:b>" , [ <NUM_LIT:2> , [ <NUM_LIT:3> ] ] ) , ( "<STR_LIT:c>" , [ { "<STR_LIT:x>" : [ "<STR_LIT:y>" ] } ] ) ) ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> encode ( test_values ) ) <EOL> self . assertEqual ( [ { u'<STR_LIT:a>' : <NUM_LIT:1> , u'<STR_LIT:c>' : [ { u'<STR_LIT:x>' : [ u'<STR_LIT:y>' ] } ] , u'<STR_LIT:b>' : [ <NUM_LIT:2> , [ <NUM_LIT:3> ] ] } ] , <EOL> list ( decode ( '<STR_LIT>' ) ) ) <EOL> class REPLTest ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> try : <EOL> self . proc = subprocess . Popen ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] , <EOL> stdout = subprocess . PIPE ) <EOL> except OSError : <EOL> self . proc = subprocess . Popen ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] , <EOL> stdout = subprocess . PIPE ) <EOL> self . port = re . findall ( b"<STR_LIT>" , self . proc . stdout . readline ( ) ) [ <NUM_LIT:0> ] <EOL> if isinstance ( self . port , port_type ) : <EOL> self . port = self . port . decode ( '<STR_LIT:utf-8>' ) <EOL> self . proc . stdout . close ( ) <EOL> def tearDown ( self ) : <EOL> c = nrepl . connect ( "<STR_LIT>" + self . port ) <EOL> c . write ( { "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT:code>" : "<STR_LIT>" } ) <EOL> self . proc . kill ( ) <EOL> def test_simple_connection ( self ) : <EOL> c = nrepl . connect ( "<STR_LIT>" + self . port ) <EOL> c . write ( { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> r = c . read ( ) <EOL> self . assertEqual ( [ "<STR_LIT>" ] , r [ "<STR_LIT:status>" ] ) <EOL> session = r [ "<STR_LIT>" ] <EOL> self . assertIsNotNone ( session ) <EOL> c . write ( { "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT:code>" : "<STR_LIT>" , "<STR_LIT>" : session } ) <EOL> r = c . read ( ) <EOL> self . assertEqual ( session , r [ "<STR_LIT>" ] ) <EOL> self . assertEqual ( "<STR_LIT:3>" , r [ "<STR_LIT:value>" ] ) <EOL> self . assertEqual ( [ "<STR_LIT>" ] , c . read ( ) [ "<STR_LIT:status>" ] ) <EOL> c . write ( { "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT:code>" : "<STR_LIT>" , "<STR_LIT>" : session } ) <EOL> self . assertEqual ( "<STR_LIT:5>" , c . read ( ) [ "<STR_LIT:value>" ] ) <EOL> c . close ( ) <EOL> def test_async_watches ( self ) : <EOL> c = nrepl . connect ( "<STR_LIT>" + self . port ) <EOL> wc = nrepl . WatchableConnection ( c ) <EOL> outs = { } <EOL> def add_resp ( session , msg ) : <EOL> out = msg . get ( "<STR_LIT>" , None ) <EOL> if out : outs [ session ] . append ( out ) <EOL> def watch_new_sessions ( msg , wc , key ) : <EOL> session = msg . get ( "<STR_LIT>" ) <EOL> outs [ session ] = [ ] <EOL> wc . watch ( "<STR_LIT>" + session , { "<STR_LIT>" : session } , <EOL> lambda msg , wc , key : add_resp ( session , msg ) ) <EOL> wc . watch ( "<STR_LIT>" , { "<STR_LIT>" : None } , watch_new_sessions ) <EOL> wc . send ( { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> wc . send ( { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> time . sleep ( <NUM_LIT:0.5> ) <EOL> for i , session in enumerate ( outs . keys ( ) ) : <EOL> wc . send ( { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : session , <EOL> "<STR_LIT:code>" : """<STR_LIT>""" % ( i , i ) } ) <EOL> time . sleep ( <NUM_LIT:2> ) <EOL> for i , ( session , _outs ) in enumerate ( outs . items ( ) ) : <EOL> self . assertEqual ( i , int ( _outs [ <NUM_LIT:0> ] ) ) <EOL> outs_values = list ( outs . values ( ) ) <EOL> self . assertTrue ( int ( outs_values [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] ) < int ( outs_values [ <NUM_LIT:1> ] [ <NUM_LIT:1> ] ) ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
