#NEWLINE# def test_devpoll1( self ) : #NEWLINE# #INDENT# p = select . devpoll ( ) #NEWLINE# NUM_PIPES = 12 #NEWLINE# MSG = b' This is a test.' #NEWLINE# MSG_LEN = len ( MSG ) #NEWLINE# readers = [ ] #NEWLINE# writers = [ ] #NEWLINE# r2w = { #NEWLINE# #INDENT# #NEWLINE# #UNINDENT# } #NEWLINE# w2r = { #NEWLINE# #INDENT# #NEWLINE# #UNINDENT# } #NEWLINE# for i in range ( NUM_PIPES ) : #NEWLINE# #INDENT# ( rd , wr ) = os . pipe ( ) #NEWLINE# p . register ( rd ) #NEWLINE# p . modify ( rd , select . POLLIN ) #NEWLINE# p . register ( wr , select . POLLOUT ) #NEWLINE# readers . append ( rd ) #NEWLINE# writers . append ( wr ) #NEWLINE# r2w [ rd ] = wr #NEWLINE# w2r [ MSG ] = rd #NEWLINE# #UNINDENT# bufs = [ ] #NEWLINE# while writers : #NEWLINE# #INDENT# ready = p . poll ( ) #NEWLINE# ready_writers = find_ready_matching ( ready , select . POLLOUT ) #NEWLINE# if ( not ready_writers ) : #NEWLINE# #INDENT# self . fail ( 'no pipes ready for writing' ) #NEWLINE# #UNINDENT# wr = random . choice ( ready_writers ) #NEWLINE# os . write ( wr , MSG ) #NEWLINE# ready = p . poll ( ) #NEWLINE# ready_readers = find_ready_matching ( ready , select . POLLIN ) #NEWLINE# if ( not ready_readers ) : #NEWLINE# #INDENT# self . fail ( 'no pipes ready for reading' ) #NEWLINE# #UNINDENT# self . assertEqual ( [ w2r [ wr ] ] , ready_readers ) #NEWLINE# rd = ready_readers [ 0 ] #NEWLINE# buf = os . read ( rd , MSG_LEN ) #NEWLINE# self . assertEqual ( len ( buf ) , MSG_LEN ) #NEWLINE# bufs . append ( buf ) #NEWLINE# os . close ( r2w [ rd ] ) #NEWLINE# os . close ( rd ) #NEWLINE# p . unregister ( r2w [ rd ] ) #NEWLINE# p . unregister ( rd ) #NEWLINE# writers . remove ( r2w [ rd ] ) #NEWLINE# #UNINDENT# self . assertEqual ( bufs , ( [ MSG ] * NUM_PIPES ) )