SearchEngineTest.test_query_KO() - Code Metrics - Inspection of "Merge pull request #70 from OnroerendErfgoed/69_re..." - OnroerendErfgoed/oe_utils - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( be44b6...3b3781 )

unknown

created 2018-05-07 10:21 UTC

SearchEngineTest.test_query_KO() A

↳ Parent: SearchEngineTest

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
dl	0
loc	9
rs	9.6666
c	0
b	0
f	0

# -*- coding: utf-8 -*-
import os
import unittest

from pyramid.httpexceptions import HTTPBadRequest
from requests import HTTPError
import responses
from webob.multidict import MultiDict

from oe_utils.data.data_transfer_objects import ResultDTO
from oe_utils.range_parser import Range
from oe_utils.search import parse_sort_string, SearchResultPager, parse_filter_params, validate_params, get_valid_params
from oe_utils.search.searchengine import SearchEngine
from tests import init_search_responses

try:
    from unittest.mock import Mock, patch, MagicMock
except ImportError:
    from mock import Mock, patch, MagicMock

test_base_url = 'http://localhost:9200'
index_name = 'dossiers'

here = os.path.dirname(__file__)

testdata = {
    "blabla": "boeboem",
    "zever": "gezever"
}


class SearchEngineTest(unittest.TestCase):
    def setUp(self):
        init_search_responses()
        self.searchengine = SearchEngine(test_base_url, index_name)
        self.system_token = 'test_token'
        self.user_acls = ["test_user", "test_role"]

    def tearDown(self):
        pass

    @responses.activate
    def test_add_to_index(self):
        object_type = 'dossier'
        object_id = 'test1'
        self.searchengine.add_to_index(self.system_token, object_type, object_id, testdata)
        self.assertEqual(responses.calls[0].request.url, 'http://localhost:9200/dossiers/dossier/test1')
        self.assertEqual(responses.calls[0].response.status_code, 201)

    @responses.activate
    def test_delete_by_query(self):
        searchengine = SearchEngine(test_base_url, index_name, '6')
        searchengine.remove_from_index_by_query(self.system_token, "id", 1)
        self.assertEqual(responses.calls[0].request.url, 'http://localhost:9200/dossiers/_delete_by_query')
        self.assertEqual(responses.calls[0].request.body, '{"query": {"match": {"id": 1}}}')
        self.assertEqual(responses.calls[0].response.status_code, 200)

    @responses.activate
    def test_add_to_index_KO(self):
        object_type = 'dossier'
        object_id = 'test2'
        self.assertRaises(HTTPError, self.searchengine.add_to_index, self.system_token, object_type, object_id,
                          testdata)

    @responses.activate
    def test_remove_from_index_KO(self):
        object_type = 'dossier'
        object_id = 'test2'
        self.assertRaises(HTTPError, self.searchengine.remove_from_index, self.system_token, object_type, object_id)

    @responses.activate
    def test_remove_from_index(self):
        object_type = 'dossier'
        object_id = 'test1'
        self.searchengine.remove_from_index(self.system_token, object_type, object_id)
        self.assertIn('OpenAmSSOID', responses.calls[0].request.headers)
        self.assertEqual(responses.calls[0].response.status_code, 200)

    @responses.activate
    def test_remove_from_index_no_system_token(self):
        object_type = 'dossier'
        object_id = 'test1'
        self.searchengine.remove_from_index(None, object_type, object_id)
        self.assertNotIn('OpenAmSSOID', responses.calls[0].request.headers)
        self.assertEqual(responses.calls[0].response.status_code, 200)

    @responses.activate
    def test_remove_index(self):
        self.searchengine.remove_index('test_token')
        self.assertEqual(responses.calls[0].response.status_code, 200)
        self.assertEqual(responses.calls[1].response.status_code, 200)

    @responses.activate
    def test_remove_index_KO(self):
        searchengine = SearchEngine(test_base_url, 'testing')
        searchengine.remove_index('test_token')
        self.assertEqual(responses.calls[0].response.status_code, 400)
        self.assertEqual(len(responses.calls), 1)

    @responses.activate
    def test_remove_index_KO_delete(self):
        searchengine = SearchEngine(test_base_url, 'test')
        self.assertRaises(HTTPError, searchengine.remove_index, 'test_token')
        self.assertEqual(len(responses.calls), 2)

    @responses.activate
    def test_create_index(self):
        self.searchengine.create_index('test_token', data='')
        self.assertEqual(responses.calls[0].response.status_code, 200)

    @responses.activate
    def test_create_index_KO(self):
        searchengine = SearchEngine(test_base_url, 'test')
        self.assertRaises(HTTPError, searchengine.create_index, 'test_token', data='')

    @responses.activate
    def test_add_type_mapping(self):
        object_type = 'dossier'
        self.searchengine.add_type_mapping(object_type, {}, 'test_token')
        self.assertEqual(responses.calls[0].response.status_code, 200)

    @responses.activate
    def test_add_type_mapping_KO(self):
        object_type = 'test'
        self.assertRaises(HTTPError, self.searchengine.add_type_mapping, object_type, {}, 'test_token')

    @responses.activate
    def test_query(self):
        query = MultiDict({
            'query': 'test'
        })
        result_dto = self.searchengine.query(self.system_token,
                                             query_params=query,
                                             user_acls=self.user_acls)
        self.assertTrue(isinstance(result_dto, ResultDTO))
        self.assertEqual(2, result_dto.total)
        self.assertIsNotNone(result_dto.data)

    @responses.activate
    def test_query_ranged(self):
        object_type = 'dossier'
        query = MultiDict({
            'query': 'test'
        })
        result_range = Range(start=0, end=4)
        result_dto = self.searchengine.query(self.system_token, object_type=object_type,
                                             query_params=query,
                                             user_acls=self.user_acls,
                                             result_range=result_range)
        self.assertTrue(isinstance(result_dto, ResultDTO))
        self.assertEqual(2, result_dto.total)
        self.assertIsNotNone(result_dto.data)
        self.assertIn('from=0', responses.calls[0].request.url)
        self.assertIn('size=5', responses.calls[0].request.url)

    @responses.activate
    def test_query_min_score(self):
        object_type = 'dossier'
        query = MultiDict({
            'query': 'test'
        })
        result_dto = self.searchengine.query(self.system_token, object_type=object_type,
                                             query_params=query,
                                             user_acls=self.user_acls,
                                             min_score=5)
        self.assertTrue(isinstance(result_dto, ResultDTO))
        self.assertEqual(2, result_dto.total)
        self.assertIsNotNone(result_dto.data)
        self.assertIn('"min_score": 5', responses.calls[0].request.body)

    @responses.activate
    def test_query_aggregations(self):
        aggregations = {
            'datum': {
                'date_histogram': {
                    'field': 'datum',
                    'interval': 'year',
                    'format': 'yyyy'
                }
            }
        }
        object_type = 'dossier'
        query = MultiDict({
            'query': 'test'
        })
        result_dto = self.searchengine.query(self.system_token, object_type=object_type,
                                             query_params=query,
                                             user_acls=self.user_acls,
                                             aggregations=aggregations)
        self.assertTrue(isinstance(result_dto, ResultDTO))
        self.assertEqual(2, result_dto.total)
        self.assertIsNotNone(result_dto.data)
        self.assertIn('aggregations', responses.calls[0].request.body)

    @responses.activate
    def test_query_no_hits(self):
        object_type = 'testing'
        query = MultiDict({
            'query': 'test'
        })
        result_dto = self.searchengine.query(self.system_token, object_type=object_type,
                                             query_params=query,
                                             user_acls=self.user_acls)
        self.assertTrue(isinstance(result_dto, ResultDTO))
        self.assertEqual(0, result_dto.total)
        self.assertIsNotNone(result_dto.data)

    @responses.activate
    def test_query_KO(self):
        object_type = 'test'
        query = MultiDict({
            'query': 'test'
        })
        self.assertRaises(HTTPError, self.searchengine.query, self.system_token, object_type=object_type,
                          query_params=query,
                          user_acls=self.user_acls)


