ParseFilterParamsTest

Complexity

Total Complexity

4

Size/Duplication

Total Lines	18
Duplicated Lines	0 %

Importance

Changes	1
Bugs	0	Features	0

4 Methods

Rating	Name	Duplication	Size	Complexity
A	setUp()	0	2	1
A	tearDown()	0	2	1
A	test_parse_filter_params_no_query_params()	0	3	1
A	test_parse_filter_params()	0	7	1

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