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