1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
|
import unittest.mock
import warnings
from concurrent.futures import Future
from AnyQt.QtCore import Qt, QCoreApplication, QThread
from Orange.widgets.utils.concurrent import ThreadExecutor, Task
class CoreAppTestCase(unittest.TestCase):
def setUp(self):
self.app = QCoreApplication.instance()
if self.app is None:
self.app = QCoreApplication([])
def tearDown(self):
self.app.processEvents()
del self.app
class TestTask(CoreAppTestCase):
def setUp(self):
# This test tests a deprecated class, so ... obviously
warnings.filterwarnings(
"ignore", "`Task` has been deprecated", PendingDeprecationWarning)
warnings.filterwarnings(
"ignore", "`submit_task` will be deprecated",
PendingDeprecationWarning)
super().setUp()
def test_task(self):
results = []
task = Task(function=QThread.currentThread)
task.resultReady.connect(results.append)
task.start()
self.app.processEvents()
self.assertSequenceEqual(results, [QThread.currentThread()])
thread = QThread()
thread.start()
try:
task = Task(function=QThread.currentThread)
task.moveToThread(thread)
self.assertIsNot(task.thread(), QThread.currentThread())
self.assertIs(task.thread(), thread)
results = Future()
def record(value):
# record the result value and the calling thread
results.set_result((QThread.currentThread(), value))
task.resultReady.connect(record, Qt.DirectConnection)
task.start()
f = task.future()
emit_thread, thread_ = results.result(3)
self.assertIs(f.result(3), thread)
self.assertIs(emit_thread, thread)
self.assertIs(thread_, thread)
finally:
thread.quit()
thread.wait()
def test_executor(self):
executor = ThreadExecutor()
f = executor.submit(QThread.currentThread)
self.assertIsNot(f.result(3), QThread.currentThread())
f = executor.submit(lambda: 1 / 0)
with self.assertRaises(ZeroDivisionError):
f.result()
results = []
task = Task(function=QThread.currentThread)
task.resultReady.connect(results.append, Qt.DirectConnection)
f = executor.submit_task(task)
self.assertIsNot(f.result(3), QThread.currentThread())
executor.shutdown()
|
