samples.tests.test_tasks.BatchUpdateSamplesTest.test_on_failure() - Code Metrics - Inspection of "Validation batch update" - cnr-ibba/IMAGE-InjectTool - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#41)

by Paolo

created 2019-09-19 12:04 UTC

BatchUpdateSamplesTest.test_on_failure() A

↳ Parent: samples.tests.test_tasks

Complexity

Conditions

Size

Total Lines	34
Code Lines	20

Duplication

Lines	34
Ratio	100 %

Importance

Changes

Metric	Value
eloc	20
dl	34
loc	34
rs	9.4
c	0
b	0
f	0
cc	1
nop	1

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Created on Mon Sep  9 17:24:54 2019

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

from billiard.einfo import ExceptionInfo

from django.core import mail
from django.test import TestCase

from image_app.models import Submission, Sample
from common.constants import NEED_REVISION, STATUSES, ERROR
from common.tests import WebSocketMixin

from .common import SampleFeaturesMixin
from ..tasks import BatchDeleteSamples, BatchUpdateSamples


class BatchDeleteSamplesTest(
        SampleFeaturesMixin, WebSocketMixin, TestCase):

    def setUp(self):
        # calling base methods
        super().setUp()

        # get a submission object
        self.submission = Submission.objects.get(pk=1)
        self.submission_id = self.submission.id

        # setting samples to delete
        self.sample_ids = ['Siems_0722_393449']

        # setting tasks
        self.my_task = BatchDeleteSamples()

    def test_on_failure(self):
        """Testing on failure methods"""

        exc = Exception("Test")
        task_id = "test_task_id"
        args = [self.submission_id, self.sample_ids]
        kwargs = {}
        einfo = ExceptionInfo

        # call on_failure method
        self.my_task.on_failure(exc, task_id, args, kwargs, einfo)

        # check submission status and message
        self.submission.refresh_from_db()

        # check submission.state changed
        self.assertEqual(self.submission.status, ERROR)
        self.assertEqual(
            self.submission.message,
            "Error in sample batch delete: Test")

        # test email sent
        self.assertGreater(len(mail.outbox), 0)

        # read email
        email = mail.outbox[0]

        self.assertEqual(
            "Error in sample batch delete for submission: 1",
            email.subject)

        self.check_message(
            message='Error',
            notification_message='Error in sample batch delete: Test')

    def test_delete_sample(self):
        # calling task and delete a sample
        res = self.my_task.run(
            submission_id=self.submission.id,
            sample_ids=self.sample_ids)

        self.assertEqual(res, "success")

        # no samples
        n_samples = Sample.objects.count()
        self.assertEqual(n_samples, 0)

        # updating validation messages

        # calling a WebSocketMixin method
        # no validation message since no data in validation table
        self.check_message(
            message=STATUSES.get_value_display(NEED_REVISION),
            notification_message=(
                "You've removed %s "
                "samples. Rerun validation please!" % len(self.sample_ids)))

    def test_delete_samples_not_exists(self):
        # calling task and delete a sample
        res = self.my_task.run(
            submission_id=self.submission_id,
            sample_ids=["meow"])

        self.assertEqual(res, "success")

        # all samples remain
        n_samples = Sample.objects.count()
        self.assertEqual(n_samples, 1)

        # updating validation messages

        # calling a WebSocketMixin method
        # no validation message since no data in validation table
        self.check_message(
            message=STATUSES.get_value_display(NEED_REVISION),
            notification_message=(
                "You've removed 0 samples. It wasn't possible to find records "
                "with these ids: meow. Rerun validation please!")
        )


class BatchUpdateSamplesTest(

        SampleFeaturesMixin, WebSocketMixin, TestCase):

    def setUp(self):
        # calling base methods
        super().setUp()

        # get a submission object
        self.submission = Submission.objects.get(pk=1)
        self.submission_id = self.submission.id

        # setting samples to delete
        self.sample_ids = {
            1: "meow",
        }

        # the attribute to change
        self.attribute = "collection_place"

        # setting tasks
        self.my_task = BatchUpdateSamples()

    def test_on_failure(self):
        """Testing on failure methods"""

        exc = Exception("Test")
        task_id = "test_task_id"
        args = [self.submission_id, self.sample_ids, self.attribute]
        kwargs = {}
        einfo = ExceptionInfo

        # call on_failure method
        self.my_task.on_failure(exc, task_id, args, kwargs, einfo)

        # check submission status and message
        self.submission.refresh_from_db()

        # check submission.state changed
        self.assertEqual(self.submission.status, ERROR)
        self.assertEqual(
            self.submission.message,
            "Error in sample batch update: Test")

        # test email sent
        self.assertGreater(len(mail.outbox), 0)

        # read email
        email = mail.outbox[0]

        self.assertEqual(
            "Error in sample batch update for submission: 1",
            email.subject)

        self.check_message(
            message='Error',
            notification_message='Error in sample batch update: Test')

    def test_update_sample(self):
        # calling task and update a sample
        res = self.my_task.run(
            submission_id=self.submission_id,
            sample_ids=self.sample_ids,
            attribute=self.attribute)

        self.assertEqual(res, "success")

        # asserting updates
        for key, value in self.sample_ids.items():
            sample = Sample.objects.get(pk=key)
            if value == "None":
                value = None
            self.assertEqual(getattr(sample, self.attribute), value)

        # calling a WebSocketMixin method
        # no validation message since no data in validation table
        self.check_message(
            message=STATUSES.get_value_display(NEED_REVISION),
            notification_message="Data updated, try to rerun validation",
        )