class ParseSortStringTest(unittest.TestCase):
    def setUp(self):
        pass

    def tearDown(self):
        pass

    def test_parse_sort_string_desc(self):
        res = parse_sort_string('-id')
        self.assertIn({'id': {'unmapped_type': 'string', 'order': 'desc', 'missing': '_last'}}, res)

    def test_parse_sort_string_asc(self):
        res = parse_sort_string('id')
        self.assertIn({'id': {'unmapped_type': 'string', 'order': 'asc', 'missing': '_last'}}, res)

    def test_parse_sort_string_noe(self):
        res = parse_sort_string(None)
        self.assertEqual(['_score'], res)


class ParseFilterParamsTest(unittest.TestCase):
    def setUp(self):
        self.filterable = ['voornaam', 'type', 'organisatie']

    def tearDown(self):
        pass

    def test_parse_filter_params(self):
        query_params = MultiDict()
        query_params.add('type', 1)
        query_params.add('type', 2)
        query_params.add('test', 22)
        res = parse_filter_params(query_params, self.filterable)
        self.assertEqual({'type': [1, 2]}, res)

    def test_parse_filter_params_no_query_params(self):
        res = parse_filter_params(None, self.filterable)
        self.assertEqual({}, res)


class SearchResultPagerTests(unittest.TestCase):
    def setUp(self):
        self.pager = SearchResultPager(5, 100, 1000)
        self.pager_five = SearchResultPager(1, 10, 5)
        self.pager_eight = SearchResultPager(1, 1, 8)

    def tearDown(self):
        del self.pager
        del self.pager_five

    def test_init(self):
        self.assertEquals(5, self.pager.page)
        self.assertEquals(100, self.pager.per_page)
        self.assertEquals(1000, self.pager.total_count)

    def test_pages(self):
        self.assertEquals(10, self.pager.pages)
        self.assertEquals(1, self.pager_five.pages)

    def test_has_prev(self):
        self.assertTrue(self.pager.has_prev)
        self.assertFalse(self.pager_five.has_prev)

    def test_prev(self):
        self.assertEquals(4, self.pager.prev)
        self.assertEquals(None, self.pager_five.prev)

    def test_has_next(self):
        self.assertTrue(self.pager.has_next)
        self.assertFalse(self.pager_five.has_next)

    def test_next(self):
        self.assertEquals(6, self.pager.next)
        self.assertEquals(None, self.pager_five.next)

    def test_iter_all_pages(self):
        pages = list(self.pager.iter_pages())
        self.assertEquals(self.pager.pages, len(pages))
        self.assertEquals(1, pages[0])
        self.assertEquals(10, pages[9])

        pages = list(self.pager_five.iter_pages())
        self.assertEquals(self.pager_five.pages, len(pages))
        self.assertEquals(1, pages[0])
        self.assertListEqual([1, 2, 3, 4, 5, None, 7, 8], list(self.pager_eight.iter_pages()))


