/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/tests/blackbox/test_serve.py

  • Committer: Andrew Bennetts
  • Date: 2009-07-20 00:52:07 UTC
  • mto: This revision was merged to the branch mainline in revision 4550.
  • Revision ID: andrew.bennetts@canonical.com-20090720005207-pttp2b3lcuh38vlz
Add test that would catch the lack of ChrootServer in cmd_serve.

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
24
25
import threading
25
26
 
26
27
from bzrlib import (
27
28
    errors,
28
29
    osutils,
29
30
    revision as _mod_revision,
 
31
    transport,
30
32
    )
31
33
from bzrlib.branch import Branch
32
34
from bzrlib.bzrdir import BzrDir
33
35
from bzrlib.errors import ParamikoNotPresent
34
 
from bzrlib.smart import medium
 
36
from bzrlib.smart import client, medium
 
37
from bzrlib.smart.server import SmartTCPServer
35
38
from bzrlib.tests import TestCaseWithTransport, TestSkipped
 
39
from bzrlib.trace import mutter
36
40
from bzrlib.transport import get_transport, remote
37
41
 
38
42
 
239
243
             % bzr_remote_path],
240
244
            self.command_executed)
241
245
 
 
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(tcp_server):
 
276
            t = threading.Thread(target=on_server_start_thread, args=(tcp_server,))
 
277
            t.start()
 
278
        def on_server_start_thread(tcp_server):
 
279
            try:
 
280
                # Run func
 
281
                self.tcp_server = tcp_server
 
282
                try:
 
283
                    func(*func_args, **func_kwargs)
 
284
                except Exception, e:
 
285
                    # Log errors to make some test failures a little less
 
286
                    # mysterious.
 
287
                    mutter('func broke: %r', e)
 
288
            finally:
 
289
                # Then stop the server
 
290
                mutter('interrupting...')
 
291
                thread.interrupt_main()
 
292
        SmartTCPServer.hooks.install_named_hook(
 
293
            'server_started_ex', on_server_start,
 
294
            'run_bzr_serve_then_func hook')
 
295
        # start a TCP server
 
296
        try:
 
297
            self.run_bzr(['serve'] + list(serve_args))
 
298
        except KeyboardInterrupt:
 
299
            pass
 
300
 
 
301
    def when_server_started(self):
 
302
        # Connect to the TCP server and issue some requests and see what comes
 
303
        # back.
 
304
        client_medium = medium.SmartTCPClientMedium(
 
305
            '127.0.0.1', self.tcp_server.port,
 
306
            'bzr://localhost:%d/' % (self.tcp_server.port,))
 
307
        smart_client = client._SmartClient(client_medium)
 
308
        resp = smart_client.call('mkdir', 'foo', '')
 
309
        resp = smart_client.call('BzrDirFormat.initialize', 'foo/')
 
310
        try:
 
311
            resp = smart_client.call('BzrDir.find_repositoryV3', 'foo/')
 
312
        except errors.ErrorFromSmartServer, e:
 
313
            resp = e.error_tuple
 
314
        self.client_resp = resp
 
315
        client_medium.disconnect()
 
316
 
 
317
 
 
318