<s> import py <EOL> from py . __ . test . dist . txnode import TXNode <EOL> class EventQueue : <EOL> def __init__ ( self , registry , queue = None ) : <EOL> if queue is None : <EOL> queue = py . std . Queue . Queue ( ) <EOL> self . queue = queue <EOL> registry . register ( self ) <EOL> def geteventargs ( self , eventname , timeout = <NUM_LIT> ) : <EOL> events = [ ] <EOL> while <NUM_LIT:1> : <EOL> try : <EOL> eventcall = self . queue . get ( timeout = timeout ) <EOL> except py . std . Queue . Empty : <EOL> print "<STR_LIT>" , events <EOL> raise IOError ( "<STR_LIT>" % ( eventname ) ) <EOL> else : <EOL> name , args , kwargs = eventcall <EOL> assert isinstance ( name , str ) <EOL> if name == eventname : <EOL> if args : <EOL> return args <EOL> return kwargs <EOL> events . append ( name ) <EOL> if name == "<STR_LIT>" : <EOL> print str ( kwargs [ "<STR_LIT>" ] ) <EOL> class MySetup : <EOL> def __init__ ( self , request ) : <EOL> self . id = <NUM_LIT:0> <EOL> self . request = request <EOL> def geteventargs ( self , eventname , timeout = <NUM_LIT> ) : <EOL> eq = EventQueue ( self . config . pluginmanager , self . queue ) <EOL> return eq . geteventargs ( eventname , timeout = timeout ) <EOL> def makenode ( self , config = None ) : <EOL> if config is None : <EOL> config = py . test . config . _reparse ( [ ] ) <EOL> self . config = config <EOL> self . queue = py . std . Queue . Queue ( ) <EOL> self . xspec = py . execnet . XSpec ( "<STR_LIT>" ) <EOL> self . gateway = py . execnet . makegateway ( self . xspec ) <EOL> self . id += <NUM_LIT:1> <EOL> self . gateway . id = str ( self . id ) <EOL> self . node = TXNode ( self . gateway , self . config , putevent = self . queue . put ) <EOL> assert not self . node . channel . isclosed ( ) <EOL> return self . node <EOL> def xfinalize ( self ) : <EOL> if hasattr ( self , '<STR_LIT>' ) : <EOL> gw = self . node . gateway <EOL> print "<STR_LIT>" , gw <EOL> gw . exit ( ) <EOL> def pytest_funcarg__mysetup ( request ) : <EOL> mysetup = MySetup ( request ) <EOL> return mysetup <EOL> def test_node_hash_equality ( mysetup ) : <EOL> node = mysetup . makenode ( ) <EOL> node2 = mysetup . makenode ( ) <EOL> assert node != node2 <EOL> assert node == node <EOL> assert not ( node != node ) <EOL> class TestMasterSlaveConnection : <EOL> def test_crash_invalid_item ( self , mysetup ) : <EOL> node = mysetup . makenode ( ) <EOL> node . send ( <NUM_LIT> ) <EOL> kwargs = mysetup . geteventargs ( "<STR_LIT>" ) <EOL> assert kwargs [ '<STR_LIT>' ] is node <EOL> assert str ( kwargs [ '<STR_LIT:error>' ] ) . find ( "<STR_LIT>" ) != - <NUM_LIT:1> <EOL> def test_crash_killed ( self , testdir , mysetup ) : <EOL> if not hasattr ( py . std . os , '<STR_LIT>' ) : <EOL> py . test . skip ( "<STR_LIT>" ) <EOL> item = testdir . getitem ( """<STR_LIT>""" ) <EOL> node = mysetup . makenode ( item . config ) <EOL> node . send ( item ) <EOL> kwargs = mysetup . geteventargs ( "<STR_LIT>" ) <EOL> assert kwargs [ '<STR_LIT>' ] is node <EOL> assert str ( kwargs [ '<STR_LIT:error>' ] ) . find ( "<STR_LIT>" ) != - <NUM_LIT:1> <EOL> def test_node_down ( self , mysetup ) : <EOL> node = mysetup . makenode ( ) <EOL> node . shutdown ( ) <EOL> kwargs = mysetup . geteventargs ( "<STR_LIT>" ) <EOL> assert kwargs [ '<STR_LIT>' ] is node <EOL> assert not kwargs [ '<STR_LIT:error>' ] <EOL> node . callback ( node . ENDMARK ) <EOL> excinfo = py . test . raises ( IOError , <EOL> "<STR_LIT>" ) <EOL> def test_send_on_closed_channel ( self , testdir , mysetup ) : <EOL> item = testdir . getitem ( "<STR_LIT>" ) <EOL> node = mysetup . makenode ( item . config ) <EOL> node . channel . close ( ) <EOL> py . test . raises ( IOError , "<STR_LIT>" ) <EOL> def test_send_one ( self , testdir , mysetup ) : <EOL> item = testdir . getitem ( "<STR_LIT>" ) <EOL> node = mysetup . makenode ( item . config ) <EOL> node . send ( item ) <EOL> kwargs = mysetup . geteventargs ( "<STR_LIT>" ) <EOL> rep = kwargs [ '<STR_LIT>' ] <EOL> assert rep . passed <EOL> print rep <EOL> assert rep . item == item <EOL> def test_send_some ( self , testdir , mysetup ) : <EOL> items = testdir . getitems ( """<STR_LIT>""" ) <EOL> node = mysetup . makenode ( items [ <NUM_LIT:0> ] . config ) <EOL> for item in items : <EOL> node . send ( item ) <EOL> for outcome in "<STR_LIT>" . split ( ) : <EOL> kwargs = mysetup . geteventargs ( "<STR_LIT>" ) <EOL> report = kwargs [ '<STR_LIT>' ] <EOL> assert getattr ( report , outcome ) <EOL> node . sendlist ( items ) <EOL> for outcome in "<STR_LIT>" . split ( ) : <EOL> rep = mysetup . geteventargs ( "<STR_LIT>" ) [ '<STR_LIT>' ] <EOL> assert getattr ( rep , outcome ) </s>
