coalib.tests.processes.TestBear.run() - Code Metrics - Inspection of "Add PyImportSortBear" - coala-analyzer/coala - Measure and Improve Code Quality continuously with Scrutinizer

Failed Conditions

Pull Request — master (#1152)

by Lasse

created 2015-12-22 10:53 UTC

coalib.tests.processes.TestBear.run() A

↳ Parent: coalib.tests.processes.TestBear

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	1
dl	0
loc	4
rs	10

import queue
import sys
import multiprocessing

sys.path.insert(0, ".")
import unittest
from coalib.processes.CONTROL_ELEMENT import CONTROL_ELEMENT
from coalib.results.Result import Result, RESULT_SEVERITY
from coalib.bears.LocalBear import LocalBear
from coalib.bears.GlobalBear import GlobalBear
from coalib.processes.BearRunning import (run,
                                          LogMessage,
                                          LOG_LEVEL,
                                          send_msg,
                                          task_done)
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)


if __name__ == '__main__':
    unittest.main(verbosity=2)


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

coala-analyzer / coala

Pull Request — master (#1152)

coalib.tests.processes.TestBear.run() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like