Passed
Pull Request — master (#35)
by Paolo
02:58
created

cryoweb.tests.test_helpers.WebSocketMixin.setUp()   A

Complexity

Conditions 1

Size

Total Lines 17
Code Lines 10

Duplication

Lines 17
Ratio 100 %

Importance

Changes 0
Metric Value
eloc 10
dl 17
loc 17
rs 9.9
c 0
b 0
f 0
cc 1
nop 1
1
#!/usr/bin/env python3
2
# -*- coding: utf-8 -*-
3
"""
4
Created on Thu Sep 20 16:01:04 2018
5
6
@author: Paolo Cozzi <[email protected]>
7
"""
8
9
# --- import
10
from unittest.mock import patch, Mock
11
12
from django.conf import settings
13
from django.core.management import call_command
14
from django.test import TestCase
15
16
from common.constants import ERROR
17
from language.models import SpecieSynonym
18
from image_app.models import (
19
    Submission, DictBreed, Name, Animal, Sample, DictSex,
20
    DictCountry, DictSpecie)
21
from common.tests import DataSourceMixinTestCase
22
23
from ..helpers import (
24
    upload_cryoweb, check_species, CryoWebImportError, cryoweb_import,
25
    check_UID)
26
from ..models import db_has_data, truncate_database, BreedsSpecies
27
28
29
class BaseTestCase():
30
    # import this file and populate database once
31
    fixtures = [
32
        'cryoweb/dictbreed',
33
        'image_app/dictcountry',
34
        'image_app/dictrole',
35
        'image_app/dictsex',
36
        'image_app/organization',
37
        'image_app/submission',
38
        'image_app/user',
39
        'language/dictspecie',
40
        'language/speciesynonym'
41
    ]
42
43
    # By default, fixtures are only loaded into the default database. If you
44
    # are using multiple databases and set multi_db=True, fixtures will be
45
    # loaded into all databases. However, this will raise problems when
46
    # managing extended user models
47
    multi_db = False
48
49
    def setUp(self):
50
        # calling my base class setup
51
        super().setUp()
52
53
        # track submission
54
        self.submission = Submission.objects.get(pk=1)
55
56
57
class CryoWebMixin(object):
58
    """Custom methods to upload cryoweb data into database for testing"""
59
60
    @classmethod
61
    def setUpClass(cls):
62
        # calling my base class setup
63
        super().setUpClass()
64
65
        # this fixture have to be loaded in a secondary (test) database,
66
        # I can't upload it using names and fixture section, so it will
67
        # be added manually using loaddata
68
        call_command(
69
            'loaddata',
70
            'cryoweb/cryoweb.json',
71
            app='cryoweb',
72
            database='cryoweb',
73
            verbosity=0)
74
75
    @classmethod
76
    def tearDownClass(cls):
77
        # truncate cryoweb database after loading
78
        if db_has_data():
79
            truncate_database()
80
81
        # calling my base class teardown class
82
        super().tearDownClass()
83
84
    def check_data_imported(self):
85
        """A method to test if data were imported from cryoweb or not"""
86
87
        # check breed upload
88
        queryset = DictBreed.objects.all()
89
90
        breeds = [(dictbreed.supplied_breed, dictbreed.country.label)
91
                  for dictbreed in queryset]
92
93
        self.assertEqual(len(queryset), 4)
94
        self.assertListEqual(
95
            breeds, [
96
                ('Bunte Bentheimer', 'United Kingdom'),
97
                ('Ostfriesisches Milchschaf', 'Italy'),
98
                ('Aberdeen Angus', 'Germany'),
99
                ('Ostfriesisches Milchschaf', 'Germany')],
100
            msg="Check breeds loaded")
101
102
        # check name upload (5 animal, 1 sample)
103
        queryset = Name.objects.filter(submission=self.submission)
104
        self.assertEqual(len(queryset), 6, msg='check name load')
105
106
        # check animal name
107
        queryset = Animal.objects.all()
108
        self.assertEqual(len(queryset), 3, msg="check animal load")
109
110
        queryset = Sample.objects.all()
111
        self.assertEqual(len(queryset), 1, msg="check sample load")
112
113
114 View Code Duplication
class WebSocketMixin(object):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
115
    """Override setUp to mock websocket objects"""
116
117
    def setUp(self):
118
        # calling my base class setup
119
        super().setUp()
120
121
        # setting channels methods
122
        self.asyncio_mock_patcher = patch(
123
            'asyncio.get_event_loop')
124
        self.asyncio_mock = self.asyncio_mock_patcher.start()
125
126
        # mocking asyncio return value
127
        self.run_until = self.asyncio_mock.return_value
128
        self.run_until.run_until_complete = Mock()
129
130
        # another patch
131
        self.send_msg_ws_patcher = patch(
132
            'cryoweb.helpers.send_message_to_websocket')
133
        self.send_msg_ws = self.send_msg_ws_patcher.start()
134
135
    def tearDown(self):
136
        # stopping mock objects
137
        self.asyncio_mock_patcher.stop()
138
        self.send_msg_ws_patcher.stop()
139
140
        # calling base methods
141
        super().tearDown()
142
143
    def check_message(
144
            self, message, notification_message, validation_message=None,
145
            pk=1):
146
        """assert message to websocket called with parameters"""
147
148
        # construct message according parameters
149
        message = {
150
            'message': message,
151
            'notification_message': notification_message
152
        }
153
154
        # in case of successful data upload, a validation message is sent
155
        if validation_message:
156
            message['validation_message'] = validation_message
157
158
        self.assertEqual(self.asyncio_mock.call_count, 1)
159
        self.assertEqual(self.run_until.run_until_complete.call_count, 1)
160
        self.assertEqual(self.send_msg_ws.call_count, 1)
161
        self.send_msg_ws.assert_called_with(
162
            message,
163
            pk)
164
165
166
class CheckSpecie(CryoWebMixin, BaseTestCase, TestCase):
167
    def test_check_species(self):
168
        """Testing species and synonyms"""
169
170
        united_kingdom = DictCountry.objects.get(label="United Kingdom")
171
172
        # test for species synonoms for cryoweb
173
        self.assertTrue(check_species(united_kingdom))
174
175
        # now delete a synonym
176
        synonym = SpecieSynonym.objects.get(
177
            language__label='United Kingdom',
178
            word='Cattle')
179
        synonym.delete()
180
181
        self.assertFalse(check_species(united_kingdom))
182
183
    def test_no_species(self):
184
        """Test no species in cryoweb database"""
185
186
        queryset = BreedsSpecies.objects.all()
187
        queryset.delete()
188
189
        self.assertRaisesRegex(
190
            CryoWebImportError,
191
            "You have no species",
192
            check_species, "United Kingdom")
193
194
195
class CheckBreed(TestCase):
196
    # import this file and populate database once
197
    fixtures = [
198
        'image_app/dictbreed',
199
        'image_app/dictcountry',
200
        'image_app/dictrole',
201
        'image_app/dictsex',
202
        'image_app/dictspecie',
203
        'image_app/organization',
204
        'image_app/submission',
205
        'image_app/user',
206
    ]
207
208
    def test_add_breed(self):
209
        italy = DictCountry.objects.get(label="Italy")
210
        united_kingdom = DictCountry.objects.get(label="United Kingdom")
211
212
        sus = DictSpecie.objects.get(label="Sus scrofa")
213
214
        # inserting an already present breed get the object without creation
215
        breed, created = DictBreed.objects.get_or_create(
216
            supplied_breed="Bunte Bentheimer",
217
            specie=sus,
218
            country=united_kingdom)
219
220
        self.assertFalse(created)
221
222
        # inserting a breed in a different country add a record
223
        breed, created = DictBreed.objects.get_or_create(
224
            supplied_breed="Bunte Bentheimer",
225
            specie=sus,
226
            country=italy)
227
228
        self.assertTrue(created)
229
230
231
class CheckUIDTest(CryoWebMixin, BaseTestCase, TestCase):
232
    def test_empty_dictsex(self):
233
        """Empty dictsex and check that I can't proceed"""
234
235
        queryset = DictSex.objects.all()
236
        queryset.delete()
237
238
        self.assertRaisesRegex(
239
            CryoWebImportError,
240
            "You have to upload DictSex data",
241
            check_UID, self.submission)
242
243
    def test_no_synonym(self):
244
        # now delete a synonym
245
        synonym = SpecieSynonym.objects.get(
246
            language__label='United Kingdom',
247
            word='Cattle')
248
        synonym.delete()
249
250
        self.assertRaisesRegex(
251
            CryoWebImportError,
252
            "Some species haven't a synonym!",
253
            check_UID, self.submission)
254
255
    def test_check_UID(self):
256
        """testing normal behaviour"""
257
258
        self.assertTrue(check_UID(self.submission))
259
260
261
class UploadCryoweb(
262
        WebSocketMixin, DataSourceMixinTestCase, BaseTestCase, TestCase):
263
    """Test upload cryoweb dump into cryoweb database"""
264
265
    # define attribute in DataSourceMixinTestCase
266
    model = Submission
267
268
    # need to clean database after testing import. Can't use CryowebMixin
269
    # since i need to test cryoweb import
270
    def tearDown(self):
271
        """Clean test database after modifications"""
272
273
        # truncate cryoweb database after loading
274
        if db_has_data():
275
            truncate_database()
276
277
        # calling my base class teardown class
278
        super().tearDown()
279
280
    def test_database_name(self):
281
        self.assertEqual(
282
            settings.DATABASES['cryoweb']['NAME'], 'test_cryoweb')
283
284
    def test_upload_cryoweb(self):
285
        """Testing uploading and importing data from cryoweb to UID"""
286
287
        self.assertTrue(upload_cryoweb(self.submission.id))
288
        self.assertTrue(db_has_data())
289
290
        # if I try again to upload cryoweb, I will get a False object and
291
        # submission message
292
        self.assertRaises(
293
            CryoWebImportError,
294
            upload_cryoweb,
295
            self.submission.id)
296
297
        # reload submission
298
        self.submission = Submission.objects.get(pk=1)
299
300
        self.assertEqual(
301
            self.submission.status,
302
            ERROR)
303
304
        self.assertIn(
305
            "Cryoweb has data",
306
            self.submission.message)
307
308
        # check async message called
309
        message = 'Error'
310
        notification_message = 'Error in importing data: Cryoweb has data'
311
312
        self.check_message(message, notification_message)
313
314
    # mock subprocess.run and raise Exception. Read it and update submission
315
    # message using helpers.upload_cryoweb
316
    def test_upload_cryoweb_errors(self):
317
        """Testing errors in uploading cryoweb data"""
318
319
        with patch('subprocess.run') as runMock:
320
            runMock.side_effect = Exception("Test upload failed")
321
            self.assertFalse(upload_cryoweb(self.submission.id))
322
323
            # reload submission
324
            self.submission = Submission.objects.get(pk=1)
325
326
            self.assertEqual(
327
                self.submission.status,
328
                ERROR)
329
330
            self.assertIn(
331
                "Test upload failed",
332
                self.submission.message)
333
334
            # check async message called
335
            message = 'Error'
336
            notification_message = ('Error in importing data: Test '
337
                                    'upload failed')
338
339
            self.check_message(message, notification_message)
340
341
342
class CryowebImport(WebSocketMixin, CryoWebMixin, BaseTestCase, TestCase):
343
    def test_database_name(self):
344
        self.assertEqual(
345
            settings.DATABASES['cryoweb']['NAME'], 'test_cryoweb')
346
347 View Code Duplication
    def test_cryoweb_import(self):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
348
        """Import from cryoweb staging database into UID"""
349
350
        self.assertTrue(cryoweb_import(self.submission))
351
352
        # test if data were loaded
353
        self.check_data_imported()
354
355
        # check async message called
356
        message = 'Loaded'
357
        notification_message = (
358
            'Cryoweb import completed for submission: 1')
359
        validation_message = {
360
            'animals': 3, 'samples': 1,
361
            'animal_unkn': 3, 'sample_unkn': 1,
362
            'animal_issues': 0, 'sample_issues': 0}
363
364
        self.check_message(message, notification_message, validation_message)
365
366
    @patch("cryoweb.helpers.check_UID", side_effect=Exception("Test message"))
367
    def test_cryoweb_import_errors(self, my_check):
368
        """Testing importing with data into UID with errors"""
369
370
        self.assertFalse(cryoweb_import(self.submission))
371
        self.assertTrue(my_check.called)
372
373
        # reload submission
374
        self.submission = Submission.objects.get(pk=1)
375
376
        self.assertEqual(
377
            self.submission.status,
378
            ERROR)
379
380
        self.assertIn(
381
            "Test message",
382
            self.submission.message)
383
384
        # check async message called
385
        message = 'Error'
386
        notification_message = (
387
            'Error in importing data: Test message')
388
389
        self.check_message(message, notification_message)
390
391
392
class CryowebReload(WebSocketMixin, CryoWebMixin, BaseTestCase, TestCase):
393
    """Simulate a cryoweb reload case. Load data as in CryowebImport, then
394
    call test which reload the same data"""
395
396
    # override fixtures
397
    fixtures = [
398
        'cryoweb/auth',
399
        'cryoweb/dictbreed',
400
        'cryoweb/image_app',
401
        'language/dictspecie',
402
        'language/speciesynonym'
403
    ]
404
405 View Code Duplication
    def test_cryoweb_import(self):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
406
        """Import from cryoweb staging database into UID, anche check that
407
        validationsummary is correct"""
408
409
        self.assertTrue(cryoweb_import(self.submission))
410
411
        # test if data were loaded
412
        self.check_data_imported()
413
414
        # check async message called
415
        message = 'Loaded'
416
        notification_message = (
417
            'Cryoweb import completed for submission: 1')
418
        validation_message = {
419
            'animals': 3, 'samples': 1,
420
            'animal_unkn': 3, 'sample_unkn': 1,
421
            'animal_issues': 0, 'sample_issues': 0}
422
423
        self.check_message(message, notification_message, validation_message)
424