source: orange/Orange/OrangeCanvas/application/tests/test_outputview.py @ 11257:cb4e4ab85ce8

Revision 11257:cb4e4ab85ce8, 2.9 KB checked in by Ales Erjavec <ales.erjavec@…>, 15 months ago (diff)

Added colored/formated output for the stdout/err output, output thread safety.

Line 
1import multiprocessing.pool
2
3from datetime import datetime
4from threading import current_thread
5
6from PyQt4.QtCore import Qt, QThread
7from ...gui.test import QAppTestCase
8
9from ..outputview import OutputView, TextStream
10
11
12class TestOutputView(QAppTestCase):
13    def test_outputview(self):
14        output = OutputView()
15        output.show()
16
17        line1 = "A line \n"
18        line2 = "A different line\n"
19        output.write(line1)
20        self.assertEqual(unicode(output.toPlainText()), line1)
21
22        output.write(line2)
23        self.assertEqual(unicode(output.toPlainText()), line1 + line2)
24
25        output.clear()
26        self.assertEqual(unicode(output.toPlainText()), "")
27
28        output.writelines([line1, line2])
29        self.assertEqual(unicode(output.toPlainText()), line1 + line2)
30
31        output.setMaximumLines(5)
32
33        def advance():
34            now = datetime.now().strftime("%c\n")
35            output.write(now)
36
37            text = unicode(output.toPlainText())
38            self.assertLessEqual(len(text.splitlines()), 5)
39
40            self.singleShot(500, advance)
41
42        advance()
43
44        self.app.exec_()
45
46    def test_formated(self):
47        output = OutputView()
48        output.show()
49
50        output.write("A sword day, ")
51        with output.formated(color=Qt.red) as f:
52            f.write("a red day...\n")
53
54            with f.formated(color=Qt.green) as f:
55                f.write("Actually sir, orcs bleed green.\n")
56
57        bold = output.formated(weight=100, underline=True)
58        bold.write("Shutup")
59
60        self.app.exec_()
61
62    def test_threadsafe(self):
63        output = OutputView()
64        output.resize(500, 300)
65        output.show()
66
67        blue_formater = output.formated(color=Qt.blue)
68        red_formater = output.formated(color=Qt.red)
69
70        correct = []
71
72        def check_thread(*args):
73            correct.append(QThread.currentThread() == self.app.thread())
74
75        blue = TextStream()
76        blue.stream.connect(blue_formater.write)
77        blue.stream.connect(check_thread)
78
79        red = TextStream()
80        red.stream.connect(red_formater.write)
81        red.stream.connect(check_thread)
82
83        def printer(i):
84            if i % 12 == 0:
85                fizzbuz = "fizzbuz"
86            elif i % 4 == 0:
87                fizzbuz = "buz"
88            elif i % 3 == 0:
89                fizzbuz = "fizz"
90            else:
91                fizzbuz = str(i)
92
93            if i % 2:
94                writer = blue
95            else:
96                writer = red
97
98            writer.write("Greetings from thread {0}. "
99                         "This is {1}\n".format(current_thread().name,
100                                                fizzbuz))
101
102        pool = multiprocessing.pool.ThreadPool(100)
103        res = pool.map_async(printer, range(10000))
104
105        self.app.exec_()
106
107        res.wait()
108
109        self.assertTrue(all(correct))
110        self.assertTrue(len(correct) == 10000)
Note: See TracBrowser for help on using the repository browser.