← 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

Abbreviate pack_stat struct format to '>6L'

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/blackbox/test_serve.py
19
19
 
20
20
import os
21
21
import signal
 
22
import sys
22
23
import thread
23
24
import threading
24
25
 
27
28
    errors,
28
29
    osutils,
29
30
    revision as _mod_revision,
 
31
    trace,
30
32
    transport,
31
33
    urlutils,
32
34
    )
41
43
    TestCaseWithMemoryTransport,
42
44
    TestCaseWithTransport,
43
45
    )
44
 
from bzrlib.trace import mutter
45
46
from bzrlib.transport import remote
46
47
 
47
48
 
56
57
 
57
58
        Returns stdout and stderr.
58
59
        """
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
60
        def on_server_start_thread(tcp_server):
 
61
            """This runs concurrently with the server thread.
 
62
 
 
63
            The server is interrupted as soon as ``func`` finishes, even if an
 
64
            exception is encountered.
 
65
            """
65
66
            try:
66
67
                # Run func if set
67
68
                self.tcp_server = tcp_server
71
72
                    except Exception, e:
72
73
                        # Log errors to make some test failures a little less
73
74
                        # mysterious.
74
 
                        mutter('func broke: %r', e)
 
75
                        trace.mutter('func broke: %r', e)
75
76
            finally:
76
77
                # Then stop the server
77
 
                mutter('interrupting...')
 
78
                trace.mutter('interrupting...')
78
79
                thread.interrupt_main()
 
80
        # When the hook is fired, it just starts ``on_server_start_thread`` and
 
81
        # return
 
82
        def on_server_start(backing_urls, tcp_server):
 
83
            t = threading.Thread(
 
84
                target=on_server_start_thread, args=(tcp_server,))
 
85
            t.start()
 
86
        # install hook
79
87
        SmartTCPServer.hooks.install_named_hook(
80
88
            'server_started_ex', on_server_start,
81
89
            'run_bzr_serve_then_func hook')
82
90
        # start a TCP server
83
91
        try:
84
 
            out, err = self.run_bzr(['serve'] + list(serve_args))
 
92
            out, err = self.run_bzr(['serve'] + list(serve_args),
 
93
                                    retcode=retcode)
85
94
        except KeyboardInterrupt, e:
86
95
            out, err = e.args
87
96
        return out, err
93
102
        super(TestBzrServe, self).setUp()
94
103
        self.disable_missing_extensions_warning()
95
104
 
 
105
    def test_server_exception_with_hook(self):
 
106
        """Catch exception from the server in the server_exception hook.
 
107
 
 
108
        We use ``run_bzr_serve_then_func`` without a ``func`` so the server
 
109
        will receive a KeyboardInterrupt exception we want to catch.
 
110
        """
 
111
        def hook(exception):
 
112
            if exception[0] is KeyboardInterrupt:
 
113
                sys.stderr.write('catching KeyboardInterrupt\n')
 
114
                return True
 
115
            else:
 
116
                return False
 
117
        SmartTCPServer.hooks.install_named_hook(
 
118
            'server_exception', hook,
 
119
            'test_server_except_hook hook')
 
120
        args = ['--port', 'localhost:0', '--quiet']
 
121
        out, err = self.run_bzr_serve_then_func(args, retcode=0)
 
122
        self.assertEqual('catching KeyboardInterrupt\n', err)
 
123
 
 
124
    def test_server_exception_no_hook(self):
 
125
        """test exception without hook returns error"""
 
126
        args = []
 
127
        out, err = self.run_bzr_serve_then_func(args, retcode=3)
 
128
 
96
129
    def assertInetServerShutsdownCleanly(self, process):
97
130
        """Shutdown the server process looking for errors."""
98
131
        # Shutdown the server: the server should shut down when it cannot read
191
224
    def test_bzr_serve_port_readonly(self):
192
225
        """bzr server should provide a read only filesystem by default."""
193
226
        process, url = self.start_server_port()
194
 
        t = transport.get_transport(url)
 
227
        t = transport.get_transport_from_url(url)
195
228
        self.assertRaises(errors.TransportNotPossible, t.mkdir, 'adir')
196
229
        self.assertServerFinishesCleanly(process)
197
230
 
274
307
 
275
308
class TestUserdirExpansion(TestCaseWithMemoryTransport):
276
309
 
277
 
    def fake_expanduser(self, path):
 
310
    @staticmethod
 
311
    def fake_expanduser(path):
278
312
        """A simple, environment-independent, function for the duration of this
279
313
        test.
280
314