crbanim.tests.test_helpers.WebSocketMixin.check_message() - Code Metrics - Inspection of "last requirements of Validation Summary Views issu..." - cnr-ibba/IMAGE-InjectTool - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#35)

by Paolo

created 2019-06-20 15:45 UTC

WebSocketMixin.check_message() A

↳ Parent: crbanim.tests.test_helpers

Complexity

Conditions

Size

Total Lines	21
Code Lines	14

Duplication

Lines	21
Ratio	100 %

Importance

Changes

Metric	Value
eloc	14
dl	21
loc	21
rs	9.7
c	0
b	0
f	0
cc	2
nop	5

#!/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 WebSocketMixin(object):

    """Override setUp to mock websocket objects"""

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

        # setting channels methods
        self.asyncio_mock_patcher = patch(
            'asyncio.get_event_loop')
        self.asyncio_mock = self.asyncio_mock_patcher.start()

        # mocking asyncio return value
        self.run_until = self.asyncio_mock.return_value
        self.run_until.run_until_complete = Mock()

        # another patch
        self.send_msg_ws_patcher = patch(
            'crbanim.helpers.send_message_to_websocket')
        self.send_msg_ws = self.send_msg_ws_patcher.start()

    def tearDown(self):
        # stopping mock objects
        self.asyncio_mock_patcher.stop()
        self.send_msg_ws_patcher.stop()

        # calling base methods
        super().tearDown()

    def check_message(
            self, message, notification_message, validation_message=None,
            pk=1):
        """assert message to websocket called with parameters"""

        # construct message according parameters
        message = {
            'message': message,
            'notification_message': notification_message
        }

        # in case of successful data upload, a validation message is sent
        if validation_message:
            message['validation_message'] = validation_message

        self.assertEqual(self.asyncio_mock.call_count, 1)
        self.assertEqual(self.run_until.run_until_complete.call_count, 1)
        self.assertEqual(self.send_msg_ws.call_count, 1)
        self.send_msg_ws.assert_called_with(
            message,
            pk)


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(WebSocketMixin, BaseTestCase, TestCase):

    def test_upload_crbanim(self):

        """Testing uploading and importing data from crbanim to UID"""

        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())

        # check async message called
        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}

        self.check_message(message, notification_message, validation_message)

    @patch("crbanim.helpers.CRBAnimReader.check_species",

           return_value=[False, 'Rainbow trout'])
    def test_upload_crbanim_errors(self, my_check):
        """Testing importing with data into UID with errors"""

        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())

        # check async message called
        message = 'Error'
        notification_message = (
            'Error in importing data: Some species '
            'are not loaded in UID database: Rainbow '
            'trout')

        self.check_message(message, notification_message)

    @patch("crbanim.helpers.CRBAnimReader.check_sex",

           return_value=[False, 'unknown'])
    def test_upload_crbanim_errors_with_sex(self, my_check):
        """Testing importing with data into UID with errors"""

        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())

        # check async message called
        message = 'Error'
        notification_message = (
            'Error in importing data: Not all Sex '
            'terms are loaded into database: check '
            'for unknown in your dataset')

        self.check_message(message, notification_message)


class ReloadCRBAnimTestCase(WebSocketMixin, 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'
    ]

    def test_upload_crbanim(self):

        """Testing uploading and importing data from crbanim to UID"""

        # 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())

        # check async message called
        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}

        self.check_message(message, notification_message, validation_message)


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 2019-06-20 15:48 UTC by Report Bug Copy Issue Report 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 2019-06-20 15:48 UTC by Report Bug Copy Issue Report 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 2019-06-20 15:48 UTC by Report Bug Copy Issue Report 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 2019-06-20 15:48 UTC by Report Bug Copy Issue Report 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 2019-06-20 15:48 UTC by Report Bug Copy Issue Report 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

cnr-ibba / IMAGE-InjectTool

Pull Request — master (#35)

WebSocketMixin.check_message() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like