ValidateParamsTests.test_validate_params() - Code Metrics - Inspection of "Merge pull request #39 from OnroerendErfgoed/38_ge..." - OnroerendErfgoed/oe_utils - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( abd5fe...1e5f43 )

by Bart

created 2017-07-27 10:08 UTC

ValidateParamsTests.test_validate_params() A

↳ Parent: ValidateParamsTests

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

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

# -*- 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
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_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)


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

OnroerendErfgoed / oe_utils

Push — master ( abd5fe...1e5f43 )

ValidateParamsTests.test_validate_params() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like