test_get_collection_info.TestInfoBase.test_info_collection_fields() - Code Metrics - Inspection of "get collection stats (#3069)" - milvus-io/milvus - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 52a593...841a75 )

unknown

created 2020-07-30 08:55 UTC

TestInfoBase.test_info_collection_fields() A

↳ Parent: test_get_collection_info

Complexity

Conditions

Size

Total Lines	15
Code Lines	9

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	9
nop	4
dl	0
loc	15
rs	9.95
c	0
b	0
f	0

import pdb
import pytest
import logging
import itertools
from time import sleep
import threading
from multiprocessing import Process
from utils import *

collection_id = "info"
default_fields = gen_default_fields() 
segment_row_count = 5000


class TestInfoBase:

    @pytest.fixture(
        scope="function",
        params=gen_single_filter_fields()
    )
    def get_filter_field(self, request):
        yield request.param

    @pytest.fixture(
        scope="function",
        params=gen_single_vector_fields()
    )
    def get_vector_field(self, request):
        yield request.param

    @pytest.fixture(
        scope="function",
        params=gen_segment_row_counts()
    )
    def get_segment_row_count(self, request):
        yield request.param

    """
    ******************************************************************
      The following cases are used to test `get_collection_info` function, no data in collection
    ******************************************************************
    """
  
    # TODO
    def test_info_collection_fields(self, connect, get_filter_field, get_vector_field):
        '''
        target: test create normal collection with different fields, check info returned
        method: create collection with diff fields: metric/field_type/..., calling `get_collection_info`
        expected: no exception raised, and value returned correct
        '''
        filter_field = get_filter_field
        vector_field = get_vector_field
        collection_name = gen_unique_str(collection_id)
        fields = {
                "fields": [filter_field, vector_field],
                "segment_row_count": segment_row_count
        }
        connect.create_collection(collection_name, fields)
        res = connect.get_collection_info(collection_name)
        # assert field_name
        # assert field_type
        # assert vector field params
        # assert metric type
        # assert dimension

    # TODO
    def test_create_collection_segment_row_count(self, connect, get_segment_row_count):
        '''
        target: test create normal collection with different fields
        method: create collection with diff segment_row_count
        expected: no exception raised
        '''
        collection_name = gen_unique_str(collection_id)
        fields = copy.deepcopy(default_fields)
        fields["segment_row_count"] = get_segment_row_count
        connect.create_collection(collection_name, fields)
        # assert segment size


    @pytest.mark.level(2)
    def test_get_collection_info_without_connection(self, collection, dis_connect):
        '''
        target: test get collection info, without connection
        method: calling get collection info with correct params, with a disconnected instance
        expected: get collection info raise exception
        '''
        with pytest.raises(Exception) as e:
            assert connect.get_collection_info(dis_connect, collection)


    def test_get_collection_info_not_existed(self, connect):
        '''
        target: test if collection not created
        method: random a collection name, which not existed in db, 
            assert the value returned by get_collection_info method
        expected: False
        '''
        collection_name = gen_unique_str(collection_id)
        with pytest.raises(Exception) as e:
            res = connect.get_collection_info(connect, collection_name)

    # TODO
    @pytest.mark.level(2)
    def test_get_collection_info_multithread(self, connect):
        '''
        target: test create collection with multithread
        method: create collection using multithread, 
        expected: collections are created
        '''
        threads_num = 4 
        threads = []
        collection_name = gen_unique_str(collection_id)
        connect.create_collection(collection_name, default_fields)

        def get_info():
            res = connect.get_collection_info(connect, collection_name)
            # assert

        for i in range(threads_num):
            t = threading.Thread(target=get_info, args=())
            threads.append(t)
            t.start()
            time.sleep(0.2)
        for t in threads:
            t.join()

    """
    ******************************************************************
      The following cases are used to test `get_collection_info` function, and insert data in collection
    ******************************************************************
    """

    # TODO
    def test_info_collection_fields_after_insert(self, connect, get_filter_field, get_vector_field):
        '''
        target: test create normal collection with different fields, check info returned
        method: create collection with diff fields: metric/field_type/..., calling `get_collection_info`
        expected: no exception raised, and value returned correct
        '''
        filter_field = get_filter_field
        vector_field = get_vector_field
        collection_name = gen_unique_str(collection_id)
        fields = {
                "fields": [filter_field, vector_field],
                "segment_row_count": segment_row_count
        }
        connect.create_collection(collection_name, fields)
        # insert
        res = connect.get_collection_info(collection_name)
        # assert field_name
        # assert field_type
        # assert vector field params
        # assert metric type
        # assert dimension

    # TODO
    def test_create_collection_segment_row_count_after_insert(self, connect, get_segment_row_count):
        '''
        target: test create normal collection with different fields
        method: create collection with diff segment_row_count
        expected: no exception raised
        '''
        collection_name = gen_unique_str(collection_id)
        fields = copy.deepcopy(default_fields)
        fields["segment_row_count"] = get_segment_row_count
        connect.create_collection(collection_name, fields)
        # insert
        # assert segment size


