<s> """<STR_LIT>""" <EOL> try : <EOL> object <EOL> except NameError : <EOL> class object : pass <EOL> import errno <EOL> import select <EOL> import socket <EOL> import os <EOL> import sys <EOL> from test import test_support <EOL> import unittest <EOL> class SelectWrapper : <EOL> def __init__ ( self ) : <EOL> self . read_fds = [ ] <EOL> self . write_fds = [ ] <EOL> self . oob_fds = [ ] <EOL> self . timeout = None <EOL> def add_read_fd ( self , fd ) : <EOL> self . read_fds . append ( fd ) <EOL> def add_write_fd ( self , fd ) : <EOL> self . write_fds . append ( fd ) <EOL> def add_oob_fd ( self , fd ) : <EOL> self . oob_fds . append ( fd ) <EOL> def set_timeout ( self , timeout ) : <EOL> self . timeout = timeout <EOL> class PollWrapper : <EOL> def __init__ ( self ) : <EOL> self . timeout = None <EOL> self . poll_object = select . poll ( ) <EOL> def add_read_fd ( self , fd ) : <EOL> self . poll_object . register ( fd , select . POLL_IN ) <EOL> def add_write_fd ( self , fd ) : <EOL> self . poll_object . register ( fd , select . POLL_OUT ) <EOL> def add_oob_fd ( self , fd ) : <EOL> self . poll_object . register ( fd , select . POLL_PRI ) <EOL> class TestSelectInvalidParameters ( unittest . TestCase ) : <EOL> def testBadSelectSetTypes ( self ) : <EOL> for bad_select_set in [ None , <NUM_LIT:1> , ] : <EOL> for pos in range ( <NUM_LIT:2> ) : <EOL> args = [ [ ] , [ ] , [ ] ] <EOL> args [ pos ] = bad_select_set <EOL> try : <EOL> timeout = <NUM_LIT:0> <EOL> rfd , wfd , xfd = select . select ( args [ <NUM_LIT:0> ] , args [ <NUM_LIT:1> ] , args [ <NUM_LIT:2> ] , timeout ) <EOL> except ( select . error , TypeError ) : <EOL> pass <EOL> except Exception , x : <EOL> self . fail ( "<STR_LIT>" % ( str ( bad_select_set ) , str ( x ) ) ) <EOL> else : <EOL> self . fail ( "<STR_LIT>" % str ( bad_select_set ) ) <EOL> def testBadSelectableTypes ( self ) : <EOL> class Nope : pass <EOL> class Almost1 : <EOL> def fileno ( self ) : <EOL> return '<STR_LIT>' <EOL> class Almost2 : <EOL> def fileno ( self ) : <EOL> return '<STR_LIT>' <EOL> for bad_selectable in [ None , <NUM_LIT:1> , object ( ) , Nope ( ) , Almost1 ( ) , Almost2 ( ) ] : <EOL> try : <EOL> timeout = <NUM_LIT:0> <EOL> rfd , wfd , xfd = select . select ( [ bad_selectable ] , [ ] , [ ] , timeout ) <EOL> except ( TypeError , select . error ) , x : <EOL> pass <EOL> else : <EOL> self . fail ( "<STR_LIT>" % str ( bad_selectable ) ) <EOL> def testInvalidTimeoutTypes ( self ) : <EOL> for invalid_timeout in [ '<STR_LIT>' ] : <EOL> try : <EOL> rfd , wfd , xfd = select . select ( [ ] , [ ] , [ ] , invalid_timeout ) <EOL> except TypeError : <EOL> pass <EOL> else : <EOL> self . fail ( "<STR_LIT>" % invalid_timeout ) <EOL> def testInvalidTimeoutValues ( self ) : <EOL> for invalid_timeout in [ - <NUM_LIT:1> ] : <EOL> try : <EOL> rfd , wfd , xfd = select . select ( [ ] , [ ] , [ ] , invalid_timeout ) <EOL> except ( ValueError , select . error ) : <EOL> pass <EOL> else : <EOL> self . fail ( "<STR_LIT>" % invalid_timeout ) <EOL> class TestSelectClientSocket ( unittest . TestCase ) : <EOL> def testUnconnectedSocket ( self ) : <EOL> sockets = [ socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) for x in range ( <NUM_LIT:5> ) ] <EOL> for pos in range ( <NUM_LIT:2> ) : <EOL> args = [ [ ] , [ ] , [ ] ] <EOL> args [ pos ] = sockets <EOL> timeout = <NUM_LIT:0> <EOL> rfd , wfd , xfd = select . select ( args [ <NUM_LIT:0> ] , args [ <NUM_LIT:1> ] , args [ <NUM_LIT:2> ] , timeout ) <EOL> for s in sockets : <EOL> self . failIf ( s in rfd ) <EOL> self . failIf ( s in wfd ) <EOL> class TestPollClientSocket ( unittest . TestCase ) : <EOL> def testEventConstants ( self ) : <EOL> for event_name in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , ] : <EOL> self . failUnless ( hasattr ( select , '<STR_LIT>' % event_name ) ) <EOL> def testUnregisterRaisesKeyError ( self ) : <EOL> s = socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) <EOL> poll_object = select . poll ( ) <EOL> try : <EOL> poll_object . unregister ( s ) <EOL> except KeyError : <EOL> pass <EOL> else : <EOL> self . fail ( "<STR_LIT>" ) <EOL> import test_socket <EOL> class ThreadedPollClientSocket ( test_socket . ThreadedTCPSocketTest ) : <EOL> def testSocketRegisteredBeforeConnected ( self ) : <EOL> self . cli_conn = self . serv . accept ( ) <EOL> def _testSocketRegisteredBeforeConnected ( self ) : <EOL> timeout = <NUM_LIT:1000> <EOL> poll_object = select . poll ( ) <EOL> poll_object . register ( self . cli , select . POLLOUT ) <EOL> result_list = poll_object . poll ( timeout ) <EOL> result_sockets = [ r [ <NUM_LIT:0> ] for r in result_list ] <EOL> self . failIf ( self . cli in result_sockets , "<STR_LIT>" ) <EOL> self . cli . setblocking ( <NUM_LIT:0> ) <EOL> self . cli . connect ( ( test_socket . HOST , test_socket . PORT ) ) <EOL> result_list = poll_object . poll ( timeout ) <EOL> result_sockets = [ r [ <NUM_LIT:0> ] for r in result_list ] <EOL> self . failUnless ( self . cli in result_sockets , "<STR_LIT>" ) <EOL> def testSocketMustBeNonBlocking ( self ) : <EOL> self . cli_conn = self . serv . accept ( ) <EOL> def _testSocketMustBeNonBlocking ( self ) : <EOL> self . cli . setblocking ( <NUM_LIT:1> ) <EOL> self . cli . connect ( ( test_socket . HOST , test_socket . PORT ) ) <EOL> timeout = <NUM_LIT:1000> <EOL> poll_object = select . poll ( ) <EOL> try : <EOL> poll_object . register ( self . cli ) <EOL> except select . error , se : <EOL> self . failUnlessEqual ( se [ <NUM_LIT:0> ] , errno . ESOCKISBLOCKING ) <EOL> except Exception , x : <EOL> self . fail ( "<STR_LIT>" % str ( x ) ) <EOL> else : <EOL> self . fail ( "<STR_LIT>" ) <EOL> class TestPipes ( unittest . TestCase ) : <EOL> verbose = <NUM_LIT:1> <EOL> def test ( self ) : <EOL> import sys <EOL> from test . test_support import verbose <EOL> if sys . platform [ : <NUM_LIT:3> ] in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) : <EOL> if verbose : <EOL> print "<STR_LIT>" , sys . platform <EOL> return <EOL> cmd = '<STR_LIT>' <EOL> p = os . popen ( cmd , '<STR_LIT:r>' ) <EOL> for tout in ( <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:4> , <NUM_LIT:8> , <NUM_LIT:16> ) + ( None , ) * <NUM_LIT:10> : <EOL> if verbose : <EOL> print '<STR_LIT>' , tout <EOL> rfd , wfd , xfd = select . select ( [ p ] , [ ] , [ ] , tout ) <EOL> if ( rfd , wfd , xfd ) == ( [ ] , [ ] , [ ] ) : <EOL> continue <EOL> if ( rfd , wfd , xfd ) == ( [ p ] , [ ] , [ ] ) : <EOL> line = p . readline ( ) <EOL> if verbose : <EOL> print repr ( line ) <EOL> if not line : <EOL> if verbose : <EOL> print '<STR_LIT>' <EOL> break <EOL> continue <EOL> self . fail ( '<STR_LIT>' % str ( rfd , wfd , xfd ) ) <EOL> p . close ( ) <EOL> def test_main ( ) : <EOL> tests = [ <EOL> TestSelectInvalidParameters , <EOL> TestSelectClientSocket , <EOL> TestPollClientSocket , <EOL> ThreadedPollClientSocket , <EOL> ] <EOL> if sys . platform [ : <NUM_LIT:4> ] != '<STR_LIT>' : <EOL> tests . append ( TestPipes ) <EOL> suites = [ unittest . makeSuite ( klass , '<STR_LIT:test>' ) for klass in tests ] <EOL> test_support . run_suite ( unittest . TestSuite ( suites ) ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> test_main ( ) </s>
