TestClass.__int__() - Code Metrics - Inspection of "Merge pull request #67 from OnroerendErfgoed/updat..." - OnroerendErfgoed/oe_utils - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( be46e6...8a6839 )

unknown

created 2018-04-03 12:44 UTC

TestClass.int() A

↳ Parent: TestClass

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	1
dl	0
loc	2
rs	10
c	1
b	0
f	0

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())

    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())

    @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)



1		import unittest
2		try:
3		from unittest.mock import Mock, patch, MagicMock, ANY, call
4		except:
5		from mock import Mock, patch, MagicMock, ANY, call
6
7		from pyramid.testing import DummySession
8
9		from oe_utils.search.indexer import Indexer
10
11		mock_oauth_helper = Mock()
12		mock_oauth_helper.get_system_token = Mock(return_value='test_token')
13
14		mock_session = Mock()
15		mock_session.index_new = {'TestClass': set()}
16		mock_session.index_dirty = {'TestClass': set()}
17		mock_session.index_deleted = {'TestClass': set()}
18
19
20		# def __init__(self, settings, index_operation, index_operation_name, cls)
21		#
22		index_operation_log = Mock()
23
24
25		class TestClass:
26		def __int__(self):
27		pass
28
29
30		def mock_index_operation(index_new, index_dirty, index_deleted, settings):
31		index_operation_log.info("starting index operation")
32		for id in index_new:
33		index_operation_log.info("index new object {}".format(id))
34		for id in index_dirty:
35		index_operation_log.info("index updated object {}".format(id))
36		for id in index_deleted:
37		index_operation_log.info("remove object {} from index".format(id))
38
39
40		class IndexerTests(unittest.TestCase):
41		@patch('oe_utils.search.indexer.event', Mock())
42		def setUp(self):
43		self.indexer = Indexer(
44		{'searchengine.baseurl': '', 'searchengine.index': '', 'redis.queue_name': 'dummy-queue'},
45		mock_index_operation,
46		'tests.test_search_indexer.mock_index_operation',
47		TestClass
48		)
49
50		def tearDown(self):
51		pass
52
53		@patch('oe_utils.search.indexer.event', Mock())
54		def test_when_registering_a_session_three_empty_index_sets_are_created(self):
55		session = Mock()
56		session.index_new = {}
57		session.index_dirty = {}
58		session.index_deleted = {}
59		self.indexer.register_session(session)
60
61		self.assertEquals(len(session.index_new['TestClass']), 0)
62		self.assertEquals(len(session.index_dirty['TestClass']), 0)
63		self.assertEquals(len(session.index_deleted['TestClass']), 0)
64
65		@patch('oe_utils.search.indexer.event', Mock())
66		def test_remove_session(self):
67		session = Mock()
68		session.index_new = {}
69		session.index_dirty = {}
70		session.index_deleted = {}
71		session.redis = set()
72		session.index_new = {'TestClass': set()}
73		session.index_dirty = {'TestClass': set()}
74		session.index_deleted = {'TestClass': set()}
75		self.indexer.sessions = {session}
76
77		self.indexer.remove_session(session)
78		self.assertEquals(len(self.indexer.sessions), 0)
79
80		@patch('oe_utils.search.indexer.event', Mock())
81		def test_remove_session_no_indexing_sets(self):
82		session = object()
83		self.indexer.sessions = {session}
84		log = Mock(warning=Mock())
85		with patch('oe_utils.search.indexer.log', log):
86		self.indexer.remove_session(session)
87		self.assertEquals(log.warning.call_count, 1)
88
89		@patch('oe_utils.search.indexer.event', Mock())
90		@patch('oe_utils.search.indexer.object_session', Mock(return_value=object()))
91		def test_update_listener_no_index_sets(self):
92		target = TestClass()
93		target.id = 1
94		log = Mock(warning=Mock())
95		with patch('oe_utils.search.indexer.log', log):
96		self.indexer._update_listener(Mock(), Mock(), target)
97		self.assertEquals(log.warning.call_count, 1)
98
99		@patch('oe_utils.search.indexer.event', Mock())
100		@patch('oe_utils.search.indexer.object_session', Mock(return_value=mock_session))
101		def test_update_listener(self):
102		target = TestClass()
103		target.id = 1
104		self.indexer._update_listener(Mock(), Mock(), target)
105		self.assertIn(1, mock_session.index_dirty['TestClass'])
106		mock_session.index_dirty['TestClass'] = set()
107
108		@patch('oe_utils.search.indexer.event', Mock())
109		@patch('oe_utils.search.indexer.object_session', Mock(return_value=mock_session))
110		def test_new_listener(self):
111		target = TestClass()
112		target.id = 1
113		self.indexer._new_listener(Mock(), Mock(), target)
114		self.assertIn(1, mock_session.index_new['TestClass'])
115		mock_session.index_new['TestClass'] = set()
116
117		@patch('oe_utils.search.indexer.event', Mock())
118		@patch('oe_utils.search.indexer.object_session', Mock(return_value=mock_session))
119		def test_delete_listener(self):
120		target = TestClass()
121		target.id = 1
122		self.indexer._delete_listener(Mock(), Mock(), target)
123		self.assertIn(1, mock_session.index_deleted['TestClass'])
124		mock_session.index_deleted['TestClass'] = set()
125
126	View Code Duplication	@patch('oe_utils.search.indexer.event', Mock())
		0 ignored issues – show Duplication introduced 2018-03-12 09:48 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
