← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_server.py

(gz) Fix test failure on alpha by correcting format string for
 gc_chk_sha1_record (Martin [gz])

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_server.py
212
212
    def start_server(self, backing_server=None):
213
213
        """Setup the Chroot on backing_server."""
214
214
        if backing_server is not None:
215
 
            self.backing_transport = transport.get_transport_from_url(
 
215
            self.backing_transport = transport.get_transport(
216
216
                backing_server.get_url())
217
217
        else:
218
 
            self.backing_transport = transport.get_transport_from_path('.')
 
218
            self.backing_transport = transport.get_transport('.')
219
219
        self.backing_transport.clone('added-by-filter').ensure_base()
220
220
        self.filter_func = lambda x: 'added-by-filter/' + x
221
221
        super(TestingPathFilteringServer, self).start_server()
233
233
    def start_server(self, backing_server=None):
234
234
        """Setup the Chroot on backing_server."""
235
235
        if backing_server is not None:
236
 
            self.backing_transport = transport.get_transport_from_url(
 
236
            self.backing_transport = transport.get_transport(
237
237
                backing_server.get_url())
238
238
        else:
239
 
            self.backing_transport = transport.get_transport_from_path('.')
 
239
            self.backing_transport = transport.get_transport('.')
240
240
        super(TestingChrootServer, self).start_server()
241
241
 
242
242
    def get_bogus_url(self):
265
265
            #raise AssertionError('thread %s hung' % (self.name,))
266
266
 
267
267
 
268
 
class TestingTCPServerMixin(object):
 
268
class TestingTCPServerMixin:
269
269
    """Mixin to support running SocketServer.TCPServer in a thread.
270
270
 
271
271
    Tests are connecting from the main thread, the server has to be run in a
589
589
    def __init__(self, request, client_address, server):
590
590
        medium.SmartServerSocketStreamMedium.__init__(
591
591
            self, request, server.backing_transport,
592
 
            server.root_client_path,
593
 
            timeout=_DEFAULT_TESTING_CLIENT_TIMEOUT)
 
592
            server.root_client_path)
594
593
        request.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
595
594
        SocketServer.BaseRequestHandler.__init__(self, request, client_address,
596
595
                                                 server)
601
600
            self._serve_one_request(server_protocol)
602
601
 
603
602
 
604
 
_DEFAULT_TESTING_CLIENT_TIMEOUT = 4.0
605
 
 
606
603
class TestingSmartServer(TestingThreadingTCPServer, server.SmartTCPServer):
607
604
 
608
605
    def __init__(self, server_address, request_handler_class,
610
607
        TestingThreadingTCPServer.__init__(self, server_address,
611
608
                                           request_handler_class)
612
609
        server.SmartTCPServer.__init__(self, backing_transport,
613
 
            root_client_path, client_timeout=_DEFAULT_TESTING_CLIENT_TIMEOUT)
614
 
 
 
610
                                       root_client_path)
615
611
    def serve(self):
616
612
        self.run_server_started_hooks()
617
613
        try:
669
665
        self.chroot_server = ChrootServer(
670
666
            self.get_backing_transport(backing_transport_server))
671
667
        self.chroot_server.start_server()
672
 
        self.backing_transport = transport.get_transport_from_url(
 
668
        self.backing_transport = transport.get_transport(
673
669
            self.chroot_server.get_url())
674
670
        super(SmartTCPServer_for_testing, self).start_server()
675
671
 
681
677
 
682
678
    def get_backing_transport(self, backing_transport_server):
683
679
        """Get a backing transport from a server we are decorating."""
684
 
        return transport.get_transport_from_url(
685
 
            backing_transport_server.get_url())
 
680
        return transport.get_transport(backing_transport_server.get_url())
686
681
 
687
682
    def get_url(self):
688
683
        url = self.server.get_url()
699
694
    def get_backing_transport(self, backing_transport_server):
700
695
        """Get a backing transport from a server we are decorating."""
701
696
        url = 'readonly+' + backing_transport_server.get_url()
702
 
        return transport.get_transport_from_url(url)
 
697
        return transport.get_transport(url)
703
698
 
704
699
 
705
700
class SmartTCPServer_for_testing_v2_only(SmartTCPServer_for_testing):
720
715
    def get_backing_transport(self, backing_transport_server):
721
716
        """Get a backing transport from a server we are decorating."""
722
717
        url = 'readonly+' + backing_transport_server.get_url()
723
 
        return transport.get_transport_from_url(url)
 
718
        return transport.get_transport(url)