<s> """<STR_LIT>""" <EOL> import socket <EOL> from twisted . trial import unittest <EOL> from twisted . internet . protocol import DatagramProtocol <EOL> from twisted . internet import udp <EOL> from twisted . python . runtime import platformType <EOL> if platformType == '<STR_LIT:win32>' : <EOL> from errno import WSAEWOULDBLOCK as EWOULDBLOCK <EOL> from errno import WSAECONNREFUSED as ECONNREFUSED <EOL> else : <EOL> from errno import EWOULDBLOCK <EOL> from errno import ECONNREFUSED <EOL> class StringUDPSocket ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , retvals ) : <EOL> self . retvals = retvals <EOL> self . connectedAddr = None <EOL> def connect ( self , addr ) : <EOL> self . connectedAddr = addr <EOL> def recvfrom ( self , size ) : <EOL> """<STR_LIT>""" <EOL> ret = self . retvals . pop ( <NUM_LIT:0> ) <EOL> if isinstance ( ret , socket . error ) : <EOL> raise ret <EOL> return ret , None <EOL> class KeepReads ( DatagramProtocol ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . reads = [ ] <EOL> def datagramReceived ( self , data , addr ) : <EOL> self . reads . append ( data ) <EOL> class ErrorsTestCase ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_socketReadNormal ( self ) : <EOL> """<STR_LIT>""" <EOL> udp . _sockErrReadIgnore . append ( - <NUM_LIT> ) <EOL> self . addCleanup ( udp . _sockErrReadIgnore . remove , - <NUM_LIT> ) <EOL> protocol = KeepReads ( ) <EOL> port = udp . Port ( None , protocol ) <EOL> port . socket = StringUDPSocket ( <EOL> [ "<STR_LIT:result>" , "<STR_LIT>" , socket . error ( - <NUM_LIT> ) , "<STR_LIT>" , <EOL> socket . error ( - <NUM_LIT> ) ] ) <EOL> port . doRead ( ) <EOL> self . assertEqual ( protocol . reads , [ "<STR_LIT:result>" , "<STR_LIT>" ] ) <EOL> port . doRead ( ) <EOL> self . assertEqual ( protocol . reads , [ "<STR_LIT:result>" , "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> def test_readImmediateError ( self ) : <EOL> """<STR_LIT>""" <EOL> udp . _sockErrReadRefuse . append ( - <NUM_LIT> ) <EOL> self . addCleanup ( udp . _sockErrReadRefuse . remove , - <NUM_LIT> ) <EOL> protocol = KeepReads ( ) <EOL> protocol . connectionRefused = lambda : <NUM_LIT:1> / <NUM_LIT:0> <EOL> port = udp . Port ( None , protocol ) <EOL> port . socket = StringUDPSocket ( [ "<STR_LIT:a>" , socket . error ( - <NUM_LIT> ) , "<STR_LIT:b>" , <EOL> socket . error ( EWOULDBLOCK ) ] ) <EOL> port . doRead ( ) <EOL> self . assertEqual ( protocol . reads , [ "<STR_LIT:a>" ] ) <EOL> port . doRead ( ) <EOL> self . assertEqual ( protocol . reads , [ "<STR_LIT:a>" , "<STR_LIT:b>" ] ) <EOL> def test_connectedReadImmediateError ( self ) : <EOL> """<STR_LIT>""" <EOL> udp . _sockErrReadRefuse . append ( - <NUM_LIT> ) <EOL> self . addCleanup ( udp . _sockErrReadRefuse . remove , - <NUM_LIT> ) <EOL> protocol = KeepReads ( ) <EOL> refused = [ ] <EOL> protocol . connectionRefused = lambda : refused . append ( True ) <EOL> port = udp . Port ( None , protocol ) <EOL> port . socket = StringUDPSocket ( [ "<STR_LIT:a>" , socket . error ( - <NUM_LIT> ) , "<STR_LIT:b>" , <EOL> socket . error ( EWOULDBLOCK ) ] ) <EOL> port . connect ( "<STR_LIT:127.0.0.1>" , <NUM_LIT> ) <EOL> port . doRead ( ) <EOL> self . assertEqual ( protocol . reads , [ "<STR_LIT:a>" ] ) <EOL> self . assertEqual ( refused , [ True ] ) <EOL> port . doRead ( ) <EOL> self . assertEqual ( protocol . reads , [ "<STR_LIT:a>" , "<STR_LIT:b>" ] ) <EOL> self . assertEqual ( refused , [ True ] ) <EOL> def test_readUnknownError ( self ) : <EOL> """<STR_LIT>""" <EOL> protocol = KeepReads ( ) <EOL> port = udp . Port ( None , protocol ) <EOL> port . socket = StringUDPSocket ( [ "<STR_LIT>" , socket . error ( - <NUM_LIT> ) ] ) <EOL> self . assertRaises ( socket . error , port . doRead ) <EOL> self . assertEqual ( protocol . reads , [ "<STR_LIT>" ] ) </s>
