MongoRepoTests.test_Ensures_text_index_for_search() - Code Metrics - ilogue/niprov - Measure and Improve Code Quality continuously with Scrutinizer

test_Ensures_text_index_for_search() A
last analyzed 2016-11-17 03:25 UTC

↳ Parent: MongoRepoTests

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	2
Bugs	0	Features	0

Metric	Value
cc	2
dl	0
loc	8
rs	9.4285
c	2
b	0
f	0

import unittest
from mock import Mock, patch, sentinel
from tests.ditest import DependencyInjectionTestBase


class MongoRepoTests(DependencyInjectionTestBase):

    def setUp(self):
        super(MongoRepoTests, self).setUp()
        self.pictureCache.getBytes.return_value = None
        self.db = Mock()
        self.db.provenance.find_one.return_value = {}
        self.db.provenance.find.return_value = {}
        self.pymongo = None

    def setupRepo(self):
        from niprov.mongo import MongoRepository
        with patch('niprov.mongo.pymongo') as self.pymongo:
            self.repo = MongoRepository(dependencies=self.dependencies)
        self.repo.db = self.db

    def test_Connection(self):
        from niprov.mongo import MongoRepository
        with patch('niprov.mongo.pymongo') as pymongo:
            repo = MongoRepository(dependencies=self.dependencies)
        pymongo.MongoClient.assert_called_with(self.config.database_url)
        self.assertEqual(pymongo.MongoClient().get_default_database(), repo.db)

    def test_byLocation_returns_img_from_record_with_path(self):
        self.setupRepo()
        p = '/p/f1'
        out = self.repo.byLocation(p)
        self.db.provenance.find_one.assert_called_with({'location':p})
        self.fileFactory.fromProvenance.assert_called_with(
            self.db.provenance.find_one())
        self.assertEqual(self.fileFactory.fromProvenance(), out)

    def test_getSeries(self):
        self.setupRepo()
        img = Mock()
        out = self.repo.getSeries(img)
        self.db.provenance.find_one.assert_called_with(
            {'seriesuid':img.getSeriesId()})
        self.fileFactory.fromProvenance.assert_called_with(
            self.db.provenance.find_one())
        self.assertEqual(self.fileFactory.fromProvenance(), out)

    def test_getSeries_returns_None_right_away_if_no_series_id(self):
        self.setupRepo()
        img = Mock()
        img.getSeriesId.return_value = None
        out = self.repo.getSeries(img)
        assert not self.db.provenance.find_one.called
        self.assertEqual(None, out)

    def test_Add(self):
        self.setupRepo()
        img = Mock()
        img.provenance = {'a':1, 'b':2}
        self.repo.add(img)
        self.db.provenance.insert_one.assert_called_with({'a':1, 'b':2})

    def test_update(self):
        self.setupRepo()
        img = Mock()
        img.provenance = {'a':1, 'b':2}
        self.repo.update(img)
        self.db.provenance.update.assert_called_with(
            {'location':img.location.toString()}, {'a':1, 'b':2})

    def test_all(self):
        self.fileFactory.fromProvenance.side_effect = lambda p: 'img_'+p
        self.db.provenance.find.return_value = ['p1', 'p2']
        self.setupRepo()
        out = self.repo.all()
        self.db.provenance.find.assert_called_with()
        self.fileFactory.fromProvenance.assert_any_call('p1')
        self.fileFactory.fromProvenance.assert_any_call('p2')
        self.assertEqual(['img_p1', 'img_p2'], out)

    def test_updateApproval(self):
        self.setupRepo()
        img = Mock()
        p = '/p/f1'

        newStatus = 'oh-oh'
        self.repo.updateApproval(p, newStatus)
        self.db.provenance.update.assert_called_with(
            {'location':p}, {'$set': {'approval': newStatus}})

    def test_latest(self):
        self.fileFactory.fromProvenance.side_effect = lambda p: 'img_'+p
        self.db.provenance.find.return_value = Mock()
        self.db.provenance.find.return_value.sort.return_value.limit.return_value = ['px','py']
        self.setupRepo()
        out = self.repo.latest()
        self.db.provenance.find.assert_called_with()
        self.db.provenance.find().sort.assert_called_with('added', -1)
        self.db.provenance.find().sort().limit.assert_called_with(20)
        self.fileFactory.fromProvenance.assert_any_call('px')
        self.fileFactory.fromProvenance.assert_any_call('py')
        self.assertEqual(['img_px', 'img_py'], out)

    def test_statistics(self):
        self.db.provenance.aggregate.return_value = [sentinel.stats,]
        self.setupRepo()
        out = self.repo.statistics()
        self.db.provenance.aggregate.assert_called_with(
           [{'$group':
                 {
                   '_id': None,
                   'totalsize': { '$sum': '$size' },
                   'count': { '$sum': 1 }
                 }
            }])
        self.assertEqual(sentinel.stats, out)

    def test_statistics_if_no_records(self):
        self.db.provenance.aggregate.return_value = []
        self.setupRepo()
        out = self.repo.statistics()
        self.assertEqual({'count':0}, out)

    def test_byId(self):
        self.setupRepo()
        ID = 'abc123'
        out = self.repo.byId(ID)
        self.db.provenance.find_one.assert_called_with({'id':ID})
        self.fileFactory.fromProvenance.assert_called_with(
            self.db.provenance.find_one())
        self.assertEqual(self.fileFactory.fromProvenance(), out)

    def test_If_db_returns_None_should_return_None_for_byId_byLocation(self):
        self.setupRepo()
        self.db.provenance.find_one.return_value = None
        out = self.repo.byId('abc123')
        assert not self.fileFactory.fromProvenance.called
        self.assertEqual(None, out)

    def test_byLocations(self):
        self.fileFactory.fromProvenance.side_effect = lambda p: 'img_'+p
        self.db.provenance.find.return_value = ['p1', 'p2']
        self.setupRepo()
        out = self.repo.byLocations(['l1','l2'])
        self.db.provenance.find.assert_called_with({'location':{'$in':['l1','l2']}})
        self.fileFactory.fromProvenance.assert_any_call('p1')
        self.fileFactory.fromProvenance.assert_any_call('p2')
        self.assertEqual(['img_p1', 'img_p2'], out)

    def test_byParents(self):
        self.fileFactory.fromProvenance.side_effect = lambda p: 'img_'+p
        self.db.provenance.find.return_value = ['p1', 'p2']
        self.setupRepo()
        out = self.repo.byParents(['x1','x2'])
        self.db.provenance.find.assert_called_with({'parents':{'$in':['x1','x2']}})
        self.fileFactory.fromProvenance.assert_any_call('p1')
        self.fileFactory.fromProvenance.assert_any_call('p2')
        self.assertEqual(['img_p1', 'img_p2'], out)

    def test_Obtains_optional_snapshot_data_from_cache_when_serializing(self):
        self.pictureCache.getBytes.return_value = sentinel.snapbytes
        with patch('niprov.mongo.bson') as bson:
            bson.Binary.return_value = sentinel.snapbson
            self.setupRepo()
            img = Mock()
            img.provenance = {'a':1}
            self.repo.add(img)
            self.pictureCache.getBytes.assert_called_with(for_=img)

            bson.Binary.assert_called_with(sentinel.snapbytes)
            self.db.provenance.insert_one.assert_called_with({'a':1, 
                '_snapshot-data':sentinel.snapbson})

    def test_If_no_snapshot_doesnt_add_data_field(self):
        self.pictureCache.getBytes.return_value = None
        with patch('niprov.mongo.bson') as bson:
            self.setupRepo()
            img = Mock()
            img.provenance = {'a':1}

            self.repo.add(img)
            assert not bson.Binary.called
            self.db.provenance.insert_one.assert_called_with({'a':1})

    def test_If_snapshotdata_hands_them_to_pictureCache_on_deserializing(self):
        img = Mock()
        self.fileFactory.fromProvenance.return_value = img
        self.setupRepo()
        self.db.provenance.find_one.return_value = {'a':3}
        out = self.repo.byLocation('/p/f1')
        assert not self.pictureCache.keep.called
        self.db.provenance.find_one.return_value = {'a':3, 
                                                    '_snapshot-data':'y7yUyS'}
        out = self.repo.byLocation('/p/f1')
        self.pictureCache.keep.assert_called_with('y7yUyS', for_=img)

    def test_Query(self):
        self.db.provenance.find.return_value = ['record1']
        self.setupRepo()
        q = Mock()
        field1 = Mock()
        field1.name = 'color'
        field1.value = 'red'
        field1.all = False
        q.getFields.return_value = [field1]
        out = self.repo.inquire(q)
        self.db.provenance.find.assert_called_with({'color':'red'})
        self.fileFactory.fromProvenance.assert_called_with('record1')

    def test_Ensures_text_index_for_search(self):
        self.setupRepo()
        self.repo.search('')
        searchfields = ['location','user','subject','project','protocol',
                  'transformation','technique','modality']
        indexspec = [(field, 'text') for field in searchfields]
        self.db.provenance.create_index.assert_called_with(indexspec, 
                                                           name='textsearch')

    def test_Search(self):
        self.db.provenance.find.return_value = ['r1','r2']
        self.setupRepo()
        self.repo.search('xyz')
        self.db.provenance.find.assert_called_with({'$text':{'$search': 'xyz'}})
        self.fileFactory.fromProvenance.assert_any_call('r1')
        self.fileFactory.fromProvenance.assert_any_call('r2')

    def test_Query_for_ALL_field(self):
        self.db.provenance.distinct.return_value = ['r1','r2']
        self.setupRepo()
        q = Mock()
        field1 = Mock()
        field1.name = 'color'
        field1.all = True
        q.getFields.return_value = [field1]
        out = self.repo.inquire(q)
        self.db.provenance.distinct.assert_called_with('color')
        assert not self.fileFactory.fromProvenance.called



