OnroerendErfgoed / oe_utils

tests/test_functional.py

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

from requests import HTTPError
from sqlalchemy import engine_from_config
from sqlalchemy.orm import sessionmaker

from webob.multidict import MultiDict
from zope.sqlalchemy import ZopeTransactionExtension

from oe_utils.data.data_transfer_objects import ResultDTO
from oe_utils.range_parser import Range
from oe_utils.search.searchengine import SearchEngine
from tests import testdata, assert_no_exception_raised, dummy_index, dummy_mapping, incorrect_testdata, DummyDossier

try:
    import configparser
except ImportError:
    import ConfigParser as configparser


def index_db(session, search_engine, system_token, object_type):
    for dd in session.query(DummyDossier):
        search_engine.add_to_index(system_token, object_type, dd.id, json.dumps(dd.dossierdata))
    time.sleep(2)


class SearchTests(unittest.TestCase):
    @classmethod

This code seems to be duplicated in your project.

    def setUpClass(cls):
        config = configparser.ConfigParser()
        config.read(os.path.join(os.path.dirname(__file__), 'test.ini'))
        settings = config.items('app:oe_utils')
        cls.settings = dict((s[0], s[1]) for s in settings)
        cls.object_type = 'dossier'
        cls.user_acls = ["test_user", "test_role"]
        cls.engine = engine_from_config(cls.settings, prefix='sqlalchemy.')
        cls.session_maker = sessionmaker(
            bind=cls.engine,
            extension=ZopeTransactionExtension()
        )

    def setUp(self):
        self.search_engine = SearchEngine(self.settings['searchengine.baseurl'],
                                          self.settings['searchengine.index'])
        self.system_token = 'test_token'
        self.search_engine.remove_index('test_token')
        self.search_engine.create_index('test_token', data=dummy_index)
        self.search_engine.add_type_mapping('actor', dummy_mapping, 'test_token')
        self.session = self.session_maker()

    def tearDown(self):
        self.session.close()
        del self.search_engine

    def test_add_to_index(self):
        object_id = 'test1'
        assert_no_exception_raised(self.search_engine.add_to_index, self.system_token, self.object_type, object_id,
                                   json.dumps(testdata))

    def test_add_to_index_KO(self):
        object_id = 'test2'
        self.assertRaises(HTTPError, self.search_engine.add_to_index, self.system_token, self.object_type, object_id,
                          incorrect_testdata)

    def test_remove_from_index_KO(self):
        object_id = 'test2'
        self.assertRaises(HTTPError, self.search_engine.remove_from_index, self.system_token, self.object_type,
                          object_id)

    def test_remove_from_index(self):
        index_db(self.session, self.search_engine, self.system_token, self.object_type)
        object_id = '1'
        assert_no_exception_raised(self.search_engine.remove_from_index, self.system_token, self.object_type, object_id)

    def test_remove_from_index_no_system_token(self):
        index_db(self.session, self.search_engine, self.system_token, self.object_type)
        object_id = '1'
        assert_no_exception_raised(self.search_engine.remove_from_index, None, self.object_type, object_id)

    def test_remove_from_index_by_query(self):
        index_db(self.session, self.search_engine, self.system_token, self.object_type)
        object_id = '1'
        assert_no_exception_raised(self.search_engine.remove_from_index_by_query, self.system_token, 'id', object_id)

    def test_remove_index(self):
        assert_no_exception_raised(self.search_engine.remove_index, 'test_token')

    def test_remove_index_KO(self):
        searchengine = SearchEngine(self.settings['searchengine.baseurl'], 'testing')
        assert_no_exception_raised(searchengine.remove_index, 'test_token')

    def test_create_index(self):
        searchengine = SearchEngine(self.settings['searchengine.baseurl'], 'test')
        assert_no_exception_raised(searchengine.remove_index, 'test_token')
        assert_no_exception_raised(searchengine.create_index, 'test_token', data=dummy_index)

    def test_create_index_KO(self):
        self.assertRaises(HTTPError, self.search_engine.create_index, 'test_token', data=dummy_index)

    def test_add_type_mapping(self):
        assert_no_exception_raised(self.search_engine.add_type_mapping, self.object_type, dummy_mapping, 'test_token')

    def test_add_type_mapping_KO(self):
        object_type = 'test'
        self.assertRaises(HTTPError, self.search_engine.add_type_mapping, object_type, {}, 'test_token')

    def test_query(self):
        index_db(self.session, self.search_engine, self.system_token, self.object_type)
        query = MultiDict({
            'query': 'test'
        })
        result_dto = self.search_engine.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)

    def test_query_ranged(self):
        index_db(self.session, self.search_engine, self.system_token, self.object_type)
        object_type = 'dossier'
        query = MultiDict({
            'query': 'test'
        })
        result_range = Range(start=0, end=4)
        result_dto = self.search_engine.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)

    def test_query_min_score(self):
        index_db(self.session, self.search_engine, self.system_token, self.object_type)
        object_type = 'dossier'
        query = MultiDict({
            'query': 'test'
        })
        result_dto = self.search_engine.query(self.system_token, object_type=object_type,
                                              query_params=query,
                                              user_acls=self.user_acls,
                                              min_score=1)
        self.assertTrue(isinstance(result_dto, ResultDTO))
        self.assertEqual(2, result_dto.total)
        self.assertIsNotNone(result_dto.data)

    def test_query_no_hits(self):
        index_db(self.session, self.search_engine, self.system_token, self.object_type)
        object_type = 'testing'
        query = MultiDict({
            'query': 'test'
        })
        result_dto = self.search_engine.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)