common.tests.mixins.WebSocketMixin.tearDown() - Code Metrics - Inspection of ":twisted_rightwards_arrows: Merge pull request #97..." - cnr-ibba/IMAGE-InjectTool - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 55dedd...16663f )

by Paolo

created 2020-06-08 15:56 UTC

common.tests.mixins.WebSocketMixin.tearDown() A

↳ Parent: common.tests.mixins

Complexity

Conditions

Size

Total Lines	6
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	3
dl	0
loc	6
rs	10
c	0
b	0
f	0
cc	1
nop	1

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Created on Wed Mar 27 14:34:02 2019

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

import os
import shutil

from unittest.mock import patch, Mock

from django.urls import reverse
from django.test import Client
from django.contrib.messages import get_messages

from ..constants import LOADED, ERROR


class LoginMixinTestCase(object):
    url = None

    def test_unathenticated(self):
        '''Non Authenticated user are directed to login page'''

        login_url = reverse("login")
        client = Client()
        response = client.get(self.url)

        self.assertRedirects(
            response, '{login_url}?next={url}'.format(
                login_url=login_url, url=self.url)
        )


class OwnerMixinTestCase(LoginMixinTestCase):
    def test_ownership(self):
        """Test a non-owner having a 404 response"""

        client = Client()
        client.login(username='test2', password='test2')

        response = client.get(self.url)
        self.assertEqual(response.status_code, 404)


class StatusMixinTestCase(object):
    response = None

    def test_status_code(self):
        self.assertEqual(self.response.status_code, 200)


class MessageMixinTestCase(object):
    def check_messages(self, response, tag, message_text):

        """Check that a response has a message of a certain type"""

        # each element is an instance
        # of django.contrib.messages.storage.base.Message
        all_messages = [msg for msg in get_messages(response.wsgi_request)]

        found = False

        # I can have moltiple message, and maybe I need to find a specific one
        for message in all_messages:
            if tag in message.tags and message_text in message.message:
                found = True

        self.assertTrue(found)


class GeneralMixinTestCase(LoginMixinTestCase, StatusMixinTestCase,
                           MessageMixinTestCase):
    pass


class FormMixinTestCase(GeneralMixinTestCase):
    form_class = None

    def test_csrf(self):
        self.assertContains(self.response, 'csrfmiddlewaretoken')

    def test_contains_form(self):
        if not self.form_class:
            raise Exception("Please set 'form_class' attribute")

        form = self.response.context.get('form')
        self.assertIsInstance(form, self.form_class)


# methods like test_unathenticated need to be called with FormMixinTestCase
class InvalidFormMixinTestCase(StatusMixinTestCase, MessageMixinTestCase):

    def test_form_errors(self):
        form = self.response.context.get('form')
        self.assertGreater(len(form.errors), 0)


# a mixin to ensure that datasource is in place
class DataSourceMixinTestCase(object):
    """Place file in data source directory"""

    # class attributes
    uploaded_file = False
    dst_path = None
    submission_model = None
    base_dir = os.path.dirname(os.path.abspath(__file__))

    # the method is use to upload data into database
    upload_method = None

    @classmethod
    def setUpClass(cls):
        """Place test data in data source if needed"""

        # calling my base class setup
        super().setUpClass()

        # ensuring that file cryoweb_test_data_only.sql is present
        submission = cls.submission_model.objects.get(pk=1)
        cls.dst_path = submission.uploaded_file.path

        # get datasource filename from submission object
        datasource_filename = os.path.basename(submission.uploaded_file.path)

        if not os.path.exists(cls.dst_path):
            src_path = os.path.join(cls.base_dir, datasource_filename)
            shutil.copy(src_path, cls.dst_path)
            cls.uploaded_file = True

    @classmethod
    def tearDownClass(cls):
        """Remove test data from data source if needed"""

        # remove file if I placed it for tests
        if cls.uploaded_file:
            os.remove(cls.dst_path)

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

    def upload_datasource(self, message):
        """Testing uploading and importing data from crbanim to UID"""

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

        # reload submission
        self.submission.refresh_from_db()

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

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

    def check_errors(self, my_check, message):
        """Common stuff for error in file loading"""

        self.assertFalse(self.upload_method(self.submission))

        # reload submission
        self.submission.refresh_from_db()

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

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

        # check for two distinct messages
        self.assertIn(
            message,
            self.submission.message)

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


