1 files changed, 161 insertions, 0 deletions
diff --git a/Lib/test/test_socketserver.py b/Lib/test/test_socketserver.py
new file mode 100644
index 0000000..2d6ee2d
--- /dev/null
+++ b/Lib/test/test_socketserver.py
@@ -0,0 +1,161 @@
+# Test suite for SocketServer.py
+
+# XXX This must be run manually -- somehow the I/O redirection of the
+# regression test breaks the test.
+
+from test_support import verbose, verify, TESTFN
+if not verbose:
+    raise ImportError, "test_socketserver can only be run manually"
+
+from SocketServer import *
+import socket
+import select
+import time
+import threading
+import os
+
+NREQ = 3
+DELAY = 0.5
+
+class MyMixinHandler:
+    def handle(self):
+        time.sleep(DELAY)
+        line = self.rfile.readline()
+        time.sleep(DELAY)
+        self.wfile.write(line)
+
+class MyStreamHandler(MyMixinHandler, StreamRequestHandler):
+    pass
+
+class MyDatagramHandler(MyMixinHandler, DatagramRequestHandler):
+    pass
+
+class MyMixinServer:
+    def serve_a_few(self):
+        for i in range(NREQ):
+            self.handle_request()
+    def handle_error(self, request, client_address):
+        self.close_request(request)
+        self.server_close()
+        raise
+
+teststring = "hello world\n"
+
+def receive(sock, n, timeout=20):
+    r, w, x = select.select([sock], [], [], timeout)
+    if sock in r:
+        return sock.recv(n)
+    else:
+        raise RuntimeError, "timed out on %s" % `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:
+        data = receive(s, 100)
+        buf += data
+    verify(buf == teststring)
+    s.close()
+
+def teststream(proto, addr):
+    s = socket.socket(proto, socket.SOCK_STREAM)
+    s.connect(addr)
+    s.send(teststring)
+    buf = data = receive(s, 100)
+    while data and '\n' not in buf:
+        data = receive(s, 100)
+        buf += data
+    verify(buf == teststring)
+    s.close()
+
+class ServerThread(threading.Thread):
+    def __init__(self, addr, svrcls, hdlrcls):
+        threading.Thread.__init__(self)
+        self.__addr = addr
+        self.__svrcls = svrcls
+        self.__hdlrcls = hdlrcls
+    def run(self):
+        class svrcls(MyMixinServer, self.__svrcls):
+            pass
+        if verbose: print "thread: creating server"
+        svr = svrcls(self.__addr, self.__hdlrcls)
+        if verbose: print "thread: serving three times"
+        svr.serve_a_few()
+        if verbose: print "thread: done"
+
+seed = 0
+def pickport():
+    global seed
+    seed += 1
+    return 10000 + (os.getpid() % 1000)*10 + seed
+
+host = ""
+testfiles = []
+def pickaddr(proto):
+    if proto == socket.AF_INET:
+        return (host, pickport())
+    else:
+        fn = TESTFN + str(pickport())
+        testfiles.append(fn)
+        return fn
+
+def cleanup():
+    for fn in testfiles:
+        try:
+            os.remove(fn)
+        except os.error:
+            pass
+    testfiles[:] = []
+
+def testloop(proto, servers, hdlrcls, testfunc):
+    for svrcls in servers:
+        addr = pickaddr(proto)
+        if verbose:
+            print "ADDR =", addr
+            print "CLASS =", svrcls
+        t = ServerThread(addr, svrcls, hdlrcls)
+        if verbose: print "server created"
+        t.start()
+        if verbose: print "server running"
+        for i in range(NREQ):
+            time.sleep(DELAY)
+            if verbose: print "test client", i
+            testfunc(proto, addr)
+        if verbose: print "waiting for server"
+        t.join()
+        if verbose: print "done"
+
+tcpservers = [TCPServer, ThreadingTCPServer]
+if hasattr(os, 'fork'):
+    tcpservers.append(ForkingTCPServer)
+udpservers = [UDPServer, ThreadingUDPServer]
+if hasattr(os, 'fork'):
+    udpservers.append(ForkingUDPServer)
+
+if not hasattr(socket, 'AF_UNIX'):
+    streamservers = []
+    dgramservers = []
+else:
+    class ForkingUnixStreamServer(ForkingMixIn, UnixStreamServer): pass
+    streamservers = [UnixStreamServer, ThreadingUnixStreamServer,
+                     ForkingUnixStreamServer]
+    class ForkingUnixDatagramServer(ForkingMixIn, UnixDatagramServer): pass
+    dgramservers = [UnixDatagramServer, ThreadingUnixDatagramServer,
+                    ForkingUnixDatagramServer]
+
+def testall():
+    testloop(socket.AF_INET, tcpservers, MyStreamHandler, teststream)
+    testloop(socket.AF_INET, udpservers, MyDatagramHandler, testdgram)
+    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)
+
+def main():
+    try:
+        testall()
+    finally:
+        cleanup()
+
+main()