Completed
Pull Request — devel (#72)
by Paolo
07:01
created

UpdateCRBAnimTestCase.test_upload_crbanim()   A

Complexity

Conditions 1

Size

Total Lines 35
Code Lines 22

Duplication

Lines 35
Ratio 100 %

Importance

Changes 0
Metric Value
eloc 22
dl 35
loc 35
rs 9.352
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
13
from collections import namedtuple
14
15
from django.test import TestCase
16
17
from common.tests import WebSocketMixin
18
from uid.models import (
19
    DictSex, DictBreed, Animal,
20
    Sample, DictUberon, DictCountry, Submission)
21
from uid.tests.mixins import (
22
    DataSourceMixinTestCase, FileReaderMixinTestCase)
23
24
from ..helpers import (
25
    logger, CRBAnimReader, upload_crbanim, fill_uid_breed,
26
    fill_uid_animal, fill_uid_sample, find_storage_type, sanitize_url)
27
from .common import BaseTestCase
28
29
30
class CRBAnimMixin(DataSourceMixinTestCase, WebSocketMixin):
31
    """Common test for CRBanim classes"""
32
33
    # define the method to upload data from. Since the function is now inside
34
    # a class it becomes a method, specifically a bound method and is supposed
35
    # to receive the self attribute by default. If we don't want to get the
36
    # self attribute, we have to declare function as a staticmetho
37
    # https://stackoverflow.com/a/35322635/4385116
38
    upload_method = staticmethod(upload_crbanim)
39
40
    def test_upload_crbanim(self):
41
        """Testing uploading and importing data from crbanim to UID"""
42
43
        # test data loaded
44
        message = "CRBAnim import completed for submission"
45
        self.upload_datasource(message)
46
47
        # check async message called
48
        notification_message = (
49
            'CRBAnim import completed for submission: 1')
50
        validation_message = {
51
            'animals': 1, 'samples': 2,
52
            'animal_unkn': 1, 'sample_unkn': 2,
53
            'animal_issues': 0, 'sample_issues': 0}
54
55
        # check async message called using WebSocketMixin.check_message
56
        self.check_message('Loaded', notification_message, validation_message)
57
58
    def check_errors(self, my_check, message, notification_message):
59
        """Common stuff for error in crbanim loading"""
60
61
        super().check_errors(my_check, message)
62
63
        # check async message called using WebSocketMixin.check_message
64
        self.check_message('Error', notification_message)
65
66
67
class CRBAnimReaderTestCase(
68
        FileReaderMixinTestCase, DataSourceMixinTestCase, BaseTestCase,
69
        TestCase):
70
    def setUp(self):
71
        # calling my base class setup
72
        super().setUp()
73
74
        # crate a CRBAnimReade object
75
        self.reader = CRBAnimReader()
76
77
        # get filenames for DataSourceMixinTestCase.dst_path
78
        self.reader.read_file(self.dst_path)
79
80
    def test_has_columns(self):
81
        """Test if class has columns"""
82
83
        reference = [
84
            'BRC_identifier',
85
            'BRC_name',
86
            'associated_research_project_name',
87
            'BRC_responsible_last_name',
88
            'BRC_responsible_first_name',
89
            'BRC_contact_email',
90
            'organization_name',
91
            'organisation_adress',
92
            'organization_url',
93
            'organization_country',
94
            'BRC_address',
95
            'BRC_url',
96
            'BRC_country',
97
            'sample_identifier',
98
            'EBI_Biosample_identifier',
99
            'sample_availability',
100
            'sample_bibliographic_references',
101
            'animal_ID',
102
            'sex',
103
            'species_latin_name',
104
            'NCBI_taxo_id',
105
            'breed_name',
106
            'country_of_origin',
107
            'sampling_protocol_url',
108
            'sampling_date',
109
            'sample_type_name',
110
            'sample_type_identifier',
111
            'sample_type_ontology',
112
            'body_part_name',
113
            'body_part_identifier',
114
            'body_part_ontology',
115
            'animal_birth_date',
116
            'sample_storage_temperature',
117
            'sample_container_type',
118
            'sample_fixer_nature']
119
120
        self.assertEqual(reference, self.reader.header)
121
122
    def test_debug_row(self):
123
        """Assert a function is callable"""
124
125
        self.reader.print_line(0)
126
        self.assertLogs(logger=logger, level=logging.DEBUG)
127
128
    def test_filter_by_column(self):
129
        """Filter records by column value"""
130
131
        # filter out biosample records from mydata
132
        data = self.reader.filter_by_column_values(
133
            "EBI_Biosample_identifier",
134
            [None])
135
        data = list(data)
136
137
        self.assertEqual(len(data), 2)
138
139
    def test_filter_by_column_case(self):
140
        """Filter records by column value case insensitive"""
141
142
        # filter out biosample records from mydata
143
        # record have capital letter. No record after filtering case sensitive
144
        data = self.reader.filter_by_column_values(
145
            "sex",
146
            ['female'],
147
            ignorecase=False)
148
        data = list(data)
149
150
        self.assertEqual(len(data), 0)
151
152
        # filtering female case insensitive
153
        data = self.reader.filter_by_column_values(
154
            "sex",
155
            ['female'],
156
            ignorecase=True)
157
        data = list(data)
158
159
        self.assertEqual(len(data), 1)
160
161
        # No record after filtering foo case insensitive
162
        data = self.reader.filter_by_column_values(
163
            "sex",
164
            ['foo'],
165
            ignorecase=True)
166
        data = list(data)
167
168
        self.assertEqual(len(data), 0)
169
170
    def test_is_valid(self):
171
        """Test recognizing a good CRBanim file"""
172
173
        with open(self.dst_path) as handle:
174
            chunk = handle.read(2048)
175
176
        check, not_found = self.reader.is_valid(chunk)
177
178
        # a good crbanim file returns True and an empty list
179
        self.assertTrue(check)
180
        self.assertEqual(not_found, [])
181
182
        # assert a not good file. self.base_dir comes from
183
        # common.tests.mixins.DataSourceMixinTestCase
184
        filename = os.path.join(
185
            self.base_dir,
186
            "Mapping_rules_CRB-Anim_InjectTool_v1.csv")
187
188
        with open(filename) as handle:
189
            chunk = handle.read(2048)
190
191
        check, not_found = self.reader.is_valid(chunk)
192
193
        # a invalid crbanim file returns False and a list
194
        self.assertFalse(check)
195
        self.assertGreater(len(not_found), 0)
196
197
198
class ProcessRecordTestCase(DataSourceMixinTestCase, BaseTestCase, TestCase):
199
    """A class to test function which process record"""
200
201
    def setUp(self):
202
        # calling my base class setup
203
        super().setUp()
204
205
        # crate a CRBAnimReade object
206
        self.reader = CRBAnimReader()
207
208
        # get filenames for DataSourceMixinTestCase.dst_path
209
        self.reader.read_file(self.dst_path)
210
211
        # filter out biosample records from mydata
212
        data = self.reader.filter_by_column_values(
213
            "EBI_Biosample_identifier",
214
            [None])
215
        self.data = list(data)
216
217
        # track the sample record for test
218
        self.record = self.data[0]
219
220
        # set a country
221
        self.country = DictCountry.objects.get(label="United Kingdom")
222
223
        # fill object to test inserts. Fill breed
224
        self.breed = fill_uid_breed(self.record, self.country)
225
226
        # filling animal and samples
227
        self.animal = fill_uid_animal(
228
            self.record,
229
            self.breed,
230
            self.submission,
231
            {})
232
233
        # testing samples
234
        self.sample = fill_uid_sample(
235
            self.record,
236
            self.animal,
237
            self.submission)
238
239
    def test_fill_uid_breed(self):
240
        """testing fill_uid_breed"""
241
242
        # testing output
243
        self.assertIsInstance(self.breed, DictBreed)
244
        self.assertEqual(self.breed.supplied_breed, self.record.breed_name)
245
        self.assertEqual(
246
            self.breed.specie.label, self.record.species_latin_name)
247
248
    def test_fill_uid_names(self):
249
        self.assertEqual(self.animal.name, self.record.animal_ID)
250
        self.assertEqual(self.sample.name, self.record.sample_identifier)
251
252
    def test_fill_uid_animals(self):
253
        # testing animal
254
        self.assertIsInstance(self.animal, Animal)
255
256
        # testing animal attributes
257
        sex = DictSex.objects.get(label__iexact=self.record.sex)
258
        self.assertEqual(self.animal.sex, sex)
259
260
    def test_fill_uid_samples(self):
261
        # testing sample
262
        self.assertIsInstance(self.sample, Sample)
263
264
        # testing sample attributes
265
        organism_part = DictUberon.objects.get(
266
            label__iexact="bone marrow")
267
        self.assertEqual(self.sample.organism_part, organism_part)
268
269
    def test_organism_part(self):
270
        """Check that an 'unknown' organims_part generate a DictUberon
271
        relying on sample_type_name (see crbanim_test_data.csv file)"""
272
273
        # get a new record
274
        record = self.data[1]
275
276
        # fill breeds
277
        breed = fill_uid_breed(record, self.country)
278
279
        # filling animal and samples
280
        animal = fill_uid_animal(
281
            record,
282
            breed,
283
            self.submission,
284
            {})
285
286
        # testing samples
287
        sample = fill_uid_sample(
288
            record,
289
            animal,
290
            self.submission)
291
292
        # testing sample attributes
293
        organism_part = DictUberon.objects.get(
294
            label__iexact="uterus")
295
        self.assertEqual(sample.organism_part, organism_part)
296
297
    def test_find_storage_type(self):
298
        """Asserting storage type conversion"""
299
300
        result = find_storage_type(self.record)
301
        self.assertEqual(result, 2)
302
303
        # create a fare record object
304
        Data = namedtuple("Data", ["sample_storage_temperature"])
305
        record = Data._make(["meow"])
306
307
        result = find_storage_type(record)
308
        self.assertIsNone(result)
309
310
311
class UploadCRBAnimTestCase(CRBAnimMixin, BaseTestCase, TestCase):
312
313
    @patch("crbanim.helpers.CRBAnimReader.check_species",
314
           return_value=[False, 'Rainbow trout'])
315
    def test_upload_crbanim_errors_with_species(self, my_check):
316
        """Testing importing with data into UID with errors"""
317
318
        message = "Some species are not loaded in UID database"
319
        notification_message = (
320
            "Error in importing data: Some species "
321
            "are not loaded in UID database: check for 'Rainbow "
322
            "trout' in your dataset")
323
324
        # check crbanim import fails
325
        self.check_errors(my_check, message, notification_message)
326
327
    @patch("crbanim.helpers.CRBAnimReader.check_sex",
328
           return_value=[False, 'unknown'])
329
    def test_upload_crbanim_errors_with_sex(self, my_check):
330
        """Testing importing with data into UID with errors"""
331
332
        message = "Not all Sex terms are loaded into database"
333
        notification_message = (
334
            "Error in importing data: %s: check "
335
            "for 'unknown' in your dataset" % (message))
336
337
        # check crbanim import fails
338
        self.check_errors(my_check, message, notification_message)
339
340
    @patch("crbanim.helpers.CRBAnimReader.check_countries",
341
           return_value=[False, 'unknown'])
342
    def test_upload_crbanim_errors_with_countries(self, my_check):
343
        """Testing importing with data into UID with errors"""
344
345
        message = "Not all countries are loaded into database"
346
        notification_message = (
347
            "Error in importing data: %s: check "
348
            "for 'unknown' in your dataset" % (message))
349
350
        # check crbanim import fails
351
        self.check_errors(my_check, message, notification_message)
352
353
354
class ReloadCRBAnimTestCase(CRBAnimMixin, BaseTestCase, TestCase):
355
356
    """Simulate a crbanim reload case. Load data as in
357
    UploadCRBAnimTestCase, then call test which reload the same data"""
358
359
    # override used fixtures
360
    fixtures = [
361
        'crbanim/auth',
362
        'crbanim/dictspecie',
363
        'crbanim/uid',
364
        'crbanim/submission',
365
        'language/speciesynonym'
366
    ]
367
368
369 View Code Duplication
class UpdateCRBAnimTestCase(CRBAnimMixin, BaseTestCase, TestCase):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
370
    """Simulate a crbanim update with the same dataset. Data already
371
    present will be ignored. I won't remove anithing from old submission,
372
    so I expect to not add any items into database"""
373
374
    # override used fixtures
375
    fixtures = [
376
        'crbanim/auth',
377
        'crbanim/dictspecie',
378
        'crbanim/uid',
379
        'crbanim/submission',
380
        'language/speciesynonym'
381
    ]
382
383
    def setUp(self):
384
        # calling my base class setup
385
        super().setUp()
386
387
        # track old submission
388
        self.old_submission = Submission.objects.get(pk=1)
389
390
        # generate a new submission from old submission object
391
        submission = self.submission
392
        submission.pk = None
393
        submission.title = "Updated database"
394
        submission.datasource_version = "Updated database"
395
        submission.save()
396
397
        # track the new submission
398
        self.submission = submission
399
400
    def test_upload_crbanim(self):
401
        """Testing uploading and importing data from crbanim to UID"""
402
403
        # test data loaded
404
        message = "CRBAnim import completed for submission"
405
        self.upload_datasource(message)
406
407
        # check animal and samples
408
        queryset = Animal.objects.all()
409
        self.assertEqual(len(queryset), 1, msg="check animal load")
410
411
        queryset = Sample.objects.all()
412
        self.assertEqual(len(queryset), 2, msg="check sample load")
413
414
        # assert data are in the proper submission
415
        self.assertEqual(self.old_submission.animal_set.count(), 1)
416
        self.assertEqual(self.old_submission.sample_set.count(), 2)
417
418
        self.assertEqual(self.submission.animal_set.count(), 0)
419
        self.assertEqual(self.submission.sample_set.count(), 0)
420
421
        # check async message called
422
        notification_message = (
423
            'CRBAnim import completed for submission: 2')
424
        validation_message = {
425
            'animals': 0, 'samples': 0,
426
            'animal_unkn': 0, 'sample_unkn': 0,
427
            'animal_issues': 0, 'sample_issues': 0}
428
429
        # check async message called using WebSocketMixin.check_message
430
        self.check_message(
431
            'Loaded',
432
            notification_message,
433
            validation_message,
434
            pk=self.submission.id)
435
436
437
class SanitizeAccessionTest(TestCase):
438
439
    def setUp(self):
440
        self.queries = [
441
            (
442
                ("https://crb-anim.fr/access-to-collection/#/simplesearch?"
443
                 "SAMPLEIDENTIFIER=Ovar-Mér-3796-1"),
444
                ("https://crb-anim.fr/access-to-collection/#/simplesearch?"
445
                 "SAMPLEIDENTIFIER=Ovar-M%C3%A9r-3796-1")
446
            ),
447
        ]
448
449
    def test_sanitize_accession(self):
450
        for query in self.queries:
451
            url, reference = query
452
            test = sanitize_url(url)
453
            self.assertEqual(reference, test)
454