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