~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_registry.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2010-09-29 22:03:03 UTC
  • mfrom: (5416.2.6 jam-integration)
  • Revision ID: pqm@pqm.ubuntu.com-20100929220303-cr95h8iwtggco721
(mbp) Add 'break-lock --force'

Show diffs side-by-side

added added

removed removed

Lines of Context:
 
1
# Copyright (C) 2006 Canonical Ltd
 
2
#
 
3
# This program is free software; you can redistribute it and/or modify
 
4
# it under the terms of the GNU General Public License as published by
 
5
# the Free Software Foundation; either version 2 of the License, or
 
6
# (at your option) any later version.
 
7
#
 
8
# This program is distributed in the hope that it will be useful,
 
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
11
# GNU General Public License for more details.
 
12
#
 
13
# You should have received a copy of the GNU General Public License
 
14
# along with this program; if not, write to the Free Software
 
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
16
 
 
17
"""Tests for the Registry classes"""
 
18
 
 
19
import os
 
20
import sys
 
21
 
 
22
from bzrlib import (
 
23
    errors,
 
24
    osutils,
 
25
    registry,
 
26
    tests,
 
27
    )
 
28
 
 
29
 
 
30
class TestRegistry(tests.TestCase):
 
31
 
 
32
    def register_stuff(self, a_registry):
 
33
        a_registry.register('one', 1)
 
34
        a_registry.register('two', 2)
 
35
        a_registry.register('four', 4)
 
36
        a_registry.register('five', 5)
 
37
 
 
38
    def test_registry(self):
 
39
        a_registry = registry.Registry()
 
40
        self.register_stuff(a_registry)
 
41
 
 
42
        self.failUnless(a_registry.default_key is None)
 
43
 
 
44
        # test get() (self.default_key is None)
 
45
        self.assertRaises(KeyError, a_registry.get)
 
46
        self.assertRaises(KeyError, a_registry.get, None)
 
47
        self.assertEqual(2, a_registry.get('two'))
 
48
        self.assertRaises(KeyError, a_registry.get, 'three')
 
49
 
 
50
        # test _set_default_key
 
51
        a_registry.default_key = 'five'
 
52
        self.failUnless(a_registry.default_key == 'five')
 
53
        self.assertEqual(5, a_registry.get())
 
54
        self.assertEqual(5, a_registry.get(None))
 
55
        # If they ask for a specific entry, they should get KeyError
 
56
        # not the default value. They can always pass None if they prefer
 
57
        self.assertRaises(KeyError, a_registry.get, 'six')
 
58
        self.assertRaises(KeyError, a_registry._set_default_key, 'six')
 
59
 
 
60
        # test keys()
 
61
        self.assertEqual(['five', 'four', 'one', 'two'], a_registry.keys())
 
62
 
 
63
    def test_registry_funcs(self):
 
64
        a_registry = registry.Registry()
 
65
        self.register_stuff(a_registry)
 
66
 
 
67
        self.failUnless('one' in a_registry)
 
68
        a_registry.remove('one')
 
69
        self.failIf('one' in a_registry)
 
70
        self.assertRaises(KeyError, a_registry.get, 'one')
 
71
 
 
72
        a_registry.register('one', 'one')
 
73
 
 
74
        self.assertEqual(['five', 'four', 'one', 'two'],
 
75
                         sorted(a_registry.keys()))
 
76
        self.assertEqual([('five', 5), ('four', 4),
 
77
                          ('one', 'one'), ('two', 2)],
 
78
                         sorted(a_registry.iteritems()))
 
79
 
 
80
    def test_register_override(self):
 
81
        a_registry = registry.Registry()
 
82
        a_registry.register('one', 'one')
 
83
        self.assertRaises(KeyError, a_registry.register, 'one', 'two')
 
84
        self.assertRaises(KeyError, a_registry.register, 'one', 'two',
 
85
                                    override_existing=False)
 
86
 
 
87
        a_registry.register('one', 'two', override_existing=True)
 
88
        self.assertEqual('two', a_registry.get('one'))
 
89
 
 
90
        self.assertRaises(KeyError, a_registry.register_lazy,
 
91
                          'one', 'three', 'four')
 
92
 
 
93
        a_registry.register_lazy('one', 'module', 'member',
 
94
                                 override_existing=True)
 
95
 
 
96
    def test_registry_help(self):
 
97
        a_registry = registry.Registry()
 
98
        a_registry.register('one', 1, help='help text for one')
 
99
        # We should not have to import the module to return the help
 
100
        # information
 
101
        a_registry.register_lazy('two', 'nonexistent_module', 'member',
 
102
                                 help='help text for two')
 
103
 
 
104
        # We should be able to handle a callable to get information
 
105
        help_calls = []
 
106
        def generic_help(reg, key):
 
107
            help_calls.append(key)
 
108
            return 'generic help for %s' % (key,)
 
109
        a_registry.register('three', 3, help=generic_help)
 
110
        a_registry.register_lazy('four', 'nonexistent_module', 'member2',
 
111
                                 help=generic_help)
 
112
        a_registry.register('five', 5)
 