class TestInfoInvalid(object):
    """
    Test get collection info with invalid params
    """
    @pytest.fixture(
        scope="function",
        params=gen_invalid_strs()
    )
    def get_collection_name(self, request):
        yield request.param


    @pytest.mark.level(2)
    def test_get_collection_info_with_invalid_collectionname(self, connect, get_collection_name):
        collection_name = get_collection_name
        with pytest.raises(Exception) as e:
            connect.get_collection_info(collection_name)

    @pytest.mark.level(2)
    def test_get_collection_info_with_empty_collectionname(self, connect):
        collection_name = ''
        with pytest.raises(Exception) as e:
            connect.get_collection_info(collection_name)

    @pytest.mark.level(2)
    def test_get_collection_info_with_none_collectionname(self, connect):
        collection_name = None
        with pytest.raises(Exception) as e:
            connect.get_collection_info(collection_name)

    def test_get_collection_info_None(self, connect):
        '''
        target: test create collection but the collection name is None
        method: create collection, param collection_name is None
        expected: create raise error
        '''
        with pytest.raises(Exception) as e:
            connect.get_collection_info(None)


1			import pdb
2			import pytest
3			import logging
4			import itertools
5			from time import sleep
6			import threading
7			from multiprocessing import Process
8			from utils import *
9
10			collection_id = "info"
11			default_fields = gen_default_fields()
12			segment_row_count = 5000
13
14
15			class TestInfoBase:
16
17			@pytest.fixture(
18			scope="function",
19			params=gen_single_filter_fields()
20			)
21			def get_filter_field(self, request):
22			yield request.param
23
24			@pytest.fixture(
25			scope="function",
26			params=gen_single_vector_fields()
27			)
28			def get_vector_field(self, request):
29			yield request.param
30
31			@pytest.fixture(
32			scope="function",
33			params=gen_segment_row_counts()
34			)
35			def get_segment_row_count(self, request):
36			yield request.param
37
38			"""
39			******************************************************************
40			The following cases are used to test `get_collection_info` function, no data in collection
41			******************************************************************
42			"""
43
44			# TODO
45			def test_info_collection_fields(self, connect, get_filter_field, get_vector_field):
46			'''
47			target: test create normal collection with different fields, check info returned
48			method: create collection with diff fields: metric/field_type/..., calling `get_collection_info`
49			expected: no exception raised, and value returned correct
50			'''
51			filter_field = get_filter_field
52			vector_field = get_vector_field
53			collection_name = gen_unique_str(collection_id)
54			fields = {
55			"fields": [filter_field, vector_field],
56			"segment_row_count": segment_row_count
57			}
58			connect.create_collection(collection_name, fields)
59			res = connect.get_collection_info(collection_name)
60			# assert field_name
61			# assert field_type
62			# assert vector field params
63			# assert metric type
64			# assert dimension
65
66			# TODO
67			def test_create_collection_segment_row_count(self, connect, get_segment_row_count):
68			'''
69			target: test create normal collection with different fields
70			method: create collection with diff segment_row_count
71			expected: no exception raised
72			'''
73			collection_name = gen_unique_str(collection_id)
74			fields = copy.deepcopy(default_fields)
75			fields["segment_row_count"] = get_segment_row_count
76			connect.create_collection(collection_name, fields)
77			# assert segment size
78
79
80			@pytest.mark.level(2)
81			def test_get_collection_info_without_connection(self, collection, dis_connect):
82			'''
83			target: test get collection info, without connection
84			method: calling get collection info with correct params, with a disconnected instance
85			expected: get collection info raise exception
86			'''
87			with pytest.raises(Exception) as e:
88			assert connect.get_collection_info(dis_connect, collection)
			0 ignored issues – show Comprehensibility Best Practice introduced 2020-07-23 02:40 UTC by Report Bug Copy Issue Report The variable `connect` does not seem to be defined. Loading history...
