← 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: Canonical.com Patch Queue Manager
  • Date: 2011-05-27 22:38:37 UTC
  • mfrom: (5920.1.1 785098-utextwrap-failure)
  • Revision ID: pqm@pqm.ubuntu.com-20110527223837-54h5nrj2y10wgj8o
(vila) Skip tests broken by sphinx monkeypatching (Vincent Ladeuil)

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), retcode=retcode)
94
85
        except KeyboardInterrupt, e:
95
86
            out, err = e.args
96
87
        return out, err
103
94
        self.disable_missing_extensions_warning()
104
95
 
105
96
    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
 
        """
 
97
        """test exception hook works to catch exceptions from server"""
111
98
        def hook(exception):
112
 
            if exception[0] is KeyboardInterrupt:
113
 
                sys.stderr.write('catching KeyboardInterrupt\n')
114
 
                return True
115
 
            else:
116
 
                return False
 
99
            from bzrlib.trace import note
 
100
            note("catching exception")
 
101
            return True
117
102
        SmartTCPServer.hooks.install_named_hook(
118
103
            'server_exception', hook,
119
104
            'test_server_except_hook hook')
120
 
        args = ['--port', 'localhost:0', '--quiet']
 
105
        args = []
121
106
        out, err = self.run_bzr_serve_then_func(args, retcode=0)
122
 
        self.assertEqual('catching KeyboardInterrupt\n', err)
 
107
        self.assertEqual('listening on port: 4155\ncatching exception\n', err)
123
108
 
124
109
    def test_server_exception_no_hook(self):
125
110
        """test exception without hook returns error"""