113
 
 
114
        def help_from_object(reg, key):
 
115
            obj = reg.get(key)
 
116
            return obj.help()
 
117
 
 
118
        class SimpleObj(object):
 
119
            def help(self):
 
120
                return 'this is my help'
 
121
        a_registry.register('six', SimpleObj(), help=help_from_object)
 
122
 
 
123
        self.assertEqual('help text for one', a_registry.get_help('one'))
 
124
        self.assertEqual('help text for two', a_registry.get_help('two'))
 
125
        self.assertEqual('generic help for three',
 
126
                         a_registry.get_help('three'))
 
127
        self.assertEqual(['three'], help_calls)
 
128
        self.assertEqual('generic help for four',
 
129
                         a_registry.get_help('four'))
 
130
        self.assertEqual(['three', 'four'], help_calls)
 
131
        self.assertEqual(None, a_registry.get_help('five'))
 
132
        self.assertEqual('this is my help', a_registry.get_help('six'))
 
133
 
 
134
        self.assertRaises(KeyError, a_registry.get_help, None)
 
135
        self.assertRaises(KeyError, a_registry.get_help, 'seven')
 
136
 
 
137
        a_registry.default_key = 'one'
 
138
        self.assertEqual('help text for one', a_registry.get_help(None))
 
139
        self.assertRaises(KeyError, a_registry.get_help, 'seven')
 
140
 
 
141
        self.assertEqual([('five', None),
 
142
                          ('four', 'generic help for four'),
 
143
                          ('one', 'help text for one'),
 
144
                          ('six', 'this is my help'),
 
145
                          ('three', 'generic help for three'),
 
146
                          ('two', 'help text for two'),
 
147
                         ], sorted((key, a_registry.get_help(key))
 
148
                                    for key in a_registry.keys()))
 
149
 
 
150
        # We don't know what order it was called in, but we should get
 
151
        # 2 more calls to three and four
 
152
        self.assertEqual(['four', 'four', 'three', 'three'],
 
153
                         sorted(help_calls))
 
154
 
 
155
    def test_registry_info(self):
 
156
        a_registry = registry.Registry()
 
157
        a_registry.register('one', 1, info='string info')
 
158
        # We should not have to import the module to return the info
 
159
        a_registry.register_lazy('two', 'nonexistent_module', 'member',
 
160
                                 info=2)
 
161
 
 
162
        # We should be able to handle a callable to get information
 
163
        a_registry.register('three', 3, info=['a', 'list'])
 
164
        obj = object()
 
165
        a_registry.register_lazy('four', 'nonexistent_module', 'member2',
 
166
                                 info=obj)
 
167
        a_registry.register('five', 5)
 
168
 
 
169
        self.assertEqual('string info', a_registry.get_info('one'))
 
170
        self.assertEqual(2, a_registry.get_info('two'))
 
171
        self.assertEqual(['a', 'list'], a_registry.get_info('three'))
 
172
        self.assertIs(obj, a_registry.get_info('four'))
 
173
        self.assertIs(None, a_registry.get_info('five'))
 
174
 
 
175
        self.assertRaises(KeyError, a_registry.get_info, None)
 
176
        self.assertRaises(KeyError, a_registry.get_info, 'six')
 
177
 
 
178
        a_registry.default_key = 'one'
 
179
        self.assertEqual('string info', a_registry.get_info(None))
 
180
        self.assertRaises(KeyError, a_registry.get_info, 'six')
 
181
 
 
182
        self.assertEqual([('five', None),
 
183
                          ('four', obj),
 
184
                          ('one', 'string info'),
 
185
                          ('three', ['a', 'list']),
 
186
                          ('two', 2),
 
187
                         ], sorted((key, a_registry.get_info(key))
 
188
                                    for key in a_registry.keys()))
 
189
 
 
190
    def test_get_prefix(self):
 
191
        my_registry = registry.Registry()
 
192
        http_object = object()
 
193
        sftp_object = object()
 
194
        my_registry.register('http:', http_object)
 
195
        my_registry.register('sftp:', sftp_object)
 
196
        found_object, suffix = my_registry.get_prefix('http://foo/bar')
 
197
        self.assertEqual('//foo/bar', suffix)
 
198
        self.assertIs(http_object, found_object)
 
199
        self.assertIsNot(sftp_object, found_object)
 
200
        found_object, suffix = my_registry.get_prefix('sftp://baz/qux')
 
201
        self.assertEqual('//baz/qux', suffix)
 
202
        self.assertIs(sftp_object, found_object)
 
203
 
 
204
 
 
205
class TestRegistryIter(tests.TestCase):
 
206
    """Test registry iteration behaviors.
 
207
 
 
208
    There are dark corner cases here when the registered objects trigger
 
209
    addition in the iterated registry.
 
210
    """
 
211
 
 
212
    def setUp(self):
 
213
        super(TestRegistryIter, self).setUp()
 
214
 
 
215
        # We create a registry with "official" objects and "hidden"
 
216
        # objects. The later represent the side effects that led to bug #277048
 
217
        # and #430510
 
218
        self.registry =  registry.Registry()
 
