<s> """<STR_LIT>""" <EOL> import os , sys <EOL> import errno <EOL> try : <EOL> import fcntl <EOL> except ImportError : <EOL> skip = "<STR_LIT>" <EOL> else : <EOL> from twisted . internet import fdesc <EOL> from twisted . python . util import untilConcludes <EOL> from twisted . trial import unittest <EOL> class NonBlockingTestCase ( unittest . SynchronousTestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_setNonBlocking ( self ) : <EOL> """<STR_LIT>""" <EOL> r , w = os . pipe ( ) <EOL> self . addCleanup ( os . close , r ) <EOL> self . addCleanup ( os . close , w ) <EOL> self . assertFalse ( fcntl . fcntl ( r , fcntl . F_GETFL ) & os . O_NONBLOCK ) <EOL> fdesc . setNonBlocking ( r ) <EOL> self . assertTrue ( fcntl . fcntl ( r , fcntl . F_GETFL ) & os . O_NONBLOCK ) <EOL> def test_setBlocking ( self ) : <EOL> """<STR_LIT>""" <EOL> r , w = os . pipe ( ) <EOL> self . addCleanup ( os . close , r ) <EOL> self . addCleanup ( os . close , w ) <EOL> fdesc . setNonBlocking ( r ) <EOL> fdesc . setBlocking ( r ) <EOL> self . assertFalse ( fcntl . fcntl ( r , fcntl . F_GETFL ) & os . O_NONBLOCK ) <EOL> class ReadWriteTestCase ( unittest . SynchronousTestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> self . r , self . w = os . pipe ( ) <EOL> fdesc . setNonBlocking ( self . r ) <EOL> fdesc . setNonBlocking ( self . w ) <EOL> def tearDown ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> os . close ( self . w ) <EOL> except OSError : <EOL> pass <EOL> try : <EOL> os . close ( self . r ) <EOL> except OSError : <EOL> pass <EOL> def write ( self , d ) : <EOL> """<STR_LIT>""" <EOL> return fdesc . writeToFD ( self . w , d ) <EOL> def read ( self ) : <EOL> """<STR_LIT>""" <EOL> l = [ ] <EOL> res = fdesc . readFromFD ( self . r , l . append ) <EOL> if res is None : <EOL> if l : <EOL> return l [ <NUM_LIT:0> ] <EOL> else : <EOL> return b"<STR_LIT>" <EOL> else : <EOL> return res <EOL> def test_writeAndRead ( self ) : <EOL> """<STR_LIT>""" <EOL> n = self . write ( b"<STR_LIT:hello>" ) <EOL> self . failUnless ( n > <NUM_LIT:0> ) <EOL> s = self . read ( ) <EOL> self . assertEqual ( len ( s ) , n ) <EOL> self . assertEqual ( b"<STR_LIT:hello>" [ : n ] , s ) <EOL> def test_writeAndReadLarge ( self ) : <EOL> """<STR_LIT>""" <EOL> orig = b"<STR_LIT>" * <NUM_LIT> <EOL> written = self . write ( orig ) <EOL> self . failUnless ( written > <NUM_LIT:0> ) <EOL> result = [ ] <EOL> resultlength = <NUM_LIT:0> <EOL> i = <NUM_LIT:0> <EOL> while resultlength < written or i < <NUM_LIT:50> : <EOL> result . append ( self . read ( ) ) <EOL> resultlength += len ( result [ - <NUM_LIT:1> ] ) <EOL> i += <NUM_LIT:1> <EOL> result = b"<STR_LIT>" . join ( result ) <EOL> self . assertEqual ( len ( result ) , written ) <EOL> self . assertEqual ( orig [ : written ] , result ) <EOL> def test_readFromEmpty ( self ) : <EOL> """<STR_LIT>""" <EOL> l = [ ] <EOL> result = fdesc . readFromFD ( self . r , l . append ) <EOL> self . assertEqual ( l , [ ] ) <EOL> self . assertEqual ( result , None ) <EOL> def test_readFromCleanClose ( self ) : <EOL> """<STR_LIT>""" <EOL> os . close ( self . w ) <EOL> self . assertEqual ( self . read ( ) , fdesc . CONNECTION_DONE ) <EOL> def test_writeToClosed ( self ) : <EOL> """<STR_LIT>""" <EOL> os . close ( self . r ) <EOL> self . assertEqual ( self . write ( b"<STR_LIT:s>" ) , fdesc . CONNECTION_LOST ) <EOL> def test_readFromInvalid ( self ) : <EOL> """<STR_LIT>""" <EOL> os . close ( self . r ) <EOL> self . assertEqual ( self . read ( ) , fdesc . CONNECTION_LOST ) <EOL> def test_writeToInvalid ( self ) : <EOL> """<STR_LIT>""" <EOL> os . close ( self . w ) <EOL> self . assertEqual ( self . write ( b"<STR_LIT:s>" ) , fdesc . CONNECTION_LOST ) <EOL> def test_writeErrors ( self ) : <EOL> """<STR_LIT>""" <EOL> oldOsWrite = os . write <EOL> def eagainWrite ( fd , data ) : <EOL> err = OSError ( ) <EOL> err . errno = errno . EAGAIN <EOL> raise err <EOL> os . write = eagainWrite <EOL> try : <EOL> self . assertEqual ( self . write ( b"<STR_LIT:s>" ) , <NUM_LIT:0> ) <EOL> finally : <EOL> os . write = oldOsWrite <EOL> def eintrWrite ( fd , data ) : <EOL> err = OSError ( ) <EOL> err . errno = errno . EINTR <EOL> raise err <EOL> os . write = eintrWrite <EOL> try : <EOL> self . assertEqual ( self . write ( b"<STR_LIT:s>" ) , <NUM_LIT:0> ) <EOL> finally : <EOL> os . write = oldOsWrite <EOL> class CloseOnExecTests ( unittest . SynchronousTestCase ) : <EOL> """<STR_LIT>""" <EOL> program = '''<STR_LIT>''' <EOL> def _execWithFileDescriptor ( self , fObj ) : <EOL> pid = os . fork ( ) <EOL> if pid == <NUM_LIT:0> : <EOL> try : <EOL> os . execv ( sys . executable , [ sys . executable , '<STR_LIT:-c>' , self . program % ( fObj . fileno ( ) , ) ] ) <EOL> except : <EOL> import traceback <EOL> traceback . print_exc ( ) <EOL> os . _exit ( <NUM_LIT:30> ) <EOL> else : <EOL> return untilConcludes ( os . waitpid , pid , <NUM_LIT:0> ) [ <NUM_LIT:1> ] <EOL> def test_setCloseOnExec ( self ) : <EOL> """<STR_LIT>""" <EOL> with open ( self . mktemp ( ) , '<STR_LIT:wb>' ) as fObj : <EOL> fdesc . _setCloseOnExec ( fObj . fileno ( ) ) <EOL> status = self . _execWithFileDescriptor ( fObj ) <EOL> self . assertTrue ( os . WIFEXITED ( status ) ) <EOL> self . assertEqual ( os . WEXITSTATUS ( status ) , <NUM_LIT:0> ) <EOL> def test_unsetCloseOnExec ( self ) : <EOL> """<STR_LIT>""" <EOL> with open ( self . mktemp ( ) , '<STR_LIT:wb>' ) as fObj : <EOL> fdesc . _setCloseOnExec ( fObj . fileno ( ) ) <EOL> fdesc . _unsetCloseOnExec ( fObj . fileno ( ) ) <EOL> status = self . _execWithFileDescriptor ( fObj ) <EOL> self . assertTrue ( os . WIFEXITED ( status ) ) <EOL> self . assertEqual ( os . WEXITSTATUS ( status ) , <NUM_LIT:20> ) </s>
