cryoweb.tests.test_helpers.ImportMixin.test_cryoweb_import() - 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-21 15:22 UTC

ImportMixin.test_cryoweb_import() A

↳ Parent: cryoweb.tests.test_helpers

Complexity

Conditions

Size

Total Lines	41
Code Lines	27

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	27
dl	0
loc	41
rs	9.232
c	0
b	0
f	0
cc	1
nop	1

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Created on Thu Sep 20 16:01:04 2018

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

# --- import
from unittest.mock import patch, Mock

from django.conf import settings
from django.core.management import call_command
from django.test import TestCase

from common.constants import ERROR
from language.models import SpecieSynonym
from image_app.models import (
    Submission, DictBreed, Name, Animal, Sample, DictSex,
    DictCountry, DictSpecie)
from common.tests import DataSourceMixinTestCase

from ..helpers import (
    upload_cryoweb, check_species, CryoWebImportError, cryoweb_import,
    check_UID)
from ..models import db_has_data, truncate_database, BreedsSpecies


class BaseMixin():
    # import this file and populate database once
    fixtures = [
        'cryoweb/dictbreed',
        'image_app/dictcountry',
        'image_app/dictrole',
        'image_app/dictsex',
        'image_app/organization',
        'image_app/submission',
        'image_app/user',
        'language/dictspecie',
        'language/speciesynonym'
    ]

    # By default, fixtures are only loaded into the default database. If you
    # are using multiple databases and set multi_db=True, fixtures will be
    # loaded into all databases. However, this will raise problems when
    # managing extended user models
    multi_db = False

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

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


