1 | import queue |
||
2 | import unittest |
||
3 | |||
4 | from coalib.misc.ContextManagers import retrieve_stdout |
||
5 | from coalib.output.printers.LogPrinter import LogPrinter |
||
6 | from coalib.processes.LogPrinterThread import LogPrinterThread |
||
7 | |||
8 | |||
9 | class TestPrinter(LogPrinter): |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
![]() |
|||
10 | |||
11 | def __init__(self): |
||
12 | LogPrinter.__init__(self, self) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
13 | |||
14 | def log_message(self, log_message, timestamp=None, **kwargs): |
||
15 | print(log_message) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
16 | |||
17 | |||
18 | class LogPrinterThreadTest(unittest.TestCase): |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
19 | |||
20 | def test_run(self): |
||
21 | log_printer = TestPrinter() |
||
22 | log_queue = queue.Queue() |
||
23 | self.uut = LogPrinterThread(log_queue, log_printer) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
Comprehensibility
Best Practice
introduced
by
|
|||
24 | log_queue.put(item="Sample message 1") |
||
25 | log_queue.put(item="Sample message 2") |
||
26 | log_queue.put(item="Sample message 3") |
||
27 | self.assertEqual(self.uut.message_queue.qsize(), 3) |
||
28 | with retrieve_stdout() as stdout: |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
29 | self.uut.start() |
||
30 | while self.uut.message_queue.qsize() > 0: |
||
31 | continue |
||
32 | self.uut.running = False |
||
33 | self.uut.join() |
||
34 | self.assertEqual(stdout.getvalue(), |
||
35 | "Sample message 1\nSample message 2\nSample " |
||
36 | "message 3\n") |
||
37 |