127		def test_when_registering_a_session_the_session_is_added_to_the_sessions(self):
128		session = Mock()
129		session.index_new = {}
130		session.index_dirty = {}
131		session.index_deleted = {}
132		self.indexer.register_session(session)
133
134		self.assertEquals(len(self.indexer.sessions), 1)
135		self.assertEquals(next(iter(self.indexer.sessions)), session)
136
137		def test_when_registering_a_session_events_for_after_commit_and_after_rollback_are_created(self):
138		event = Mock()
139		with patch('oe_utils.search.indexer.event', event):
140		session = Mock()
141		session.index_new = {}
142		session.index_dirty = {}
143		session.index_deleted = {}
144		self.indexer.register_session(session)
145
146		self.assertEquals(event.listen.call_args_list[0][0][1], 'after_commit')
147		self.assertEquals(event.listen.call_args_list[1][0][1], 'after_rollback')
148
149		@patch('oe_utils.search.indexer.event', Mock())
150		def test_the_listener_after_rollback_clears_the_three_index_sets(self):
151		session = Mock()
152		session.index_new = {}
153		session.index_dirty = {}
154		session.index_deleted = {}
155		self.indexer.register_session(session)
156		session.index_new['TestClass'].add(Mock())
157		session.index_dirty['TestClass'].add(Mock())
158		session.index_deleted['TestClass'].add(Mock())
159
160		self.indexer.after_rollback_listener(session)
161
162		self.assertEquals(len(session.index_new['TestClass']), 0)
163		self.assertEquals(len(session.index_dirty['TestClass']), 0)
164		self.assertEquals(len(session.index_deleted['TestClass']), 0)
165
166		def test_the_listener_after_rollback_logs_a_warning_when_no_sets_were_created(self):
167		log = Mock(warning=Mock())
168		with patch('oe_utils.search.indexer.log', log):
169		self.indexer.after_rollback_listener(DummySession())
170
171		self.assertEquals(log.warning.call_count, 1)
172
173		@patch('oe_utils.search.indexer.event', Mock())
174		def test_the_listener_after_commit_calls_the_index_functions(self):
175		index_operation_log.reset_mock()
176		session = Mock()
177		session.index_new = {}
178		session.index_dirty = {}
179		session.index_deleted = {}
180		self.indexer.register_session(session)
181		session.index_new['TestClass'].add(1)
182		session.index_dirty['TestClass'].add(2)
183		session.index_deleted['TestClass'].add(3)
184
185		self.indexer.after_commit_listener(session)
186
187		mock_calls = [call("starting index operation"),
188		call("index new object 1"), call("index updated object 2"), call("remove object 3 from index")]
189
190		self.assertEquals(index_operation_log.info.mock_calls, mock_calls)
191
192		index_operation_log.reset_mock()
193
194		@patch('oe_utils.search.indexer.event', Mock())
195		def test_the_listener_after_commit_no_index_sets(self):
196		session = object()
197		log = Mock(warning=Mock())
198		with patch('oe_utils.search.indexer.log', log):
199		self.indexer.after_commit_listener(session)
200		self.assertEquals(log.warning.call_count, 1)
201
202	View Code Duplication	@patch('oe_utils.search.indexer.event', Mock())
		0 ignored issues – show Duplication introduced 2018-03-12 09:48 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
203		@patch('oe_utils.search.indexer.Queue.enqueue', Mock())
204		def test_if_redis_is_available_then_the_index_operation_will_be_queued(self):
205		session = Mock()
206		session.index_new = {}
207		session.index_dirty = {}
208		session.index_deleted = {}
209		redis = Mock()
210		enqueue = Mock()
211		with patch('oe_utils.search.indexer.Queue.enqueue', enqueue):
212		self.indexer.register_session(session, redis)
213
214		self.indexer.after_commit_listener(session)
215
216		self.assertEquals(enqueue.call_count, 1)
217
218		@patch('oe_utils.search.indexer.event', Mock())
219		def test_if_redis_queue_name_is_not_filled_in_then_a_KeyError_will_be_raised(self):
220		session = Mock()
221		session.index_new = {}
222		session.index_dirty = {}
223		session.index_deleted = {}
224		session.redis = Mock()
225
226		self.assertRaises(KeyError, Indexer({}, '', '', TestClass).after_commit_listener, session)
227
228

OnroerendErfgoed / oe_utils

Push — master ( be46e6...8a6839 )

TestClass.__int__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

TestClass.int() A