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

UploadCRBAnimTestCase.test_upload_crbanim_errors()   A

Complexity

Conditions 1

Size

Total Lines 42
Code Lines 28

Duplication

Lines 42
Ratio 100 %

Importance

Changes 0
Metric Value
eloc 28
dl 42
loc 42
rs 9.208
c 0
b 0
f 0
cc 1
nop 4
1
#!/usr/bin/env python3
2
# -*- coding: utf-8 -*-
3
"""
4
Created on Fri Feb 22 15:49:18 2019
5
6
@author: Paolo Cozzi <[email protected]>
7
"""
8
9
import os
10
import logging
11
12
from unittest.mock import patch, Mock
13
from collections import namedtuple
14
15
from django.test import TestCase
16
17
from common.constants import ERROR, LOADED
18
from image_app.models import (
19
    Submission, db_has_data, DictSpecie, DictSex, DictBreed, Name, Animal,
20
    Sample, DictUberon, DictCountry)
21
22
from ..helpers import (
23
    logger, CRBAnimReader, upload_crbanim, fill_uid_breed, fill_uid_names,
24
    fill_uid_animal, fill_uid_sample, find_storage_type)
25
from .common import BaseTestCase
26
27
28 View Code Duplication
class WebSocketMixin(object):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
29
    """Override setUp to mock websocket objects"""
30
31
    def setUp(self):
32
        # calling my base class setup
33
        super().setUp()
34
35
        # setting channels methods
36
        self.asyncio_mock_patcher = patch(
37
            'asyncio.get_event_loop')
38
        self.asyncio_mock = self.asyncio_mock_patcher.start()
39
40
        # mocking asyncio return value
41
        self.run_until = self.asyncio_mock.return_value
42
        self.run_until.run_until_complete = Mock()
43
44
        # another patch
45
        self.send_msg_ws_patcher = patch(
46
            'crbanim.helpers.send_message_to_websocket')
47
        self.send_msg_ws = self.send_msg_ws_patcher.start()
48
49
    def tearDown(self):
50
        # stopping mock objects
51
        self.asyncio_mock_patcher.stop()
52
        self.send_msg_ws_patcher.stop()
53
54
        # calling base methods
55
        super().tearDown()
56
57
    def check_message(
58
            self, message, notification_message, validation_message=None,
59
            pk=1):
60
        """assert message to websocket called with parameters"""
61
62
        # construct message according parameters
63
        message = {
64
            'message': message,
65
            'notification_message': notification_message
66
        }
67
68
        # in case of successful data upload, a validation message is sent
69
        if validation_message:
70
            message['validation_message'] = validation_message
71
72
        self.assertEqual(self.asyncio_mock.call_count, 1)
73
        self.assertEqual(self.run_until.run_until_complete.call_count, 1)
74
        self.assertEqual(self.send_msg_ws.call_count, 1)
75
        self.send_msg_ws.assert_called_with(
76
            message,
77
            pk)
78
79
80
class CRBAnimMixin(WebSocketMixin):
81
    """Common test for CRBanim classes"""
82
83
    def test_upload_crbanim(self):
84
        """Testing uploading and importing data from crbanim to UID"""
85
86
        # assert upload
87
        self.assertTrue(upload_crbanim(self.submission))
88
89
        # reload submission
90
        self.submission.refresh_from_db()
91
92
        # assert submission messages
93
        self.assertEqual(
94
            self.submission.status,
95
            LOADED)
96
97
        self.assertIn(
98
            "CRBAnim import completed for submission",
99
            self.submission.message)
100
101
        # assert data into database
102
        self.assertTrue(db_has_data())
103
        self.assertTrue(Animal.objects.exists())
104
        self.assertTrue(Sample.objects.exists())
105
106
        # check async message called
107
        message = 'Loaded'
108
        notification_message = (
109
            'CRBAnim import completed for submission: 1')
110
        validation_message = {
111
            'animals': 1, 'samples': 2,
112
            'animal_unkn': 1, 'sample_unkn': 2,
113
            'animal_issues': 0, 'sample_issues': 0}
