Completed
Push — master ( 4f7ee6...646424 )
by Paolo
08:30 queued 06:53
created

CryowebUpdate.test_cryoweb_import()   A

Complexity

Conditions 1

Size

Total Lines 48
Code Lines 33

Duplication

Lines 0
Ratio 0 %

Importance

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