219
 
 
220
        def register_more():
 
221
            self.registry.register('hidden', None)
 
222
 
 
223
        self.registry.register('passive', None)
 
224
        self.registry.register('active', register_more)
 
225
        self.registry.register('passive-too', None)
 
226
 
 
227
        class InvasiveGetter(registry._ObjectGetter):
 
228
 
 
229
            def get_obj(inner_self):
 
230
                # Surprise ! Getting a registered object (think lazy loaded
 
231
                # module) register yet another object !
 
232
                self.registry.register('more hidden', None)
 
233
                return inner_self._obj
 
234
 
 
235
        self.registry.register('hacky', None)
 
236
        # We peek under the covers because the alternative is to use lazy
 
237
        # registration and create a module that can reference our test registry
 
238
        # it's too much work for such a corner case -- vila 090916
 
239
        self.registry._dict['hacky'] = InvasiveGetter(None)
 
240
 
 
241
    def _iter_them(self, iter_func_name):
 
242
        iter_func = getattr(self.registry, iter_func_name, None)
 
243
        self.assertIsNot(None, iter_func)
 
244
        count = 0
 
245
        for name, func in iter_func():
 
246
            count += 1
 
247
            self.assertFalse(name in ('hidden', 'more hidden'))
 
248
            if func is not None:
 
249
                # Using an object register another one as a side effect
 
250
                func()
 
251
        self.assertEqual(4, count)
 
252
 
 
253
    def test_iteritems(self):
 
254
        # the dict is modified during the iteration
 
255
        self.assertRaises(RuntimeError, self._iter_them, 'iteritems')
 
256
 
 
257
    def test_items(self):
 
258
        # we should be able to iterate even if one item modify the dict
 
259
        self._iter_them('items')
 
260
 
 
261
 
 
262
class TestRegistryWithDirs(tests.TestCaseInTempDir):
 
263
    """Registry tests that require temporary dirs"""
 
264
 
 
265
    def create_plugin_file(self, contents):
 
266
        """Create a file to be used as a plugin.
 
267
 
 
268
        This is created in a temporary directory, so that we
 
269
        are sure that it doesn't start in the plugin path.
 
270
        """
 
271
        os.mkdir('tmp')
 
272
        plugin_name = 'bzr_plugin_a_%s' % (osutils.rand_chars(4),)
 
273
        open('tmp/'+plugin_name+'.py', 'wb').write(contents)
 
274
        return plugin_name
 
275
 
 
276
    def create_simple_plugin(self):
 
277
        return self.create_plugin_file(
 
278
            'object1 = "foo"\n'
 
279
            '\n\n'
 
280
            'def function(a,b,c):\n'
 
281
            '    return a,b,c\n'
 
282
            '\n\n'
 
283
            'class MyClass(object):\n'
 
284
            '    def __init__(self, a):\n'
 
285
            '      self.a = a\n'
 
286
            '\n\n'
 
287
        )
 
288
 
 
289
    def test_lazy_import_registry(self):
 
290
        plugin_name = self.create_simple_plugin()
 
291
        a_registry = registry.Registry()
 
292
        a_registry.register_lazy('obj', plugin_name, 'object1')
 
293
        a_registry.register_lazy('function', plugin_name, 'function')
 
294
        a_registry.register_lazy('klass', plugin_name, 'MyClass')
 
295
        a_registry.register_lazy('module', plugin_name, None)
 
296
 
 
297
        self.assertEqual(['function', 'klass', 'module', 'obj'],
 
298
                         sorted(a_registry.keys()))
 
299
        # The plugin should not be loaded until we grab the first object
 
300
        self.failIf(plugin_name in sys.modules)
 
301
 
 
302
        # By default the plugin won't be in the search path
 
303
        self.assertRaises(ImportError, a_registry.get, 'obj')
 
304
 
 
305
        plugin_path = os.getcwd() + '/tmp'
 
306
        sys.path.append(plugin_path)
 
307
        try:
 
308
            obj = a_registry.get('obj')
 
309
            self.assertEqual('foo', obj)
 
310
            self.failUnless(plugin_name in sys.modules)
 
311
 
 
312
            # Now grab another object
 
313
            func = a_registry.get('function')
 
314
            self.assertEqual(plugin_name, func.__module__)
 
315
            self.assertEqual('function', func.__name__)
 
316
            self.assertEqual((1, [], '3'), func(1, [], '3'))
 
317
 
 
318
            # And finally a class
 
319
            klass = a_registry.get('klass')
 
320
            self.assertEqual(plugin_name, klass.__module__)
 
321
            self.assertEqual('MyClass', klass.__name__)
 
322
 
 
323
            inst = klass(1)
 
324
            self.assertIsInstance(inst, klass)
 
325
            self.assertEqual(1, inst.a)
 
326
 
 
327
            module = a_registry.get('module')
 
328
            self.assertIs(obj, module.object1)
 
329
            self.assertIs(func, module.function)
 
330
            self.assertIs(klass, module.MyClass)
 
331
        finally:
 
332
            sys.path.remove(plugin_path)
 
333