89
90			def test_get_collection_info_not_existed(self, connect):
91			'''
92			target: test if collection not created
93			method: random a collection name, which not existed in db,
94			assert the value returned by get_collection_info method
95			expected: False
96			'''
97			collection_name = gen_unique_str(collection_id)
98			with pytest.raises(Exception) as e:
99			res = connect.get_collection_info(connect, collection_name)
100
101			# TODO
102			@pytest.mark.level(2)
103			def test_get_collection_info_multithread(self, connect):
104			'''
105			target: test create collection with multithread
106			method: create collection using multithread,
107			expected: collections are created
108			'''
109			threads_num = 4
110			threads = []
111			collection_name = gen_unique_str(collection_id)
112			connect.create_collection(collection_name, default_fields)
113
114			def get_info():
115			res = connect.get_collection_info(connect, collection_name)
116			# assert
117
118			for i in range(threads_num):
119			t = threading.Thread(target=get_info, args=())
120			threads.append(t)
121			t.start()
122			time.sleep(0.2)
123			for t in threads:
124			t.join()
125
126			"""
127			******************************************************************
128			The following cases are used to test `get_collection_info` function, and insert data in collection
129			******************************************************************
130			"""
131
132			# TODO
133			def test_info_collection_fields_after_insert(self, connect, get_filter_field, get_vector_field):
134			'''
135			target: test create normal collection with different fields, check info returned
136			method: create collection with diff fields: metric/field_type/..., calling `get_collection_info`
137			expected: no exception raised, and value returned correct
138			'''
139			filter_field = get_filter_field
140			vector_field = get_vector_field
141			collection_name = gen_unique_str(collection_id)
142			fields = {
143			"fields": [filter_field, vector_field],
144			"segment_row_count": segment_row_count
145			}
146			connect.create_collection(collection_name, fields)
147			# insert
148			res = connect.get_collection_info(collection_name)
149			# assert field_name
150			# assert field_type
151			# assert vector field params
152			# assert metric type
153			# assert dimension
154
155			# TODO
156			def test_create_collection_segment_row_count_after_insert(self, connect, get_segment_row_count):
157			'''
158			target: test create normal collection with different fields
159			method: create collection with diff segment_row_count
160			expected: no exception raised
161			'''
162			collection_name = gen_unique_str(collection_id)
163			fields = copy.deepcopy(default_fields)
164			fields["segment_row_count"] = get_segment_row_count
165			connect.create_collection(collection_name, fields)
166			# insert
167			# assert segment size
168
169
170			class TestInfoInvalid(object):
171			"""
172			Test get collection info with invalid params
173			"""
174			@pytest.fixture(
175			scope="function",
176			params=gen_invalid_strs()
177			)
178			def get_collection_name(self, request):
179			yield request.param
180
181
182			@pytest.mark.level(2)
183			def test_get_collection_info_with_invalid_collectionname(self, connect, get_collection_name):
184			collection_name = get_collection_name
185			with pytest.raises(Exception) as e:
186			connect.get_collection_info(collection_name)
187
188			@pytest.mark.level(2)
189			def test_get_collection_info_with_empty_collectionname(self, connect):
190			collection_name = ''
191			with pytest.raises(Exception) as e:
192			connect.get_collection_info(collection_name)
193
194			@pytest.mark.level(2)
195			def test_get_collection_info_with_none_collectionname(self, connect):
196			collection_name = None
197			with pytest.raises(Exception) as e:
198			connect.get_collection_info(collection_name)
199
200			def test_get_collection_info_None(self, connect):
201			'''
202			target: test create collection but the collection name is None
203			method: create collection, param collection_name is None
204			expected: create raise error
205			'''
206			with pytest.raises(Exception) as e:
207			connect.get_collection_info(None)
208

milvus-io / milvus

Push — master ( 52a593...841a75 )

TestInfoBase.test_info_collection_fields() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like