crbanim.tests.test_helpers.ReloadCRBAnimTestCase.test_upload_crbanim() - Code Metrics - Inspection of "Issue 15" - cnr-ibba/IMAGE-InjectTool - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#35)

by Paolo

created 2019-06-18 17:50 UTC

ReloadCRBAnimTestCase.test_upload_crbanim() A

↳ Parent: crbanim.tests.test_helpers

Complexity

Conditions

Size

Total Lines	39
Code Lines	26

Duplication

Lines	39
Ratio	100 %

Importance

Changes

Metric	Value
eloc	26
dl	39
loc	39
rs	9.256
c	0
b	0
f	0
cc	1
nop	3

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Created on Fri Feb 22 15:49:18 2019

@author: Paolo Cozzi <[email protected]>
"""

import os
import logging

from unittest.mock import patch, Mock
from collections import namedtuple

from django.test import TestCase

from common.constants import ERROR, LOADED
from image_app.models import (
    Submission, db_has_data, DictSpecie, DictSex, DictBreed, Name, Animal,
    Sample, DictUberon, DictCountry)

from ..helpers import (
    logger, CRBAnimReader, upload_crbanim, fill_uid_breed, fill_uid_names,
    fill_uid_animal, fill_uid_sample, find_storage_type)
from .common import BaseTestCase


class CRBAnimReaderTestCase(BaseTestCase, TestCase):
    def setUp(self):
        # calling my base class setup
        super().setUp()

        # crate a CRBAnimReade object
        self.reader = CRBAnimReader()

        # get filenames for DataSourceMixinTestCase.dst_path
        self.reader.read_file(self.dst_path)

    def test_has_columns(self):
        """Test if class has columns"""

        reference = [
            'BRC_identifier',
            'BRC_name',
            'associated_research_project_name',
            'BRC_responsible_last_name',
            'BRC_responsible_first_name',
            'BRC_contact_email',
            'organization_name',
            'organisation_adress',
            'organization_url',
            'organization_country',
            'BRC_address',
            'BRC_url',
            'BRC_country',
            'sample_identifier',
            'EBI_Biosample_identifier',
            'sample_availability',
            'sample_bibliographic_references',
            'animal_ID',
            'sex',
            'species_latin_name',
            'NCBI_taxo_id',
            'breed_name',
            'country_of_origin',
            'sampling_protocol_url',
            'sampling_date',
            'sample_type_name',
            'sample_type_identifier',
            'sample_type_ontology',
            'body_part_name',
            'body_part_identifier',
            'body_part_ontology',
            'animal_birth_date',
            'sample_storage_temperature',
            'sample_container_type',
            'sample_fixer_nature']

        self.assertEqual(reference, self.reader.header)

    def test_debug_row(self):
        """Assert a function is callable"""

        self.reader.print_line(0)
        self.assertLogs(logger=logger, level=logging.DEBUG)

    def test_check_species(self):
        """Test check species method"""

        # get a country
        country = DictCountry.objects.get(label="United Kingdom")

        check, not_found = self.reader.check_species(country)

        self.assertTrue(check)
        self.assertEqual(len(not_found), 0)

        # changing species set
        DictSpecie.objects.filter(label='Bos taurus').delete()

        check, not_found = self.reader.check_species(country)

        # the read species are not included in fixtures
        self.assertFalse(check)
        self.assertGreater(len(not_found), 0)

    def test_check_sex(self):
        """Test check sex method"""

        check, not_found = self.reader.check_sex()

        self.assertTrue(check)
        self.assertEqual(len(not_found), 0)

        # changing sex set
        DictSex.objects.filter(label='female').delete()

        check, not_found = self.reader.check_sex()

        # the read species are not included in fixtures
        self.assertFalse(check)
        self.assertGreater(len(not_found), 0)

    def test_filter_by_column(self):
        """Filter records by column value"""

        # filter out biosample records from mydata
        data = self.reader.filter_by_column_values(
            "EBI_Biosample_identifier",
            [None])
        data = list(data)

        self.assertEqual(len(data), 2)

    def test_filter_by_column_case(self):
        """Filter records by column value case insensitive"""

        # filter out biosample records from mydata
        # record have capital letter. No record after filtering case sensitive
        data = self.reader.filter_by_column_values(
            "sex",
            ['female'],
            ignorecase=False)
        data = list(data)

        self.assertEqual(len(data), 0)

        # filtering female case insensitive
        data = self.reader.filter_by_column_values(
            "sex",
            ['female'],
            ignorecase=True)
        data = list(data)

        self.assertEqual(len(data), 1)

        # No record after filtering foo case insensitive
        data = self.reader.filter_by_column_values(
            "sex",
            ['foo'],
            ignorecase=True)
        data = list(data)

        self.assertEqual(len(data), 0)

    def test_is_valid(self):
        """Test recognizing a good CRBanim file"""

        with open(self.dst_path) as handle:
            chunk = handle.read(2048)

        check, not_found = self.reader.is_valid(chunk)

        # a good crbanim file returns True and an empty list
        self.assertTrue(check)
        self.assertEqual(not_found, [])

        # assert a not good file. self.base_dir comes from
        # common.tests.mixins.DataSourceMixinTestCase
        filename = os.path.join(
            self.base_dir,
            "Mapping_rules_CRB-Anim_InjectTool_v1.csv")

        with open(filename) as handle:
            chunk = handle.read(2048)

        check, not_found = self.reader.is_valid(chunk)

        # a invalid crbanim file returns False and a list
        self.assertFalse(check)
        self.assertGreater(len(not_found), 0)


class ProcessRecordTestCase(BaseTestCase, TestCase):
    """A class to test function which process record"""

    def setUp(self):
        # calling my base class setup
        super().setUp()

        # crate a CRBAnimReade object
        self.reader = CRBAnimReader()

        # get filenames for DataSourceMixinTestCase.dst_path
        self.reader.read_file(self.dst_path)

        # filter out biosample records from mydata
        data = self.reader.filter_by_column_values(
            "EBI_Biosample_identifier",
            [None])
        self.data = list(data)

        # track the sample record for test
        self.record = self.data[0]

        # set a country
        self.country = DictCountry.objects.get(label="United Kingdom")

        # fill object to test inserts. Fill breed
        self.breed = fill_uid_breed(self.record, self.country)

        # fill names
        self.animal_name, self.sample_name = fill_uid_names(
            self.record, self.submission)

        # filling animal and samples
        self.animal = fill_uid_animal(
            self.record,
            self.animal_name,
            self.breed,
            self.submission,
            {})

        # testing samples
        self.sample = fill_uid_sample(
            self.record,
            self.sample_name,
            self.animal,
            self.submission)

    def test_fill_uid_breed(self):
        """testing fill_uid_breed"""

        # testing output
        self.assertIsInstance(self.breed, DictBreed)
        self.assertEqual(self.breed.supplied_breed, self.record.breed_name)
        self.assertEqual(
            self.breed.specie.label, self.record.species_latin_name)

    def test_fill_uid_names(self):
        # testing output
        self.assertIsInstance(self.animal_name, Name)
        self.assertIsInstance(self.sample_name, Name)

        self.assertEqual(self.animal_name.name, self.record.animal_ID)
        self.assertEqual(self.sample_name.name, self.record.sample_identifier)

    def test_fill_uid_animals(self):
        # testing animal
        self.assertIsInstance(self.animal, Animal)

        # testing animal attributes
        sex = DictSex.objects.get(label__iexact=self.record.sex)
        self.assertEqual(self.animal.sex, sex)

    def test_fill_uid_samples(self):
        # testing sample
        self.assertIsInstance(self.sample, Sample)

        # testing sample attributes
        organism_part = DictUberon.objects.get(
            label__iexact="bone marrow")
        self.assertEqual(self.sample.organism_part, organism_part)

    def test_organism_part(self):
        """Check that an 'unknown' organims_part generate a DictUberon
        relying on sample_type_name (see crbanim_test_data.csv file)"""

        # get a new record
        record = self.data[1]

        # fill breeds
        breed = fill_uid_breed(record, self.country)

        # creating name
        animal_name, sample_name = fill_uid_names(
            record, self.submission)

        # filling animal and samples
        animal = fill_uid_animal(
            record,
            animal_name,
            breed,
            self.submission,
            {})

        # testing samples
        sample = fill_uid_sample(
            record,
            sample_name,
            animal,
            self.submission)

        # testing sample attributes
        organism_part = DictUberon.objects.get(
            label__iexact="uterus")
        self.assertEqual(sample.organism_part, organism_part)

    def test_find_storage_type(self):
        result = find_storage_type(self.record)
        self.assertEqual(result, "frozen, -80 degrees Celsius freezer")

        # create a fare record object
        Data = namedtuple("Data", ["sample_storage_temperature"])
        record = Data._make(["meow"])

        result = find_storage_type(record)
        self.assertIsNone(result)


class UploadCRBAnimTestCase(BaseTestCase, TestCase):

    @patch('crbanim.helpers.send_message_to_websocket')

    @patch('asyncio.get_event_loop')
    def test_upload_crbanim(self, asyncio_mock, send_message_to_websocket_mock):
        """Testing uploading and importing data from crbanim to UID"""
        tmp = asyncio_mock.return_value
        tmp.run_until_complete = Mock()

        self.assertTrue(upload_crbanim(self.submission))

        # reload submission
        self.submission.refresh_from_db()

        # assert submission messages
        self.assertEqual(
            self.submission.status,
            LOADED)

        self.assertIn(
            "CRBAnim import completed for submission",
            self.submission.message)

        # assert data into database
        self.assertTrue(db_has_data())
        self.assertTrue(Animal.objects.exists())
        self.assertTrue(Sample.objects.exists())

        self.assertEqual(asyncio_mock.call_count, 1)
        self.assertEqual(tmp.run_until_complete.call_count, 1)
        self.assertEqual(send_message_to_websocket_mock.call_count, 1)
        send_message_to_websocket_mock.assert_called_with(
            {'message': 'Loaded',
             'notification_message': 'CRBAnim import completed for '
                                     'submission: 1',
             'validation_message': {'animals': 1, 'samples': 2,
                                    'animal_unkn': 1, 'sample_unkn': 2,
                                    'animal_issues': 0, 'sample_issues': 0}}, 1)

    @patch('crbanim.helpers.send_message_to_websocket')

    @patch('asyncio.get_event_loop')
    @patch("crbanim.helpers.CRBAnimReader.check_species",
           return_value=[False, 'Rainbow trout'])
    def test_upload_crbanim_errors(self, my_check, asyncio_mock,
                                   send_message_to_websocket_mock):
        """Testing importing with data into UID with errors"""
        tmp = asyncio_mock.return_value
        tmp.run_until_complete = Mock()

        self.assertFalse(upload_crbanim(self.submission))

        # reload submission
        self.submission.refresh_from_db()

        # test my mock method called
        self.assertTrue(my_check.called)

        # reload submission
        self.submission = Submission.objects.get(pk=1)

        self.assertEqual(
            self.submission.status,
            ERROR)

        self.assertIn(
            "Some species are not loaded in UID database",
            self.submission.message)

        # assert data into database
        self.assertFalse(db_has_data())
        self.assertFalse(Animal.objects.exists())
        self.assertFalse(Sample.objects.exists())

        self.assertEqual(asyncio_mock.call_count, 1)
        self.assertEqual(tmp.run_until_complete.call_count, 1)
        self.assertEqual(send_message_to_websocket_mock.call_count, 1)
        send_message_to_websocket_mock.assert_called_with(
            {'message': 'Error',
             'notification_message': 'Error in importing data: Some species '
                                     'are not loaded in UID database: Rainbow '
                                     'trout'}, 1)

    @patch('crbanim.helpers.send_message_to_websocket')

    @patch('asyncio.get_event_loop')
    @patch("crbanim.helpers.CRBAnimReader.check_sex",
           return_value=[False, 'unknown'])
    def test_upload_crbanim_errors_with_sex(self, my_check, asyncio_mock,
                                            send_message_to_websocket_mock):
        """Testing importing with data into UID with errors"""
        tmp = asyncio_mock.return_value
        tmp.run_until_complete = Mock()

        self.assertFalse(upload_crbanim(self.submission))

        # reload submission
        self.submission.refresh_from_db()

        # test my mock method called
        self.assertTrue(my_check.called)

        # reload submission
        self.submission = Submission.objects.get(pk=1)

        self.assertEqual(
            self.submission.status,
            ERROR)

        # check for two distinct messages
        self.assertIn(
            "Not all Sex terms are loaded into database",
            self.submission.message)

        self.assertNotIn(
            "CRBAnim import completed for submission",
            self.submission.message)

        # assert data into database
        self.assertFalse(db_has_data())
        self.assertFalse(Animal.objects.exists())
        self.assertFalse(Sample.objects.exists())

        self.assertEqual(asyncio_mock.call_count, 1)
        self.assertEqual(tmp.run_until_complete.call_count, 1)
        self.assertEqual(send_message_to_websocket_mock.call_count, 1)
        send_message_to_websocket_mock.assert_called_with(
            {'message': 'Error',
             'notification_message': 'Error in importing data: Not all Sex '
                                     'terms are loaded into database: check '
                                     'for unknown in your dataset'}, 1)


class ReloadCRBAnimTestCase(BaseTestCase, TestCase):
    """Simulate a crbanim reload case. Load data as in
    UploadCRBAnimTestCase, then call test which reload the same data"""

    # override useal fixtures
    fixtures = [
        'crbanim/auth',
        'crbanim/dictspecie',
        'crbanim/image_app',
        'crbanim/submission',
        'language/speciesynonym'
    ]

    @patch('crbanim.helpers.send_message_to_websocket')

    @patch('asyncio.get_event_loop')
    def test_upload_crbanim(
            self, asyncio_mock, send_message_to_websocket_mock):
        """Testing uploading and importing data from crbanim to UID"""

        tmp = asyncio_mock.return_value
        tmp.run_until_complete = Mock()

        # assert upload
        self.assertTrue(upload_crbanim(self.submission))

        # reload submission
        self.submission.refresh_from_db()

        # assert submission messages
        self.assertEqual(
            self.submission.status,
            LOADED)

        self.assertIn(
            "CRBAnim import completed for submission",
            self.submission.message)

        # assert data into database
        self.assertTrue(db_has_data())
        self.assertTrue(Animal.objects.exists())
        self.assertTrue(Sample.objects.exists())

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

cnr-ibba / IMAGE-InjectTool

Pull Request — master (#35)

ReloadCRBAnimTestCase.test_upload_crbanim() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like