zooma.tests.test_tasks.TestAnnotateAll.test_annotateall_nb() - Code Metrics - Inspection of ":sparkles: Structure ZOOMA calls into InjectTool" - cnr-ibba/IMAGE-InjectTool - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#51)

by Paolo

created 2019-09-26 15:16 UTC

TestAnnotateAll.test_annotateall_nb() A

↳ Parent: zooma.tests.test_tasks

Complexity

Conditions

Size

Total Lines	11
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	5
dl	0
loc	11
rs	10
c	0
b	0
f	0
cc	1
nop	2

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Created on Thu Oct 25 13:34:43 2018

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

from unittest.mock import patch, Mock

from django.test import TestCase

from image_app.models import (
    DictBreed, DictCountry, DictSpecie, DictUberon, DictDevelStage,
    DictPhysioStage)

from ..tasks import (
    AnnotateBreeds, AnnotateCountries, AnnotateSpecies, AnnotateUberon,
    AnnotateDictDevelStage, AnnotateDictPhysioStage, AnnotateAll)


class TestAnnotateBreeds(TestCase):
    """A class to test annotate breeds"""

    fixtures = [
        "image_app/dictbreed",
        "image_app/dictcountry",
        "image_app/dictspecie"
    ]

    def setUp(self):
        self.my_task = AnnotateBreeds()

        # get a breed object
        breed = DictBreed.objects.get(pk=1)

        # erase attributes
        breed.mapped_breed = None
        breed.mapped_breed_term = None
        breed.confidence = None
        breed.save()

    @patch("zooma.tasks.annotate_breed")
    def test_task(self, my_func):
        res = self.my_task.run()

        # assert a success
        self.assertEqual(res, "success")
        self.assertTrue(my_func.called)


class TestAnnotateCountries(TestCase):
    """A class to test annotate countries"""

    fixtures = [
        "image_app/dictcountry",
    ]

    def setUp(self):
        self.my_task = AnnotateCountries()

        # get a country object
        country = DictCountry.objects.get(pk=1)

        # erase attributes
        country.term = None
        country.confidence = None
        country.save()

    @patch("zooma.tasks.AnnotateCountries.annotate_func")
    def test_task(self, my_func):
        res = self.my_task.run()

        # assert a success
        self.assertEqual(res, "success")
        self.assertTrue(my_func.called)


class TestAnnotateSpecies(TestCase):
    """A class to test annotate species"""

    fixtures = [
        "image_app/dictspecie",
    ]

    def setUp(self):
        self.my_task = AnnotateSpecies()

        # get a specie object
        specie = DictSpecie.objects.get(pk=1)

        # erase attributes
        specie.term = None
        specie.confidence = None
        specie.save()

    @patch("zooma.tasks.AnnotateSpecies.annotate_func")
    def test_task(self, my_func):
        res = self.my_task.run()

        # assert a success
        self.assertEqual(res, "success")
        self.assertTrue(my_func.called)


class TestAnnotateUberon(TestCase):
    """A class to test annotate uberon"""

    fixtures = [
        "image_app/dictuberon",
    ]

    def setUp(self):
        self.my_task = AnnotateUberon()

        # get a specie object
        part = DictUberon.objects.get(pk=1)

        # erase attributes
        part.term = None
        part.confidence = None
        part.save()

    @patch("zooma.tasks.AnnotateUberon.annotate_func")
    def test_task(self, my_func):
        res = self.my_task.run()

        # assert a success
        self.assertEqual(res, "success")
        self.assertTrue(my_func.called)


class TestAnnotateDictDevelStage(TestCase):
    """A class to test annotate developmental stages"""

    fixtures = [
        "image_app/dictstage",
    ]

    def setUp(self):
        self.my_task = AnnotateDictDevelStage()

        # get a specie object
        stage = DictDevelStage.objects.get(pk=1)

        # erase attributes
        stage.term = None
        stage.confidence = None
        stage.save()

    @patch("zooma.tasks.AnnotateDictDevelStage.annotate_func")
    def test_task(self, my_func):
        res = self.my_task.run()

        # assert a success
        self.assertEqual(res, "success")
        self.assertTrue(my_func.called)


