← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_smart_transport.py

  • Committer: Andrew Bennetts
  • Date: 2010-01-13 23:16:20 UTC
  • mfrom: (4957 +trunk)
  • mto: This revision was merged to the branch mainline in revision 4960.
  • Revision ID: andrew.bennetts@canonical.com-20100113231620-n6in2yjib2v6z03g
Merge lp:bzr.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_smart_transport.py
987
987
    All of these tests are run with a server running on another thread serving
988
988
    a MemoryTransport, and a connection to it already open.
989
989
 
990
 
    the server is obtained by calling self.setUpServer(readonly=False).
 
990
    the server is obtained by calling self.start_server(readonly=False).
991
991
    """
992
992
 
993
 
    def setUpServer(self, readonly=False, backing_transport=None):
 
993
    def start_server(self, readonly=False, backing_transport=None):
994
994
        """Setup the server.
995
995
 
996
996
        :param readonly: Create a readonly server.
1000
1000
        # self.vfs_transport_factory etc.
1001
1001
        if not backing_transport:
1002
1002
            mem_server = memory.MemoryServer()
1003
 
            mem_server.setUp()
1004
 
            self.addCleanup(mem_server.tearDown)
 
1003
            mem_server.start_server()
 
1004
            self.addCleanup(mem_server.stop_server)
1005
1005
            self.permit_url(mem_server.get_url())
1006
1006
            self.backing_transport = get_transport(mem_server.get_url())
1007
1007
        else:
1021
1021
            del self.transport
1022
1022
        if getattr(self, 'server', None):
1023
1023
            self.server.stop_background_thread()
 
1024
            # XXX: why not .stop_server() -- mbp 20100106
1024
1025
            del self.server
1025
1026
 
1026
1027
 
1028
1029
 
1029
1030
    def test_server_setup_teardown(self):
1030
1031
        """It should be safe to teardown the server with no requests."""
1031
 
        self.setUpServer()
 
1032
        self.start_server()
1032
1033
        server = self.server
1033
1034
        transport = remote.RemoteTCPTransport(self.server.get_url())
1034
1035
        self.tearDownServer()
1036
1037
 
1037
1038
    def test_server_closes_listening_sock_on_shutdown_after_request(self):
1038
1039
        """The server should close its listening socket when it's stopped."""
1039
 
        self.setUpServer()
 
1040
        self.start_server()
1040
1041
        server = self.server
1041
1042
        self.transport.has('.')
1042
1043
        self.tearDownServer()
1051
1052
 
1052
1053
    def setUp(self):
1053
1054
        super(WritableEndToEndTests, self).setUp()
1054
 
        self.setUpServer()
 
1055
        self.start_server()
1055
1056
 
1056
1057
    def test_start_tcp_server(self):
1057
1058
        url = self.server.get_url()
1130
1131
    def test_mkdir_error_readonly(self):
1131
1132
        """TransportNotPossible should be preserved from the backing transport."""
1132
1133
        self._captureVar('BZR_NO_SMART_VFS', None)
1133
 
        self.setUpServer(readonly=True)
 
1134
        self.start_server(readonly=True)
1134
1135
        self.assertRaises(errors.TransportNotPossible, self.transport.mkdir,
1135
1136
            'foo')
1136
1137
 
1146
1147
        self.hook_calls = []
1147
1148
        server.SmartTCPServer.hooks.install_named_hook('server_started',
1148
1149
            self.capture_server_call, None)
1149
 
        self.setUpServer()
 
1150
        self.start_server()
1150
1151
        # at this point, the server will be starting a thread up.
1151
1152
        # there is no indicator at the moment, so bodge it by doing a request.
1152
1153
        self.transport.has('.')
1160
1161
        self.hook_calls = []
1161
1162
        server.SmartTCPServer.hooks.install_named_hook('server_started',
1162
1163
            self.capture_server_call, None)
1163
 
        self.setUpServer(backing_transport=get_transport("."))
 
1164
        self.start_server(backing_transport=get_transport("."))
1164
1165
        # at this point, the server will be starting a thread up.
1165
1166
        # there is no indicator at the moment, so bodge it by doing a request.
1166
1167
        self.transport.has('.')
1176
1177
        self.hook_calls = []
1177
1178
        server.SmartTCPServer.hooks.install_named_hook('server_stopped',
1178
1179
            self.capture_server_call, None)
1179
 
        self.setUpServer()
 
1180
        self.start_server()
1180
1181
        result = [([self.backing_transport.base], self.transport.base)]
1181
1182
        # check the stopping message isn't emitted up front.
1182
1183
        self.assertEqual([], self.hook_calls)
1193
1194
        self.hook_calls = []
1194
1195
        server.SmartTCPServer.hooks.install_named_hook('server_stopped',
1195
1196
            self.capture_server_call, None)
1196
 
        self.setUpServer(backing_transport=get_transport("."))
 
1197
        self.start_server(backing_transport=get_transport("."))
1197
1198
        result = [(
1198
1199
            [self.backing_transport.base, self.backing_transport.external_url()]
1199
1200
            , self.transport.base)]