← 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: Martin
  • Date: 2011-05-21 16:29:38 UTC
  • mto: This revision was merged to the branch mainline in revision 5907.
  • Revision ID: gzlist@googlemail.com-20110521162938-1vrw3hp0197l3vrl
Add tests for non-ascii conflict serialisation

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
23
22
import thread
24
23
import threading
25
24
 
28
27
    errors,
29
28
    osutils,
30
29
    revision as _mod_revision,
31
 
    trace,
32
30
    transport,
33
31
    urlutils,
34
32
    )
43
41
    TestCaseWithMemoryTransport,
44
42
    TestCaseWithTransport,
45
43
    )
 
44
from bzrlib.trace import mutter
46
45
from bzrlib.transport import remote
47
46
 
48
47
 
57
56
 
58
57
        Returns stdout and stderr.
59
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()
60
64
        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
 
            """
66
65
            try:
67
66
                # Run func if set
68
67
                self.tcp_server = tcp_server
72
71
                    except Exception, e:
73
72
                        # Log errors to make some test failures a little less
74
73
                        # mysterious.
75
 
                        trace.mutter('func broke: %r', e)
 
74
                        mutter('func broke: %r', e)
76
75
            finally:
77
76
                # Then stop the server
78
 
                trace.mutter('interrupting...')
 
77
                mutter('interrupting...')
79
78
                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
87
79
        SmartTCPServer.hooks.install_named_hook(
88
80
            'server_started_ex', on_server_start,
89
81
            'run_bzr_serve_then_func hook')
90
82
        # start a TCP server
91
83
        try:
92
 
            out, err = self.run_bzr(['serve'] + list(serve_args),
93
 
                                    retcode=retcode)
 
84
            out, err = self.run_bzr(['serve'] + list(serve_args))
94
85
        except KeyboardInterrupt, e:
95
86
            out, err = e.args
96
87
        return out, err
102
93
        super(TestBzrServe, self).setUp()
103
94
        self.disable_missing_extensions_warning()
104
95
 
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
 
 
129
96
    def assertInetServerShutsdownCleanly(self, process):
130
97
        """Shutdown the server process looking for errors."""
131
98
        # Shutdown the server: the server should shut down when it cannot read
307
274
 
308
275
class TestUserdirExpansion(TestCaseWithMemoryTransport):
309
276
 
310
 
    @staticmethod
311
 
    def fake_expanduser(path):
 
277
    def fake_expanduser(self, path):
312
278
        """A simple, environment-independent, function for the duration of this
313
279
        test.
314
280