114
115
        self.check_message(message, notification_message, validation_message)
116
117
    def check_errors(self, my_check, message, notification_message):
118
        """Common stuff for error in crbanim loading"""
119
120
        self.assertFalse(upload_crbanim(self.submission))
121
122
        # reload submission
123
        self.submission.refresh_from_db()
124
125
        # test my mock method called
126
        self.assertTrue(my_check.called)
127
128
        # reload submission
129
        self.submission = Submission.objects.get(pk=1)
130
131
        self.assertEqual(
132
            self.submission.status,
133
            ERROR)
134
135
        # check for two distinct messages
136
        self.assertIn(
137
            message,
138
            self.submission.message)
139
140
        self.assertNotIn(
141
            "CRBAnim import completed for submission",
142
            self.submission.message)
143
144
        # assert data into database
145
        self.assertFalse(db_has_data())
146
        self.assertFalse(Animal.objects.exists())
147
        self.assertFalse(Sample.objects.exists())
148
149
        # check async message called
150
        self.check_message('Error', notification_message)
151
152
153
class CRBAnimReaderTestCase(BaseTestCase, TestCase):
154
    def setUp(self):
155
        # calling my base class setup
156
        super().setUp()
157
158
        # crate a CRBAnimReade object
159
        self.reader = CRBAnimReader()
160
161
        # get filenames for DataSourceMixinTestCase.dst_path
162
        self.reader.read_file(self.dst_path)
163
164
    def test_has_columns(self):
165
        """Test if class has columns"""
166
167
        reference = [
168
            'BRC_identifier',
169
            'BRC_name',
170
            'associated_research_project_name',
171
            'BRC_responsible_last_name',
172
            'BRC_responsible_first_name',
173
            'BRC_contact_email',
174
            'organization_name',
175
            'organisation_adress',
176
            'organization_url',
177
            'organization_country',
178
            'BRC_address',
179
            'BRC_url',
180
            'BRC_country',
181
            'sample_identifier',
182
            'EBI_Biosample_identifier',
183
            'sample_availability',
184
            'sample_bibliographic_references',
185
            'animal_ID',
186
            'sex',
187
            'species_latin_name',
188
            'NCBI_taxo_id',
189
            'breed_name',
190
            'country_of_origin',
191
            'sampling_protocol_url',
192
            'sampling_date',
193
            'sample_type_name',
194
            'sample_type_identifier',
195
            'sample_type_ontology',
196
            'body_part_name',
197
            'body_part_identifier',
198
            'body_part_ontology',
199
            'animal_birth_date',
200
            'sample_storage_temperature',
201
            'sample_container_type',
202
            'sample_fixer_nature']
203
204
        self.assertEqual(reference, self.reader.header)
205
206
    def test_debug_row(self):
207
        """Assert a function is callable"""
208
209
        self.reader.print_line(0)
210
        self.assertLogs(logger=logger, level=logging.DEBUG)
211
212
    def test_check_species(self):
213
        """Test check species method"""
214
215
        # get a country
216
        country = DictCountry.objects.get(label="United Kingdom")
217
218
        check, not_found = self.reader.check_species(country)
219
220
        self.assertTrue(check)
221
        self.assertEqual(len(not_found), 0)
222
223
        # changing species set
224
        DictSpecie.objects.filter(label='Bos taurus').delete()
225
226
        check, not_found = self.reader.check_species(country)
227
228
        # the read species are not included in fixtures
229
        self.assertFalse(check)
230
        self.assertGreater(len(not_found), 0)
231
232
    def test_check_sex(self):
233
        """Test check sex method"""
234
235
        check, not_found = self.reader.check_sex()
236
237
        self.assertTrue(check)
238
        self.assertEqual(len(not_found), 0)
239
240
        # changing sex set
241
        DictSex.objects.filter(label='female').delete()
242
243
        check, not_found = self.reader.check_sex()
244
245
        # the read species are not included in fixtures
