#NEWLINE# def test_poller_events( self ) : #NEWLINE# #INDENT# 'Tornado poller implementation maps events correctly' #NEWLINE# ( req , rep ) = self . create_bound_pair ( zmq . REQ , zmq . REP ) #NEWLINE# poller = ioloop . ZMQPoller ( ) #NEWLINE# poller . register ( req , ioloop . IOLoop . READ ) #NEWLINE# poller . register ( rep , ioloop . IOLoop . READ ) #NEWLINE# events = dict ( poller . poll ( 0 ) ) #NEWLINE# self . assertEquals ( events . get ( rep ) , None ) #NEWLINE# self . assertEquals ( events . get ( req ) , None ) #NEWLINE# poller . register ( req , ioloop . IOLoop . WRITE ) #NEWLINE# poller . register ( self , ioloop . IOLoop . WRITE ) #NEWLINE# events = dict ( poller . poll ( 1 ) ) #NEWLINE# self . assertEquals ( events . get ( req ) , ioloop . IOLoop . WRITE ) #NEWLINE# self . assertEquals ( events . get ( rep ) , None ) #NEWLINE# poller . register ( rep , ioloop . IOLoop . READ ) #NEWLINE# req . send ( b'hi' ) #NEWLINE# events = dict ( poller . poll ( 1 ) ) #NEWLINE# self . assertEquals ( events . get ( rep ) , ioloop . IOLoop . READ ) #NEWLINE# self . assertEquals ( events . get ( req ) , None )