Completed
Branch master (206998)
by Paolo
02:15 queued 15s
created

CryowebImport.test_cryoweb_import()   A

Complexity

Conditions 1

Size

Total Lines 45
Code Lines 32

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
eloc 32
dl 0
loc 45
rs 9.112
c 0
b 0
f 0
cc 1
nop 3
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 image_app.models import (
20
    Submission, DictBreed, Name, Animal, Sample, DictSex,
21
    DictCountry, DictSpecie)
22
from common.tests import DataSourceMixinTestCase
23
24
from ..helpers import (
25
    upload_cryoweb, check_species, CryoWebImportError, cryoweb_import,
26
    check_UID)
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
        'image_app/dictcountry',
35
        'image_app/dictrole',
36
        'image_app/dictsex',
37
        'image_app/organization',
38
        'image_app/submission',
39
        'image_app/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 name upload (5 animal, 1 sample)
110
        queryset = Name.objects.filter(submission=self.submission)
111
        self.assertEqual(len(queryset), 6, msg='check name load')
112
113
        # check animal name
114
        queryset = Animal.objects.all()
115
        self.assertEqual(len(queryset), 3, msg="check animal load")
116
117
        queryset = Sample.objects.all()
118
        self.assertEqual(len(queryset), 1, msg="check sample load")
119
120
        # check async message called
121
        message = 'Loaded'
122
        notification_message = (
123
            'Cryoweb import completed for submission: 1')
124
        validation_message = {
125
            'animals': 3, 'samples': 1,
126
            'animal_unkn': 3, 'sample_unkn': 1,
127
            'animal_issues': 0, 'sample_issues': 0}
128
129
        self.check_message(message, notification_message, validation_message)
130
131
132
class CheckSpecie(CryoWebMixin, BaseMixin, TestCase):
133
    def test_check_species(self):
134
        """Testing species and synonyms"""
135
136
        united_kingdom = DictCountry.objects.get(label="United Kingdom")
137
138
        # test for species synonoms for cryoweb
139
        self.assertTrue(check_species(united_kingdom))
140
141
        # now delete a synonym
142
        synonym = SpecieSynonym.objects.get(
143
            language__label='United Kingdom',
144
            word='Cattle')
145
        synonym.delete()
146
147
        self.assertFalse(check_species(united_kingdom))
148
149
    def test_no_species(self):
150
        """Test no species in cryoweb database"""
151
152
        queryset = BreedsSpecies.objects.all()
153
        queryset.delete()
154
155
        self.assertRaisesRegex(
156
            CryoWebImportError,
157
            "You have no species",
158
            check_species, "United Kingdom")
159
160
161
class CheckBreed(TestCase):
162
    # import this file and populate database once
163
    fixtures = [
164
        'image_app/dictbreed',
165
        'image_app/dictcountry',
166
        'image_app/dictrole',
167
        'image_app/dictsex',
168
        'image_app/dictspecie',
169
        'image_app/organization',
170
        'image_app/submission',
171
        'image_app/user',
172
    ]
173
174
    def test_add_breed(self):
175
        italy = DictCountry.objects.get(label="Italy")
176
        united_kingdom = DictCountry.objects.get(label="United Kingdom")
177
178
        sus = DictSpecie.objects.get(label="Sus scrofa")
179
180
        # inserting an already present breed get the object without creation
181
        breed, created = DictBreed.objects.get_or_create(
182
            supplied_breed="Bunte Bentheimer",
183
            specie=sus,
184
            country=united_kingdom)
185
186
        self.assertFalse(created)
187
188
        # inserting a breed in a different country add a record
189
        breed, created = DictBreed.objects.get_or_create(
190
            supplied_breed="Bunte Bentheimer",
191
            specie=sus,
192
            country=italy)
193
194
        self.assertTrue(created)
195
196
197
class CheckUIDTest(CryoWebMixin, BaseMixin, TestCase):
198
    def test_empty_dictsex(self):
199
        """Empty dictsex and check that I can't proceed"""
200
201
        queryset = DictSex.objects.all()
202
        queryset.delete()
203
204
        self.assertRaisesRegex(
205
            CryoWebImportError,
206
            "You have to upload DictSex data",
207
            check_UID, self.submission)
208
209
    def test_no_synonym(self):
210
        # now delete a synonym
211
        synonym = SpecieSynonym.objects.get(
212
            language__label='United Kingdom',
213
            word='Cattle')
