/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/smart/server.py

  • Committer: John Arbash Meinel
  • Date: 2009-12-22 16:28:47 UTC
  • mto: This revision was merged to the branch mainline in revision 4922.
  • Revision ID: john@arbash-meinel.com-20091222162847-tvnsc69to4l4uf5r
Implement a permute_for_extension helper.

Use it for all of the 'simple' extension permutations.
It basically permutes all tests in the current module, by setting TestCase.module.
Which works well for most of our extension tests. Some had more advanced
handling of permutations (extra permutations, custom vars, etc.)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/smart/server.py
1
 
# Copyright (C) 2006-2010 Canonical Ltd
 
1
# Copyright (C) 2006, 2007, 2008 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
138
138
                        if e.args[0] != errno.EBADF:
139
139
                            trace.warning("listening socket error: %s", e)
140
140
                    else:
141
 
                        if self._should_terminate:
142
 
                            break
143
141
                        self.serve_conn(conn, thread_name_suffix)
144
142
            except KeyboardInterrupt:
145
143
                # dont log when CTRL-C'd.
238
236
SmartTCPServer.hooks = SmartServerHooks()
239
237
 
240
238
 
 
239
class SmartTCPServer_for_testing(SmartTCPServer):
 
240
    """Server suitable for use by transport tests.
 
241
 
 
242
    This server is backed by the process's cwd.
 
243
    """
 
244
 
 
245
    def __init__(self, thread_name_suffix=''):
 
246
        SmartTCPServer.__init__(self, None)
 
247
        self.client_path_extra = None
 
248
        self.thread_name_suffix = thread_name_suffix
 
249
 
 
250
    def get_backing_transport(self, backing_transport_server):
 
251
        """Get a backing transport from a server we are decorating."""
 
252
        return transport.get_transport(backing_transport_server.get_url())
 
253
 
 
254
    def setUp(self, backing_transport_server=None,
 
255
              client_path_extra='/extra/'):
 
256
        """Set up server for testing.
 
257
 
 
258
        :param backing_transport_server: backing server to use.  If not
 
259
            specified, a LocalURLServer at the current working directory will
 
260
            be used.
 
261
        :param client_path_extra: a path segment starting with '/' to append to
 
262
            the root URL for this server.  For instance, a value of '/foo/bar/'
 
263
            will mean the root of the backing transport will be published at a
 
264
            URL like `bzr://127.0.0.1:nnnn/foo/bar/`, rather than
 
265
            `bzr://127.0.0.1:nnnn/`.  Default value is `extra`, so that tests
 
266
            by default will fail unless they do the necessary path translation.
 
267
        """
 
268
        if not client_path_extra.startswith('/'):
 
269
            raise ValueError(client_path_extra)
 
270
        from bzrlib.transport.chroot import ChrootServer
 
271
        if backing_transport_server is None:
 
272
            from bzrlib.transport.local import LocalURLServer
 
273
            backing_transport_server = LocalURLServer()
 
274
        self.chroot_server = ChrootServer(
 
275
            self.get_backing_transport(backing_transport_server))
 
276
        self.chroot_server.setUp()
 
277
        self.backing_transport = transport.get_transport(
 
278
            self.chroot_server.get_url())
 
279
        self.root_client_path = self.client_path_extra = client_path_extra
 
280
        self.start_background_thread(self.thread_name_suffix)
 
281
 
 
282
    def tearDown(self):
 
283
        self.stop_background_thread()
 
284
        self.chroot_server.tearDown()
 
285
 
 
286
    def get_url(self):
 
287
        url = super(SmartTCPServer_for_testing, self).get_url()
 
288
        return url[:-1] + self.client_path_extra
 
289
 
 
290
    def get_bogus_url(self):
 
291
        """Return a URL which will fail to connect"""
 
292
        return 'bzr://127.0.0.1:1/'
 
293
 
 
294
 
 
295
class ReadonlySmartTCPServer_for_testing(SmartTCPServer_for_testing):
 
296
    """Get a readonly server for testing."""
 
297
 
 
298
    def get_backing_transport(self, backing_transport_server):
 
299
        """Get a backing transport from a server we are decorating."""
 
300
        url = 'readonly+' + backing_transport_server.get_url()
 
301
        return transport.get_transport(url)
 
302
 
 
303
 
 
304
class SmartTCPServer_for_testing_v2_only(SmartTCPServer_for_testing):
 
305
    """A variation of SmartTCPServer_for_testing that limits the client to
 
306
    using RPCs in protocol v2 (i.e. bzr <= 1.5).
 
307
    """
 
308
 
 
309
    def get_url(self):
 
310
        url = super(SmartTCPServer_for_testing_v2_only, self).get_url()
 
311
        url = 'bzr-v2://' + url[len('bzr://'):]
 
312
        return url
 
313
 
 
314
 
 
315
class ReadonlySmartTCPServer_for_testing_v2_only(SmartTCPServer_for_testing_v2_only):
 
316
    """Get a readonly server for testing."""
 
317
 
 
318
    def get_backing_transport(self, backing_transport_server):
 
319
        """Get a backing transport from a server we are decorating."""
 
320
        url = 'readonly+' + backing_transport_server.get_url()
 
321
        return transport.get_transport(url)
 
322
 
 
323
 
241
324
def _local_path_for_transport(transport):
242
325
    """Return a local path for transport, if reasonably possible.
243
326
    
303
386
        """Chroot transport, and decorate with userdir expander."""
304
387
        self.base_path = self.get_base_path(transport)
305
388
        chroot_server = chroot.ChrootServer(transport)
306
 
        chroot_server.start_server()
307
 
        self.cleanups.append(chroot_server.stop_server)
 
389
        chroot_server.setUp()
 
390
        self.cleanups.append(chroot_server.tearDown)
308
391
        transport = get_transport(chroot_server.get_url())
309
392
        if self.base_path is not None:
310
393
            # Decorate the server's backing transport with a filter that can
311
394
            # expand homedirs.
312
395
            expand_userdirs = self._make_expand_userdirs_filter(transport)
313
 
            expand_userdirs.start_server()
314
 
            self.cleanups.append(expand_userdirs.stop_server)
 
396
            expand_userdirs.setUp()
 
397
            self.cleanups.append(expand_userdirs.tearDown)
315
398
            transport = get_transport(expand_userdirs.get_url())
316
399
        self.transport = transport
317
400