test_asyncore.py 文件源码

def test_dispatcher(self):
        fd = os.open(support.TESTFN, os.O_RDONLY)
        data = []
        class FileDispatcher(asyncore.file_dispatcher):
            def handle_read(self):
                data.append(self.recv(29))
        s = FileDispatcher(fd)
        os.close(fd)
        asyncore.loop(timeout=0.01, use_poll=True, count=2)
        self.assertEqual(b"".join(data), self.d)