class ValidateParamsTests(unittest.TestCase):
    def setUp(self):
        self.params = MultiDict()
        self.params.add('voornaam', 'jos')
        self.params.add('gemeente_naam', 'merksem')

    def tearDown(self):
        pass

    def test_validate_params(self):
        valid_keys = ['query', 'sort', 'naam', 'voornaam', 'omschrijving', 'type', 'organisatie',
                      'persid', 'rrn', 'kbo', 'uid',
                      'telefoon', 'email', 'land', 'postcode', 'gemeente', 'gemeente_naam', 'straat', 'straat_naam',
                      'huisnummer', 'huisnummer_label', 'subadres', 'subadres_label', 'niscode', 'acdcode', 'status',
                      'erkenningsnr']
        res = validate_params(self.params, valid_keys)
        self.assertEqual(2, len(res))
        self.assertIn('voornaam', res)

    def test_validate_params_invalid_param(self):
        valid_keys = ['query', 'sort']
        self.assertRaises(HTTPBadRequest, validate_params, self.params, valid_keys)

    def test_get_valid_params_only_shows_those_that_are_valid(self):
        valid_params = get_valid_params(self.params, ['voornaam', 'naam'])

        self.assertEqual(1, len(valid_params))
        self.assertIn('voornaam', valid_params)




