/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: Martin Pool
  • Date: 2009-12-09 05:47:32 UTC
  • mfrom: (4879 +trunk)
  • mto: This revision was merged to the branch mainline in revision 4880.
  • Revision ID: mbp@sourcefrog.net-20091209054732-7414e9uma23mfv6x
trivial merge of trunk

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/blackbox/test_serve.py
30
30
    errors,
31
31
    osutils,
32
32
    revision as _mod_revision,
 
33
    urlutils,
33
34
    )
34
35
from bzrlib.branch import Branch
35
36
from bzrlib.bzrdir import BzrDir
44
45
from bzrlib.trace import mutter
45
46
from bzrlib.transport import get_transport, remote
46
47
 
47
 
 
48
 
class TestBzrServe(TestCaseWithTransport):
 
48
class TestBzrServeBase(TestCaseWithTransport):
 
49
 
 
50
    def run_bzr_serve_then_func(self, serve_args, retcode=0, func=None,
 
51
                                *func_args, **func_kwargs):
 
52
        """Run 'bzr serve', and run the given func in a thread once the server
 
53
        has started.
 
54
        
 
55
        When 'func' terminates, the server will be terminated too.
 
56
        
 
57
        Returns stdout and stderr.
 
58
        """
 
59
        # install hook
 
60
        def on_server_start(backing_urls, tcp_server):
 
61
            t = threading.Thread(
 
62
                target=on_server_start_thread, args=(tcp_server,))
 
63
            t.start()
 
64
        def on_server_start_thread(tcp_server):
 
65
            try:
 
66
                # Run func if set
 
67
                self.tcp_server = tcp_server
 
68
                if not func is None:
 
69
                    try:
 
70
                        func(*func_args, **func_kwargs)
 
71
                    except Exception, e:
 
72
                        # Log errors to make some test failures a little less
 
73
                        # mysterious.
 
74
                        mutter('func broke: %r', e)
 
75
            finally:
 
76
                # Then stop the server
 
77
                mutter('interrupting...')
 
78
                thread.interrupt_main()
 
79
        SmartTCPServer.hooks.install_named_hook(
 
80
            'server_started_ex', on_server_start,
 
81
            'run_bzr_serve_then_func hook')
 
82
        # start a TCP server
 
83
        try:
 
84
            out, err = self.run_bzr(['serve'] + list(serve_args))
 
85
        except KeyboardInterrupt, e:
 
86
            out, err = e.args
 
87
        return out, err
 
88
 
 
89
 
 
90
class TestBzrServe(TestBzrServeBase):
49
91
 
50
92
    def setUp(self):
51
93
        super(TestBzrServe, self).setUp()
118
160
        url = 'bzr://localhost:%d/' % port
119
161
        self.permit_url(url)
120
162
        return process, url
 
163
    
 
164
    def test_bzr_serve_quiet(self):
 
165
        self.make_branch('.')
 
166
        args = ['--port', 'localhost:0', '--quiet']
 
167
        out, err = self.run_bzr_serve_then_func(args, retcode=3)
 
168
        self.assertEqual('', out)
 
169
        self.assertEqual('', err)
121
170
 
122
171
    def test_bzr_serve_inet_readonly(self):
123
172
        """bzr server should provide a read only filesystem by default."""
167
216
        self.assertServerFinishesCleanly(process)
168
217
 
169
218
 
170
 
class TestCmdServeChrooting(TestCaseWithTransport):
 
219
class TestCmdServeChrooting(TestBzrServeBase):
171
220
 
172
221
    def test_serve_tcp(self):
173
222
        """'bzr serve' wraps the given --directory in a ChrootServer.
182
231
            ['--port', '127.0.0.1:0',
183
232
             '--directory', t.local_abspath('server-root'),
184
233
             '--allow-writes'],
185
 
            self.when_server_started)
 
234
            func=self.when_server_started)
186
235
        # The when_server_started method issued a find_repositoryV3 that should
187
236
        # fail with 'norepository' because there are no repositories inside the
188
237
        # --directory.
189
238
        self.assertEqual(('norepository',), self.client_resp)
190
239
 
191
 
    def run_bzr_serve_then_func(self, serve_args, func, *func_args,
192
 
            **func_kwargs):
193
 
        """Run 'bzr serve', and run the given func in a thread once the server
194
 
        has started.
195
 
        
196
 
        When 'func' terminates, the server will be terminated too.
197
 
        """
198
 
        # install hook
199
 
        def on_server_start(backing_urls, tcp_server):
200
 
            t = threading.Thread(
201
 
                target=on_server_start_thread, args=(tcp_server,))
202
 
            t.start()
203
 
        def on_server_start_thread(tcp_server):
204
 
            try:
205
 
                # Run func
206
 
                self.tcp_server = tcp_server
207
 
                try:
208
 
                    func(*func_args, **func_kwargs)
209
 
                except Exception, e:
210
 
                    # Log errors to make some test failures a little less
211
 
                    # mysterious.
212
 
                    mutter('func broke: %r', e)
213
 
            finally:
214
 
                # Then stop the server
215
 
                mutter('interrupting...')
216
 
                thread.interrupt_main()
217
 
        SmartTCPServer.hooks.install_named_hook(
218
 
            'server_started_ex', on_server_start,
219
 
            'run_bzr_serve_then_func hook')
220
 
        # start a TCP server
221
 
        try:
222
 
            self.run_bzr(['serve'] + list(serve_args))
223
 
        except KeyboardInterrupt:
224
 
            pass
225
 
 
226
240
    def when_server_started(self):
227
241
        # Connect to the TCP server and issue some requests and see what comes
228
242
        # back.
278
292
        (optionally decorated with 'readonly+').  BzrServerFactory can
279
293
        determine the original --directory from that transport.
280
294
        """
 
295
        # URLs always include the trailing slash, and get_base_path returns it
 
296
        base_dir = osutils.abspath('/a/b/c') + '/'
 
297
        base_url = urlutils.local_path_to_url(base_dir) + '/'
281
298
        # Define a fake 'protocol' to capture the transport that cmd_serve
282
299
        # passes to serve_bzr.
283
300
        def capture_transport(transport, host, port, inet):
284
301
            self.bzr_serve_transport = transport
285
302
        cmd = builtins.cmd_serve()
286
303
        # Read-only
287
 
        cmd.run(directory='/a/b/c', protocol=capture_transport)
 
304
        cmd.run(directory=base_dir, protocol=capture_transport)
288
305
        server_maker = BzrServerFactory()
289
306
        self.assertEqual(
290
 
            'readonly+file:///a/b/c/', self.bzr_serve_transport.base)
 
307
            'readonly+%s' % base_url, self.bzr_serve_transport.base)
291
308
        self.assertEqual(
292
 
            u'/a/b/c/', server_maker.get_base_path(self.bzr_serve_transport))
 
309
            base_dir, server_maker.get_base_path(self.bzr_serve_transport))
293
310
        # Read-write
294
 
        cmd.run(directory='/a/b/c', protocol=capture_transport,
 
311
        cmd.run(directory=base_dir, protocol=capture_transport,
295
312
            allow_writes=True)
296
313
        server_maker = BzrServerFactory()
297
 
        self.assertEqual('file:///a/b/c/', self.bzr_serve_transport.base)
298
 
        self.assertEqual(
299
 
            u'/a/b/c/', server_maker.get_base_path(self.bzr_serve_transport))
 
314
        self.assertEqual(base_url, self.bzr_serve_transport.base)
 
315
        self.assertEqual(base_dir,
 
316
            server_maker.get_base_path(self.bzr_serve_transport))
300
317