1 | import multiprocessing |
||
2 | import os |
||
3 | import platform |
||
4 | import queue |
||
5 | import re |
||
6 | import subprocess |
||
7 | import sys |
||
8 | import unittest |
||
9 | |||
10 | from pyprint.ConsolePrinter import ConsolePrinter |
||
11 | |||
12 | from coalib.output.printers.LogPrinter import LogPrinter |
||
13 | from coalib.processes.CONTROL_ELEMENT import CONTROL_ELEMENT |
||
14 | from coalib.processes.Processing import ( |
||
15 | ACTIONS, autoapply_actions, check_result_ignore, create_process_group, |
||
16 | execute_section, filter_raising_callables, get_default_actions, |
||
17 | get_file_dict, print_result, process_queues, simplify_section_result, |
||
18 | yield_ignore_ranges) |
||
19 | from coalib.results.HiddenResult import HiddenResult |
||
20 | from coalib.results.Result import RESULT_SEVERITY, Result |
||
21 | from coalib.results.result_actions.ApplyPatchAction import ApplyPatchAction |
||
22 | from coalib.results.result_actions.PrintDebugMessageAction import ( |
||
23 | PrintDebugMessageAction) |
||
24 | from coalib.results.result_actions.ResultAction import ResultAction |
||
25 | from coalib.results.SourceRange import SourceRange |
||
26 | from coalib.settings.ConfigurationGathering import gather_configuration |
||
27 | from coalib.settings.Section import Section |
||
28 | from coalib.settings.Setting import Setting |
||
29 | |||
30 | |||
31 | process_group_test_code = """ |
||
32 | import time, subprocess, os, platform, sys; |
||
33 | p=subprocess.Popen([sys.executable, |
||
34 | "-c", |
||
35 | "import time; time.sleep(0.1)"]); |
||
36 | pgid = p.pid if platform.system() == "Windows" else os.getpgid(p.pid); |
||
37 | print(p.pid, pgid) |
||
38 | p.terminate() |
||
39 | """ |
||
40 | |||
41 | |||
42 | class DummyProcess(multiprocessing.Process): |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
![]() |
|||
43 | |||
44 | def __init__(self, control_queue, starts_dead=False): |
||
45 | multiprocessing.Process.__init__(self) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
46 | self.control_queue = control_queue |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
47 | self.starts_dead = starts_dead |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
48 | |||
49 | def is_alive(self): |
||
50 | return not self.control_queue.empty() and not self.starts_dead |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
51 | |||
52 | |||
53 | class ProcessingTestLogPrinter(LogPrinter): |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
54 | |||
55 | def __init__(self, log_queue): |
||
56 | LogPrinter.__init__(self, self) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
57 | self.log_queue = log_queue |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
58 | self.set_up = False |
||
59 | |||
60 | def log_message(self, log_message, timestamp=None, **kwargs): |
||
61 | self.log_queue.put(log_message) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
62 | |||
63 | |||
64 | class ProcessingTest(unittest.TestCase): |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
65 | |||
66 | def setUp(self): |
||
67 | config_path = os.path.abspath(os.path.join( |
||
68 | os.path.dirname(__file__), |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
69 | "section_executor_test_files", |
||
70 | ".coafile")) |
||
71 | self.testcode_c_path = os.path.join(os.path.dirname(config_path), |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
72 | "testcode.c") |
||
73 | |||
74 | self.result_queue = queue.Queue() |
||
75 | self.queue = queue.Queue() |
||
76 | self.log_queue = queue.Queue() |
||
77 | log_printer = LogPrinter(ConsolePrinter()) |
||
78 | self.log_printer = ProcessingTestLogPrinter(self.log_queue) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
79 | |||
80 | (self.sections, |
||
81 | self.local_bears, |
||
82 | self.global_bears, |
||
83 | targets) = gather_configuration(lambda *args: True, |
||
84 | log_printer, |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
85 | arg_list=["--config", |
||
86 | re.escape(config_path)]) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
87 | self.assertEqual(len(self.local_bears["default"]), 1) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
88 | self.assertEqual(len(self.global_bears["default"]), 1) |
||
89 | self.assertEqual(targets, []) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
90 | |||
91 | def test_run(self): |
||
92 | self.sections['default'].append(Setting('jobs', "1")) |
||
93 | results = execute_section(self.sections["default"], |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
94 | self.global_bears["default"], |
||
95 | self.local_bears["default"], |
||
96 | lambda *args: self.result_queue.put(args[2]), |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
97 | self.log_printer) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
98 | self.assertTrue(results[0]) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
99 | |||
100 | local_results = self.result_queue.get(timeout=0) |
||
101 | global_results = self.result_queue.get(timeout=0) |
||
102 | self.assertTrue(self.result_queue.empty()) |
||
103 | |||
104 | self.assertEqual(len(local_results), 1) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
105 | self.assertEqual(len(global_results), 1) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
106 | # Result dict also returned |
||
107 | # One file |
||
108 | self.assertEqual(len(results[1]), 1) |
||
109 | # One global bear |
||
110 | self.assertEqual(len(results[2]), 1) |
||
111 | |||
112 | local_result = local_results[0] |
||
113 | global_result = global_results[0] |
||
114 | |||
115 | self.assertRegex(repr(local_result), |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
116 | "<Result object\\(id={}, origin='LocalTestBear', " |
||
117 | "affected_code=\\(\\), severity=NORMAL, message='test" |
||
118 | " msg'\\) at 0x[0-9a-fA-F]+>".format( |
||
119 | hex(local_result.id))) |
||
120 | self.assertRegex(repr(global_result), |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
121 | "<Result object\\(id={}, origin='GlobalTestBear', " |
||
122 | "affected_code=\\(.*start=.*file=.*section_executor_" |
||
123 | "test_files.*line=None.*end=.*\\), severity=NORMAL, " |
||
124 | "message='test message'\\) at " |
||
125 | "0x[0-9a-fA-F]+>".format(hex(global_result.id))) |
||
126 | |||
127 | def test_empty_run(self): |
||
128 | self.sections['default'].append(Setting('jobs', "bogus!")) |
||
129 | results = execute_section(self.sections["default"], |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
130 | [], |
||
131 | [], |
||
132 | lambda *args: self.result_queue.put(args[2]), |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
133 | self.log_printer) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
134 | # No results |
||
135 | self.assertFalse(results[0]) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
136 | # One file |
||
137 | self.assertEqual(len(results[1]), 1) |
||
138 | # No global bear |
||
139 | self.assertEqual(len(results[2]), 0) |
||
140 | |||
141 | def test_process_queues(self): |
||
142 | ctrlq = queue.Queue() |
||
143 | |||
144 | # Append custom controlling sequences. |
||
145 | |||
146 | # Simulated process 1 |
||
147 | ctrlq.put((CONTROL_ELEMENT.LOCAL, 1)) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
148 | ctrlq.put((CONTROL_ELEMENT.LOCAL_FINISHED, None)) |
||
149 | ctrlq.put((CONTROL_ELEMENT.GLOBAL, 1)) |
||
150 | |||
151 | # Simulated process 2 |
||
152 | ctrlq.put((CONTROL_ELEMENT.LOCAL, 2)) |
||
153 | |||
154 | # Simulated process 1 |
||
155 | ctrlq.put((CONTROL_ELEMENT.GLOBAL_FINISHED, None)) |
||
156 | |||
157 | # Simulated process 2 |
||
158 | ctrlq.put((CONTROL_ELEMENT.LOCAL_FINISHED, None)) |
||
159 | ctrlq.put((CONTROL_ELEMENT.GLOBAL, 1)) |
||
160 | ctrlq.put((CONTROL_ELEMENT.GLOBAL_FINISHED, None)) |
||
161 | |||
162 | first_local = Result.from_values("o", "The first result.", file="f") |
||
163 | second_local = Result.from_values("ABear", |
||
164 | "The second result.", |
||
165 | file="f", |
||
166 | line=1) |
||
167 | third_local = Result.from_values("ABear", |
||
168 | "The second result.", |
||
169 | file="f", |
||
170 | line=4) |
||
171 | fourth_local = Result.from_values("ABear", |
||
172 | "Another result.", |
||
173 | file="f", |
||
174 | line=7) |
||
175 | first_global = Result("o", "The one and only global result.") |
||
176 | section = Section("") |
||
177 | section.append(Setting('min_severity', "normal")) |
||
178 | process_queues( |
||
179 | [DummyProcess(control_queue=ctrlq) for i in range(3)], |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
Comprehensibility
Best Practice
introduced
by
|
|||
180 | ctrlq, |
||
181 | {1: [first_local, |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
182 | second_local, |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
183 | third_local, |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
184 | # The following are to be ignored |
||
185 | Result('o', 'm', severity=RESULT_SEVERITY.INFO), |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
186 | Result.from_values("ABear", "u", "f", 2, 1), |
||
187 | Result.from_values("ABear", "u", "f", 3, 1)], |
||
188 | 2: [fourth_local, |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
189 | # The following are to be ignored |
||
190 | HiddenResult("t", "c"), |
||
191 | Result.from_values("ABear", "u", "f", 5, 1), |
||
192 | Result.from_values("ABear", "u", "f", 6, 1)]}, |
||
193 | {1: [first_global]}, |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
194 | {"f": ["first line # stop ignoring, invalid ignore range\n", |
||
195 | "second line # ignore all\n", |
||
196 | "third line\n", |
||
197 | "fourth line\n", |
||
198 | "# Start ignoring ABear, BBear and CBear\n", |
||
199 | "# Stop ignoring\n", |
||
200 | "seventh"]}, |
||
201 | lambda *args: self.queue.put(args[2]), |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
202 | section, |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
203 | self.log_printer) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
204 | |||
205 | self.assertEqual(self.queue.get(timeout=0), ([first_local, |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
206 | second_local, |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
207 | third_local])) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
208 | self.assertEqual(self.queue.get(timeout=0), ([fourth_local])) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
209 | self.assertEqual(self.queue.get(timeout=0), ([first_global])) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
210 | self.assertEqual(self.queue.get(timeout=0), ([first_global])) |
||
211 | |||
212 | def test_dead_processes(self): |
||
213 | ctrlq = queue.Queue() |
||
214 | # Not enough FINISH elements in the queue, processes start already dead |
||
215 | # Also queue elements are reversed |
||
216 | ctrlq.put((CONTROL_ELEMENT.GLOBAL_FINISHED, None)) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
217 | ctrlq.put((CONTROL_ELEMENT.LOCAL_FINISHED, None)) |
||
218 | |||
219 | process_queues( |
||
220 | [DummyProcess(ctrlq, starts_dead=True) for i in range(3)], |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
Comprehensibility
Best Practice
introduced
by
|
|||
221 | ctrlq, {}, {}, {}, |
||
222 | lambda *args: self.queue.put(args[2]), |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
223 | Section(""), |
||
224 | self.log_printer) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
225 | with self.assertRaises(queue.Empty): |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
226 | self.queue.get(timeout=0) |
||
227 | |||
228 | # Not enough FINISH elements in the queue, processes start already dead |
||
229 | ctrlq.put((CONTROL_ELEMENT.LOCAL_FINISHED, None)) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
230 | ctrlq.put((CONTROL_ELEMENT.GLOBAL_FINISHED, None)) |
||
231 | |||
232 | process_queues( |
||
233 | [DummyProcess(ctrlq, starts_dead=True) for i in range(3)], |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
Comprehensibility
Best Practice
introduced
by
|
|||
234 | ctrlq, {}, {}, {}, |
||
235 | lambda *args: self.queue.put(args[2]), |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
236 | Section(""), |
||
237 | self.log_printer) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
238 | with self.assertRaises(queue.Empty): |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
239 | self.queue.get(timeout=0) |
||
240 | |||
241 | def test_create_process_group(self): |
||
242 | p = create_process_group([sys.executable, |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
243 | "-c", |
||
244 | process_group_test_code], |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
245 | stdout=subprocess.PIPE, |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
246 | stderr=subprocess.PIPE) |
||
247 | retval = p.wait() |
||
248 | if retval != 0: |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
249 | for line in p.stderr: |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
250 | print(line, end='') |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
251 | raise Exception("Subprocess did not exit correctly") |
||
252 | output = [i for i in p.stdout] |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
253 | p.stderr.close() |
||
254 | p.stdout.close() |
||
255 | pid, pgid = [int(i.strip()) for i_out in output for i in i_out.split()] |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
Comprehensibility
Best Practice
introduced
by
|
|||
256 | if platform.system() != "Windows": |
||
257 | # There is no way of testing this on windows with the current |
||
258 | # python modules subprocess and os |
||
259 | self.assertEqual(p.pid, pgid) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
260 | |||
261 | def test_filter_raising_callables(self): |
||
262 | class A(Exception): |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
263 | pass |
||
264 | |||
265 | class B(Exception): |
||
266 | pass |
||
267 | |||
268 | class C(Exception): |
||
269 | pass |
||
270 | |||
271 | def create_exception_raiser(exception): |
||
272 | def raiser(exc): |
||
273 | if exception in exc: |
||
274 | raise exception |
||
275 | return exception |
||
276 | return raiser |
||
277 | |||
278 | raiseA, raiseB, raiseC = (create_exception_raiser(exc) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
279 | for exc in [A, B, C]) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
Comprehensibility
Best Practice
introduced
by
Comprehensibility
Best Practice
introduced
by
|
|||
280 | |||
281 | test_list = [raiseA, raiseC, raiseB, raiseC] |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
Comprehensibility
Best Practice
introduced
by
Comprehensibility
Best Practice
introduced
by
|
|||
282 | self.assertEqual(list(filter_raising_callables(test_list, A, (A,))), |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
283 | [C, B, C]) |
||
284 | |||
285 | self.assertEqual(list(filter_raising_callables(test_list, |
||
286 | (B, C), |
||
287 | exc=(B, C))), |
||
288 | [A]) |
||
289 | |||
290 | # Test whether non filtered exceptions bubble up. |
||
291 | with self.assertRaises(B): |
||
292 | list(filter_raising_callables(test_list, C, exc=(B, C))) |
||
293 | |||
294 | def test_get_file_dict(self): |
||
295 | file_dict = get_file_dict([self.testcode_c_path], self.log_printer) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
296 | self.assertEqual(len(file_dict), 1) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
297 | self.assertEqual(type(file_dict[self.testcode_c_path]), |
||
298 | tuple, |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
299 | msg="files in file_dict should not be editable") |
||
300 | self.assertEqual("Files that will be checked:\n" + self.testcode_c_path, |
||
301 | self.log_printer.log_queue.get().message) |
||
302 | |||
303 | def test_get_file_dict_non_existent_file(self): |
||
304 | file_dict = get_file_dict(["non_existent_file"], self.log_printer) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
305 | self.assertEqual(file_dict, {}) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
306 | self.assertIn(("Failed to read file 'non_existent_file' because of " |
||
307 | "an unknown error."), |
||
308 | self.log_printer.log_queue.get().message) |
||
309 | |||
310 | def test_simplify_section_result(self): |
||
311 | results = (True, |
||
312 | {"file1": [Result("a", "b")], "file2": None}, |
||
313 | {"file3": [Result("a", "c")]}, |
||
314 | None) |
||
315 | yielded, yielded_unfixed, all_results = simplify_section_result(results) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
316 | self.assertEqual(yielded, True) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
317 | self.assertEqual(yielded_unfixed, True) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
318 | self.assertEqual(len(all_results), 2) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
319 | |||
320 | def test_ignore_results(self): |
||
321 | ranges = [([], SourceRange.from_values("f", 1, 1, 2, 2))] |
||
322 | result = Result.from_values("origin", |
||
323 | "message", |
||
324 | file="e", |
||
325 | line=1, |
||
326 | column=1, |
||
327 | end_line=2, |
||
328 | end_column=2) |
||
329 | |||
330 | self.assertFalse(check_result_ignore(result, ranges)) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
Comprehensibility
Best Practice
introduced
by
|
|||
331 | |||
332 | ranges.append(([], SourceRange.from_values("e", 2, 3, 3, 3))) |
||
333 | self.assertFalse(check_result_ignore(result, ranges)) |
||
334 | |||
335 | ranges.append(([], SourceRange.from_values("e", 1, 1, 2, 2))) |
||
336 | self.assertTrue(check_result_ignore(result, ranges)) |
||
337 | |||
338 | result1 = Result.from_values("origin", "message", file="e") |
||
339 | self.assertFalse(check_result_ignore(result1, ranges)) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
340 | |||
341 | ranges = [(['something', 'else', 'not origin'], |
||
342 | SourceRange.from_values("e", 1, 1, 2, 2))] |
||
343 | self.assertFalse(check_result_ignore(result, ranges)) |
||
344 | |||
345 | ranges = [(['something', 'else', 'origin'], |
||
346 | SourceRange.from_values("e", 1, 1, 2, 2))] |
||
347 | self.assertTrue(check_result_ignore(result, ranges)) |
||
348 | |||
349 | def test_ignore_glob(self): |
||
350 | result = Result.from_values("LineLengthBear", |
||
351 | "message", |
||
352 | file="d", |
||
353 | line=1, |
||
354 | column=1, |
||
355 | end_line=2, |
||
356 | end_column=2) |
||
357 | ranges = [(["(line*|space*)", "py*"], |
||
358 | SourceRange.from_values("d", 1, 1, 2, 2))] |
||
359 | self.assertTrue(check_result_ignore(result, ranges)) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
Comprehensibility
Best Practice
introduced
by
|
|||
360 | |||
361 | result = Result.from_values("SpaceConsistencyBear", |
||
362 | "message", |
||
363 | file="d", |
||
364 | line=1, |
||
365 | column=1, |
||
366 | end_line=2, |
||
367 | end_column=2) |
||
368 | ranges = [(["(line*|space*)", "py*"], |
||
369 | SourceRange.from_values("d", 1, 1, 2, 2))] |
||
370 | self.assertTrue(check_result_ignore(result, ranges)) |
||
371 | |||
372 | result = Result.from_values("XMLBear", |
||
373 | "message", |
||
374 | file="d", |
||
375 | line=1, |
||
376 | column=1, |
||
377 | end_line=2, |
||
378 | end_column=2) |
||
379 | ranges = [(["(line*|space*)", "py*"], |
||
380 | SourceRange.from_values("d", 1, 1, 2, 2))] |
||
381 | self.assertFalse(check_result_ignore(result, ranges)) |
||
382 | |||
383 | def test_yield_ignore_ranges(self): |
||
384 | test_file_dict_a = {'f': |
||
385 | ('# Ignore aBear\n', |
||
386 | 'a_string = "This string should be ignored"\n')} |
||
387 | test_ignore_range_a = list(yield_ignore_ranges(test_file_dict_a)) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
388 | for test_bears, test_source_range in test_ignore_range_a: |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
389 | self.assertEqual(test_bears, ['abear']) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
390 | self.assertEqual(test_source_range.start.line, 1) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
391 | self.assertEqual(test_source_range.start.column, 1) |
||
392 | self.assertEqual(test_source_range.end.line, 2) |
||
393 | self.assertEqual(test_source_range.end.column, 43) |
||
394 | |||
395 | test_file_dict_b = {'f': |
||
396 | ('# start Ignoring bBear\n', |
||
397 | 'b_string = "This string should be ignored"\n', |
||
398 | '# stop ignoring\n')} |
||
399 | test_ignore_range_b = list(yield_ignore_ranges(test_file_dict_b)) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
400 | for test_bears, test_source_range in test_ignore_range_b: |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
401 | self.assertEqual(test_bears, ['bbear']) |
||
402 | self.assertEqual(test_source_range.start.line, 1) |
||
403 | self.assertEqual(test_source_range.start.column, 1) |
||
404 | self.assertEqual(test_source_range.end.line, 3) |
||
405 | self.assertEqual(test_source_range.end.column, 16) |
||
406 | |||
407 | test_file_dict_c = {'f': |
||
408 | ('# Start ignoring cBear\n', |
||
409 | '# Stop ignoring cBear This & prev ignored\n')} |
||
410 | test_ignore_range_c = list(yield_ignore_ranges(test_file_dict_c)) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
411 | for test_bears, test_source_range in test_ignore_range_c: |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
412 | self.assertEqual(test_bears, ['cbear']) |
||
413 | self.assertEqual(test_source_range.start.line, 1) |
||
414 | self.assertEqual(test_source_range.start.column, 1) |
||
415 | self.assertEqual(test_source_range.end.line, 2) |
||
416 | self.assertEqual(test_source_range.end.column, 42) |
||
417 | |||
418 | test_file_dict_d = {'f': |
||
419 | ('# Start ignoring cBear\n', |
||
420 | 'All of this ignored\n')} |
||
421 | test_ignore_range_d = list(yield_ignore_ranges(test_file_dict_d)) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
422 | for test_bears, test_source_range in test_ignore_range_d: |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
423 | self.assertEqual(test_bears, ['cbear']) |
||
424 | self.assertEqual(test_source_range.start.line, 1) |
||
425 | self.assertEqual(test_source_range.start.column, 1) |
||
426 | self.assertEqual(test_source_range.end.line, 2) |
||
427 | self.assertEqual(test_source_range.end.column, 20) |
||
428 | |||
429 | |||
430 | class ProcessingTest_GetDefaultActions(unittest.TestCase): |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
431 | |||
432 | def setUp(self): |
||
433 | self.section = Section("X") |
||
434 | |||
435 | def test_no_key(self): |
||
436 | self.assertEqual(get_default_actions(self.section), ({}, {})) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
437 | |||
438 | def test_no_value(self): |
||
439 | self.section.append(Setting("default_actions", "")) |
||
440 | self.assertEqual(get_default_actions(self.section), ({}, {})) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
441 | |||
442 | def test_only_valid_actions(self): |
||
443 | self.section.append(Setting( |
||
444 | "default_actions", |
||
445 | "MyBear: PrintDebugMessageAction, ValidBear: ApplyPatchAction")) |
||
446 | self.assertEqual( |
||
447 | get_default_actions(self.section), |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
448 | ({"MyBear": PrintDebugMessageAction, |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
449 | "ValidBear": ApplyPatchAction}, |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
450 | {})) |
||
451 | |||
452 | def test_valid_and_invalid_actions(self): |
||
453 | self.section.append(Setting( |
||
454 | "default_actions", |
||
455 | "MyBear: INVALID_action, ValidBear: ApplyPatchAction, XBear: ABC")) |
||
456 | self.assertEqual(get_default_actions(self.section), |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
457 | ({"ValidBear": ApplyPatchAction}, |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
458 | {"MyBear": "INVALID_action", "XBear": "ABC"})) |
||
459 | |||
460 | |||
461 | class ProcessingTest_AutoapplyActions(unittest.TestCase): |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
462 | |||
463 | def setUp(self): |
||
464 | self.log_queue = queue.Queue() |
||
465 | self.log_printer = ProcessingTestLogPrinter(self.log_queue) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
466 | |||
467 | self.resultY = Result("YBear", "msg1") |
||
468 | self.resultZ = Result("ZBear", "msg2") |
||
469 | self.results = [self.resultY, self.resultZ] |
||
470 | self.section = Section("A") |
||
471 | |||
472 | def test_no_default_actions(self): |
||
473 | ret = autoapply_actions(self.results, |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
474 | {}, |
||
475 | {}, |
||
476 | self.section, |
||
477 | self.log_printer) |
||
478 | self.assertEqual(ret, self.results) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
479 | self.assertTrue(self.log_queue.empty()) |
||
480 | |||
481 | def test_with_invalid_action(self): |
||
482 | self.section.append(Setting("default_actions", |
||
483 | "XBear: nonSENSE_action")) |
||
484 | ret = autoapply_actions(self.results, |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
485 | {}, |
||
486 | {}, |
||
487 | self.section, |
||
488 | self.log_printer) |
||
489 | self.assertEqual(ret, self.results) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
490 | self.assertEqual(self.log_queue.get().message, |
||
491 | "Selected default action 'nonSENSE_action' for bear " |
||
492 | "'XBear' does not exist. Ignoring action.") |
||
493 | self.assertTrue(self.log_queue.empty()) |
||
494 | |||
495 | def test_without_default_action_and_unapplicable(self): |
||
496 | # Use a result where no default action is supplied for and another one |
||
497 | # where the action is not applicable. |
||
498 | old_is_applicable = ApplyPatchAction.is_applicable |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
499 | ApplyPatchAction.is_applicable = lambda *args: False |
||
500 | |||
501 | self.section.append(Setting( |
||
502 | "default_actions", |
||
503 | "NoBear: ApplyPatchAction, YBear: ApplyPatchAction")) |
||
504 | ret = autoapply_actions(self.results, |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
505 | {}, |
||
506 | {}, |
||
507 | self.section, |
||
508 | self.log_printer) |
||
509 | self.assertEqual(ret, self.results) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
510 | self.assertEqual(self.log_queue.get().message, |
||
511 | "Selected default action 'ApplyPatchAction' for bear " |
||
512 | "'YBear' is not applicable. Action not applied.") |
||
513 | self.assertTrue(self.log_queue.empty()) |
||
514 | |||
515 | ApplyPatchAction.is_applicable = old_is_applicable |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
516 | |||
517 | View Code Duplication | def test_applicable_action(self): |
|
518 | # Use a result whose action can be successfully applied. |
||
519 | log_printer = self.log_printer |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
520 | |||
521 | class TestAction(ResultAction): |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
522 | |||
523 | def apply(self, *args, **kwargs): |
||
524 | log_printer.debug("ACTION APPLIED SUCCESSFULLY.") |
||
525 | |||
526 | ACTIONS.append(TestAction) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
527 | |||
528 | self.section.append(Setting("default_actions", "ZBear: TestAction")) |
||
529 | ret = autoapply_actions(self.results, |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
530 | {}, |
||
531 | {}, |
||
532 | self.section, |
||
533 | log_printer) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
534 | self.assertEqual(ret, [self.resultY]) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
535 | self.assertEqual(self.log_queue.get().message, |
||
536 | "ACTION APPLIED SUCCESSFULLY.") |
||
537 | self.assertEqual(self.log_queue.get().message, |
||
538 | "Applied 'TestAction' " |
||
539 | "on the whole project from 'ZBear'.") |
||
540 | self.assertTrue(self.log_queue.empty()) |
||
541 | |||
542 | ACTIONS.pop() |
||
543 | |||
544 | View Code Duplication | def test_failing_action(self): |
|
545 | class FailingTestAction(ResultAction): |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
546 | |||
547 | def apply(self, *args, **kwargs): |
||
548 | raise RuntimeError("YEAH THAT'S A FAILING BEAR") |
||
549 | |||
550 | ACTIONS.append(FailingTestAction) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
551 | |||
552 | self.section.append(Setting("default_actions", |
||
553 | "YBear: FailingTestAction")) |
||
554 | ret = autoapply_actions(self.results, |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
555 | {}, |
||
556 | {}, |
||
557 | self.section, |
||
558 | self.log_printer) |
||
559 | self.assertEqual(ret, self.results) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
560 | self.assertEqual(self.log_queue.get().message, |
||
561 | "Failed to execute action 'FailingTestAction'" |
||
562 | " with error: YEAH THAT'S A FAILING BEAR.") |
||
563 | self.assertIn("YEAH THAT'S A FAILING BEAR", |
||
564 | self.log_queue.get().message) |
||
565 | self.assertEqual(self.log_queue.get().message, |
||
566 | "-> for result " + repr(self.resultY) + ".") |
||
567 | self.assertTrue(self.log_queue.empty()) |
||
568 | |||
569 | ACTIONS.pop() |
||
570 | |||
571 | |||
572 | class ProcessingTest_PrintResult(unittest.TestCase): |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
573 | |||
574 | def setUp(self): |
||
575 | self.section = Section('name') |
||
576 | self.log_printer = LogPrinter(ConsolePrinter(), log_level=0) |
||
577 | |||
578 | def test_autoapply_override(self): |
||
579 | """ |
||
580 | Tests that the default_actions aren't automatically applied when the |
||
581 | autoapply setting overrides that. |
||
582 | """ |
||
583 | self.section.append(Setting('default_actions', |
||
584 | 'somebear: PrintDebugMessageAction')) |
||
585 | |||
586 | # Verify that it would apply the action, i.e. remove the result |
||
587 | results = [5, HiddenResult('origin', []), |
||
588 | Result('somebear', 'message', debug_msg='debug')] |
||
589 | retval, newres = print_result(results, {}, 0, lambda *args: None, |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
590 | self.section, self.log_printer, {}, []) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
591 | self.assertEqual(newres, []) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
592 | |||
593 | # Override and verify that result is unprocessed, i.e. not gone |
||
594 | self.section.append(Setting('autoapply', 'false')) |
||
595 | retval, newres = print_result(results, {}, 0, lambda *args: None, |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
596 | self.section, self.log_printer, {}, []) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
597 | self.assertNotEqual(newres, []) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
598 |