MongoRepoTests.test_bySubject() - Code Metrics - Inspection of "Browse and export by project, user, modality" - ilogue/niprov - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#143)

by Jasper

created 2016-06-14 23:53 UTC

MongoRepoTests.test_bySubject() A

↳ Parent: MongoRepoTests

Complexity

Conditions

Size

Total Lines

Duplication

Lines	9
Ratio	90 %

Importance

Changes

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

import unittest
from mock import Mock, patch, sentinel
from datetime import timedelta
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 = {}

    def setupRepo(self):
        from niprov.mongo import MongoRepository
        with patch('niprov.mongo.pymongo') as 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_knowsByLocation(self):
        self.setupRepo()
        p = '/p/f1'
        self.db.provenance.find_one.return_value = None
        self.assertFalse(self.repo.knowsByLocation(p))
        self.db.provenance.find_one.assert_called_with({'location':p})
        self.db.provenance.find_one.return_value = 1
        self.assertTrue(self.repo.knowsByLocation(p))

    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_knowsSeries_returns_False_if_no_series_id(self):
        self.setupRepo()
        img = Mock()
        img.getSeriesId.return_value = None
        self.assertFalse(self.repo.knowsSeries(img))

    def test_knowsSeries(self):
        self.setupRepo()
        img = Mock()
        self.assertTrue(self.repo.knowsSeries(img))
        self.db.provenance.find_one.return_value = None
        self.assertFalse(self.repo.knowsSeries(img))

    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_byApproval(self):
        self.fileFactory.fromProvenance.side_effect = lambda p: 'img_'+p
        self.db.provenance.find.return_value = ['p1', 'p2']
        self.setupRepo()
        a = 'AOk'
        out = self.repo.byApproval(a)
        self.db.provenance.find.assert_called_with({'approval':a})
        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_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_Converts_timedelta_to_float_when_serializing(self):

        self.setupRepo()
        img = Mock()
        img.provenance = {'a':1, 'duration':timedelta(seconds=67.89)}
        self.repo.add(img)
        self.db.provenance.insert_one.assert_called_with(
            {'a':1, 'duration':67.89})

    def test_Converts_duration_to_timedelta_when_deserializing(self):
        self.setupRepo()
        self.db.provenance.find_one.return_value = {'a':3, 'duration':89.01}
        out = self.repo.byLocation('/p/f1')
        self.fileFactory.fromProvenance.assert_called_with(
            {'a':3, 'duration':timedelta(seconds=89.01)})

    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_If_no_record_returned_byLocation_byId_getSeries_raise_alarm(self):
        self.setupRepo()
        self.db.provenance.find_one.return_value = None
        out = self.repo.byLocation('nowhere')
        self.listener.unknownFile.assert_called_with('nowhere')
        out = self.repo.byId('xxxx')
        self.listener.unknownFile.assert_called_with('id: xxxx')
        img = Mock()
        img.getSeriesId.return_value = '123abc'
        out = self.repo.getSeries(img)
        self.listener.unknownFile.assert_called_with('seriesuid: 123abc')

    def test_Query(self):
        self.db.provenance.find.return_value = ['record1']
        self.setupRepo()
        q = Mock()
        field1 = Mock()
        field1.name = 'color'
        field1.value = 'red'
        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')




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

ilogue / niprov

Pull Request — master (#143)

MongoRepoTests.test_bySubject() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like