← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/blackbox/test_serve.py

  • Committer: Vincent Ladeuil
  • Date: 2009-05-05 15:31:34 UTC
  • mto: (4343.1.1 integration)
  • mto: This revision was merged to the branch mainline in revision 4344.
  • Revision ID: v.ladeuil+lp@free.fr-20090505153134-q4bp4is9gywsmzrv
Clean up test for log formats.

* bzrlib/tests/blackbox/test_logformats.py:
Update tests to actual style.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/blackbox/test_serve.py
21
21
import signal
22
22
import subprocess
23
23
import sys
24
 
import thread
25
24
import threading
26
25
 
27
26
from bzrlib import (
28
27
    errors,
29
28
    osutils,
30
29
    revision as _mod_revision,
31
 
    transport,
32
30
    )
33
31
from bzrlib.branch import Branch
34
32
from bzrlib.bzrdir import BzrDir
35
33
from bzrlib.errors import ParamikoNotPresent
36
 
from bzrlib.smart import client, medium
37
 
from bzrlib.smart.server import SmartTCPServer
 
34
from bzrlib.smart import medium
38
35
from bzrlib.tests import TestCaseWithTransport, TestSkipped
39
 
from bzrlib.trace import mutter
40
36
from bzrlib.transport import get_transport, remote
41
37
 
42
38
 
142
138
        self.make_read_requests(branch)
143
139
        self.assertServerFinishesCleanly(process)
144
140
 
145
 
    def test_bzr_serve_supports_protocol(self):
146
 
        # Make a branch
147
 
        self.make_branch('.')
148
 
 
149
 
        process, url = self.start_server_port(['--allow-writes',
150
 
                                               '--protocol=bzr'])
151
 
 
152
 
        # Connect to the server
153
 
        branch = Branch.open(url)
154
 
        self.make_read_requests(branch)
155
 
        self.assertServerFinishesCleanly(process)
156
 
 
157
141
    def test_bzr_connect_to_bzr_ssh(self):
158
142
        """User acceptance that get_transport of a bzr+ssh:// behaves correctly.
159
143
 
243
227
             % bzr_remote_path],
244
228
            self.command_executed)
245
229
 
246
 
 
247
 
class TestCmdServeChrooting(TestCaseWithTransport):
248
 
 
249
 
    def test_serve_tcp(self):
250
 
        """'bzr serve' wraps the given --directory in a ChrootServer.
251
 
 
252
 
        So requests that search up through the parent directories (like
253
 
        find_repositoryV3) will give "not found" responses, rather than
254
 
        InvalidURLJoin or jail break errors.
255
 
        """
256
 
        t = self.get_transport()
257
 
        t.mkdir('server-root')
258
 
        self.run_bzr_serve_then_func(
259
 
            ['--port', '0', '--directory', t.local_abspath('server-root'),
260
 
             '--allow-writes'],
261
 
            self.when_server_started)
262
 
        # The when_server_started method issued a find_repositoryV3 that should
263
 
        # fail with 'norepository' because there are no repositories inside the
264
 
        # --directory.
265
 
        self.assertEqual(('norepository',), self.client_resp)
266
 
        
267
 
    def run_bzr_serve_then_func(self, serve_args, func, *func_args,
268
 
            **func_kwargs):
269
 
        """Run 'bzr serve', and run the given func in a thread once the server
270
 
        has started.
271
 
        
272
 
        When 'func' terminates, the server will be terminated too.
273
 
        """
274
 
        # install hook
275
 
        def on_server_start(backing_urls, tcp_server):
276
 
            t = threading.Thread(
277
 
                target=on_server_start_thread, args=(tcp_server,))
278
 
            t.start()
279
 
        def on_server_start_thread(tcp_server):
280
 
            try:
281
 
                # Run func
282
 
                self.tcp_server = tcp_server
283
 
                try:
284
 
                    func(*func_args, **func_kwargs)
285
 
                except Exception, e:
286
 
                    # Log errors to make some test failures a little less
287
 
                    # mysterious.
288
 
                    mutter('func broke: %r', e)
289
 
            finally:
290
 
                # Then stop the server
291
 
                mutter('interrupting...')
292
 
                thread.interrupt_main()
293
 
        SmartTCPServer.hooks.install_named_hook(
294
 
            'server_started_ex', on_server_start,
295
 
            'run_bzr_serve_then_func hook')
296
 
        # start a TCP server
297
 
        try:
298
 
            self.run_bzr(['serve'] + list(serve_args))
299
 
        except KeyboardInterrupt:
300
 
            pass
301
 
 
302
 
    def when_server_started(self):
303
 
        # Connect to the TCP server and issue some requests and see what comes
304
 
        # back.
305
 
        client_medium = medium.SmartTCPClientMedium(
306
 
            '127.0.0.1', self.tcp_server.port,
307
 
            'bzr://localhost:%d/' % (self.tcp_server.port,))
308
 
        smart_client = client._SmartClient(client_medium)
309
 
        resp = smart_client.call('mkdir', 'foo', '')
310
 
        resp = smart_client.call('BzrDirFormat.initialize', 'foo/')
311
 
        try:
312
 
            resp = smart_client.call('BzrDir.find_repositoryV3', 'foo/')
313
 
        except errors.ErrorFromSmartServer, e:
314
 
            resp = e.error_tuple
315
 
        self.client_resp = resp
316
 
        client_medium.disconnect()
317
 
 
318
 
 
319