class TestAnnotateDictPhysioStage(TestCase):
    """A class to test annotate physiological stages"""

    fixtures = [
        "image_app/dictstage",
    ]

    def setUp(self):
        self.my_task = AnnotateDictPhysioStage()

        # get a specie object
        stage = DictPhysioStage.objects.get(pk=1)

        # erase attributes
        stage.term = None
        stage.confidence = None
        stage.save()

    @patch("zooma.tasks.AnnotateDictPhysioStage.annotate_func")
    def test_task(self, my_func):
        res = self.my_task.run()

        # assert a success
        self.assertEqual(res, "success")
        self.assertTrue(my_func.called)


class TestAnnotateAll(TestCase):

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

        # patching objects
        self.mock_group_patcher = patch('zooma.tasks.group')
        self.mock_group = self.mock_group_patcher.start()

        # define a group result object
        result = Mock()
        result.waiting.side_effect = [True, False]
        result.join.return_value = ["success"] * 6
        self.mock_group.return_value.delay.return_value = result

        # define my task
        self.my_task = AnnotateAll()

        # change lock_id (useful when running test during cron)
        self.my_task.lock_id = "test-AnnotateAll"

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

        # calling base object
        super().tearDown()

    @patch("time.sleep")
    def test_annotateall(self, my_time):
        """Test AnnotateAll while a lock is present"""

        res = self.my_task.run()

        # assert success in annotation
        self.assertEqual(res, "success")

        # assert mock objects called
        self.assertTrue(self.mock_group.called)
        self.assertTrue(self.mock_group.return_value.delay.called)
        self.assertTrue(my_time.called)

    # Test a non blocking instance
    @patch("redis.lock.Lock.acquire", return_value=False)
    def test_annotateall_nb(self, my_lock):
        """Test AnnotateAll while a lock is present"""

        res = self.my_task.run()

        # assert database is locked
        self.assertEqual(res, "%s already running!" % (self.my_task.name))

        # assert mock objects called
        self.assertFalse(self.mock_group.called)