1		import unittest
2		from mock import Mock, patch, sentinel
3		from tests.ditest import DependencyInjectionTestBase
4
5
6		class MongoRepoTests(DependencyInjectionTestBase):
7
8		def setUp(self):
9		super(MongoRepoTests, self).setUp()
10		self.pictureCache.getBytes.return_value = None
11		self.db = Mock()
12		self.db.provenance.find_one.return_value = {}
13		self.db.provenance.find.return_value = {}
14		self.pymongo = None
15
16		def setupRepo(self):
17		from niprov.mongo import MongoRepository
18		with patch('niprov.mongo.pymongo') as self.pymongo:
19		self.repo = MongoRepository(dependencies=self.dependencies)
20		self.repo.db = self.db
21
22		def test_Connection(self):
23		from niprov.mongo import MongoRepository
24		with patch('niprov.mongo.pymongo') as pymongo:
25		repo = MongoRepository(dependencies=self.dependencies)
26		pymongo.MongoClient.assert_called_with(self.config.database_url)
27		self.assertEqual(pymongo.MongoClient().get_default_database(), repo.db)
28
29		def test_byLocation_returns_img_from_record_with_path(self):
30		self.setupRepo()
31		p = '/p/f1'
32		out = self.repo.byLocation(p)
33		self.db.provenance.find_one.assert_called_with({'location':p})
34		self.fileFactory.fromProvenance.assert_called_with(
35		self.db.provenance.find_one())
36		self.assertEqual(self.fileFactory.fromProvenance(), out)
37
38		def test_getSeries(self):
39		self.setupRepo()
40		img = Mock()
41		out = self.repo.getSeries(img)
42		self.db.provenance.find_one.assert_called_with(
43		{'seriesuid':img.getSeriesId()})
44		self.fileFactory.fromProvenance.assert_called_with(
45		self.db.provenance.find_one())
46		self.assertEqual(self.fileFactory.fromProvenance(), out)
47
48		def test_getSeries_returns_None_right_away_if_no_series_id(self):
49		self.setupRepo()
50		img = Mock()
51		img.getSeriesId.return_value = None
52		out = self.repo.getSeries(img)
53		assert not self.db.provenance.find_one.called
54		self.assertEqual(None, out)
55
56		def test_Add(self):
57		self.setupRepo()
58		img = Mock()
59		img.provenance = {'a':1, 'b':2}
60		self.repo.add(img)
61		self.db.provenance.insert_one.assert_called_with({'a':1, 'b':2})
62
63		def test_update(self):
64		self.setupRepo()
65		img = Mock()
66		img.provenance = {'a':1, 'b':2}
67		self.repo.update(img)
68		self.db.provenance.update.assert_called_with(
69		{'location':img.location.toString()}, {'a':1, 'b':2})
70
71		def test_all(self):
72		self.fileFactory.fromProvenance.side_effect = lambda p: 'img_'+p
73		self.db.provenance.find.return_value = ['p1', 'p2']
74		self.setupRepo()
75		out = self.repo.all()
76		self.db.provenance.find.assert_called_with()
77		self.fileFactory.fromProvenance.assert_any_call('p1')
78		self.fileFactory.fromProvenance.assert_any_call('p2')
79		self.assertEqual(['img_p1', 'img_p2'], out)
80
81		def test_updateApproval(self):
82		self.setupRepo()
83		img = Mock()
84	View Code Duplication	p = '/p/f1'
		0 ignored issues – show Duplication introduced 2016-07-04 18:53 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
