test_kazoo_utils.py 文件源码

def test_serverset_destroy(self):
        testutil.initialize_kazoo_client_manager(ZK_HOSTS)
        client = KazooClientManager().get_client()
        server_set = ServerSet(ServerSetTestCase.SERVER_SET_DESTROY_PATH,
                               ZK_HOSTS,
                               waiting_in_secs=0.01)
        server_set.join(ServerSetTestCase.PORT_1, use_ip=False)
        server_set.join(ServerSetTestCase.PORT_2, use_ip=False)
        # Give time to let server set join to do its magic.
        gevent.sleep(1)
        server_set._destroy(ServerSetTestCase.END_POINT_1)
        gevent.sleep(1)
        children = client.get_children(
            ServerSetTestCase.SERVER_SET_DESTROY_PATH)
        for child in children:
            self.assertFalse(child.endswith(ServerSetTestCase.END_POINT_1))