source: orange/Orange/OrangeCanvas/application/tests/test_outputview.py @ 11258:4c6f5d7a456b

Revision 11258:4c6f5d7a456b, 3.7 KB checked in by Ales Erjavec <ales.erjavec@…>, 15 months ago (diff)

Override sys.excepthook with a custom handler.

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