85		newStatus = 'oh-oh'
86		self.repo.updateApproval(p, newStatus)
87		self.db.provenance.update.assert_called_with(
88		{'location':p}, {'$set': {'approval': newStatus}})
89
90		def test_latest(self):
91		self.fileFactory.fromProvenance.side_effect = lambda p: 'img_'+p
92		self.db.provenance.find.return_value = Mock()
93		self.db.provenance.find.return_value.sort.return_value.limit.return_value = ['px','py']
94		self.setupRepo()
95		out = self.repo.latest()
96		self.db.provenance.find.assert_called_with()
97		self.db.provenance.find().sort.assert_called_with('added', -1)
98		self.db.provenance.find().sort().limit.assert_called_with(20)
99		self.fileFactory.fromProvenance.assert_any_call('px')
100		self.fileFactory.fromProvenance.assert_any_call('py')
101		self.assertEqual(['img_px', 'img_py'], out)
102
103		def test_statistics(self):
104		self.db.provenance.aggregate.return_value = [sentinel.stats,]
105		self.setupRepo()
106		out = self.repo.statistics()
107		self.db.provenance.aggregate.assert_called_with(
108		[{'$group':
109		{
110		'_id': None,
111		'totalsize': { '$sum': '$size' },
112		'count': { '$sum': 1 }
113		}
114		}])
115		self.assertEqual(sentinel.stats, out)
116
117		def test_statistics_if_no_records(self):
118		self.db.provenance.aggregate.return_value = []
119		self.setupRepo()
120		out = self.repo.statistics()
121		self.assertEqual({'count':0}, out)
122
123		def test_byId(self):
124		self.setupRepo()
125		ID = 'abc123'
126		out = self.repo.byId(ID)
127		self.db.provenance.find_one.assert_called_with({'id':ID})
128		self.fileFactory.fromProvenance.assert_called_with(
129		self.db.provenance.find_one())
130		self.assertEqual(self.fileFactory.fromProvenance(), out)
131
132		def test_If_db_returns_None_should_return_None_for_byId_byLocation(self):
133		self.setupRepo()
134		self.db.provenance.find_one.return_value = None
135		out = self.repo.byId('abc123')
136		assert not self.fileFactory.fromProvenance.called
137		self.assertEqual(None, out)
138
139		def test_byLocations(self):
140		self.fileFactory.fromProvenance.side_effect = lambda p: 'img_'+p
141		self.db.provenance.find.return_value = ['p1', 'p2']
142		self.setupRepo()
143		out = self.repo.byLocations(['l1','l2'])
144		self.db.provenance.find.assert_called_with({'location':{'$in':['l1','l2']}})
145		self.fileFactory.fromProvenance.assert_any_call('p1')
146		self.fileFactory.fromProvenance.assert_any_call('p2')
147		self.assertEqual(['img_p1', 'img_p2'], out)
148
149		def test_byParents(self):
150		self.fileFactory.fromProvenance.side_effect = lambda p: 'img_'+p
151		self.db.provenance.find.return_value = ['p1', 'p2']
152		self.setupRepo()
153		out = self.repo.byParents(['x1','x2'])
154		self.db.provenance.find.assert_called_with({'parents':{'$in':['x1','x2']}})
155		self.fileFactory.fromProvenance.assert_any_call('p1')
156		self.fileFactory.fromProvenance.assert_any_call('p2')
157		self.assertEqual(['img_p1', 'img_p2'], out)
158
159		def test_Obtains_optional_snapshot_data_from_cache_when_serializing(self):
160		self.pictureCache.getBytes.return_value = sentinel.snapbytes
161		with patch('niprov.mongo.bson') as bson:
162		bson.Binary.return_value = sentinel.snapbson
163		self.setupRepo()
164		img = Mock()
165		img.provenance = {'a':1}
166		self.repo.add(img)
167	View Code Duplication	self.pictureCache.getBytes.assert_called_with(for_=img)
		0 ignored issues – show Duplication introduced 2016-07-04 18:53 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