class CryoWebMixin(object):
    """Custom methods to upload cryoweb data into database for testing"""

    @classmethod
    def setUpClass(cls):
        # calling my base class setup
        super().setUpClass()

        # this fixture have to be loaded in a secondary (test) database,
        # I can't upload it using names and fixture section, so it will
        # be added manually using loaddata
        call_command(
            'loaddata',
            'cryoweb/cryoweb.json',
            app='cryoweb',
            database='cryoweb',
            verbosity=0)

    @classmethod
    def tearDownClass(cls):
        # truncate cryoweb database after loading
        if db_has_data():
            truncate_database()

        # calling my base class teardown class
        super().tearDownClass()


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(
            'cryoweb.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 ImportMixin(WebSocketMixin, CryoWebMixin):
    """Mixing to test cryoweb_import method"""

    def test_cryoweb_import(self):
        """A method to test if data were imported from cryoweb or not"""

        self.assertTrue(cryoweb_import(self.submission))

        # check breed upload
        queryset = DictBreed.objects.all()

        breeds = [(dictbreed.supplied_breed, dictbreed.country.label)
                  for dictbreed in queryset]

        self.assertEqual(len(queryset), 4)
        self.assertListEqual(
            breeds, [
                ('Bunte Bentheimer', 'United Kingdom'),
                ('Ostfriesisches Milchschaf', 'Italy'),
                ('Aberdeen Angus', 'Germany'),
                ('Ostfriesisches Milchschaf', 'Germany')],
            msg="Check breeds loaded")

        # check name upload (5 animal, 1 sample)
        queryset = Name.objects.filter(submission=self.submission)
        self.assertEqual(len(queryset), 6, msg='check name load')

        # check animal name
        queryset = Animal.objects.all()
        self.assertEqual(len(queryset), 3, msg="check animal load")

        queryset = Sample.objects.all()
        self.assertEqual(len(queryset), 1, msg="check sample load")

        # check async message called
        message = 'Loaded'
        notification_message = (
            'Cryoweb import completed for submission: 1')
        validation_message = {
            'animals': 3, 'samples': 1,
            'animal_unkn': 3, 'sample_unkn': 1,
            'animal_issues': 0, 'sample_issues': 0}

        self.check_message(message, notification_message, validation_message)


class CheckSpecie(CryoWebMixin, BaseMixin, TestCase):
    def test_check_species(self):
        """Testing species and synonyms"""

        united_kingdom = DictCountry.objects.get(label="United Kingdom")

        # test for species synonoms for cryoweb
        self.assertTrue(check_species(united_kingdom))

        # now delete a synonym
        synonym = SpecieSynonym.objects.get(
            language__label='United Kingdom',
            word='Cattle')
        synonym.delete()

        self.assertFalse(check_species(united_kingdom))

    def test_no_species(self):
        """Test no species in cryoweb database"""

        queryset = BreedsSpecies.objects.all()
        queryset.delete()

        self.assertRaisesRegex(
            CryoWebImportError,
            "You have no species",
            check_species, "United Kingdom")


class CheckBreed(TestCase):
    # import this file and populate database once
    fixtures = [
        'image_app/dictbreed',
        'image_app/dictcountry',
        'image_app/dictrole',
        'image_app/dictsex',
        'image_app/dictspecie',
        'image_app/organization',
        'image_app/submission',
        'image_app/user',
    ]

    def test_add_breed(self):
        italy = DictCountry.objects.get(label="Italy")
        united_kingdom = DictCountry.objects.get(label="United Kingdom")

        sus = DictSpecie.objects.get(label="Sus scrofa")

        # inserting an already present breed get the object without creation
        breed, created = DictBreed.objects.get_or_create(
            supplied_breed="Bunte Bentheimer",
            specie=sus,
            country=united_kingdom)

        self.assertFalse(created)

        # inserting a breed in a different country add a record
        breed, created = DictBreed.objects.get_or_create(
            supplied_breed="Bunte Bentheimer",
            specie=sus,
            country=italy)

        self.assertTrue(created)


class CheckUIDTest(CryoWebMixin, BaseMixin, TestCase):
    def test_empty_dictsex(self):
        """Empty dictsex and check that I can't proceed"""

        queryset = DictSex.objects.all()
        queryset.delete()

        self.assertRaisesRegex(
            CryoWebImportError,
            "You have to upload DictSex data",
            check_UID, self.submission)

    def test_no_synonym(self):
        # now delete a synonym
        synonym = SpecieSynonym.objects.get(
            language__label='United Kingdom',
            word='Cattle')
        synonym.delete()

        self.assertRaisesRegex(
            CryoWebImportError,
            "Some species haven't a synonym!",
            check_UID, self.submission)

    def test_check_UID(self):
        """testing normal behaviour"""

        self.assertTrue(check_UID(self.submission))


class UploadCryoweb(
        WebSocketMixin, DataSourceMixinTestCase, BaseMixin, TestCase):
    """Test upload cryoweb dump into cryoweb database"""

    # define attribute in DataSourceMixinTestCase
    model = Submission

    # need to clean database after testing import. Can't use CryowebMixin
    # since i need to test cryoweb import
    def tearDown(self):
        """Clean test database after modifications"""

        # truncate cryoweb database after loading
        if db_has_data():
            truncate_database()

        # calling my base class teardown class
        super().tearDown()

    def test_database_name(self):
        self.assertEqual(
            settings.DATABASES['cryoweb']['NAME'], 'test_cryoweb')

    def test_upload_cryoweb(self):
        """Testing uploading and importing data from cryoweb to UID"""

        self.assertTrue(upload_cryoweb(self.submission.id))
        self.assertTrue(db_has_data())

        # if I try again to upload cryoweb, I will get a False object and
        # submission message
        self.assertRaises(
            CryoWebImportError,
            upload_cryoweb,
            self.submission.id)

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

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

        self.assertIn(
            "Cryoweb has data",
            self.submission.message)

        # check async message called
        message = 'Error'
        notification_message = 'Error in importing data: Cryoweb has data'

        self.check_message(message, notification_message)

    # mock subprocess.run and raise Exception. Read it and update submission
    # message using helpers.upload_cryoweb
    def test_upload_cryoweb_errors(self):
        """Testing errors in uploading cryoweb data"""

        with patch('subprocess.run') as runMock:
            runMock.side_effect = Exception("Test upload failed")
            self.assertFalse(upload_cryoweb(self.submission.id))

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

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

            self.assertIn(
                "Test upload failed",
                self.submission.message)

            # check async message called
            message = 'Error'
            notification_message = ('Error in importing data: Test '
                                    'upload failed')

            self.check_message(message, notification_message)


class CryowebImport(ImportMixin, BaseMixin, TestCase):
    def test_database_name(self):
        self.assertEqual(
            settings.DATABASES['cryoweb']['NAME'], 'test_cryoweb')

    @patch("cryoweb.helpers.check_UID", side_effect=Exception("Test message"))
    def test_cryoweb_import_errors(self, my_check):
        """Testing importing with data into UID with errors"""

        self.assertFalse(cryoweb_import(self.submission))
        self.assertTrue(my_check.called)

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

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

        self.assertIn(
            "Test message",
            self.submission.message)

        # check async message called
        message = 'Error'
        notification_message = (
            'Error in importing data: Test message')

        self.check_message(message, notification_message)


class CryowebReload(ImportMixin, BaseMixin, TestCase):
    """Simulate a cryoweb reload case. Load data as in CryowebImport, then
    call test which reload the same data"""

    # override fixtures
    fixtures = [
        'cryoweb/auth',
        'cryoweb/dictbreed',
        'cryoweb/image_app',
        'language/dictspecie',
        'language/speciesynonym'
    ]


1		#!/usr/bin/env python3
2		# -- coding: utf-8 --
3		"""
4		Created on Thu Sep 20 16:01:04 2018
5
6		@author: Paolo Cozzi <[email protected]>
7		"""
8
9		# --- import
10		from unittest.mock import patch, Mock
11
12		from django.conf import settings
13		from django.core.management import call_command
14		from django.test import TestCase
15
16		from common.constants import ERROR
17		from language.models import SpecieSynonym
18		from image_app.models import (
19		Submission, DictBreed, Name, Animal, Sample, DictSex,
20		DictCountry, DictSpecie)
21		from common.tests import DataSourceMixinTestCase
22
23		from ..helpers import (
24		upload_cryoweb, check_species, CryoWebImportError, cryoweb_import,
25		check_UID)
26		from ..models import db_has_data, truncate_database, BreedsSpecies
27
28
29		class BaseMixin():
30		# import this file and populate database once
31		fixtures = [
32		'cryoweb/dictbreed',
33		'image_app/dictcountry',
34		'image_app/dictrole',
35		'image_app/dictsex',
36		'image_app/organization',
37		'image_app/submission',
38		'image_app/user',
39		'language/dictspecie',
40		'language/speciesynonym'
41		]
42
43		# By default, fixtures are only loaded into the default database. If you
44		# are using multiple databases and set multi_db=True, fixtures will be
45		# loaded into all databases. However, this will raise problems when
46		# managing extended user models
47		multi_db = False
48
49		def setUp(self):
50		# calling my base class setup
51		super().setUp()
52
53		# track submission
54		self.submission = Submission.objects.get(pk=1)
55
56
57		class CryoWebMixin(object):
58		"""Custom methods to upload cryoweb data into database for testing"""
59
60		@classmethod
61		def setUpClass(cls):
62		# calling my base class setup
63		super().setUpClass()
64
65		# this fixture have to be loaded in a secondary (test) database,
66		# I can't upload it using names and fixture section, so it will
67		# be added manually using loaddata
68		call_command(
69		'loaddata',
70		'cryoweb/cryoweb.json',
71		app='cryoweb',
72		database='cryoweb',
73		verbosity=0)
74
75		@classmethod
76		def tearDownClass(cls):
77		# truncate cryoweb database after loading
78		if db_has_data():
79		truncate_database()
80
81		# calling my base class teardown class
82		super().tearDownClass()
83
84
85	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...
86		"""Override setUp to mock websocket objects"""
87
88		def setUp(self):
89		# calling my base class setup
90		super().setUp()
91
92		# setting channels methods
93		self.asyncio_mock_patcher = patch(
94		'asyncio.get_event_loop')
95		self.asyncio_mock = self.asyncio_mock_patcher.start()
96
97		# mocking asyncio return value
98		self.run_until = self.asyncio_mock.return_value
99		self.run_until.run_until_complete = Mock()
100
101		# another patch
102		self.send_msg_ws_patcher = patch(
103		'cryoweb.helpers.send_message_to_websocket')
104		self.send_msg_ws = self.send_msg_ws_patcher.start()
105
106		def tearDown(self):
107		# stopping mock objects
108		self.asyncio_mock_patcher.stop()
109		self.send_msg_ws_patcher.stop()
110
111		# calling base methods
112		super().tearDown()
113
114		def check_message(
115		self, message, notification_message, validation_message=None,
116		pk=1):
117		"""assert message to websocket called with parameters"""
118
119		# construct message according parameters
120		message = {
121		'message': message,
122		'notification_message': notification_message
123		}
124
125		# in case of successful data upload, a validation message is sent
126		if validation_message:
127		message['validation_message'] = validation_message
128
129		self.assertEqual(self.asyncio_mock.call_count, 1)
130		self.assertEqual(self.run_until.run_until_complete.call_count, 1)
131		self.assertEqual(self.send_msg_ws.call_count, 1)
132		self.send_msg_ws.assert_called_with(
133		message,
134		pk)
135
136
137		class ImportMixin(WebSocketMixin, CryoWebMixin):
138		"""Mixing to test cryoweb_import method"""
139
140		def test_cryoweb_import(self):
141		"""A method to test if data were imported from cryoweb or not"""
142
143		self.assertTrue(cryoweb_import(self.submission))
144
145		# check breed upload
146		queryset = DictBreed.objects.all()
147
148		breeds = [(dictbreed.supplied_breed, dictbreed.country.label)
149		for dictbreed in queryset]
150
151		self.assertEqual(len(queryset), 4)
152		self.assertListEqual(
153		breeds, [
154		('Bunte Bentheimer', 'United Kingdom'),
155		('Ostfriesisches Milchschaf', 'Italy'),
156		('Aberdeen Angus', 'Germany'),
157		('Ostfriesisches Milchschaf', 'Germany')],
158		msg="Check breeds loaded")
159
160		# check name upload (5 animal, 1 sample)
161		queryset = Name.objects.filter(submission=self.submission)
162		self.assertEqual(len(queryset), 6, msg='check name load')
163
164		# check animal name
165		queryset = Animal.objects.all()
166		self.assertEqual(len(queryset), 3, msg="check animal load")
167
168		queryset = Sample.objects.all()
169		self.assertEqual(len(queryset), 1, msg="check sample load")
170
171		# check async message called
172		message = 'Loaded'
173		notification_message = (
174		'Cryoweb import completed for submission: 1')
175		validation_message = {
176		'animals': 3, 'samples': 1,
177		'animal_unkn': 3, 'sample_unkn': 1,
178		'animal_issues': 0, 'sample_issues': 0}
179
180		self.check_message(message, notification_message, validation_message)
181
182
183		class CheckSpecie(CryoWebMixin, BaseMixin, TestCase):
184		def test_check_species(self):
185		"""Testing species and synonyms"""
186
187		united_kingdom = DictCountry.objects.get(label="United Kingdom")
188
189		# test for species synonoms for cryoweb
190		self.assertTrue(check_species(united_kingdom))
191
192		# now delete a synonym
193		synonym = SpecieSynonym.objects.get(
194		language__label='United Kingdom',
195		word='Cattle')
196		synonym.delete()
197
198		self.assertFalse(check_species(united_kingdom))
199
200		def test_no_species(self):
201		"""Test no species in cryoweb database"""
202
203		queryset = BreedsSpecies.objects.all()
204		queryset.delete()
205
206		self.assertRaisesRegex(
207		CryoWebImportError,
208		"You have no species",
209		check_species, "United Kingdom")
210
211
212		class CheckBreed(TestCase):
213		# import this file and populate database once
214		fixtures = [
215		'image_app/dictbreed',
216		'image_app/dictcountry',
217		'image_app/dictrole',
218		'image_app/dictsex',
219		'image_app/dictspecie',
220		'image_app/organization',
221		'image_app/submission',
222		'image_app/user',
223		]
224
225		def test_add_breed(self):
226		italy = DictCountry.objects.get(label="Italy")
227		united_kingdom = DictCountry.objects.get(label="United Kingdom")
228
229		sus = DictSpecie.objects.get(label="Sus scrofa")
230
231		# inserting an already present breed get the object without creation
232		breed, created = DictBreed.objects.get_or_create(
233		supplied_breed="Bunte Bentheimer",
234		specie=sus,
235		country=united_kingdom)
236
237		self.assertFalse(created)
238
239		# inserting a breed in a different country add a record
240		breed, created = DictBreed.objects.get_or_create(
241		supplied_breed="Bunte Bentheimer",
242		specie=sus,
243		country=italy)
244
245		self.assertTrue(created)
246
247
248		class CheckUIDTest(CryoWebMixin, BaseMixin, TestCase):
249		def test_empty_dictsex(self):
250		"""Empty dictsex and check that I can't proceed"""
251
252		queryset = DictSex.objects.all()
253		queryset.delete()
254
255		self.assertRaisesRegex(
256		CryoWebImportError,
257		"You have to upload DictSex data",
258		check_UID, self.submission)
259
260		def test_no_synonym(self):
261		# now delete a synonym
262		synonym = SpecieSynonym.objects.get(
263		language__label='United Kingdom',
264		word='Cattle')
265		synonym.delete()
266
267		self.assertRaisesRegex(
268		CryoWebImportError,
269		"Some species haven't a synonym!",
270		check_UID, self.submission)
271
272		def test_check_UID(self):
273		"""testing normal behaviour"""
274
275		self.assertTrue(check_UID(self.submission))
276
277
278		class UploadCryoweb(
279		WebSocketMixin, DataSourceMixinTestCase, BaseMixin, TestCase):
280		"""Test upload cryoweb dump into cryoweb database"""
281
282		# define attribute in DataSourceMixinTestCase
283		model = Submission
284
285		# need to clean database after testing import. Can't use CryowebMixin
286		# since i need to test cryoweb import
287		def tearDown(self):
288		"""Clean test database after modifications"""
289
290		# truncate cryoweb database after loading
291		if db_has_data():
292		truncate_database()
293
294		# calling my base class teardown class
295		super().tearDown()
296
297		def test_database_name(self):
298		self.assertEqual(
299		settings.DATABASES['cryoweb']['NAME'], 'test_cryoweb')
300
301		def test_upload_cryoweb(self):
302		"""Testing uploading and importing data from cryoweb to UID"""
303
304		self.assertTrue(upload_cryoweb(self.submission.id))
305		self.assertTrue(db_has_data())
306
307		# if I try again to upload cryoweb, I will get a False object and
308		# submission message
309		self.assertRaises(
310		CryoWebImportError,
311		upload_cryoweb,
312		self.submission.id)
313
314		# reload submission
315		self.submission = Submission.objects.get(pk=1)
316
317		self.assertEqual(
318		self.submission.status,
319		ERROR)
320
321		self.assertIn(
322		"Cryoweb has data",
323		self.submission.message)
324
325		# check async message called
326		message = 'Error'
327		notification_message = 'Error in importing data: Cryoweb has data'
328
329		self.check_message(message, notification_message)
330
331		# mock subprocess.run and raise Exception. Read it and update submission
332		# message using helpers.upload_cryoweb
333		def test_upload_cryoweb_errors(self):
334		"""Testing errors in uploading cryoweb data"""
335
336		with patch('subprocess.run') as runMock:
337		runMock.side_effect = Exception("Test upload failed")
338		self.assertFalse(upload_cryoweb(self.submission.id))
339
340		# reload submission
341		self.submission = Submission.objects.get(pk=1)
342
343		self.assertEqual(
344		self.submission.status,
345		ERROR)
346
347		self.assertIn(
348		"Test upload failed",
349		self.submission.message)
350
351		# check async message called
352		message = 'Error'
353		notification_message = ('Error in importing data: Test '
354		'upload failed')
355
356		self.check_message(message, notification_message)
357
358
359		class CryowebImport(ImportMixin, BaseMixin, TestCase):
360		def test_database_name(self):
361		self.assertEqual(
362		settings.DATABASES['cryoweb']['NAME'], 'test_cryoweb')
363
364		@patch("cryoweb.helpers.check_UID", side_effect=Exception("Test message"))
365		def test_cryoweb_import_errors(self, my_check):
366		"""Testing importing with data into UID with errors"""
367
368		self.assertFalse(cryoweb_import(self.submission))
369		self.assertTrue(my_check.called)
370
371		# reload submission
372		self.submission = Submission.objects.get(pk=1)
373
374		self.assertEqual(
375		self.submission.status,
376		ERROR)
377
378		self.assertIn(
379		"Test message",
380		self.submission.message)
381
382		# check async message called
383		message = 'Error'
384		notification_message = (
385		'Error in importing data: Test message')
386
387		self.check_message(message, notification_message)
388
389
390		class CryowebReload(ImportMixin, BaseMixin, TestCase):
391		"""Simulate a cryoweb reload case. Load data as in CryowebImport, then
392		call test which reload the same data"""
393
394		# override fixtures
395		fixtures = [
396		'cryoweb/auth',
397		'cryoweb/dictbreed',
398		'cryoweb/image_app',
399		'language/dictspecie',
400		'language/speciesynonym'
401		]
402

cnr-ibba / IMAGE-InjectTool

Pull Request — master (#35)

ImportMixin.test_cryoweb_import() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like