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
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
|
# (C) Copyright 2005-2023 Enthought, Inc., Austin, TX
# All rights reserved.
#
# This software is provided without warranty under the terms of the BSD
# license included in LICENSE.txt and may be redistributed only under
# the conditions described in the aforementioned license. The license
# is also available online at http://www.enthought.com/licenses/BSD.txt
#
# Thanks for using Enthought open source!
""" Tests for dynamic notifiers with `dispatch='ui'`.
Dynamic notifiers created with the `dispatch='ui'` option dispatch event
notifications on the UI thread. The class handling the dispatch,
`FastUITraitChangeNotifyWrapper`, is a subclass of `TraitChangeNotifyWrapper`.
Most of the functionality of the class is thus already covered by the
`TestDynamicNotifiers` test case, and we only need to test that the
notification really occurs on the UI thread.
At present, `dispatch='ui'` and `dispatch='fast_ui'` have the same effect.
"""
import threading
import time
import unittest
# Preamble: Try importing Qt, and set QT_FOUND to True on success.
try:
from pyface.util.guisupport import get_app_qt4
# This import is necessary to set the `ui_handler` global variable in
# `traits.trait_notifiers`, which is responsible for dispatching the events
# to the UI thread.
from traitsui.qt4 import toolkit # noqa: F401
qt4_app = get_app_qt4()
except Exception:
QT_FOUND = False
else:
QT_FOUND = True
from traits import trait_notifiers
from traits.api import Callable, Float, HasTraits, on_trait_change
class CalledAsMethod(HasTraits):
foo = Float
class CalledAsDecorator(HasTraits):
foo = Float
callback = Callable
@on_trait_change("foo", dispatch="ui")
def on_foo_change(self, obj, name, old, new):
self.callback(obj, name, old, new)
class BaseTestUINotifiers(object):
""" Tests for dynamic notifiers with `dispatch='ui'`.
"""
#### 'TestCase' protocol ##################################################
def setUp(self):
self.notifications = []
#### 'TestUINotifiers' protocol ###########################################
def flush_event_loop(self):
""" Post and process the Qt events. """
qt4_app.sendPostedEvents()
qt4_app.processEvents()
def on_foo_notifications(self, obj, name, old, new):
thread_id = threading.current_thread().ident
event = (thread_id, (obj, name, old, new))
self.notifications.append(event)
#### Tests ################################################################
@unittest.skipIf(
not QT_FOUND, "Qt event loop not found, UI dispatch not possible."
)
def test_notification_from_main_thread(self):
obj = self.obj_factory()
obj.foo = 3
self.flush_event_loop()
notifications = self.notifications
self.assertEqual(len(notifications), 1)
thread_id, event = notifications[0]
self.assertEqual(event, (obj, "foo", 0, 3))
ui_thread = trait_notifiers.ui_thread
self.assertEqual(thread_id, ui_thread)
@unittest.skipIf(
not QT_FOUND, "Qt event loop not found, UI dispatch not possible."
)
def test_notification_from_separate_thread(self):
obj = self.obj_factory()
# Set obj.foo to 3 on a separate thread.
def set_foo_to_3(obj):
obj.foo = 3
threading.Thread(target=set_foo_to_3, args=(obj,)).start()
# Wait for a while to make sure the function has finished.
time.sleep(0.1)
self.flush_event_loop()
notifications = self.notifications
self.assertEqual(len(notifications), 1)
thread_id, event = notifications[0]
self.assertEqual(event, (obj, "foo", 0, 3))
ui_thread = trait_notifiers.ui_thread
self.assertEqual(thread_id, ui_thread)
class TestMethodUINotifiers(BaseTestUINotifiers, unittest.TestCase):
""" Tests for dynamic notifiers with `dispatch='ui'` set by method call.
"""
def obj_factory(self):
obj = CalledAsMethod()
obj.on_trait_change(self.on_foo_notifications, "foo", dispatch="ui")
return obj
class TestDecoratorUINotifiers(BaseTestUINotifiers, unittest.TestCase):
""" Tests for dynamic notifiers with `dispatch='ui'` set by decorator. """
def obj_factory(self):
return CalledAsDecorator(callback=self.on_foo_notifications)
|
