OnroerendErfgoed / oe_utils

tests/test_datamanagers.py

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

from sqlalchemy import engine_from_config

from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm.exc import NoResultFound
import transaction

from zope.sqlalchemy import ZopeTransactionExtension

from oe_utils.data.data_managers import DataManager
from oe_utils.data.data_transfer_objects import ResultDTO
from oe_utils.range_parser import Range
from tests import init_test_db, DummyDossier, fill_db, DossierEmail, Dummy

try:
    from unittest.mock import Mock
except ImportError:
    from mock import Mock
try:
    import configparser
except ImportError:
    import ConfigParser as configparser


class DataManagerTest(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')
        settings = dict((s[0], s[1]) for s in settings)
        cls.engine = engine_from_config(settings, prefix='sqlalchemy.')
        cls.session_maker = sessionmaker(
            bind=cls.engine,
            extension=ZopeTransactionExtension()
        )

    def setUp(self):
        init_test_db(self.engine)
        self.session = self.session_maker()
        self.data_manager = DataManager(self.session, DummyDossier)

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

    def test_get_dossier(self):
        dossier = self.data_manager.get(1)
        self.assertIsNotNone(dossier)
        self.assertEqual(1, dossier.id)

    def test_get_dossier_for_update(self):
        dossier = self.data_manager.get_for_update(1)
        self.assertIsNotNone(dossier)
        self.assertEqual(1, dossier.id)

    def test_get_dummy(self):
        dummy = self.data_manager.get(1, Dummy)
        self.assertIsNotNone(dummy)
        self.assertEqual(1, dummy.id)
        self.assertIsInstance(dummy, Dummy)

    def test_get_one_unexisting_dossier(self):
        self.assertRaises(NoResultFound, self.data_manager.get_one, 1112)

    def test_get_one_for_update_unexisting_dossier(self):
        self.assertRaises(NoResultFound, self.data_manager.get_one_for_update, 1112)

    def test_get_one_for_update(self):
        dossier = self.data_manager.get_one_for_update(1)
        self.assertIsNotNone(dossier)
        self.assertEqual(1, dossier.id)

    def test_get_unexisting_dossier(self):
        res = self.data_manager.get(1112)
        self.assertIsNone(res)

    def test_delete_dossier(self):
        with transaction.manager as manager:
            self.data_manager.delete(1)
            manager.commit()
        self.assertRaises(NoResultFound, self.data_manager.get_one, 1)

    def test_delete_unexisting_dossier(self):
        self.assertRaises(NoResultFound, self.data_manager.delete, 11234)

    def test_save_dossier_new(self):
        dossier = DummyDossier(id=3)
        dossier_saved = self.data_manager.save(dossier)
        self.assertEqual(3, dossier_saved.id)

    def test_save_dossier_edit(self):
        dossier = DummyDossier(id=22)
        dossier.dossierdata = {"test": "blabla"}
        self.data_manager.save(dossier)
        self.session.flush()
        saved_dossier = self.data_manager.get(22)
        self.assertEqual({"test": "blabla"}, saved_dossier.dossierdata)

    def test_save_dossier_new_and_edit(self):
        dossier = DummyDossier(id=3)
        dossier_saved = self.data_manager.save(dossier)
        dossier.dossierdata = {"test": "blabla"}
        dossier_saved = self.data_manager.save(dossier)
        self.assertEqual(3, dossier_saved.id)

    def test_process_ranged_query(self):
        fill_db(self.session)
        query = self.session.query(DummyDossier)
        result_range = Range(start=0, end=4)
        result_dto = self.data_manager.process_ranged_query(query, result_range)
        self.assertTrue(isinstance(result_dto, ResultDTO))
        self.assertEqual(77, result_dto.total)
        self.assertEqual(5, len(result_dto.data))

    def test_process_empty_range_query(self):
        fill_db(self.session)
        query = self.session.query(DummyDossier)
        result_dto = self.data_manager.process_ranged_query(query, None)
        self.assertTrue(isinstance(result_dto, ResultDTO))
        self.assertEqual(77, result_dto.total)
        self.assertEqual(77, len(result_dto.data))

    def test_insert_append_object(self):
        dossier = DummyDossier(id=66)
        email = DossierEmail()
        email.email = 'test@test.test'
        dossier.emails.append(email)
        self.data_manager.save(dossier)

    def test_ordered_result_list_test_asc(self):
        query = self.session.query(Dummy)
        sortlist = [('test', 'asc')]
        query = self.data_manager.ordered_result_list(query, Dummy, sortlist)
        res = query.all()
        self.assertEqual(4, len(res))
        self.assertEqual('alpha', res[0].test)

    def test_ordered_result_list_test_desc(self):
        query = self.session.query(Dummy)
        sortlist = [('test', 'desc')]
        query = self.data_manager.ordered_result_list(query, Dummy, sortlist)
        res = query.all()
        self.assertEqual(4, len(res))
        self.assertEqual('zero', res[0].test)

    def test_ordered_result_list_id_asc(self):
        query = self.session.query(Dummy)
        sortlist = [('id', 'asc')]
        query = self.data_manager.ordered_result_list(query, Dummy, sortlist)
        res = query.all()
        self.assertEqual(4, len(res))
        self.assertEqual(1, res[0].id)