1			#!/usr/bin/env python3
2			# -- coding: utf-8 --
3			"""
4			Created on Thu Oct 25 13:34:43 2018
5
6			@author: Paolo Cozzi <[email protected]>
7			"""
8
9			from unittest.mock import patch, Mock
10
11			from django.test import TestCase
12
13			from image_app.models import (
14			DictBreed, DictCountry, DictSpecie, DictUberon, DictDevelStage,
15			DictPhysioStage)
16
17			from ..tasks import (
18			AnnotateBreeds, AnnotateCountries, AnnotateSpecies, AnnotateUberon,
19			AnnotateDictDevelStage, AnnotateDictPhysioStage, AnnotateAll)
20
21
22			class TestAnnotateBreeds(TestCase):
23			"""A class to test annotate breeds"""
24
25			fixtures = [
26			"image_app/dictbreed",
27			"image_app/dictcountry",
28			"image_app/dictspecie"
29			]
30
31			def setUp(self):
32			self.my_task = AnnotateBreeds()
33
34			# get a breed object
35			breed = DictBreed.objects.get(pk=1)
36
37			# erase attributes
38			breed.mapped_breed = None
39			breed.mapped_breed_term = None
40			breed.confidence = None
41			breed.save()
42
43			@patch("zooma.tasks.annotate_breed")
44			def test_task(self, my_func):
45			res = self.my_task.run()
46
47			# assert a success
48			self.assertEqual(res, "success")
49			self.assertTrue(my_func.called)
50
51
52			class TestAnnotateCountries(TestCase):
53			"""A class to test annotate countries"""
54
55			fixtures = [
56			"image_app/dictcountry",
57			]
58
59			def setUp(self):
60			self.my_task = AnnotateCountries()
61
62			# get a country object
63			country = DictCountry.objects.get(pk=1)
64
65			# erase attributes
66			country.term = None
67			country.confidence = None
68			country.save()
69
70			@patch("zooma.tasks.AnnotateCountries.annotate_func")
71			def test_task(self, my_func):
72			res = self.my_task.run()
73
74			# assert a success
75			self.assertEqual(res, "success")
76			self.assertTrue(my_func.called)
77
78
79			class TestAnnotateSpecies(TestCase):
80			"""A class to test annotate species"""
81
82			fixtures = [
83			"image_app/dictspecie",
84			]
85
86			def setUp(self):
87			self.my_task = AnnotateSpecies()
88
89			# get a specie object
90			specie = DictSpecie.objects.get(pk=1)
91
92			# erase attributes
93			specie.term = None
94			specie.confidence = None
95			specie.save()
96
97			@patch("zooma.tasks.AnnotateSpecies.annotate_func")
98			def test_task(self, my_func):
99			res = self.my_task.run()
100
101			# assert a success
102			self.assertEqual(res, "success")
103			self.assertTrue(my_func.called)
104
105
106			class TestAnnotateUberon(TestCase):
107			"""A class to test annotate uberon"""
108
109			fixtures = [
110			"image_app/dictuberon",
111			]
112
113			def setUp(self):
114			self.my_task = AnnotateUberon()
115
116			# get a specie object
117			part = DictUberon.objects.get(pk=1)
118
119			# erase attributes
120			part.term = None
121			part.confidence = None
122			part.save()
123
124			@patch("zooma.tasks.AnnotateUberon.annotate_func")
125			def test_task(self, my_func):
126			res = self.my_task.run()
127
128			# assert a success
129			self.assertEqual(res, "success")
130			self.assertTrue(my_func.called)
131
132
133			class TestAnnotateDictDevelStage(TestCase):
134			"""A class to test annotate developmental stages"""
135
136			fixtures = [
137			"image_app/dictstage",
138			]
139
140			def setUp(self):
141			self.my_task = AnnotateDictDevelStage()
142
143			# get a specie object
144			stage = DictDevelStage.objects.get(pk=1)
145
146			# erase attributes
147			stage.term = None
148			stage.confidence = None
149			stage.save()
150
151			@patch("zooma.tasks.AnnotateDictDevelStage.annotate_func")
152			def test_task(self, my_func):
153			res = self.my_task.run()
154
155			# assert a success
156			self.assertEqual(res, "success")
157			self.assertTrue(my_func.called)
158
159
160			class TestAnnotateDictPhysioStage(TestCase):
161			"""A class to test annotate physiological stages"""
162
163			fixtures = [
164			"image_app/dictstage",
165			]
166
167			def setUp(self):
168			self.my_task = AnnotateDictPhysioStage()
169
170			# get a specie object
171			stage = DictPhysioStage.objects.get(pk=1)
172
173			# erase attributes
174			stage.term = None
175			stage.confidence = None
176			stage.save()
177
178			@patch("zooma.tasks.AnnotateDictPhysioStage.annotate_func")
179			def test_task(self, my_func):
180			res = self.my_task.run()
181
182			# assert a success
183			self.assertEqual(res, "success")
184			self.assertTrue(my_func.called)
185
186
187			class TestAnnotateAll(TestCase):
188
189			def setUp(self):
190			# calling my base setup
191			super().setUp()
192
193			# patching objects
194			self.mock_group_patcher = patch('zooma.tasks.group')
195			self.mock_group = self.mock_group_patcher.start()
196
197			# define a group result object
198			result = Mock()
199			result.waiting.side_effect = [True, False]
200			result.join.return_value = ["success"] * 6
201			self.mock_group.return_value.delay.return_value = result
202
203			# define my task
204			self.my_task = AnnotateAll()
205
206			# change lock_id (useful when running test during cron)
207			self.my_task.lock_id = "test-AnnotateAll"
208
209			def tearDown(self):
210			# stopping mock objects
211			self.mock_group_patcher.stop()
212
213			# calling base object
214			super().tearDown()
215
216			@patch("time.sleep")
217			def test_annotateall(self, my_time):
218			"""Test AnnotateAll while a lock is present"""
219
220			res = self.my_task.run()
221
222			# assert success in annotation
223			self.assertEqual(res, "success")
224
225			# assert mock objects called
226			self.assertTrue(self.mock_group.called)
227			self.assertTrue(self.mock_group.return_value.delay.called)
228			self.assertTrue(my_time.called)
229
230			# Test a non blocking instance
231			@patch("redis.lock.Lock.acquire", return_value=False)
232			def test_annotateall_nb(self, my_lock):
233			"""Test AnnotateAll while a lock is present"""
234
235			res = self.my_task.run()
236
237			# assert database is locked
238			self.assertEqual(res, "%s already running!" % (self.my_task.name))
239
240			# assert mock objects called
241			self.assertFalse(self.mock_group.called)
242

cnr-ibba / IMAGE-InjectTool

Pull Request — master (#51)

TestAnnotateAll.test_annotateall_nb() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like