milvus-io / milvus

        #     with pytest.raises(Exception) as e:

        #         status, result = connect.search(collection, top_k, query_vecs, params=search_param)

    @pytest.mark.level(2)

    def test_search_with_invalid_nprobe_ip(self, connect, ip_collection, get_nprobes):

        '''

        target: test search fuction, with the wrong top_k

        method: search with top_k

        expected: raise an error, and the connection is normal

        '''

        index_type = IndexType.IVF_SQ8

        index_param = {"nlist": 16384}

        connect.create_index(ip_collection, index_type, index_param)

        nprobe = get_nprobes

        search_param = {"nprobe": nprobe}

        logging.getLogger().info(nprobe)

        query_vecs = gen_vectors(1, dim)

        # if isinstance(nprobe, int):

        status, result = connect.search(ip_collection, top_k, query_vecs, params=search_param)

        assert not status.OK()

        # else:

        #     with pytest.raises(Exception) as e:

        #         status, result = connect.search(ip_collection, top_k, query_vecs, params=search_param)

    def get_nprobes(self, request):

        yield request.param

    @pytest.mark.level(1)

    def test_search_with_invalid_nprobe(self, connect, collection, get_nprobes):

        '''

        target: test search fuction, with the wrong nprobe

        method: search with nprobe

        expected: raise an error, and the connection is normal

        '''

        index_type = IndexType.IVF_SQ8

        index_param = {"nlist": 16384}

        connect.create_index(collection, index_type, index_param)

        nprobe = get_nprobes

        search_param = {"nprobe": nprobe}

        logging.getLogger().info(nprobe)

        query_vecs = gen_vectors(1, dim)

        # if isinstance(nprobe, int):

        status, result = connect.search(collection, top_k, query_vecs, params=search_param)

        assert not status.OK()

        # else:

        #     with pytest.raises(Exception) as e:

        #         status, result = connect.search(collection, top_k, query_vecs, params=search_param)