1			# -- coding: utf-8 --
2			import os
3			import unittest
4
5			from pyramid.httpexceptions import HTTPBadRequest
6			from requests import HTTPError
7			import responses
8			from webob.multidict import MultiDict
9
10			from oe_utils.data.data_transfer_objects import ResultDTO
11			from oe_utils.range_parser import Range
12			from oe_utils.search import parse_sort_string, SearchResultPager, parse_filter_params, validate_params, get_valid_params
13			from oe_utils.search.searchengine import SearchEngine
14			from tests import init_search_responses
15
16			try:
17			from unittest.mock import Mock, patch, MagicMock
18			except ImportError:
19			from mock import Mock, patch, MagicMock
20
21			test_base_url = 'http://localhost:9200'
22			index_name = 'dossiers'
23
24			here = os.path.dirname(__file__)
25
26			testdata = {
27			"blabla": "boeboem",
28			"zever": "gezever"
29			}
30
31
32			class SearchEngineTest(unittest.TestCase):
33			def setUp(self):
34			init_search_responses()
35			self.searchengine = SearchEngine(test_base_url, index_name)
36			self.system_token = 'test_token'
37			self.user_acls = ["test_user", "test_role"]
38
39			def tearDown(self):
40			pass
41
42			@responses.activate
43			def test_add_to_index(self):
44			object_type = 'dossier'
45			object_id = 'test1'
46			self.searchengine.add_to_index(self.system_token, object_type, object_id, testdata)
47			self.assertEqual(responses.calls[0].request.url, 'http://localhost:9200/dossiers/dossier/test1')
48			self.assertEqual(responses.calls[0].response.status_code, 201)
49
50			@responses.activate
51			def test_delete_by_query(self):
52			searchengine = SearchEngine(test_base_url, index_name, '6')
53			searchengine.remove_from_index_by_query(self.system_token, "id", 1)
54			self.assertEqual(responses.calls[0].request.url, 'http://localhost:9200/dossiers/_delete_by_query')
55			self.assertEqual(responses.calls[0].request.body, '{"query": {"match": {"id": 1}}}')
56			self.assertEqual(responses.calls[0].response.status_code, 200)
57
58			@responses.activate
59			def test_add_to_index_KO(self):
60			object_type = 'dossier'
61			object_id = 'test2'
62			self.assertRaises(HTTPError, self.searchengine.add_to_index, self.system_token, object_type, object_id,
63			testdata)
64
65			@responses.activate
66			def test_remove_from_index_KO(self):
67			object_type = 'dossier'
68			object_id = 'test2'
69			self.assertRaises(HTTPError, self.searchengine.remove_from_index, self.system_token, object_type, object_id)
70
71			@responses.activate
72			def test_remove_from_index(self):
73			object_type = 'dossier'
74			object_id = 'test1'
75			self.searchengine.remove_from_index(self.system_token, object_type, object_id)
76			self.assertIn('OpenAmSSOID', responses.calls[0].request.headers)
77			self.assertEqual(responses.calls[0].response.status_code, 200)
78
79			@responses.activate
80			def test_remove_from_index_no_system_token(self):
81			object_type = 'dossier'
82			object_id = 'test1'
83			self.searchengine.remove_from_index(None, object_type, object_id)
84			self.assertNotIn('OpenAmSSOID', responses.calls[0].request.headers)
85			self.assertEqual(responses.calls[0].response.status_code, 200)
86
87			@responses.activate
88			def test_remove_index(self):
89			self.searchengine.remove_index('test_token')
90			self.assertEqual(responses.calls[0].response.status_code, 200)
91			self.assertEqual(responses.calls[1].response.status_code, 200)
92
93			@responses.activate
94			def test_remove_index_KO(self):
95			searchengine = SearchEngine(test_base_url, 'testing')
96			searchengine.remove_index('test_token')
97			self.assertEqual(responses.calls[0].response.status_code, 400)
98			self.assertEqual(len(responses.calls), 1)
99
100			@responses.activate
101			def test_remove_index_KO_delete(self):
102			searchengine = SearchEngine(test_base_url, 'test')
103			self.assertRaises(HTTPError, searchengine.remove_index, 'test_token')
104			self.assertEqual(len(responses.calls), 2)
105
106			@responses.activate
107			def test_create_index(self):
108			self.searchengine.create_index('test_token', data='')
109			self.assertEqual(responses.calls[0].response.status_code, 200)
110
111			@responses.activate
112			def test_create_index_KO(self):
113			searchengine = SearchEngine(test_base_url, 'test')
114			self.assertRaises(HTTPError, searchengine.create_index, 'test_token', data='')
115
116			@responses.activate
117			def test_add_type_mapping(self):
118			object_type = 'dossier'
119			self.searchengine.add_type_mapping(object_type, {}, 'test_token')
120			self.assertEqual(responses.calls[0].response.status_code, 200)
121
122			@responses.activate
123			def test_add_type_mapping_KO(self):
124			object_type = 'test'
125			self.assertRaises(HTTPError, self.searchengine.add_type_mapping, object_type, {}, 'test_token')
126
127			@responses.activate
128			def test_query(self):
129			query = MultiDict({
130			'query': 'test'
131			})
132			result_dto = self.searchengine.query(self.system_token,
133			query_params=query,
134			user_acls=self.user_acls)
135			self.assertTrue(isinstance(result_dto, ResultDTO))
136			self.assertEqual(2, result_dto.total)
137			self.assertIsNotNone(result_dto.data)
138
139			@responses.activate
140			def test_query_ranged(self):
141			object_type = 'dossier'
142			query = MultiDict({
143			'query': 'test'
144			})
145			result_range = Range(start=0, end=4)
146			result_dto = self.searchengine.query(self.system_token, object_type=object_type,
147			query_params=query,
148			user_acls=self.user_acls,
149			result_range=result_range)
150			self.assertTrue(isinstance(result_dto, ResultDTO))
151			self.assertEqual(2, result_dto.total)
152			self.assertIsNotNone(result_dto.data)
153			self.assertIn('from=0', responses.calls[0].request.url)
154			self.assertIn('size=5', responses.calls[0].request.url)
155
156			@responses.activate
157			def test_query_min_score(self):
158			object_type = 'dossier'
159			query = MultiDict({
160			'query': 'test'
161			})
162			result_dto = self.searchengine.query(self.system_token, object_type=object_type,
163			query_params=query,
164			user_acls=self.user_acls,
165			min_score=5)
166			self.assertTrue(isinstance(result_dto, ResultDTO))
167			self.assertEqual(2, result_dto.total)
168			self.assertIsNotNone(result_dto.data)
169			self.assertIn('"min_score": 5', responses.calls[0].request.body)
170
171			@responses.activate
172			def test_query_aggregations(self):
173			aggregations = {
174			'datum': {
175			'date_histogram': {
176			'field': 'datum',
177			'interval': 'year',
178			'format': 'yyyy'
179			}
180			}
181			}
182			object_type = 'dossier'
183			query = MultiDict({
184			'query': 'test'
185			})
186			result_dto = self.searchengine.query(self.system_token, object_type=object_type,
187			query_params=query,
188			user_acls=self.user_acls,
189			aggregations=aggregations)
190			self.assertTrue(isinstance(result_dto, ResultDTO))
191			self.assertEqual(2, result_dto.total)
192			self.assertIsNotNone(result_dto.data)
193			self.assertIn('aggregations', responses.calls[0].request.body)
194
195			@responses.activate
196			def test_query_no_hits(self):
197			object_type = 'testing'
198			query = MultiDict({
199			'query': 'test'
200			})
201			result_dto = self.searchengine.query(self.system_token, object_type=object_type,
202			query_params=query,
203			user_acls=self.user_acls)
204			self.assertTrue(isinstance(result_dto, ResultDTO))
205			self.assertEqual(0, result_dto.total)
206			self.assertIsNotNone(result_dto.data)
207
208			@responses.activate
209			def test_query_KO(self):
210			object_type = 'test'
211			query = MultiDict({
212			'query': 'test'
213			})
214			self.assertRaises(HTTPError, self.searchengine.query, self.system_token, object_type=object_type,
215			query_params=query,
216			user_acls=self.user_acls)
217
218
219			class ParseSortStringTest(unittest.TestCase):
220			def setUp(self):
221			pass
222
223			def tearDown(self):
224			pass
225
226			def test_parse_sort_string_desc(self):
227			res = parse_sort_string('-id')
228			self.assertIn({'id': {'unmapped_type': 'string', 'order': 'desc', 'missing': '_last'}}, res)
229
230			def test_parse_sort_string_asc(self):
231			res = parse_sort_string('id')
232			self.assertIn({'id': {'unmapped_type': 'string', 'order': 'asc', 'missing': '_last'}}, res)
233
234			def test_parse_sort_string_noe(self):
235			res = parse_sort_string(None)
236			self.assertEqual(['_score'], res)
237
238
239			class ParseFilterParamsTest(unittest.TestCase):
240			def setUp(self):
241			self.filterable = ['voornaam', 'type', 'organisatie']
242
243			def tearDown(self):
244			pass
245
246			def test_parse_filter_params(self):
247			query_params = MultiDict()
248			query_params.add('type', 1)
249			query_params.add('type', 2)
250			query_params.add('test', 22)
251			res = parse_filter_params(query_params, self.filterable)
252			self.assertEqual({'type': [1, 2]}, res)
253
254			def test_parse_filter_params_no_query_params(self):
255			res = parse_filter_params(None, self.filterable)
256			self.assertEqual({}, res)
257
258
259			class SearchResultPagerTests(unittest.TestCase):
260			def setUp(self):
261			self.pager = SearchResultPager(5, 100, 1000)
262			self.pager_five = SearchResultPager(1, 10, 5)
263			self.pager_eight = SearchResultPager(1, 1, 8)
264
265			def tearDown(self):
266			del self.pager
267			del self.pager_five
268
269			def test_init(self):
270			self.assertEquals(5, self.pager.page)
271			self.assertEquals(100, self.pager.per_page)
272			self.assertEquals(1000, self.pager.total_count)
273
274			def test_pages(self):
275			self.assertEquals(10, self.pager.pages)
276			self.assertEquals(1, self.pager_five.pages)
277
278			def test_has_prev(self):
279			self.assertTrue(self.pager.has_prev)
280			self.assertFalse(self.pager_five.has_prev)
281
282			def test_prev(self):
283			self.assertEquals(4, self.pager.prev)
284			self.assertEquals(None, self.pager_five.prev)
285
286			def test_has_next(self):
287			self.assertTrue(self.pager.has_next)
288			self.assertFalse(self.pager_five.has_next)
289
290			def test_next(self):
291			self.assertEquals(6, self.pager.next)
292			self.assertEquals(None, self.pager_five.next)
293
294			def test_iter_all_pages(self):
295			pages = list(self.pager.iter_pages())
296			self.assertEquals(self.pager.pages, len(pages))
297			self.assertEquals(1, pages[0])
298			self.assertEquals(10, pages[9])
299
300			pages = list(self.pager_five.iter_pages())
301			self.assertEquals(self.pager_five.pages, len(pages))
302			self.assertEquals(1, pages[0])
303			self.assertListEqual([1, 2, 3, 4, 5, None, 7, 8], list(self.pager_eight.iter_pages()))
304
305
306			class ValidateParamsTests(unittest.TestCase):
307			def setUp(self):
308			self.params = MultiDict()
309			self.params.add('voornaam', 'jos')
310			self.params.add('gemeente_naam', 'merksem')
311
312			def tearDown(self):
313			pass
314
315			def test_validate_params(self):
316			valid_keys = ['query', 'sort', 'naam', 'voornaam', 'omschrijving', 'type', 'organisatie',
317			'persid', 'rrn', 'kbo', 'uid',
318			'telefoon', 'email', 'land', 'postcode', 'gemeente', 'gemeente_naam', 'straat', 'straat_naam',
319			'huisnummer', 'huisnummer_label', 'subadres', 'subadres_label', 'niscode', 'acdcode', 'status',
320			'erkenningsnr']
321			res = validate_params(self.params, valid_keys)
322			self.assertEqual(2, len(res))
323			self.assertIn('voornaam', res)
324
325			def test_validate_params_invalid_param(self):
326			valid_keys = ['query', 'sort']
327			self.assertRaises(HTTPBadRequest, validate_params, self.params, valid_keys)
328
329			def test_get_valid_params_only_shows_those_that_are_valid(self):
330			valid_params = get_valid_params(self.params, ['voornaam', 'naam'])
331
332			self.assertEqual(1, len(valid_params))
333			self.assertIn('voornaam', valid_params)
334
335
336

OnroerendErfgoed / oe_utils

Push — master ( be44b6...3b3781 )

SearchEngineTest.test_query_KO() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like