TestClass

Complexity

Total Complexity

1

Size/Duplication

Total Lines	3
Duplicated Lines	0 %

Importance

Changes	1
Bugs	0	Features	0

1 Method

Rating	Name	Duplication	Size	Complexity
A	__int__()	0	2	1

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

from pyramid.testing import DummySession

from oe_utils.search.indexer import Indexer

mock_oauth_helper = Mock()
mock_oauth_helper.get_system_token = Mock(return_value='test_token')

mock_session = Mock()
mock_session.index_new = {'TestClass': set()}
mock_session.index_dirty = {'TestClass': set()}
mock_session.index_deleted = {'TestClass': set()}


# def __init__(self, settings, index_operation, index_operation_name, cls)
#
index_operation_log = Mock()


class TestClass:
    def __int__(self):
        pass


def mock_index_operation(index_new, index_dirty, index_deleted, settings):
    index_operation_log.info("starting index operation")
    for id in index_new:
        index_operation_log.info("index new object {}".format(id))
    for id in index_dirty:
        index_operation_log.info("index updated object {}".format(id))
    for id in index_deleted:
        index_operation_log.info("remove object {} from index".format(id))


class IndexerTests(unittest.TestCase):
    @patch('oe_utils.search.indexer.event', Mock())
    def setUp(self):
        self.indexer = Indexer(
            {'searchengine.baseurl': '', 'searchengine.index': '', 'redis.queue_name': 'dummy-queue'},
            mock_index_operation,
            'tests.test_search_indexer.mock_index_operation',
            TestClass
        )

    def tearDown(self):
        pass

    @patch('oe_utils.search.indexer.event', Mock())
    def test_when_registering_a_session_three_empty_index_sets_are_created(self):
        session = Mock()
        session.index_new = {}
        session.index_dirty = {}
        session.index_deleted = {}
        self.indexer.register_session(session)

        self.assertEquals(len(session.index_new['TestClass']), 0)
        self.assertEquals(len(session.index_dirty['TestClass']), 0)
        self.assertEquals(len(session.index_deleted['TestClass']), 0)

    @patch('oe_utils.search.indexer.event', Mock())
    def test_remove_session(self):
        session = Mock()
        session.index_new = {}
        session.index_dirty = {}
        session.index_deleted = {}
        session.redis = set()
        session.index_new = {'TestClass': set()}
        session.index_dirty = {'TestClass': set()}
        session.index_deleted = {'TestClass': set()}
        self.indexer.sessions = {session}

        self.indexer.remove_session(session)
        self.assertEquals(len(self.indexer.sessions), 0)

    @patch('oe_utils.search.indexer.event', Mock())
    def test_remove_session_no_indexing_sets(self):
        session = object()
        self.indexer.sessions = {session}
        log = Mock(warning=Mock())
        with patch('oe_utils.search.indexer.log', log):
            self.indexer.remove_session(session)
            self.assertEquals(log.warning.call_count, 1)

    @patch('oe_utils.search.indexer.event', Mock())
    @patch('oe_utils.search.indexer.object_session', Mock(return_value=object()))
    def test_update_listener_no_index_sets(self):
        target = TestClass()
        target.id = 1
        log = Mock(warning=Mock())
        with patch('oe_utils.search.indexer.log', log):
            self.indexer._update_listener(Mock(), Mock(), target)
            self.assertEquals(log.warning.call_count, 1)

    @patch('oe_utils.search.indexer.event', Mock())
    @patch('oe_utils.search.indexer.object_session', Mock(return_value=mock_session))
    def test_update_listener(self):
        target = TestClass()
        target.id = 1
        self.indexer._update_listener(Mock(), Mock(), target)
        self.assertIn(1, mock_session.index_dirty['TestClass'])
        mock_session.index_dirty['TestClass'] = set()

    @patch('oe_utils.search.indexer.event', Mock())
    @patch('oe_utils.search.indexer.object_session', Mock(return_value=mock_session))
    def test_new_listener(self):
        target = TestClass()
        target.id = 1
        self.indexer._new_listener(Mock(), Mock(), target)
        self.assertIn(1, mock_session.index_new['TestClass'])
        mock_session.index_new['TestClass'] = set()

    @patch('oe_utils.search.indexer.event', Mock())
    @patch('oe_utils.search.indexer.object_session', Mock(return_value=mock_session))
    def test_delete_listener(self):
        target = TestClass()
        target.id = 1
        self.indexer._delete_listener(Mock(), Mock(), target)
        self.assertIn(1, mock_session.index_deleted['TestClass'])
        mock_session.index_deleted['TestClass'] = set()

    @patch('oe_utils.search.indexer.event', Mock())