246
        self.assertFalse(check)
247
        self.assertGreater(len(not_found), 0)
248
249
    def test_filter_by_column(self):
250
        """Filter records by column value"""
251
252
        # filter out biosample records from mydata
253
        data = self.reader.filter_by_column_values(
254
            "EBI_Biosample_identifier",
255
            [None])
256
        data = list(data)
257
258
        self.assertEqual(len(data), 2)
259
260
    def test_filter_by_column_case(self):
261
        """Filter records by column value case insensitive"""
262
263
        # filter out biosample records from mydata
264
        # record have capital letter. No record after filtering case sensitive
265
        data = self.reader.filter_by_column_values(
266
            "sex",
267
            ['female'],
268
            ignorecase=False)
269
        data = list(data)
270
271
        self.assertEqual(len(data), 0)
272
273
        # filtering female case insensitive
274
        data = self.reader.filter_by_column_values(
275
            "sex",
276
            ['female'],
277
            ignorecase=True)
278
        data = list(data)
279
280
        self.assertEqual(len(data), 1)
281
282
        # No record after filtering foo case insensitive
283
        data = self.reader.filter_by_column_values(
284
            "sex",
285
            ['foo'],
286
            ignorecase=True)
287
        data = list(data)
288
289
        self.assertEqual(len(data), 0)
290
291
    def test_is_valid(self):
292
        """Test recognizing a good CRBanim file"""
293
294
        with open(self.dst_path) as handle:
295
            chunk = handle.read(2048)
296
297
        check, not_found = self.reader.is_valid(chunk)
298
299
        # a good crbanim file returns True and an empty list
300
        self.assertTrue(check)
301
        self.assertEqual(not_found, [])
302
303
        # assert a not good file. self.base_dir comes from
304
        # common.tests.mixins.DataSourceMixinTestCase
305
        filename = os.path.join(
306
            self.base_dir,
307
            "Mapping_rules_CRB-Anim_InjectTool_v1.csv")
308
309
        with open(filename) as handle:
310
            chunk = handle.read(2048)
311
312
        check, not_found = self.reader.is_valid(chunk)
313
314
        # a invalid crbanim file returns False and a list
315
        self.assertFalse(check)
316
        self.assertGreater(len(not_found), 0)
317
318
319
class ProcessRecordTestCase(BaseTestCase, TestCase):
320
    """A class to test function which process record"""
321
322
    def setUp(self):
323
        # calling my base class setup
324
        super().setUp()
325
326
        # crate a CRBAnimReade object
327
        self.reader = CRBAnimReader()
328
329
        # get filenames for DataSourceMixinTestCase.dst_path
330
        self.reader.read_file(self.dst_path)
331
332
        # filter out biosample records from mydata
333
        data = self.reader.filter_by_column_values(
334
            "EBI_Biosample_identifier",
335
            [None])
336
        self.data = list(data)
337
338
        # track the sample record for test
339
        self.record = self.data[0]
340
341
        # set a country
342
        self.country = DictCountry.objects.get(label="United Kingdom")
343
344
        # fill object to test inserts. Fill breed
345
        self.breed = fill_uid_breed(self.record, self.country)
346
347
        # fill names
348
        self.animal_name, self.sample_name = fill_uid_names(
349
            self.record, self.submission)
350
351
        # filling animal and samples
352
        self.animal = fill_uid_animal(
353
            self.record,
354
            self.animal_name,
355
            self.breed,
356
            self.submission,
357
            {})
358
359
        # testing samples
360
        self.sample = fill_uid_sample(
361
            self.record,
362
            self.sample_name,
363
            self.animal,
364
            self.submission)
365
366
    def test_fill_uid_breed(self):
367
        """testing fill_uid_breed"""
368
369
        # testing output
370
        self.assertIsInstance(self.breed, DictBreed)
371
        self.assertEqual(self.breed.supplied_breed, self.record.breed_name)
372
        self.assertEqual(
373
            self.breed.specie.label, self.record.species_latin_name)
374
375
    def test_fill_uid_names(self):
