Code Duplication    Length = 32-33 lines in 2 locations

tests/milvus_python_test/entity/test_search.py 2 locations

@@ 328-360 (lines=33) @@
325
            assert res[1]._distances[0] > epsilon
326
327
    # TODO:
328
    @pytest.mark.level(2)
329
    def _test_search_index_partitions_B(self, connect, collection, get_simple_index, get_top_k):
330
        '''
331
        target: test basic search fuction, all the search params is corrent, test all index params, and build
332
        method: search collection with the given vectors and tags, check the result
333
        expected: the length of the result is top_k
334
        '''
335
        top_k = get_top_k
336
        nq = 2
337
        tag = "tag"
338
        new_tag = "new_tag"
339
        index_type = get_simple_index["index_type"]
340
        if index_type == "IVF_PQ":
341
            pytest.skip("Skip PQ")
342
        connect.create_partition(collection, tag)
343
        connect.create_partition(collection, new_tag)
344
        entities, ids = init_data(connect, collection, partition_tags=tag)
345
        new_entities, new_ids = init_data(connect, collection, nb=6001, partition_tags=new_tag)
346
        connect.create_index(collection, field_name, get_simple_index)
347
        search_param = get_search_param(index_type)
348
        query, vecs = gen_query_vectors(field_name, new_entities, top_k, nq, search_params=search_param)
349
        if top_k > top_k_limit:
350
            with pytest.raises(Exception) as e:
351
                res = connect.search(collection, query)
352
        else:
353
            res = connect.search(collection, query, partition_tags=["(.*)tag"])
354
            assert not check_id_result(res[0], ids[0])
355
            assert check_id_result(res[1], new_ids[0])
356
            assert res[0]._distances[0] > epsilon
357
            assert res[1]._distances[0] < epsilon
358
            res = connect.search(collection, query, partition_tags=["new(.*)"])
359
            assert res[0]._distances[0] > epsilon
360
            assert res[1]._distances[0] < epsilon
361
362
    # 
363
    # test for ip metric
@@ 294-325 (lines=32) @@
291
            assert len(res) == nq
292
            assert len(res[0]) == 0
293
294
    @pytest.mark.level(2)
295
    def test_search_index_partitions(self, connect, collection, get_simple_index, get_top_k):
296
        '''
297
        target: test basic search fuction, all the search params is corrent, test all index params, and build
298
        method: search collection with the given vectors and tags, check the result
299
        expected: the length of the result is top_k
300
        '''
301
        top_k = get_top_k
302
        nq = 2
303
        new_tag = "new_tag"
304
        index_type = get_simple_index["index_type"]
305
        if index_type == "IVF_PQ":
306
            pytest.skip("Skip PQ")
307
        connect.create_partition(collection, tag)
308
        connect.create_partition(collection, new_tag)
309
        entities, ids = init_data(connect, collection, partition_tags=tag)
310
        new_entities, new_ids = init_data(connect, collection, nb=6001, partition_tags=new_tag)
311
        connect.create_index(collection, field_name, get_simple_index)
312
        search_param = get_search_param(index_type)
313
        query, vecs = gen_query_vectors(field_name, entities, top_k, nq, search_params=search_param)
314
        if top_k > top_k_limit:
315
            with pytest.raises(Exception) as e:
316
                res = connect.search(collection, query)
317
        else:
318
            res = connect.search(collection, query)
319
            assert check_id_result(res[0], ids[0])
320
            assert not check_id_result(res[1], new_ids[0])
321
            assert res[0]._distances[0] < epsilon
322
            assert res[1]._distances[0] < epsilon
323
            res = connect.search(collection, query, partition_tags=["new_tag"])
324
            assert res[0]._distances[0] > epsilon
325
            assert res[1]._distances[0] > epsilon
326
327
    # TODO:
328
    @pytest.mark.level(2)