<s> """<STR_LIT>""" <EOL> import errno , socket <EOL> from zope . interface import verify <EOL> from twisted . python . log import err <EOL> from twisted . internet . interfaces import IReactorSocket <EOL> from twisted . internet . error import UnsupportedAddressFamily <EOL> from twisted . internet . protocol import DatagramProtocol , ServerFactory <EOL> from twisted . internet . test . reactormixins import ( <EOL> ReactorBuilder , needsRunningReactor ) <EOL> class IReactorSocketVerificationTestsBuilder ( ReactorBuilder ) : <EOL> """<STR_LIT>""" <EOL> requiredInterfaces = [ IReactorSocket ] <EOL> def test_provider ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = self . buildReactor ( ) <EOL> self . assertTrue ( <EOL> verify . verifyObject ( IReactorSocket , reactor ) ) <EOL> class AdoptStreamPortErrorsTestsBuilder ( ReactorBuilder ) : <EOL> """<STR_LIT>""" <EOL> requiredInterfaces = [ IReactorSocket ] <EOL> def test_invalidDescriptor ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = self . buildReactor ( ) <EOL> probe = socket . socket ( ) <EOL> fileno = probe . fileno ( ) <EOL> probe . close ( ) <EOL> exc = self . assertRaises ( <EOL> socket . error , <EOL> reactor . adoptStreamPort , fileno , socket . AF_INET , ServerFactory ( ) ) <EOL> self . assertEqual ( exc . args [ <NUM_LIT:0> ] , errno . EBADF ) <EOL> def test_invalidAddressFamily ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = self . buildReactor ( ) <EOL> port = socket . socket ( ) <EOL> port . listen ( <NUM_LIT:1> ) <EOL> self . addCleanup ( port . close ) <EOL> arbitrary = <NUM_LIT:2> ** <NUM_LIT:16> + <NUM_LIT:7> <EOL> self . assertRaises ( <EOL> UnsupportedAddressFamily , <EOL> reactor . adoptStreamPort , port . fileno ( ) , arbitrary , ServerFactory ( ) ) <EOL> def test_stopOnlyCloses ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = self . buildReactor ( ) <EOL> portSocket = socket . socket ( ) <EOL> self . addCleanup ( portSocket . close ) <EOL> portSocket . listen ( <NUM_LIT:1> ) <EOL> portSocket . setblocking ( False ) <EOL> port = reactor . adoptStreamPort ( <EOL> portSocket . fileno ( ) , portSocket . family , ServerFactory ( ) ) <EOL> d = port . stopListening ( ) <EOL> def stopped ( ignored ) : <EOL> exc = self . assertRaises ( socket . error , portSocket . accept ) <EOL> self . assertEqual ( exc . args [ <NUM_LIT:0> ] , errno . EAGAIN ) <EOL> d . addCallback ( stopped ) <EOL> d . addErrback ( err , "<STR_LIT>" ) <EOL> needsRunningReactor ( <EOL> reactor , <EOL> lambda : d . addCallback ( lambda ignored : reactor . stop ( ) ) ) <EOL> reactor . run ( ) <EOL> class AdoptStreamConnectionErrorsTestsBuilder ( ReactorBuilder ) : <EOL> """<STR_LIT>""" <EOL> requiredInterfaces = [ IReactorSocket ] <EOL> def test_invalidAddressFamily ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = self . buildReactor ( ) <EOL> connection = socket . socket ( ) <EOL> self . addCleanup ( connection . close ) <EOL> arbitrary = <NUM_LIT:2> ** <NUM_LIT:16> + <NUM_LIT:7> <EOL> self . assertRaises ( <EOL> UnsupportedAddressFamily , <EOL> reactor . adoptStreamConnection , connection . fileno ( ) , arbitrary , <EOL> ServerFactory ( ) ) <EOL> class AdoptDatagramPortErrorsTestsBuilder ( ReactorBuilder ) : <EOL> """<STR_LIT>""" <EOL> requiredInterfaces = [ IReactorSocket ] <EOL> def test_invalidDescriptor ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = self . buildReactor ( ) <EOL> probe = socket . socket ( ) <EOL> fileno = probe . fileno ( ) <EOL> probe . close ( ) <EOL> exc = self . assertRaises ( <EOL> socket . error , <EOL> reactor . adoptDatagramPort , fileno , socket . AF_INET , <EOL> DatagramProtocol ( ) ) <EOL> self . assertEqual ( exc . args [ <NUM_LIT:0> ] , errno . EBADF ) <EOL> def test_invalidAddressFamily ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = self . buildReactor ( ) <EOL> port = socket . socket ( socket . AF_INET , socket . SOCK_DGRAM ) <EOL> self . addCleanup ( port . close ) <EOL> arbitrary = <NUM_LIT:2> ** <NUM_LIT:16> + <NUM_LIT:7> <EOL> self . assertRaises ( <EOL> UnsupportedAddressFamily , <EOL> reactor . adoptDatagramPort , port . fileno ( ) , arbitrary , <EOL> DatagramProtocol ( ) ) <EOL> def test_stopOnlyCloses ( self ) : <EOL> """<STR_LIT>""" <EOL> reactor = self . buildReactor ( ) <EOL> portSocket = socket . socket ( socket . AF_INET , socket . SOCK_DGRAM ) <EOL> self . addCleanup ( portSocket . close ) <EOL> portSocket . setblocking ( False ) <EOL> port = reactor . adoptDatagramPort ( <EOL> portSocket . fileno ( ) , portSocket . family , DatagramProtocol ( ) ) <EOL> d = port . stopListening ( ) <EOL> def stopped ( ignored ) : <EOL> exc = self . assertRaises ( socket . error , portSocket . recvfrom , <NUM_LIT:1> ) <EOL> self . assertEqual ( exc . args [ <NUM_LIT:0> ] , errno . EAGAIN ) <EOL> d . addCallback ( stopped ) <EOL> d . addErrback ( err , "<STR_LIT>" ) <EOL> needsRunningReactor ( <EOL> reactor , <EOL> lambda : d . addCallback ( lambda ignored : reactor . stop ( ) ) ) <EOL> reactor . run ( ) <EOL> globals ( ) . update ( IReactorSocketVerificationTestsBuilder . makeTestCaseClasses ( ) ) <EOL> globals ( ) . update ( AdoptStreamPortErrorsTestsBuilder . makeTestCaseClasses ( ) ) <EOL> globals ( ) . update ( AdoptStreamConnectionErrorsTestsBuilder . makeTestCaseClasses ( ) ) <EOL> globals ( ) . update ( AdoptDatagramPortErrorsTestsBuilder . makeTestCaseClasses ( ) ) </s>
