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
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
|
# (C) Copyright 2007-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 the composite plugin manager. """
import unittest
from traits.api import Bool
from envisage.application import Application
from envisage.composite_plugin_manager import CompositePluginManager
from envisage.plugin import Plugin
from envisage.plugin_manager import PluginManager
class SimplePlugin(Plugin):
"""A simple plugin."""
#### 'SimplePlugin' protocol ##############################################
started = Bool(False)
stopped = Bool(False)
#### 'IPlugin' protocol ###################################################
def start(self):
"""Start the plugin."""
self.started = True
self.stopped = False
def stop(self):
"""Stop the plugin."""
self.started = False
self.stopped = True
class CustomException(Exception):
"""Custom exception used for testing purposes."""
pass
class RaisingPluginManager(PluginManager):
"""A PluginManager that raises on iteration."""
def __iter__(self):
raise CustomException("Something went wrong.")
class CompositePluginManagerTestCase(unittest.TestCase):
"""Tests for the composite plugin manager."""
def test_find_no_plugins_if_there_are_no_plugin_managers(self):
plugin_manager = CompositePluginManager()
ids = [plugin.id for plugin in plugin_manager]
self.assertEqual(0, len(ids))
def test_find_no_plugins_if_there_are_no_plugins_in_plugin_managers(self):
plugin_manager = CompositePluginManager(
plugin_managers=[PluginManager(), PluginManager()]
)
ids = [plugin.id for plugin in plugin_manager]
self.assertEqual(0, len(ids))
def test_find_plugins_in_a_single_plugin_manager(self):
plugin_manager = CompositePluginManager(
plugin_managers=[
PluginManager(
plugins=[SimplePlugin(id="red"), SimplePlugin(id="yellow")]
)
]
)
ids = [plugin.id for plugin in plugin_manager]
self.assertEqual(2, len(ids))
self.assertIn("red", ids)
self.assertIn("yellow", ids)
self._test_start_and_stop(plugin_manager, ["red", "yellow"])
def test_find_plugins_in_a_multiple_plugin_managers(self):
plugin_manager = CompositePluginManager(
plugin_managers=[
PluginManager(
plugins=[SimplePlugin(id="red"), SimplePlugin(id="yellow")]
),
PluginManager(plugins=[SimplePlugin(id="green")]),
]
)
ids = [plugin.id for plugin in plugin_manager]
self.assertEqual(3, len(ids))
self.assertIn("red", ids)
self.assertIn("yellow", ids)
self.assertIn("green", ids)
self._test_start_and_stop(plugin_manager, ["red", "yellow", "green"])
def test_application_gets_propogated_to_plugin_managers(self):
application = Application()
composite_plugin_manager = CompositePluginManager(
application=application,
plugin_managers=[PluginManager(), PluginManager()],
)
for plugin_manager in composite_plugin_manager.plugin_managers:
self.assertEqual(application, plugin_manager.application)
def test_propogate_plugin_added_or_remove_events_from_plugin_managers(
self,
):
a = PluginManager()
b = PluginManager()
composite_plugin_manager = CompositePluginManager(
plugin_managers=[a, b]
)
composite_plugin_manager._plugins
def added(obj, trait_name, old, new):
added.count += 1
added.count = 0
composite_plugin_manager.on_trait_change(added, "plugin_added")
def removed(obj, trait_name, old, new):
removed.count += 1
removed.count = 0
composite_plugin_manager.on_trait_change(removed, "plugin_removed")
a.add_plugin(Plugin(id="foo"))
self.assertEqual(1, self._plugin_count(composite_plugin_manager))
a.remove_plugin(a.get_plugin("foo"))
self.assertEqual(0, self._plugin_count(composite_plugin_manager))
def test_correct_exception_propagated_from_plugin_manager(self):
plugin_manager = CompositePluginManager(
plugin_managers=[RaisingPluginManager()]
)
with self.assertRaises(CustomException):
plugin_manager.start()
#### Private protocol #####################################################
def _plugin_count(self, plugin_manager):
"""Return how many plugins the plugin manager contains."""
count = 0
for plugin in plugin_manager:
count += 1
return count
def _test_start_and_stop(self, plugin_manager, expected):
"""
Make sure the plugin manager starts and stops the expected plugins.
"""
# Make sure the plugin manager found only the required plugins.
self.assertEqual(expected, [plugin.id for plugin in plugin_manager])
# Start the plugin manager. This starts all of the plugin manager's
# plugins.
plugin_manager.start()
# Make sure all of the the plugins were started.
for id in expected:
plugin = plugin_manager.get_plugin(id)
self.assertNotEqual(None, plugin)
self.assertEqual(True, plugin.started)
# Stop the plugin manager. This stops all of the plugin manager's
# plugins.
plugin_manager.stop()
# Make sure all of the the plugins were stopped.
for id in expected:
plugin = plugin_manager.get_plugin(id)
self.assertNotEqual(None, plugin)
self.assertEqual(True, plugin.stopped)
|