class AsyncIOMixin(object):
    """A common mixing for async 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()

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

        # calling base methods
        super().tearDown()


# TODO: move into submission.tests (since related to submission.helpers)
class WebSocketMixin(AsyncIOMixin):
    """Override setUp to mock websocket objects"""

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

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

    def tearDown(self):
        # stopping mock objects
        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)

    def check_message_not_called(self):
        """Check django channels async messages not called"""

        self.assertEqual(self.asyncio_mock.call_count, 0)
        self.assertEqual(self.run_until.run_until_complete.call_count, 0)
        self.assertEqual(self.send_msg_ws.call_count, 0)


1		#!/usr/bin/env python3
2		# -- coding: utf-8 --
3		"""
4		Created on Wed Mar 27 14:34:02 2019
5
6		@author: Paolo Cozzi <[email protected]>
7		"""
8
9		import os
10		import shutil
11
12		from unittest.mock import patch, Mock
13
14		from django.urls import reverse
15		from django.test import Client
16		from django.contrib.messages import get_messages
17
18		from ..constants import LOADED, ERROR
19
20
21		class LoginMixinTestCase(object):
22		url = None
23
24		def test_unathenticated(self):
25		'''Non Authenticated user are directed to login page'''
26
27		login_url = reverse("login")
28		client = Client()
29		response = client.get(self.url)
30
31		self.assertRedirects(
32		response, '{login_url}?next={url}'.format(
33		login_url=login_url, url=self.url)
34		)
35
36
37		class OwnerMixinTestCase(LoginMixinTestCase):
38		def test_ownership(self):
39		"""Test a non-owner having a 404 response"""
40
41		client = Client()
42		client.login(username='test2', password='test2')
43
44		response = client.get(self.url)
45		self.assertEqual(response.status_code, 404)
46
47
48		class StatusMixinTestCase(object):
49		response = None
50
51		def test_status_code(self):
52		self.assertEqual(self.response.status_code, 200)
53
54
55		class MessageMixinTestCase(object):
56	View Code Duplication	def check_messages(self, response, tag, message_text):
		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...
57		"""Check that a response has a message of a certain type"""
58
59		# each element is an instance
60		# of django.contrib.messages.storage.base.Message
61		all_messages = [msg for msg in get_messages(response.wsgi_request)]
62
63		found = False
64
65		# I can have moltiple message, and maybe I need to find a specific one
66		for message in all_messages:
67		if tag in message.tags and message_text in message.message:
68		found = True
69
70		self.assertTrue(found)
71
72
73		class GeneralMixinTestCase(LoginMixinTestCase, StatusMixinTestCase,
74		MessageMixinTestCase):
75		pass
76
77
78		class FormMixinTestCase(GeneralMixinTestCase):
79		form_class = None
80
81		def test_csrf(self):
82		self.assertContains(self.response, 'csrfmiddlewaretoken')
83
84		def test_contains_form(self):
85		if not self.form_class:
86		raise Exception("Please set 'form_class' attribute")
87
88		form = self.response.context.get('form')
89		self.assertIsInstance(form, self.form_class)
90
91
92		# methods like test_unathenticated need to be called with FormMixinTestCase
93		class InvalidFormMixinTestCase(StatusMixinTestCase, MessageMixinTestCase):
94
95		def test_form_errors(self):
96		form = self.response.context.get('form')
97		self.assertGreater(len(form.errors), 0)
98
99
100		# a mixin to ensure that datasource is in place
101		class DataSourceMixinTestCase(object):
102		"""Place file in data source directory"""
103
104		# class attributes
105		uploaded_file = False
106		dst_path = None
107		submission_model = None
108		base_dir = os.path.dirname(os.path.abspath(__file__))
109
110		# the method is use to upload data into database
111		upload_method = None
112
113		@classmethod
114		def setUpClass(cls):
115		"""Place test data in data source if needed"""
116
117		# calling my base class setup
118		super().setUpClass()
119
120		# ensuring that file cryoweb_test_data_only.sql is present
121		submission = cls.submission_model.objects.get(pk=1)
122		cls.dst_path = submission.uploaded_file.path
123
124		# get datasource filename from submission object
125		datasource_filename = os.path.basename(submission.uploaded_file.path)
126
127		if not os.path.exists(cls.dst_path):
128		src_path = os.path.join(cls.base_dir, datasource_filename)
129		shutil.copy(src_path, cls.dst_path)
130		cls.uploaded_file = True
131
132		@classmethod
133		def tearDownClass(cls):
134		"""Remove test data from data source if needed"""
135
136		# remove file if I placed it for tests
137		if cls.uploaded_file:
138		os.remove(cls.dst_path)
139
140		# calling my base class teardown class
141		super().tearDownClass()
142
143		def upload_datasource(self, message):
144		"""Testing uploading and importing data from crbanim to UID"""
145
146		# assert upload
147		self.assertTrue(self.upload_method(self.submission))
148
149		# reload submission
150		self.submission.refresh_from_db()
151
152		# assert submission messages
153		self.assertEqual(
154		self.submission.status,
155		LOADED)
156
157		self.assertIn(
158		message,
159		self.submission.message)
160
161		def check_errors(self, my_check, message):
162		"""Common stuff for error in file loading"""
163
164		self.assertFalse(self.upload_method(self.submission))
165
166		# reload submission
167		self.submission.refresh_from_db()
168
169		# test my mock method called
170		self.assertTrue(my_check.called)
171
172		self.assertEqual(
173		self.submission.status,
174		ERROR)
175
176		# check for two distinct messages
177		self.assertIn(
178		message,
179		self.submission.message)
180
181		self.assertNotIn(
182		"import completed for submission",
183		self.submission.message)
184
185
186		class AsyncIOMixin(object):
187		"""A common mixing for async objects"""
188
189		def setUp(self):
190		# calling my base class setup
191		super().setUp()
192
193		# setting channels methods
194		self.asyncio_mock_patcher = patch(
195		'asyncio.get_event_loop')
196		self.asyncio_mock = self.asyncio_mock_patcher.start()
197
198		# mocking asyncio return value
199		self.run_until = self.asyncio_mock.return_value
200		self.run_until.run_until_complete = Mock()
201
202		def tearDown(self):
203		# stopping mock objects
204		self.asyncio_mock_patcher.stop()
205
206		# calling base methods
207		super().tearDown()
208
209
210		# TODO: move into submission.tests (since related to submission.helpers)
211		class WebSocketMixin(AsyncIOMixin):
212		"""Override setUp to mock websocket objects"""
213
214		def setUp(self):
215		# calling my base class setup
216		super().setUp()
217
218		# another patch
219		self.send_msg_ws_patcher = patch(
220		'submissions.helpers.send_message_to_websocket')
221		self.send_msg_ws = self.send_msg_ws_patcher.start()
222
223		def tearDown(self):
224		# stopping mock objects
225		self.send_msg_ws_patcher.stop()
226
227		# calling base methods
228		super().tearDown()
229
230		def check_message(
231		self, message, notification_message, validation_message=None,
232		pk=1):
233		"""assert message to websocket called with parameters"""
234
235		# construct message according parameters
236		message = {
237		'message': message,
238		'notification_message': notification_message
239		}
240
241		# in case of successful data upload, a validation message is sent
242		if validation_message:
243		message['validation_message'] = validation_message
244
245		self.assertEqual(self.asyncio_mock.call_count, 1)
246		self.assertEqual(self.run_until.run_until_complete.call_count, 1)
247		self.assertEqual(self.send_msg_ws.call_count, 1)
248		self.send_msg_ws.assert_called_with(
249		message,
250		pk)
251
252		def check_message_not_called(self):
253		"""Check django channels async messages not called"""
254
255		self.assertEqual(self.asyncio_mock.call_count, 0)
256		self.assertEqual(self.run_until.run_until_complete.call_count, 0)
257		self.assertEqual(self.send_msg_ws.call_count, 0)
258

cnr-ibba / IMAGE-InjectTool

Push — master ( 55dedd...16663f )

common.tests.mixins.WebSocketMixin.tearDown() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like