/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: 2010-01-12 22:51:31 UTC
  • mto: This revision was merged to the branch mainline in revision 4955.
  • Revision ID: john@arbash-meinel.com-20100112225131-he8h411p6aeeb947
Delay grabbing an output stream until we actually go to show a diff.

This makes the test suite happy, but it also seems to be reasonable.
If we aren't going to write anything, we don't need to hold an
output stream open.

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 start_server(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.start_server()
 
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 stop_server(self):
 
283
        self.stop_background_thread()
 
284
        self.chroot_server.stop_server()
 
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