← 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: Andrew Bennetts
  • Date: 2009-12-03 05:57:41 UTC
  • mfrom: (4857 +trunk)
  • mto: This revision was merged to the branch mainline in revision 4869.
  • Revision ID: andrew.bennetts@canonical.com-20091203055741-vmmg0fmjgjw2pwvu
Merge lp:bzr.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/blackbox/test_serve.py
45
45
from bzrlib.trace import mutter
46
46
from bzrlib.transport import get_transport, remote
47
47
 
48
 
 
49
 
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):
50
91
 
51
92
    def setUp(self):
52
93
        super(TestBzrServe, self).setUp()
119
160
        url = 'bzr://localhost:%d/' % port
120
161
        self.permit_url(url)
121
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)
122
170
 
123
171
    def test_bzr_serve_inet_readonly(self):
124
172
        """bzr server should provide a read only filesystem by default."""
168
216
        self.assertServerFinishesCleanly(process)
169
217
 
170
218
 
171
 
class TestCmdServeChrooting(TestCaseWithTransport):
 
219
class TestCmdServeChrooting(TestBzrServeBase):
172
220
 
173
221
    def test_serve_tcp(self):
174
222
        """'bzr serve' wraps the given --directory in a ChrootServer.
183
231
            ['--port', '127.0.0.1:0',
184
232
             '--directory', t.local_abspath('server-root'),
185
233
             '--allow-writes'],
186
 
            self.when_server_started)
 
234
            func=self.when_server_started)
187
235
        # The when_server_started method issued a find_repositoryV3 that should
188
236
        # fail with 'norepository' because there are no repositories inside the
189
237
        # --directory.
190
238
        self.assertEqual(('norepository',), self.client_resp)
191
239
 
192
 
    def run_bzr_serve_then_func(self, serve_args, func, *func_args,
193
 
            **func_kwargs):
194
 
        """Run 'bzr serve', and run the given func in a thread once the server
195
 
        has started.
196
 
        
197
 
        When 'func' terminates, the server will be terminated too.
198
 
        """
199
 
        # install hook
200
 
        def on_server_start(backing_urls, tcp_server):
201
 
            t = threading.Thread(
202
 
                target=on_server_start_thread, args=(tcp_server,))
203
 
            t.start()
204
 
        def on_server_start_thread(tcp_server):
205
 
            try:
206
 
                # Run func
207
 
                self.tcp_server = tcp_server
208
 
                try:
209
 
                    func(*func_args, **func_kwargs)
210
 
                except Exception, e:
211
 
                    # Log errors to make some test failures a little less
212
 
                    # mysterious.
213
 
                    mutter('func broke: %r', e)
214
 
            finally:
215
 
                # Then stop the server
216
 
                mutter('interrupting...')
217
 
                thread.interrupt_main()
218
 
        SmartTCPServer.hooks.install_named_hook(
219
 
            'server_started_ex', on_server_start,
220
 
            'run_bzr_serve_then_func hook')
221
 
        # start a TCP server
222
 
        try:
223
 
            self.run_bzr(['serve'] + list(serve_args))
224
 
        except KeyboardInterrupt:
225
 
            pass
226
 
 
227
240
    def when_server_started(self):
228
241
        # Connect to the TCP server and issue some requests and see what comes
229
242
        # back.