DependentBear - Code Metrics - coala-analyzer/coala - Measure and Improve Code Quality continuously with Scrutinizer

DependentBear A
last analyzed 2016-09-16 10:08 UTC

↳ Parent: Project

Complexity

Total Complexity

Size/Duplication

Total Lines	11
Duplicated Lines	0 %

Importance

Changes	2
Bugs	0	Features	0

Metric	Value
dl	0
loc	11
rs	10
c	2
b	0
f	0
wmc	2

1 Method

Rating	Name	Duplication	Size	Complexity
A	run()	0	7	2

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):

    BEAR_DEPS = {SimpleBear}

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


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):

    BEAR_DEPS = {SimpleGlobalBear}

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


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

coala-analyzer / coala

DependentBear A last analyzed 2016-09-16 10:08 UTC

Complexity

Size/Duplication

Importance

1 Method

Duplication Side-by-Side

Filter issues like

DependentBear A
last analyzed 2016-09-16 10:08 UTC