← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_server.py

  • Committer: Vincent Ladeuil
  • Date: 2010-02-09 20:28:26 UTC
  • mto: (5029.1.1 integration)
  • mto: This revision was merged to the branch mainline in revision 5030.
  • Revision ID: v.ladeuil+lp@free.fr-20100209202826-0u7aj2x51uzm79pv
Move SmartTCPServer_for_testing and friends to bzrlib.tests.test_server

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_server.py
18
18
    transport,
19
19
    urlutils,
20
20
    )
 
21
from bzrlib.smart import server
21
22
 
22
23
 
23
24
class TestServer(transport.Server):
205
206
        return unlistable.UnlistableTransportDecorator
206
207
 
207
208
 
 
209
class SmartTCPServer_for_testing(server.SmartTCPServer):
 
210
    """Server suitable for use by transport tests.
 
211
 
 
212
    This server is backed by the process's cwd.
 
213
    """
 
214
 
 
215
    def __init__(self, thread_name_suffix=''):
 
216
        super(SmartTCPServer_for_testing, self).__init__(None)
 
217
        self.client_path_extra = None
 
218
        self.thread_name_suffix = thread_name_suffix
 
219
 
 
220
    def get_backing_transport(self, backing_transport_server):
 
221
        """Get a backing transport from a server we are decorating."""
 
222
        return transport.get_transport(backing_transport_server.get_url())
 
223
 
 
224
    def start_server(self, backing_transport_server=None,
 
225
              client_path_extra='/extra/'):
 
226
        """Set up server for testing.
 
227
 
 
228
        :param backing_transport_server: backing server to use.  If not
 
229
            specified, a LocalURLServer at the current working directory will
 
230
            be used.
 
231
        :param client_path_extra: a path segment starting with '/' to append to
 
232
            the root URL for this server.  For instance, a value of '/foo/bar/'
 
233
            will mean the root of the backing transport will be published at a
 
234
            URL like `bzr://127.0.0.1:nnnn/foo/bar/`, rather than
 
235
            `bzr://127.0.0.1:nnnn/`.  Default value is `extra`, so that tests
 
236
            by default will fail unless they do the necessary path translation.
 
237
        """
 
238
        if not client_path_extra.startswith('/'):
 
239
            raise ValueError(client_path_extra)
 
240
        from bzrlib.transport.chroot import ChrootServer
 
241
        if backing_transport_server is None:
 
242
            backing_transport_server = LocalURLServer()
 
243
        self.chroot_server = ChrootServer(
 
244
            self.get_backing_transport(backing_transport_server))
 
245
        self.chroot_server.start_server()
 
246
        self.backing_transport = transport.get_transport(
 
247
            self.chroot_server.get_url())
 
248
        self.root_client_path = self.client_path_extra = client_path_extra
 
249
        self.start_background_thread(self.thread_name_suffix)
 
250
 
 
251
    def stop_server(self):
 
252
        self.stop_background_thread()
 
253
        self.chroot_server.stop_server()
 
254
 
 
255
    def get_url(self):
 
256
        url = super(SmartTCPServer_for_testing, self).get_url()
 
257
        return url[:-1] + self.client_path_extra
 
258
 
 
259
    def get_bogus_url(self):
 
260
        """Return a URL which will fail to connect"""
 
261
        return 'bzr://127.0.0.1:1/'
 
262
 
 
263
 
 
264
class ReadonlySmartTCPServer_for_testing(SmartTCPServer_for_testing):
 
265
    """Get a readonly server for testing."""
 
266
 
 
267
    def get_backing_transport(self, backing_transport_server):
 
268
        """Get a backing transport from a server we are decorating."""
 
269
        url = 'readonly+' + backing_transport_server.get_url()
 
270
        return transport.get_transport(url)
 
271
 
 
272
 
 
273
class SmartTCPServer_for_testing_v2_only(SmartTCPServer_for_testing):
 
274
    """A variation of SmartTCPServer_for_testing that limits the client to
 
275
    using RPCs in protocol v2 (i.e. bzr <= 1.5).
 
276
    """
 
277
 
 
278
    def get_url(self):
 
279
        url = super(SmartTCPServer_for_testing_v2_only, self).get_url()
 
280
        url = 'bzr-v2://' + url[len('bzr://'):]
 
281
        return url
 
282
 
 
283
 
 
284
class ReadonlySmartTCPServer_for_testing_v2_only(
 
285
    SmartTCPServer_for_testing_v2_only):
 
286
    """Get a readonly server for testing."""
 
287
 
 
288
    def get_backing_transport(self, backing_transport_server):
 
289
        """Get a backing transport from a server we are decorating."""
 
290
        url = 'readonly+' + backing_transport_server.get_url()
 
291
        return transport.get_transport(url)
 
292
 
 
293
 
 
294
 
 
295