Failed Conditions
Pull Request — master (#2076)
by Abdeali
02:04
created

TestPrinter.log_message()   A

Complexity

Conditions 1

Size

Total Lines 2

Duplication

Lines 0
Ratio 0 %
Metric Value
cc 1
dl 0
loc 2
rs 10
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
The variable LogPrinter does not seem to be defined.
Loading history...
10
11
    def __init__(self):
12
        LogPrinter.__init__(self, self)
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable self does not seem to be defined.
Loading history...
13
14
    def log_message(self, log_message, timestamp=None, **kwargs):
15
        print(log_message)
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable log_message does not seem to be defined.
Loading history...
16
17
18
class LogPrinterThreadTest(unittest.TestCase):
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable unittest does not seem to be defined.
Loading history...
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
The variable log_queue does not seem to be defined.
Loading history...
Comprehensibility Best Practice introduced by
The variable log_printer does not seem to be defined.
Loading history...
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
The variable stdout does not seem to be defined.
Loading history...
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