This code seems to be duplicated in your project.

    def test_when_registering_a_session_the_session_is_added_to_the_sessions(self):
        session = Mock()
        session.index_new = {}
        session.index_dirty = {}
        session.index_deleted = {}
        self.indexer.register_session(session)

        self.assertEquals(len(self.indexer.sessions), 1)
        self.assertEquals(next(iter(self.indexer.sessions)), session)

    def test_when_registering_a_session_events_for_after_commit_and_after_rollback_are_created(self):
        event = Mock()
        with patch('oe_utils.search.indexer.event', event):
            session = Mock()
            session.index_new = {}
            session.index_dirty = {}
            session.index_deleted = {}
            self.indexer.register_session(session)

            self.assertEquals(event.listen.call_args_list[0][0][1], 'after_commit')
            self.assertEquals(event.listen.call_args_list[1][0][1], 'after_rollback')

    @patch('oe_utils.search.indexer.event', Mock())
    def test_the_listener_after_rollback_clears_the_three_index_sets(self):
        session = Mock()
        session.index_new = {}
        session.index_dirty = {}
        session.index_deleted = {}
        self.indexer.register_session(session)
        session.index_new['TestClass'].add(Mock())
        session.index_dirty['TestClass'].add(Mock())
        session.index_deleted['TestClass'].add(Mock())

        self.indexer.after_rollback_listener(session)

        self.assertEquals(len(session.index_new['TestClass']), 0)
        self.assertEquals(len(session.index_dirty['TestClass']), 0)
        self.assertEquals(len(session.index_deleted['TestClass']), 0)

    def test_the_listener_after_rollback_logs_a_warning_when_no_sets_were_created(self):
        log = Mock(warning=Mock())
        with patch('oe_utils.search.indexer.log', log):
            self.indexer.after_rollback_listener(DummySession())

            self.assertEquals(log.warning.call_count, 1)

    @patch('oe_utils.search.indexer.event', Mock())
    def test_the_listener_after_commit_calls_the_index_functions(self):
        index_operation_log.reset_mock()
        session = Mock()
        session.index_new = {}
        session.index_dirty = {}
        session.index_deleted = {}
        self.indexer.register_session(session)
        session.index_new['TestClass'].add(1)
        session.index_dirty['TestClass'].add(2)
        session.index_deleted['TestClass'].add(3)

        self.indexer.after_commit_listener(session)

        mock_calls = [call("starting index operation"),
            call("index new object 1"), call("index updated object 2"), call("remove object 3 from index")]

        self.assertEquals(index_operation_log.info.mock_calls, mock_calls)

        index_operation_log.reset_mock()

    @patch('oe_utils.search.indexer.event', Mock())
    def test_the_listener_after_commit_no_index_sets(self):
        session = object()
        log = Mock(warning=Mock())
        with patch('oe_utils.search.indexer.log', log):
            self.indexer.after_commit_listener(session)
            self.assertEquals(log.warning.call_count, 1)

    @patch('oe_utils.search.indexer.event', Mock())

This code seems to be duplicated in your project.

    @patch('oe_utils.search.indexer.Queue.enqueue', Mock())
    def test_if_redis_is_available_then_the_index_operation_will_be_queued(self):
        session = Mock()
        session.index_new = {}
        session.index_dirty = {}
        session.index_deleted = {}
        redis = Mock()
        enqueue = Mock()
        with patch('oe_utils.search.indexer.Queue.enqueue', enqueue):
            self.indexer.register_session(session, redis)

            self.indexer.after_commit_listener(session)

            self.assertEquals(enqueue.call_count, 1)

    @patch('oe_utils.search.indexer.event', Mock())
    def test_if_redis_queue_name_is_not_filled_in_then_a_KeyError_will_be_raised(self):
        session = Mock()
        session.index_new = {}
        session.index_dirty = {}
        session.index_deleted = {}
        session.redis = Mock()

        self.assertRaises(KeyError, Indexer({}, '', '', TestClass).after_commit_listener, session)