PyXR

c:\python24\lib \ test \ test_socketserver.py



0001 # Test suite for SocketServer.py
0002 
0003 from test import test_support
0004 from test.test_support import verbose, verify, TESTFN, TestSkipped
0005 test_support.requires('network')
0006 
0007 from SocketServer import *
0008 import socket
0009 import select
0010 import time
0011 import threading
0012 import os
0013 
0014 NREQ = 3
0015 DELAY = 0.5
0016 
0017 class MyMixinHandler:
0018     def handle(self):
0019         time.sleep(DELAY)
0020         line = self.rfile.readline()
0021         time.sleep(DELAY)
0022         self.wfile.write(line)
0023 
0024 class MyStreamHandler(MyMixinHandler, StreamRequestHandler):
0025     pass
0026 
0027 class MyDatagramHandler(MyMixinHandler, DatagramRequestHandler):
0028     pass
0029 
0030 class MyMixinServer:
0031     def serve_a_few(self):
0032         for i in range(NREQ):
0033             self.handle_request()
0034     def handle_error(self, request, client_address):
0035         self.close_request(request)
0036         self.server_close()
0037         raise
0038 
0039 teststring = "hello world\n"
0040 
0041 def receive(sock, n, timeout=20):
0042     r, w, x = select.select([sock], [], [], timeout)
0043     if sock in r:
0044         return sock.recv(n)
0045     else:
0046         raise RuntimeError, "timed out on %r" % (sock,)
0047 
0048 def testdgram(proto, addr):
0049     s = socket.socket(proto, socket.SOCK_DGRAM)
0050     s.sendto(teststring, addr)
0051     buf = data = receive(s, 100)
0052     while data and '\n' not in buf:
0053         data = receive(s, 100)
0054         buf += data
0055     verify(buf == teststring)
0056     s.close()
0057 
0058 def teststream(proto, addr):
0059     s = socket.socket(proto, socket.SOCK_STREAM)
0060     s.connect(addr)
0061     s.sendall(teststring)
0062     buf = data = receive(s, 100)
0063     while data and '\n' not in buf:
0064         data = receive(s, 100)
0065         buf += data
0066     verify(buf == teststring)
0067     s.close()
0068 
0069 class ServerThread(threading.Thread):
0070     def __init__(self, addr, svrcls, hdlrcls):
0071         threading.Thread.__init__(self)
0072         self.__addr = addr
0073         self.__svrcls = svrcls
0074         self.__hdlrcls = hdlrcls
0075     def run(self):
0076         class svrcls(MyMixinServer, self.__svrcls):
0077             pass
0078         if verbose: print "thread: creating server"
0079         svr = svrcls(self.__addr, self.__hdlrcls)
0080         if verbose: print "thread: serving three times"
0081         svr.serve_a_few()
0082         if verbose: print "thread: done"
0083 
0084 seed = 0
0085 def pickport():
0086     global seed
0087     seed += 1
0088     return 10000 + (os.getpid() % 1000)*10 + seed
0089 
0090 host = "localhost"
0091 testfiles = []
0092 def pickaddr(proto):
0093     if proto == socket.AF_INET:
0094         return (host, pickport())
0095     else:
0096         fn = TESTFN + str(pickport())
0097         if os.name == 'os2':
0098             # AF_UNIX socket names on OS/2 require a specific prefix
0099             # which can't include a drive letter and must also use
0100             # backslashes as directory separators
0101             if fn[1] == ':':
0102                 fn = fn[2:]
0103             if fn[0] in (os.sep, os.altsep):
0104                 fn = fn[1:]
0105             fn = os.path.join('\socket', fn)
0106             if os.sep == '/':
0107                 fn = fn.replace(os.sep, os.altsep)
0108             else:
0109                 fn = fn.replace(os.altsep, os.sep)
0110         testfiles.append(fn)
0111         return fn
0112 
0113 def cleanup():
0114     for fn in testfiles:
0115         try:
0116             os.remove(fn)
0117         except os.error:
0118             pass
0119     testfiles[:] = []
0120 
0121 def testloop(proto, servers, hdlrcls, testfunc):
0122     for svrcls in servers:
0123         addr = pickaddr(proto)
0124         if verbose:
0125             print "ADDR =", addr
0126             print "CLASS =", svrcls
0127         t = ServerThread(addr, svrcls, hdlrcls)
0128         if verbose: print "server created"
0129         t.start()
0130         if verbose: print "server running"
0131         for i in range(NREQ):
0132             time.sleep(DELAY)
0133             if verbose: print "test client", i
0134             testfunc(proto, addr)
0135         if verbose: print "waiting for server"
0136         t.join()
0137         if verbose: print "done"
0138 
0139 tcpservers = [TCPServer, ThreadingTCPServer]
0140 if hasattr(os, 'fork') and os.name not in ('os2',):
0141     tcpservers.append(ForkingTCPServer)
0142 udpservers = [UDPServer, ThreadingUDPServer]
0143 if hasattr(os, 'fork') and os.name not in ('os2',):
0144     udpservers.append(ForkingUDPServer)
0145 
0146 if not hasattr(socket, 'AF_UNIX'):
0147     streamservers = []
0148     dgramservers = []
0149 else:
0150     class ForkingUnixStreamServer(ForkingMixIn, UnixStreamServer): pass
0151     streamservers = [UnixStreamServer, ThreadingUnixStreamServer]
0152     if hasattr(os, 'fork') and os.name not in ('os2',):
0153         streamservers.append(ForkingUnixStreamServer)
0154     class ForkingUnixDatagramServer(ForkingMixIn, UnixDatagramServer): pass
0155     dgramservers = [UnixDatagramServer, ThreadingUnixDatagramServer]
0156     if hasattr(os, 'fork') and os.name not in ('os2',):
0157         dgramservers.append(ForkingUnixDatagramServer)
0158 
0159 def testall():
0160     testloop(socket.AF_INET, tcpservers, MyStreamHandler, teststream)
0161     testloop(socket.AF_INET, udpservers, MyDatagramHandler, testdgram)
0162     if hasattr(socket, 'AF_UNIX'):
0163         testloop(socket.AF_UNIX, streamservers, MyStreamHandler, teststream)
0164         # Alas, on Linux (at least) recvfrom() doesn't return a meaningful
0165         # client address so this cannot work:
0166         ##testloop(socket.AF_UNIX, dgramservers, MyDatagramHandler, testdgram)
0167 
0168 def test_main():
0169     import imp
0170     if imp.lock_held():
0171         # If the import lock is held, the threads will hang.
0172         raise TestSkipped("can't run when import lock is held")
0173 
0174     try:
0175         testall()
0176     finally:
0177         cleanup()
0178 
0179 if __name__ == "__main__":
0180     test_main()
0181 

Generated by PyXR 0.9.4
SourceForge.net Logo