BearRunningUnitTest - Code Metrics - Inspection of "Makman2/linterv4" - coala-analyzer/coala - Measure and Improve Code Quality continuously with Scrutinizer

Failed Conditions

Pull Request — master (#1990)

by Mischa

created 2016-04-12 21:41 UTC

BearRunningUnitTest A

↳ Parent: Project

Complexity

Total Complexity

Size/Duplication

Total Lines	110
Duplicated Lines	0 %

Metric	Value
dl	0
loc	110
rs	10
wmc	9

6 Methods

Rating	Name	Size	Complexity
A	test_evil_bear()	15	1
A	test_messaging()	11	1
B	test_strange_bear()	26	2
B	test_dependencies()	32	3
A	test_queue_done_marking()	6	1
A	setUp()	13	1

import multiprocessing
import queue
import unittest

from coalib.bears.GlobalBear import GlobalBear
from coalib.bears.LocalBear import LocalBear
from coalib.processes.BearRunning import (
    LOG_LEVEL, LogMessage, run, send_msg, task_done)
from coalib.processes.CONTROL_ELEMENT import CONTROL_ELEMENT
from coalib.results.Result import RESULT_SEVERITY, Result
from coalib.settings.Section import Section


class LocalTestBear(LocalBear):

    def run(self, filename, file):
        if filename == "file1":
            raise Exception("Just to throw anything here.")
        return [Result.from_values("LocalTestBear",
                                   "something went wrong",
                                   filename)]


class SimpleBear(LocalBear):

    def run(self,
            filename,
            file,
            *args,
            dependency_results=None,
            **kwargs):
        return [Result.from_values("SimpleBear",
                                   "something went wrong",
                                   filename),
                # This result should not be passed to DependentBear
                Result.from_values("FakeBear",
                                   "something went wrong",
                                   filename),
                Result.from_values("SimpleBear",
                                   "another thing went wrong",
                                   filename)]


class DependentBear(LocalBear):

    def run(self,
            filename,
            file,
            *args,
            dependency_results=None,
            **kwargs):
        assert len(dependency_results["SimpleBear"]) == 2

    @staticmethod
    def get_dependencies():
        return [SimpleBear]


class SimpleGlobalBear(GlobalBear):

    def run(self,
            *args,
            dependency_results=None,
            **kwargs):
        return [Result("SimpleGlobalBear", "something went wrong"),
                # This result should not be passed to DependentBear
                Result("FakeBear", "something went wrong"),
                Result("SimpleGlobalBear", "another thing went wrong")]


class DependentGlobalBear(GlobalBear):

    def run(self,
            *args,
            dependency_results=None,
            **kwargs):
        assert len(dependency_results["SimpleGlobalBear"]) == 3

    @staticmethod
    def get_dependencies():
        return [SimpleGlobalBear]


class GlobalTestBear(GlobalBear):

    def run(self):
        result = []
        for file, contents in self.file_dict.items():
            result.append(Result.from_values("GlobalTestBear",
                                             "Files are bad in general!",
                                             file,
                                             severity=RESULT_SEVERITY.INFO))
        return result


class EvilBear(LocalBear):


    def execute(self, *args, **kwargs):
        raise NotImplementedError


class UnexpectedBear1(LocalBear):

    def run(self, filename, file):
        return [1,
                Result("UnexpectedBear1", "test result")]


class UnexpectedBear2(LocalBear):

    def run(self, filename, file):
        return 1


class BearRunningUnitTest(unittest.TestCase):

    def setUp(self):
        self.settings = Section("name")

        self.file_name_queue = queue.Queue()
        self.local_bear_list = []
        self.global_bear_list = []
        self.global_bear_queue = queue.Queue()
        self.file_dict = {}
        manager = multiprocessing.Manager()
        self.local_result_dict = manager.dict()
        self.global_result_dict = manager.dict()
        self.message_queue = queue.Queue()
        self.control_queue = queue.Queue()

    def test_queue_done_marking(self):
        self.message_queue.put("test")
        task_done(self.message_queue)  # Should make the queue joinable
        self.message_queue.join()

        task_done("test")  # Should pass silently

    def test_messaging(self):
        send_msg(self.message_queue,
                 0,
                 LOG_LEVEL.DEBUG,
                 "test",
                 "messag",
                 delimiter="-",
                 end="e")

        self.assertEqual(self.message_queue.get(),
                         LogMessage(LOG_LEVEL.DEBUG, "test-message"))

    def test_dependencies(self):
        self.local_bear_list.append(SimpleBear(self.settings,
                                               self.message_queue))
        self.local_bear_list.append(DependentBear(self.settings,
                                                  self.message_queue))
        self.global_bear_list.append(SimpleGlobalBear({},
                                                      self.settings,
                                                      self.message_queue))
        self.global_bear_list.append(DependentGlobalBear({},
                                                         self.settings,
                                                         self.message_queue))
        self.global_bear_queue.put(1)
        self.global_bear_queue.put(0)
        self.file_name_queue.put("t")
        self.file_dict["t"] = []

        run(self.file_name_queue,
            self.local_bear_list,
            self.global_bear_list,
            self.global_bear_queue,
            self.file_dict,
            self.local_result_dict,
            self.global_result_dict,
            self.message_queue,
            self.control_queue)

        try:
            while True:
                msg = self.message_queue.get(timeout=0)
                self.assertEqual(msg.log_level, LOG_LEVEL.DEBUG)
        except queue.Empty:
            pass

    def test_evil_bear(self):
        self.local_bear_list.append(EvilBear(self.settings,
                                             self.message_queue))
        self.file_name_queue.put("t")
        self.file_dict["t"] = []

        run(self.file_name_queue,
            self.local_bear_list,
            self.global_bear_list,
            self.global_bear_queue,
            self.file_dict,
            self.local_result_dict,
            self.global_result_dict,
            self.message_queue,
            self.control_queue)

    def test_strange_bear(self):
        self.local_bear_list.append(UnexpectedBear1(self.settings,
                                                    self.message_queue))
        self.local_bear_list.append(UnexpectedBear2(self.settings,
                                                    self.message_queue))
        self.file_name_queue.put("t")
        self.file_dict["t"] = []

        run(self.file_name_queue,
            self.local_bear_list,
            self.global_bear_list,
            self.global_bear_queue,
            self.file_dict,
            self.local_result_dict,
            self.global_result_dict,
            self.message_queue,
            self.control_queue)

        expected_messages = [LOG_LEVEL.DEBUG,
                             LOG_LEVEL.ERROR,
                             LOG_LEVEL.DEBUG,
                             LOG_LEVEL.DEBUG,
                             LOG_LEVEL.WARNING]

        for msg in expected_messages:
            self.assertEqual(msg, self.message_queue.get(timeout=0).log_level)


class BearRunningIntegrationTest(unittest.TestCase):
    example_file = """a
b
c
d
"""

    def setUp(self):
        self.settings = Section("name")

        self.file_name_queue = queue.Queue()
        self.local_bear_list = []
        self.global_bear_list = []
        self.global_bear_queue = queue.Queue()
        self.file_dict = {}
        manager = multiprocessing.Manager()
        self.local_result_dict = manager.dict()
        self.global_result_dict = manager.dict()
        self.message_queue = queue.Queue()
        self.control_queue = queue.Queue()

        self.file1 = "file1"
        self.file2 = "arbitrary"

        self.file_name_queue.put(self.file1)
        self.file_name_queue.put(self.file2)
        self.file_name_queue.put("invalid file")
        self.local_bear_list.append(LocalTestBear(self.settings,
                                                  self.message_queue))
        self.local_bear_list.append("not a valid bear")
        self.file_dict[self.file1] = self.example_file
        self.file_dict[self.file2] = self.example_file
        self.global_bear_list.append(GlobalTestBear(self.file_dict,
                                                    self.settings,
                                                    self.message_queue))
        self.global_bear_list.append("not a valid bear")
        self.global_bear_queue.put(0)
        self.global_bear_queue.put(1)

    def test_run(self):
        run(self.file_name_queue,
            self.local_bear_list,
            self.global_bear_list,
            self.global_bear_queue,
            self.file_dict,
            self.local_result_dict,
            self.global_result_dict,
            self.message_queue,
            self.control_queue)

        expected_messages = [LOG_LEVEL.DEBUG,
                             LOG_LEVEL.WARNING,
                             LOG_LEVEL.DEBUG,
                             LOG_LEVEL.WARNING,
                             LOG_LEVEL.DEBUG,
                             LOG_LEVEL.WARNING,
                             LOG_LEVEL.ERROR,
                             LOG_LEVEL.DEBUG,
                             LOG_LEVEL.DEBUG,
                             LOG_LEVEL.WARNING]
        for msg in expected_messages:
            self.assertEqual(msg, self.message_queue.get(timeout=0).log_level)

        local_result_expected = [[],
                                 [Result.from_values("LocalTestBear",
                                                     "something went wrong",
                                                     'arbitrary')]
                                 ]
        for expected in local_result_expected:
            control_elem, index = self.control_queue.get()
            self.assertEqual(control_elem, CONTROL_ELEMENT.LOCAL)
            real = self.local_result_dict[index]
            self.assertEqual(real, expected)

        global_results_expected = [Result.from_values(
                                       "GlobalTestBear",
                                       "Files are bad in general!",
                                       "file1",
                                       severity=RESULT_SEVERITY.INFO),
                                   Result.from_values(
                                       "GlobalTestBear",
                                       "Files are bad in general!",
                                       "arbitrary",
                                       severity=RESULT_SEVERITY.INFO)]

        control_elem, index = self.control_queue.get()
        self.assertEqual(control_elem, CONTROL_ELEMENT.LOCAL_FINISHED)
        control_elem, index = self.control_queue.get()
        self.assertEqual(control_elem, CONTROL_ELEMENT.GLOBAL)
        real = self.global_result_dict[index]
        self.assertEqual(sorted(global_results_expected), sorted(real))

        control_elem, none = self.control_queue.get(timeout=0)
        self.assertEqual(control_elem, CONTROL_ELEMENT.GLOBAL_FINISHED)
        self.assertEqual(none, None)

        # The invalid bear gets a None in that dict for dependency resolution
        self.assertEqual(len(self.global_result_dict), 2)
        self.assertEqual(len(self.local_result_dict),
                         len(local_result_expected))
        self.assertRaises(queue.Empty, self.message_queue.get, timeout=0)
        self.assertRaises(queue.Empty, self.control_queue.get, timeout=0)


1			import multiprocessing
2			import queue
3			import unittest
4
5			from coalib.bears.GlobalBear import GlobalBear
6			from coalib.bears.LocalBear import LocalBear
7			from coalib.processes.BearRunning import (
8			LOG_LEVEL, LogMessage, run, send_msg, task_done)
9			from coalib.processes.CONTROL_ELEMENT import CONTROL_ELEMENT
10			from coalib.results.Result import RESULT_SEVERITY, Result
11			from coalib.settings.Section import Section
12
13
14			class LocalTestBear(LocalBear):
15
16			def run(self, filename, file):
17			if filename == "file1":
18			raise Exception("Just to throw anything here.")
19			return [Result.from_values("LocalTestBear",
20			"something went wrong",
21			filename)]
22
23
24			class SimpleBear(LocalBear):
25
26			def run(self,
27			filename,
28			file,
29			*args,
30			dependency_results=None,
31			**kwargs):
32			return [Result.from_values("SimpleBear",
33			"something went wrong",
34			filename),
35			# This result should not be passed to DependentBear
36			Result.from_values("FakeBear",
37			"something went wrong",
38			filename),
39			Result.from_values("SimpleBear",
40			"another thing went wrong",
41			filename)]
42
43
44			class DependentBear(LocalBear):
45
46			def run(self,
47			filename,
48			file,
49			*args,
50			dependency_results=None,
51			**kwargs):
52			assert len(dependency_results["SimpleBear"]) == 2
53
54			@staticmethod
55			def get_dependencies():
56			return [SimpleBear]
57
58
59			class SimpleGlobalBear(GlobalBear):
60
61			def run(self,
62			*args,
63			dependency_results=None,
64			**kwargs):
65			return [Result("SimpleGlobalBear", "something went wrong"),
66			# This result should not be passed to DependentBear
67			Result("FakeBear", "something went wrong"),
68			Result("SimpleGlobalBear", "another thing went wrong")]
69
70
71			class DependentGlobalBear(GlobalBear):
72
73			def run(self,
74			*args,
75			dependency_results=None,
76			**kwargs):
77			assert len(dependency_results["SimpleGlobalBear"]) == 3
78
79			@staticmethod
80			def get_dependencies():
81			return [SimpleGlobalBear]
82
83
84			class GlobalTestBear(GlobalBear):
85
86			def run(self):
87			result = []
88			for file, contents in self.file_dict.items():
89			result.append(Result.from_values("GlobalTestBear",
90			"Files are bad in general!",
91			file,
92			severity=RESULT_SEVERITY.INFO))
93			return result
94
95
96			class EvilBear(LocalBear):
			0 ignored issues – show Unused Code introduced 2016-02-22 07:01 UTC by Report Bug Copy Issue Report This abstract class does not seem to be used anywhere. Loading history...
97
98			def execute(self, args, *kwargs):
99			raise NotImplementedError
100
101
102			class UnexpectedBear1(LocalBear):
103
104			def run(self, filename, file):
105			return [1,
106			Result("UnexpectedBear1", "test result")]
107
108
109			class UnexpectedBear2(LocalBear):
110
111			def run(self, filename, file):
112			return 1
113
114
115			class BearRunningUnitTest(unittest.TestCase):
116
117			def setUp(self):
118			self.settings = Section("name")
119
120			self.file_name_queue = queue.Queue()
121			self.local_bear_list = []
122			self.global_bear_list = []
123			self.global_bear_queue = queue.Queue()
124			self.file_dict = {}
125			manager = multiprocessing.Manager()
126			self.local_result_dict = manager.dict()
127			self.global_result_dict = manager.dict()
128			self.message_queue = queue.Queue()
129			self.control_queue = queue.Queue()
130
131			def test_queue_done_marking(self):
132			self.message_queue.put("test")
133			task_done(self.message_queue) # Should make the queue joinable
134			self.message_queue.join()
135
136			task_done("test") # Should pass silently
137
138			def test_messaging(self):
139			send_msg(self.message_queue,
140			0,
141			LOG_LEVEL.DEBUG,
142			"test",
143			"messag",
144			delimiter="-",
145			end="e")
146
147			self.assertEqual(self.message_queue.get(),
148			LogMessage(LOG_LEVEL.DEBUG, "test-message"))
149
150			def test_dependencies(self):
151			self.local_bear_list.append(SimpleBear(self.settings,
152			self.message_queue))
153			self.local_bear_list.append(DependentBear(self.settings,
154			self.message_queue))
155			self.global_bear_list.append(SimpleGlobalBear({},
156			self.settings,
157			self.message_queue))
158			self.global_bear_list.append(DependentGlobalBear({},
159			self.settings,
160			self.message_queue))
161			self.global_bear_queue.put(1)
162			self.global_bear_queue.put(0)
163			self.file_name_queue.put("t")
164			self.file_dict["t"] = []
165
166			run(self.file_name_queue,
167			self.local_bear_list,
168			self.global_bear_list,
169			self.global_bear_queue,
170			self.file_dict,
171			self.local_result_dict,
172			self.global_result_dict,
173			self.message_queue,
174			self.control_queue)
175
176			try:
177			while True:
178			msg = self.message_queue.get(timeout=0)
179			self.assertEqual(msg.log_level, LOG_LEVEL.DEBUG)
180			except queue.Empty:
181			pass
182
183			def test_evil_bear(self):
184			self.local_bear_list.append(EvilBear(self.settings,
185			self.message_queue))
186			self.file_name_queue.put("t")
187			self.file_dict["t"] = []
188
189			run(self.file_name_queue,
190			self.local_bear_list,
191			self.global_bear_list,
192			self.global_bear_queue,
193			self.file_dict,
194			self.local_result_dict,
195			self.global_result_dict,
196			self.message_queue,
197			self.control_queue)
198
199			def test_strange_bear(self):
200			self.local_bear_list.append(UnexpectedBear1(self.settings,
201			self.message_queue))
202			self.local_bear_list.append(UnexpectedBear2(self.settings,
203			self.message_queue))
204			self.file_name_queue.put("t")
205			self.file_dict["t"] = []
206
207			run(self.file_name_queue,
208			self.local_bear_list,
209			self.global_bear_list,
210			self.global_bear_queue,
211			self.file_dict,
212			self.local_result_dict,
213			self.global_result_dict,
214			self.message_queue,
215			self.control_queue)
216
217			expected_messages = [LOG_LEVEL.DEBUG,
218			LOG_LEVEL.ERROR,
219			LOG_LEVEL.DEBUG,
220			LOG_LEVEL.DEBUG,
221			LOG_LEVEL.WARNING]
222
223			for msg in expected_messages:
224			self.assertEqual(msg, self.message_queue.get(timeout=0).log_level)
225
226
227			class BearRunningIntegrationTest(unittest.TestCase):
228			example_file = """a
229			b
230			c
231			d
232			"""
233
234			def setUp(self):
235			self.settings = Section("name")
236
237			self.file_name_queue = queue.Queue()
238			self.local_bear_list = []
239			self.global_bear_list = []
240			self.global_bear_queue = queue.Queue()
241			self.file_dict = {}
242			manager = multiprocessing.Manager()
243			self.local_result_dict = manager.dict()
244			self.global_result_dict = manager.dict()
245			self.message_queue = queue.Queue()
246			self.control_queue = queue.Queue()
247
248			self.file1 = "file1"
249			self.file2 = "arbitrary"
250
251			self.file_name_queue.put(self.file1)
252			self.file_name_queue.put(self.file2)
253			self.file_name_queue.put("invalid file")
254			self.local_bear_list.append(LocalTestBear(self.settings,
255			self.message_queue))
256			self.local_bear_list.append("not a valid bear")
257			self.file_dict[self.file1] = self.example_file
258			self.file_dict[self.file2] = self.example_file
259			self.global_bear_list.append(GlobalTestBear(self.file_dict,
260			self.settings,
261			self.message_queue))
262			self.global_bear_list.append("not a valid bear")
263			self.global_bear_queue.put(0)
264			self.global_bear_queue.put(1)
265
266			def test_run(self):
267			run(self.file_name_queue,
268			self.local_bear_list,
269			self.global_bear_list,
270			self.global_bear_queue,
271			self.file_dict,
272			self.local_result_dict,
273			self.global_result_dict,
274			self.message_queue,
275			self.control_queue)
276
277			expected_messages = [LOG_LEVEL.DEBUG,
278			LOG_LEVEL.WARNING,
279			LOG_LEVEL.DEBUG,
280			LOG_LEVEL.WARNING,
281			LOG_LEVEL.DEBUG,
282			LOG_LEVEL.WARNING,
283			LOG_LEVEL.ERROR,
284			LOG_LEVEL.DEBUG,
285			LOG_LEVEL.DEBUG,
286			LOG_LEVEL.WARNING]
287			for msg in expected_messages:
288			self.assertEqual(msg, self.message_queue.get(timeout=0).log_level)
289
290			local_result_expected = [[],
291			[Result.from_values("LocalTestBear",
292			"something went wrong",
293			'arbitrary')]
294			]
295			for expected in local_result_expected:
296			control_elem, index = self.control_queue.get()
297			self.assertEqual(control_elem, CONTROL_ELEMENT.LOCAL)
298			real = self.local_result_dict[index]
299			self.assertEqual(real, expected)
300
301			global_results_expected = [Result.from_values(
302			"GlobalTestBear",
303			"Files are bad in general!",
304			"file1",
305			severity=RESULT_SEVERITY.INFO),
306			Result.from_values(
307			"GlobalTestBear",
308			"Files are bad in general!",
309			"arbitrary",
310			severity=RESULT_SEVERITY.INFO)]
311
312			control_elem, index = self.control_queue.get()
313			self.assertEqual(control_elem, CONTROL_ELEMENT.LOCAL_FINISHED)
314			control_elem, index = self.control_queue.get()
315			self.assertEqual(control_elem, CONTROL_ELEMENT.GLOBAL)
316			real = self.global_result_dict[index]
317			self.assertEqual(sorted(global_results_expected), sorted(real))
318
319			control_elem, none = self.control_queue.get(timeout=0)
320			self.assertEqual(control_elem, CONTROL_ELEMENT.GLOBAL_FINISHED)
321			self.assertEqual(none, None)
322
323			# The invalid bear gets a None in that dict for dependency resolution
324			self.assertEqual(len(self.global_result_dict), 2)
325			self.assertEqual(len(self.local_result_dict),
326			len(local_result_expected))
327			self.assertRaises(queue.Empty, self.message_queue.get, timeout=0)
328			self.assertRaises(queue.Empty, self.control_queue.get, timeout=0)
329

coala-analyzer / coala

Pull Request — master (#1990)

BearRunningUnitTest A

Complexity

Size/Duplication

6 Methods

Duplication Side-by-Side

Filter issues like