168		bson.Binary.assert_called_with(sentinel.snapbytes)
169		self.db.provenance.insert_one.assert_called_with({'a':1,
170		'_snapshot-data':sentinel.snapbson})
171
172		def test_If_no_snapshot_doesnt_add_data_field(self):
173		self.pictureCache.getBytes.return_value = None
174		with patch('niprov.mongo.bson') as bson:
175		self.setupRepo()
176		img = Mock()
177	View Code Duplication	img.provenance = {'a':1}
		0 ignored issues – show Duplication introduced 2016-07-04 18:53 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
178		self.repo.add(img)
179		assert not bson.Binary.called
180		self.db.provenance.insert_one.assert_called_with({'a':1})
181
182		def test_If_snapshotdata_hands_them_to_pictureCache_on_deserializing(self):
183		img = Mock()
184		self.fileFactory.fromProvenance.return_value = img
185		self.setupRepo()
186		self.db.provenance.find_one.return_value = {'a':3}
187		out = self.repo.byLocation('/p/f1')
188		assert not self.pictureCache.keep.called
189		self.db.provenance.find_one.return_value = {'a':3,
190		'_snapshot-data':'y7yUyS'}
191		out = self.repo.byLocation('/p/f1')
192		self.pictureCache.keep.assert_called_with('y7yUyS', for_=img)
193
194		def test_Query(self):
195		self.db.provenance.find.return_value = ['record1']
196		self.setupRepo()
197		q = Mock()
198		field1 = Mock()
199		field1.name = 'color'
200		field1.value = 'red'
201		field1.all = False
202		q.getFields.return_value = [field1]
203		out = self.repo.inquire(q)
204		self.db.provenance.find.assert_called_with({'color':'red'})
205		self.fileFactory.fromProvenance.assert_called_with('record1')
206
207		def test_Ensures_text_index_for_search(self):
208		self.setupRepo()
209		self.repo.search('')
210		searchfields = ['location','user','subject','project','protocol',
211		'transformation','technique','modality']
212		indexspec = [(field, 'text') for field in searchfields]
213		self.db.provenance.create_index.assert_called_with(indexspec,
214		name='textsearch')
215
216		def test_Search(self):
217		self.db.provenance.find.return_value = ['r1','r2']
218		self.setupRepo()
219		self.repo.search('xyz')
220		self.db.provenance.find.assert_called_with({'$text':{'$search': 'xyz'}})
221		self.fileFactory.fromProvenance.assert_any_call('r1')
222		self.fileFactory.fromProvenance.assert_any_call('r2')
223
224		def test_Query_for_ALL_field(self):
225		self.db.provenance.distinct.return_value = ['r1','r2']
226		self.setupRepo()
227		q = Mock()
228		field1 = Mock()
229		field1.name = 'color'
230		field1.all = True
231		q.getFields.return_value = [field1]
232		out = self.repo.inquire(q)
233		self.db.provenance.distinct.assert_called_with('color')
234		assert not self.fileFactory.fromProvenance.called
235
236

ilogue / niprov

test_Ensures_text_index_for_search() A last analyzed 2016-11-17 03:25 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

test_Ensures_text_index_for_search() A
last analyzed 2016-11-17 03:25 UTC