Completed
Pull Request — devel (#74)
by Paolo
07:56
created

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

Complexity

Conditions 1

Size

Total Lines 6
Code Lines 3

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
eloc 3
dl 0
loc 6
rs 10
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
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
                "Test upload failed",
362
                self.submission.message)
363
364
            # check async message called
365
            message = 'Error'
366
            notification_message = ('Error in importing data: Test '
367
                                    'upload failed')
368
369
            self.check_message(message, notification_message)
370
371
372
class CryowebImport(
373
        ImportMixin, WebSocketMixin, CryoWebMixin, TestCase):
374
    def test_database_name(self):
375
        self.assertEqual(
376
            settings.DATABASES['cryoweb']['NAME'], 'test_cryoweb')
377
378
    @patch("cryoweb.helpers.check_UID", side_effect=Exception("Test message"))
379
    def test_cryoweb_import_errors(self, my_check):
380
        """Testing importing with data into UID with errors"""
381
382
        self.assertFalse(cryoweb_import(self.submission))
383
        self.assertTrue(my_check.called)
384
385
        # reload submission
386
        self.submission = Submission.objects.get(pk=1)
387
388
        self.assertEqual(
389
            self.submission.status,
390
            ERROR)
391
392
        self.assertIn(
393
            "Test message",
394
            self.submission.message)
395
396
        # check async message called
397
        message = 'Error'
398
        notification_message = (
399
            'Error in importing data: Test message')
400
401
        self.check_message(message, notification_message)
402
403
404
class CryowebReload(
405
        ImportMixin, WebSocketMixin, CryoWebMixin, TestCase):
406
    """Simulate a cryoweb reload case. Load data as in CryowebImport, then
407
    call test which reload the same data"""
408
409
    # change fixtures in order to upload data to different databases
410
    fixtures = {
411
        'cryoweb': [
412
            'cryoweb/cryoweb',
413
        ],
414
        'default': [
415
            'cryoweb/auth',
416
            'cryoweb/dictbreed',
417
            'cryoweb/uid',
418
            'language/dictspecie',
419
            'language/speciesynonym'
420
        ]
421
    }
422
423
424
class CryowebUpdate(
425
        ImportMixin, WebSocketMixin, CryoWebMixin, TestCase):
426
    """Simulate a cryoweb update with the same dataset. Data already
427
    present will be ignored"""
428
429
    # change fixtures in order to upload data to different databases
430
    fixtures = {
431
        'cryoweb': [
432
            'cryoweb/cryoweb',
433
        ],
434
        'default': [
435
            'cryoweb/auth',
436
            'cryoweb/dictbreed',
437
            'cryoweb/uid',
438
            'language/dictspecie',
439
            'language/speciesynonym'
440
        ]
441
    }
442
443
    def setUp(self):
444
        # calling my base class setup
445
        super().setUp()
446
447
        # track old submission
448
        self.old_submission = Submission.objects.get(pk=1)
449
450
        # generate a new submission from old submission object
451
        submission = self.submission
452
        submission.pk = None
453
        submission.title = "Updated database"
454
        submission.datasource_version = "Updated database"
455
        submission.save()
456
457
        # track the new submission
458
        self.submission = submission
459
460
        # now delete an animal and its associated sample
461
        sample = Sample.objects.get(pk=1)
462
        animal = sample.animal
463
        animal.delete()
464
465
    # override ImportMixin.test_cryoweb_import
466
    def test_cryoweb_import(self):
467
        """A method to test if data were imported from cryoweb or not"""
468
469
        self.assertTrue(cryoweb_import(self.submission))
470
471
        # check breed upload
472
        queryset = DictBreed.objects.all()
473
474
        breeds = [(dictbreed.supplied_breed, dictbreed.country.label)
475
                  for dictbreed in queryset]
476
477
        self.assertEqual(len(queryset), 4)
478
        self.assertListEqual(
479
            breeds, [
480
                ('Bunte Bentheimer', 'United Kingdom'),
481
                ('Ostfriesisches Milchschaf', 'Italy'),
482
                ('Aberdeen Angus', 'Germany'),
483
                ('Ostfriesisches Milchschaf', 'Germany')],
484
            msg="Check breeds loaded")
485
486
        # check animal and samples
487
        queryset = Animal.objects.all()
488
        self.assertEqual(len(queryset), 3, msg="check animal load")
489
490
        queryset = Sample.objects.all()
491
        self.assertEqual(len(queryset), 1, msg="check sample load")
492
493
        # assert data are in the proper submission
494
        self.assertEqual(self.old_submission.animal_set.count(), 2)
495
        self.assertEqual(self.old_submission.sample_set.count(), 0)
496
497
        self.assertEqual(self.submission.animal_set.count(), 1)
498
        self.assertEqual(self.submission.sample_set.count(), 1)
499
500
        # check async message called
501
        message = 'Loaded'
502
        notification_message = (
503
            'Cryoweb import completed for submission: 2')
504
        validation_message = {
505
            'animals': 1, 'samples': 1,
506
            'animal_unkn': 1, 'sample_unkn': 1,
507
            'animal_issues': 0, 'sample_issues': 0}
508
509
        self.check_message(
510
            message,
511
            notification_message,
512
            validation_message,
513
            pk=self.submission.id)
514