376
        # testing output
377
        self.assertIsInstance(self.animal_name, Name)
378
        self.assertIsInstance(self.sample_name, Name)
379
380
        self.assertEqual(self.animal_name.name, self.record.animal_ID)
381
        self.assertEqual(self.sample_name.name, self.record.sample_identifier)
382
383
    def test_fill_uid_animals(self):
384
        # testing animal
385
        self.assertIsInstance(self.animal, Animal)
386
387
        # testing animal attributes
388
        sex = DictSex.objects.get(label__iexact=self.record.sex)
389
        self.assertEqual(self.animal.sex, sex)
390
391
    def test_fill_uid_samples(self):
392
        # testing sample
393
        self.assertIsInstance(self.sample, Sample)
394
395
        # testing sample attributes
396
        organism_part = DictUberon.objects.get(
397
            label__iexact="bone marrow")
398
        self.assertEqual(self.sample.organism_part, organism_part)
399
400
    def test_organism_part(self):
401
        """Check that an 'unknown' organims_part generate a DictUberon
402
        relying on sample_type_name (see crbanim_test_data.csv file)"""
403
404
        # get a new record
405
        record = self.data[1]
406
407
        # fill breeds
408
        breed = fill_uid_breed(record, self.country)
409
410
        # creating name
411
        animal_name, sample_name = fill_uid_names(
412
            record, self.submission)
413
414
        # filling animal and samples
415
        animal = fill_uid_animal(
416
            record,
417
            animal_name,
418
            breed,
419
            self.submission,
420
            {})
421
422
        # testing samples
423
        sample = fill_uid_sample(
424
            record,
425
            sample_name,
426
            animal,
427
            self.submission)
428
429
        # testing sample attributes
430
        organism_part = DictUberon.objects.get(
431
            label__iexact="uterus")
432
        self.assertEqual(sample.organism_part, organism_part)
433
434
    def test_find_storage_type(self):
435
        result = find_storage_type(self.record)
436
        self.assertEqual(result, "frozen, -80 degrees Celsius freezer")
437
438
        # create a fare record object
439
        Data = namedtuple("Data", ["sample_storage_temperature"])
440
        record = Data._make(["meow"])
441
442
        result = find_storage_type(record)
443
        self.assertIsNone(result)
444
445
446
class UploadCRBAnimTestCase(CRBAnimMixin, BaseTestCase, TestCase):
447
448
    @patch("crbanim.helpers.CRBAnimReader.check_species",
449
           return_value=[False, 'Rainbow trout'])
450
    def test_upload_crbanim_errors_with_species(self, my_check):
451
        """Testing importing with data into UID with errors"""
452
453
        message = "Some species are not loaded in UID database"
454
        notification_message = (
455
            'Error in importing data: Some species '
456
            'are not loaded in UID database: Rainbow '
457
            'trout')
458
459
        # check crbanim import fails
460
        self.check_errors(my_check, message, notification_message)
461
462
    @patch("crbanim.helpers.CRBAnimReader.check_sex",
463
           return_value=[False, 'unknown'])
464
    def test_upload_crbanim_errors_with_sex(self, my_check):
465
        """Testing importing with data into UID with errors"""
466
467
        message = "Not all Sex terms are loaded into database"
468
        notification_message = (
469
            'Error in importing data: Not all Sex '
470
            'terms are loaded into database: check '
471
            'for unknown in your dataset')
472
473
        # check crbanim import fails
474
        self.check_errors(my_check, message, notification_message)
475
476
477
class ReloadCRBAnimTestCase(CRBAnimMixin, BaseTestCase, TestCase):
478
479
    """Simulate a crbanim reload case. Load data as in
480
    UploadCRBAnimTestCase, then call test which reload the same data"""
481
482
    # override useal fixtures
483
    fixtures = [
484
        'crbanim/auth',
485
        'crbanim/dictspecie',
486
        'crbanim/image_app',
487
        'crbanim/submission',
488
        'language/speciesynonym'
489
    ]
490