test_asyncore.py 文件源码

def test_handle_accepted(self):
        # make sure handle_accepted() is called when a client connects

        class TestListener(BaseTestHandler):

            def __init__(self):
                BaseTestHandler.__init__(self)
                self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
                self.bind((HOST, 0))
                self.listen(5)
                self.address = self.socket.getsockname()[:2]

            def handle_accept(self):
                asyncore.dispatcher.handle_accept(self)

            def handle_accepted(self, sock, addr):
                sock.close()
                self.flag = True

        server = TestListener()
        client = BaseClient(server.address)
        self.loop_waiting_for_flag(server)