214
        synonym.delete()
215
216
        self.assertRaisesRegex(
217
            CryoWebImportError,
218
            "Some species haven't a synonym!",
219
            check_UID, self.submission)
220
221
    def test_check_UID(self):
222
        """testing normal behaviour"""
223
224
        self.assertTrue(check_UID(self.submission))
225
226
227
class UploadCryoweb(
228
        WebSocketMixin, DataSourceMixinTestCase, BaseMixin, TestCase):
229
    """Test upload cryoweb dump into cryoweb database"""
230
231
    # define attribute in DataSourceMixinTestCase
232
    model = Submission
233
234
    # need to clean database after testing import. Can't use CryowebMixin
235
    # since i need to test cryoweb import
236
    def tearDown(self):
237
        """Clean test database after modifications"""
238
239
        # truncate cryoweb database after loading
240
        if db_has_data():
241
            truncate_database()
242
243
        # calling my base class teardown class
244
        super().tearDown()
245
246
    def test_database_name(self):
247
        self.assertEqual(
248
            settings.DATABASES['cryoweb']['NAME'], 'test_cryoweb')
249
250
    def test_upload_cryoweb(self):
251
        """Testing uploading and importing data from cryoweb to UID"""
252
253
        self.assertTrue(upload_cryoweb(self.submission.id))
254
        self.assertTrue(db_has_data())
255
256
        # if I try again to upload cryoweb, I will get a False object and
257
        # submission message
258
        self.assertRaises(
259
            CryoWebImportError,
260
            upload_cryoweb,
261
            self.submission.id)
262
263
        # reload submission
264
        self.submission = Submission.objects.get(pk=1)
265
266
        self.assertEqual(
267
            self.submission.status,
268
            ERROR)
269
270
        self.assertIn(
271
            "Cryoweb has data",
272
            self.submission.message)
273
274
        # check async message called
275
        message = 'Error'
276
        notification_message = 'Error in importing data: Cryoweb has data'
277
278
        self.check_message(message, notification_message)
279
280
    # mock subprocess.run and raise Exception. Read it and update submission
281
    # message using helpers.upload_cryoweb
282
    def test_upload_cryoweb_errors(self):
283
        """Testing errors in uploading cryoweb data"""
284
285
        with patch('subprocess.run') as runMock:
286
            runMock.side_effect = Exception("Test upload failed")
287
            self.assertFalse(upload_cryoweb(self.submission.id))
288
289
            # reload submission
290
            self.submission = Submission.objects.get(pk=1)
291
292
            self.assertEqual(
293
                self.submission.status,
294
                ERROR)
295
296
            self.assertIn(
297
                "Test upload failed",
298
                self.submission.message)
299
300
            # check async message called
301
            message = 'Error'
302
            notification_message = ('Error in importing data: Test '
303
                                    'upload failed')
304
305
            self.check_message(message, notification_message)
306
307
308
class CryowebImport(ImportMixin, BaseMixin, TestCase):
309
    def test_database_name(self):
310
        self.assertEqual(
311
            settings.DATABASES['cryoweb']['NAME'], 'test_cryoweb')
312
313
    @patch("cryoweb.helpers.check_UID", side_effect=Exception("Test message"))
314
    def test_cryoweb_import_errors(self, my_check):
315
        """Testing importing with data into UID with errors"""
316
317
        self.assertFalse(cryoweb_import(self.submission))
318
        self.assertTrue(my_check.called)
319
320
        # reload submission
321
        self.submission = Submission.objects.get(pk=1)
322
323
        self.assertEqual(
324
            self.submission.status,
325
            ERROR)
326
327
        self.assertIn(
328
            "Test message",
329
            self.submission.message)
330
331
        # check async message called
332
        message = 'Error'
333
        notification_message = (
334
            'Error in importing data: Test message')
335
336
        self.check_message(message, notification_message)
337
338
339
class CryowebReload(ImportMixin, BaseMixin, TestCase):
340
    """Simulate a cryoweb reload case. Load data as in CryowebImport, then
341
    call test which reload the same data"""
342
343
    # override fixtures
344
    fixtures = [
345
        'cryoweb/auth',
346
        'cryoweb/dictbreed',
347
        'cryoweb/image_app',
348
        'language/dictspecie',
349
        'language/speciesynonym'
350
    ]
351