CryoWebMixin.setUpClass()   B
last analyzed

Complexity

Conditions 7

Size

Total Lines 39
Code Lines 29

Duplication

Lines 0
Ratio 0 %

Importance

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