

@reap_threads
def test_shutdown(self):

    class MyServer(socketserver.TCPServer):
        pass

    class MyHandler(socketserver.StreamRequestHandler):
        pass
    threads = []
    for i in range(20):
        s = MyServer((HOST, 0), MyHandler)
        t = threading.Thread(name='MyServer serving', target=s.serve_forever, kwargs={
            'poll_interval': 0.01,
        })
        t.daemon = True
        threads.append((t, s))
    for (t, s) in threads:
        t.start()
        s.shutdown()
    for (t, s) in threads:
        t.join()
        s.server_close()

dataset/ETHPy150Open amrdraz/kodr/app/brython/www/src/Lib/test/test_socketserver.py SocketServerTest.test_shutdown/original