# Test suite for SocketServer.py
from test
import test_support
from test
.test_support
import verbose
, verify
, TESTFN
, TestSkipped
test_support
.requires('network')
from SocketServer
import *
line
= self
.rfile
.readline()
class MyStreamHandler(MyMixinHandler
, StreamRequestHandler
):
class MyDatagramHandler(MyMixinHandler
, DatagramRequestHandler
):
def handle_error(self
, request
, client_address
):
self
.close_request(request
)
teststring
= "hello world\n"
def receive(sock
, n
, timeout
=20):
r
, w
, x
= select
.select([sock
], [], [], timeout
)
raise RuntimeError, "timed out on %r" % (sock
,)
def testdgram(proto
, addr
):
s
= socket
.socket(proto
, socket
.SOCK_DGRAM
)
s
.sendto(teststring
, addr
)
buf
= data
= receive(s
, 100)
while data
and '\n' not in buf
:
verify(buf
== teststring
)
def teststream(proto
, addr
):
s
= socket
.socket(proto
, socket
.SOCK_STREAM
)
buf
= data
= receive(s
, 100)
while data
and '\n' not in buf
:
verify(buf
== teststring
)
class ServerThread(threading
.Thread
):
def __init__(self
, addr
, svrcls
, hdlrcls
):
threading
.Thread
.__init
__(self
)
class svrcls(MyMixinServer
, self
.__svrcls
):
if verbose
: print "thread: creating server"
svr
= svrcls(self
.__addr
, self
.__hdlrcls
)
if verbose
: print "thread: serving three times"
if verbose
: print "thread: done"
return 10000 + (os
.getpid() % 1000)*10 + seed
if proto
== socket
.AF_INET
:
return (host
, pickport())
fn
= TESTFN
+ str(pickport())
# AF_UNIX socket names on OS/2 require a specific prefix
# which can't include a drive letter and must also use
# backslashes as directory separators
if fn
[0] in (os
.sep
, os
.altsep
):
fn
= os
.path
.join('\socket', fn
)
fn
= fn
.replace(os
.sep
, os
.altsep
)
fn
= fn
.replace(os
.altsep
, os
.sep
)
def testloop(proto
, servers
, hdlrcls
, testfunc
):
t
= ServerThread(addr
, svrcls
, hdlrcls
)
if verbose
: print "server created"
if verbose
: print "server running"
if verbose
: print "test client", i
if verbose
: print "waiting for server"
tcpservers
= [TCPServer
, ThreadingTCPServer
]
if hasattr(os
, 'fork') and os
.name
not in ('os2',):
tcpservers
.append(ForkingTCPServer
)
udpservers
= [UDPServer
, ThreadingUDPServer
]
if hasattr(os
, 'fork') and os
.name
not in ('os2',):
udpservers
.append(ForkingUDPServer
)
if not hasattr(socket
, 'AF_UNIX'):
class ForkingUnixStreamServer(ForkingMixIn
, UnixStreamServer
): pass
streamservers
= [UnixStreamServer
, ThreadingUnixStreamServer
]
if hasattr(os
, 'fork') and os
.name
not in ('os2',):
streamservers
.append(ForkingUnixStreamServer
)
class ForkingUnixDatagramServer(ForkingMixIn
, UnixDatagramServer
): pass
dgramservers
= [UnixDatagramServer
, ThreadingUnixDatagramServer
]
if hasattr(os
, 'fork') and os
.name
not in ('os2',):
dgramservers
.append(ForkingUnixDatagramServer
)
testloop(socket
.AF_INET
, tcpservers
, MyStreamHandler
, teststream
)
testloop(socket
.AF_INET
, udpservers
, MyDatagramHandler
, testdgram
)
if hasattr(socket
, 'AF_UNIX'):
testloop(socket
.AF_UNIX
, streamservers
, MyStreamHandler
, teststream
)
# Alas, on Linux (at least) recvfrom() doesn't return a meaningful
# client address so this cannot work:
##testloop(socket.AF_UNIX, dgramservers, MyDatagramHandler, testdgram)
# If the import lock is held, the threads will hang.
raise TestSkipped("can't run when import lock is held")
if __name__
== "__main__":