testcase.assertEqual(72 << 8, status)

def close_server(server):

server.server_close()

if hasattr(server, 'active_children'):

# ForkingMixIn: Manually reap all child processes, since server_close()

# calls waitpid() in non-blocking mode using the WNOHANG flag.

for pid in server.active_children.copy():

try:

os.waitpid(pid, 0)

except ChildProcessError:

pass

server.active_children.clear()

@unittest.skipUnless(threading, 'Threading required for this test.')

class SocketServerTest(unittest.TestCase):

"""Test all socket servers."""

class MyServer(svrcls):

def handle_error(self, request, client_address):

self.close_request(request)

self.server_close()

close_server(self)

raise

class MyHandler(hdlrbase):

if verbose: print "waiting for server"

server.shutdown()

t.join()

server.server_close()

close_server(server)

self.assertRaises(socket.error, server.socket.fileno)

if verbose: print "done"

s.shutdown()

for t, s in threads:

t.join()

close_server(s)

def test_tcpserver_bind_leak(self):

# Issue #22435: the server socket wouldn't be closed if bind()/listen()

s.close()

server.handle_request()

self.assertEqual(server.shutdown_called, 1)

server.server_close()

close_server(server)

def test_main():