1		#!/usr/bin/env python3
2		# -- coding: utf-8 --
3		"""
4		Created on Mon Sep 9 17:24:54 2019
5
6		@author: Paolo Cozzi <[email protected]>
7		"""
8
9		from billiard.einfo import ExceptionInfo
10
11		from django.core import mail
12		from django.test import TestCase
13
14		from image_app.models import Submission, Sample
15		from common.constants import NEED_REVISION, STATUSES, ERROR
16		from common.tests import WebSocketMixin
17
18		from .common import SampleFeaturesMixin
19		from ..tasks import BatchDeleteSamples, BatchUpdateSamples
20
21
22		class BatchDeleteSamplesTest(
23		SampleFeaturesMixin, WebSocketMixin, TestCase):
24
25		def setUp(self):
26		# calling base methods
27		super().setUp()
28
29		# get a submission object
30		self.submission = Submission.objects.get(pk=1)
31		self.submission_id = self.submission.id
32
33		# setting samples to delete
34		self.sample_ids = ['Siems_0722_393449']
35
36		# setting tasks
37		self.my_task = BatchDeleteSamples()
38
39		def test_on_failure(self):
40		"""Testing on failure methods"""
41
42		exc = Exception("Test")
43		task_id = "test_task_id"
44		args = [self.submission_id, self.sample_ids]
45		kwargs = {}
46		einfo = ExceptionInfo
47
48		# call on_failure method
49		self.my_task.on_failure(exc, task_id, args, kwargs, einfo)
50
51		# check submission status and message
52		self.submission.refresh_from_db()
53
54		# check submission.state changed
55		self.assertEqual(self.submission.status, ERROR)
56		self.assertEqual(
57		self.submission.message,
58		"Error in sample batch delete: Test")
59
60		# test email sent
61		self.assertGreater(len(mail.outbox), 0)
62
63		# read email
64		email = mail.outbox[0]
65
66		self.assertEqual(
67		"Error in sample batch delete for submission: 1",
68		email.subject)
69
70		self.check_message(
71		message='Error',
72		notification_message='Error in sample batch delete: Test')
73
74		def test_delete_sample(self):
75		# calling task and delete a sample
76		res = self.my_task.run(
77		submission_id=self.submission.id,
78		sample_ids=self.sample_ids)
79
80		self.assertEqual(res, "success")
81
82		# no samples
83		n_samples = Sample.objects.count()
84		self.assertEqual(n_samples, 0)
85
86		# updating validation messages
87
88		# calling a WebSocketMixin method
89		# no validation message since no data in validation table
90		self.check_message(
91		message=STATUSES.get_value_display(NEED_REVISION),
92		notification_message=(
93		"You've removed %s "
94		"samples. Rerun validation please!" % len(self.sample_ids)))
95
96		def test_delete_samples_not_exists(self):
97		# calling task and delete a sample
98		res = self.my_task.run(
99		submission_id=self.submission_id,
100		sample_ids=["meow"])
101
102		self.assertEqual(res, "success")
103
104		# all samples remain
105		n_samples = Sample.objects.count()
106		self.assertEqual(n_samples, 1)
107
108		# updating validation messages
109
110		# calling a WebSocketMixin method
111		# no validation message since no data in validation table
112		self.check_message(
113		message=STATUSES.get_value_display(NEED_REVISION),
114		notification_message=(
115		"You've removed 0 samples. It wasn't possible to find records "
116		"with these ids: meow. Rerun validation please!")
117		)
118
119
120	View Code Duplication	class BatchUpdateSamplesTest(
		0 ignored issues – show Duplication introduced 2019-09-18 15:24 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
121		SampleFeaturesMixin, WebSocketMixin, TestCase):
122
123		def setUp(self):
124		# calling base methods
125		super().setUp()
126
127		# get a submission object
128		self.submission = Submission.objects.get(pk=1)
129		self.submission_id = self.submission.id
130
131		# setting samples to delete
132		self.sample_ids = {
133		1: "meow",
134		}
135
136		# the attribute to change
137		self.attribute = "collection_place"
138
139		# setting tasks
140		self.my_task = BatchUpdateSamples()
141
142		def test_on_failure(self):
143		"""Testing on failure methods"""
144
145		exc = Exception("Test")
146		task_id = "test_task_id"
147		args = [self.submission_id, self.sample_ids, self.attribute]
148		kwargs = {}
149		einfo = ExceptionInfo
150
151		# call on_failure method
152		self.my_task.on_failure(exc, task_id, args, kwargs, einfo)
153
154		# check submission status and message
155		self.submission.refresh_from_db()
156
157		# check submission.state changed
158		self.assertEqual(self.submission.status, ERROR)
159		self.assertEqual(
160		self.submission.message,
161		"Error in sample batch update: Test")
162
163		# test email sent
164		self.assertGreater(len(mail.outbox), 0)
165
166		# read email
167		email = mail.outbox[0]
168
169		self.assertEqual(
170		"Error in sample batch update for submission: 1",
171		email.subject)
172
173		self.check_message(
174		message='Error',
175		notification_message='Error in sample batch update: Test')
176
177		def test_update_sample(self):
178		# calling task and update a sample
179		res = self.my_task.run(
180		submission_id=self.submission_id,
181		sample_ids=self.sample_ids,
182		attribute=self.attribute)
183
184		self.assertEqual(res, "success")
185
186		# asserting updates
187		for key, value in self.sample_ids.items():
188		sample = Sample.objects.get(pk=key)
189		if value == "None":
190		value = None
191		self.assertEqual(getattr(sample, self.attribute), value)
192
193		# calling a WebSocketMixin method
194		# no validation message since no data in validation table
195		self.check_message(
196		message=STATUSES.get_value_display(NEED_REVISION),
197		notification_message="Data updated, try to rerun validation",
198		)
199

cnr-ibba / IMAGE-InjectTool

Pull Request — master (#41)

BatchUpdateSamplesTest.test_on_failure() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like