Skip to content

C
cpython
Commit 1176c6ed authored by Georg Brandl's avatar Georg Brandl
Browse files
Options

Rewrite test_socketserver as unittest, written for GHOP by Benjamin Petersen.

parent 48b9ef77
Hide whitespace changes
Inline Side-by-side
Showing with 177 additions and 146 deletions
Lib/test/test_socketserver.py
View file @ 1176c6ed
# Test suite for SocketServer.py """
Test suite for SocketServer.py.
"""
from test import test_support import os
from test.test_support import (verbose, verify, TESTFN, TestSkipped,
reap_children)
test_support.requires('network')
from SocketServer import *
import socket import socket
import errno import errno
import imp
import select import select
import time import time
import threading import threading
import os from functools import wraps
import unittest
import SocketServer
import test.test_support
from test.test_support import reap_children, verbose, TestSkipped
from test.test_support import TESTFN as TEST_FILE
test.test_support.requires("network")
NREQ = 3 NREQ = 3
DELAY = 0.5 DELAY = 0.5
TEST_STR = "hello world\n"
HOST = "localhost"
HAVE_UNIX_SOCKETS = hasattr(socket, "AF_UNIX")
HAVE_FORKING = hasattr(os, "fork") and os.name != "os2"
class MyMixinHandler: class MyMixinHandler:
def handle(self): def handle(self):
...@@ -23,50 +35,41 @@ class MyMixinHandler: ...@@ -23,50 +35,41 @@ class MyMixinHandler:
time.sleep(DELAY) time.sleep(DELAY)
self.wfile.write(line) self.wfile.write(line)
class MyStreamHandler(MyMixinHandler, StreamRequestHandler):
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 %r" % (sock,)
class MyStreamHandler(MyMixinHandler, SocketServer.StreamRequestHandler):
pass pass
class MyDatagramHandler(MyMixinHandler, DatagramRequestHandler): class MyDatagramHandler(MyMixinHandler,
SocketServer.DatagramRequestHandler):
pass pass
class ForkingUnixStreamServer(SocketServer.ForkingMixIn,
SocketServer.UnixStreamServer):
pass
class ForkingUnixDatagramServer(SocketServer.ForkingMixIn,
SocketServer.UnixDatagramServer):
pass
class MyMixinServer: class MyMixinServer:
def serve_a_few(self): def serve_a_few(self):
for i in range(NREQ): for i in range(NREQ):
self.handle_request() self.handle_request()
def handle_error(self, request, client_address): def handle_error(self, request, client_address):
self.close_request(request) self.close_request(request)
self.server_close() self.server_close()
raise 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 %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:
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.sendall(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): class ServerThread(threading.Thread):
def __init__(self, addr, svrcls, hdlrcls): def __init__(self, addr, svrcls, hdlrcls):
...@@ -75,6 +78,7 @@ class ServerThread(threading.Thread): ...@@ -75,6 +78,7 @@ class ServerThread(threading.Thread):
self.__svrcls = svrcls self.__svrcls = svrcls
self.__hdlrcls = hdlrcls self.__hdlrcls = hdlrcls
self.ready = threading.Event() self.ready = threading.Event()
def run(self): def run(self):
class svrcls(MyMixinServer, self.__svrcls): class svrcls(MyMixinServer, self.__svrcls):
pass pass
...@@ -93,64 +97,8 @@ class ServerThread(threading.Thread): ...@@ -93,64 +97,8 @@ class ServerThread(threading.Thread):
svr.serve_a_few() svr.serve_a_few()
if verbose: print "thread: done" if verbose: print "thread: done"
seed = 0
def pickport(): class ForgivingTCPServer(SocketServer.TCPServer):
global seed
seed += 1
return 10000 + (os.getpid() % 1000)*10 + seed
host = "localhost"
testfiles = []
def pickaddr(proto):
if proto == socket.AF_INET:
return (host, pickport())
else:
fn = TESTFN + str(pickport())
if os.name == 'os2':
# 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[1] == ':':
fn = fn[2:]
if fn[0] in (os.sep, os.altsep):
fn = fn[1:]
fn = os.path.join('\socket', fn)
if os.sep == '/':
fn = fn.replace(os.sep, os.altsep)
else:
fn = fn.replace(os.altsep, os.sep)
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):
t.ready.wait(10*DELAY)
if not t.ready.isSet():
raise RuntimeError("Server not ready within a reasonable time")
if verbose: print "test client", i
testfunc(proto, addr)
if verbose: print "waiting for server"
t.join()
if verbose: print "done"
class ForgivingTCPServer(TCPServer):
# prevent errors if another process is using the port we want # prevent errors if another process is using the port we want
def server_bind(self): def server_bind(self):
host, default_port = self.server_address host, default_port = self.server_address
...@@ -160,64 +108,147 @@ class ForgivingTCPServer(TCPServer): ...@@ -160,64 +108,147 @@ class ForgivingTCPServer(TCPServer):
for port in [default_port, 3434, 8798, 23833]: for port in [default_port, 3434, 8798, 23833]:
try: try:
self.server_address = host, port self.server_address = host, port
TCPServer.server_bind(self) SocketServer.TCPServer.server_bind(self)
break break
except socket.error, (err, msg): except socket.error, (err, msg):
if err != errno.EADDRINUSE: if err != errno.EADDRINUSE:
raise raise
print >>sys.__stderr__, \ print >> sys.__stderr__, \
' WARNING: failed to listen on port %d, trying another' % port "WARNING: failed to listen on port %d, trying another: " % port
tcpservers = [ForgivingTCPServer, ThreadingTCPServer]
if hasattr(os, 'fork') and os.name not in ('os2',): class SocketServerTest(unittest.TestCase):
tcpservers.append(ForkingTCPServer) """Test all socket servers."""
udpservers = [UDPServer, ThreadingUDPServer]
if hasattr(os, 'fork') and os.name not in ('os2',): def setUp(self):
udpservers.append(ForkingUDPServer) self.port_seed = 0
self.test_files = []
if not hasattr(socket, 'AF_UNIX'):
streamservers = [] def tearDown(self):
dgramservers = [] reap_children()
else:
class ForkingUnixStreamServer(ForkingMixIn, UnixStreamServer): pass for fn in self.test_files:
streamservers = [UnixStreamServer, ThreadingUnixStreamServer] try:
if hasattr(os, 'fork') and os.name not in ('os2',): os.remove(fn)
streamservers.append(ForkingUnixStreamServer) except os.error:
class ForkingUnixDatagramServer(ForkingMixIn, UnixDatagramServer): pass pass
dgramservers = [UnixDatagramServer, ThreadingUnixDatagramServer] self.test_files[:] = []
if hasattr(os, 'fork') and os.name not in ('os2',):
dgramservers.append(ForkingUnixDatagramServer) def pickport(self):
self.port_seed += 1
def sloppy_cleanup(): return 10000 + (os.getpid() % 1000)*10 + self.port_seed
# See http://python.org/sf/1540386
# We need to reap children here otherwise a child from one server def pickaddr(self, proto):
# can be left running for the next server and cause a test failure. if proto == socket.AF_INET:
time.sleep(DELAY) return (HOST, self.pickport())
reap_children() else:
fn = TEST_FILE + str(self.pickport())
def testall(): if os.name == 'os2':
testloop(socket.AF_INET, tcpservers, MyStreamHandler, teststream) # AF_UNIX socket names on OS/2 require a specific prefix
sloppy_cleanup() # which can't include a drive letter and must also use
testloop(socket.AF_INET, udpservers, MyDatagramHandler, testdgram) # backslashes as directory separators
if hasattr(socket, 'AF_UNIX'): if fn[1] == ':':
sloppy_cleanup() fn = fn[2:]
testloop(socket.AF_UNIX, streamservers, MyStreamHandler, teststream) if fn[0] in (os.sep, os.altsep):
# Alas, on Linux (at least) recvfrom() doesn't return a meaningful fn = fn[1:]
# client address so this cannot work: fn = os.path.join('\socket', fn)
##testloop(socket.AF_UNIX, dgramservers, MyDatagramHandler, testdgram) if os.sep == '/':
fn = fn.replace(os.sep, os.altsep)
else:
fn = fn.replace(os.altsep, os.sep)
self.test_files.append(fn)
return fn
def run_servers(self, proto, servers, hdlrcls, testfunc):
for svrcls in servers:
addr = self.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):
t.ready.wait(10*DELAY)
self.assert_(t.ready.isSet(),
"Server not ready within a reasonable time")
if verbose: print "test client", i
testfunc(proto, addr)
if verbose: print "waiting for server"
t.join()
if verbose: print "done"
def stream_examine(self, proto, addr):
s = socket.socket(proto, socket.SOCK_STREAM)
s.connect(addr)
s.sendall(TEST_STR)
buf = data = receive(s, 100)
while data and '\n' not in buf:
data = receive(s, 100)
buf += data
self.assertEquals(buf, TEST_STR)
s.close()
def dgram_examine(self, proto, addr):
s = socket.socket(proto, socket.SOCK_DGRAM)
s.sendto(TEST_STR, addr)
buf = data = receive(s, 100)
while data and '\n' not in buf:
data = receive(s, 100)
buf += data
self.assertEquals(buf, TEST_STR)
s.close()
def test_TCPServers(self):
# Test SocketServer.TCPServer
servers = [ForgivingTCPServer, SocketServer.ThreadingTCPServer]
if HAVE_FORKING:
servers.append(SocketServer.ForkingTCPServer)
self.run_servers(socket.AF_INET, servers,
MyStreamHandler, self.stream_examine)
def test_UDPServers(self):
# Test SocketServer.UPDServer
servers = [SocketServer.UDPServer,
SocketServer.ThreadingUDPServer]
if HAVE_FORKING:
servers.append(SocketServer.ForkingUDPServer)
self.run_servers(socket.AF_INET, servers, MyDatagramHandler,
self.dgram_examine)
def test_stream_servers(self):
# Test SocketServer's stream servers
if not HAVE_UNIX_SOCKETS:
return
servers = [SocketServer.UnixStreamServer,
SocketServer.ThreadingUnixStreamServer]
if HAVE_FORKING:
servers.append(ForkingUnixStreamServer)
self.run_servers(socket.AF_UNIX, servers, MyStreamHandler,
self.stream_examine)
# Alas, on Linux (at least) recvfrom() doesn't return a meaningful
# client address so this cannot work:
# def test_dgram_servers(self):
# # Test SocketServer.UnixDatagramServer
# if not HAVE_UNIX_SOCKETS:
# return
# servers = [SocketServer.UnixDatagramServer,
# SocketServer.ThreadingUnixDatagramServer]
# if HAVE_FORKING:
# servers.append(ForkingUnixDatagramServer)
# self.run_servers(socket.AF_UNIX, servers, MyDatagramHandler,
# self.dgram_examine)
def test_main(): def test_main():
import imp
if imp.lock_held(): if imp.lock_held():
# If the import lock is held, the threads will hang. # If the import lock is held, the threads will hang
raise TestSkipped("can't run when import lock is held") raise TestSkipped("can't run when import lock is held")
reap_children() test.test_support.run_unittest(SocketServerTest)
try:
testall()
finally:
cleanup()
reap_children()
if __name__ == "__main__": if __name__ == "__main__":
test_main() test_main()
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7