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

ReloadCRBAnimTestCase.test_upload_crbanim()   A

Complexity

Conditions 1

Size

Total Lines 33
Code Lines 20

Duplication

Lines 33
Ratio 100 %

Importance

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