/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 breezy/tests/test_smart_signals.py

  • Committer: Marius Kruger
  • Date: 2010-07-10 21:28:56 UTC
  • mto: (5384.1.1 integration)
  • mto: This revision was merged to the branch mainline in revision 5385.
  • Revision ID: marius.kruger@enerweb.co.za-20100710212856-uq4ji3go0u5se7hx
* Update documentation
* add NEWS

Show diffs side-by-side

added added

removed removed

Lines of Context:
breezy/tests/test_smart_signals.py
1
 
# Copyright (C) 2011 Canonical Ltd
2
 
#
3
 
# This program is free software; you can redistribute it and/or modify
4
 
# it under the terms of the GNU General Public License as published by
5
 
# the Free Software Foundation; either version 2 of the License, or
6
 
# (at your option) any later version.
7
 
#
8
 
# This program is distributed in the hope that it will be useful,
9
 
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
 
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
11
 
# GNU General Public License for more details.
12
 
#
13
 
# You should have received a copy of the GNU General Public License
14
 
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
 
 
17
 
 
18
 
import os
19
 
import signal
20
 
import threading
21
 
import weakref
22
 
 
23
 
from breezy import tests, transport
24
 
from breezy.bzr.smart import client, medium, server, signals
25
 
 
26
 
# Windows doesn't define SIGHUP. And while we could just skip a lot of these
27
 
# tests, we often don't actually care about interaction with 'signal', so we
28
 
# can still run the tests for code coverage.
29
 
SIGHUP = getattr(signal, 'SIGHUP', 1)
30
 
 
31
 
 
32
 
class TestSignalHandlers(tests.TestCase):
33
 
 
34
 
    def setUp(self):
35
 
        super(TestSignalHandlers, self).setUp()
36
 
        # This allows us to mutate the signal handler callbacks, but leave it
37
 
        # 'pristine' after the test case.
38
 
        # TODO: Arguably, this could be put into the base test.TestCase, along
39
 
        #       with a tearDown that asserts that all the entries have been
40
 
        #       removed properly. Global state is always a bit messy. A shame
41
 
        #       that we need it for signal handling.
42
 
        orig = signals._setup_on_hangup_dict()
43
 
        self.assertIs(None, orig)
44
 
 
45
 
        def cleanup():
46
 
            signals._on_sighup = None
47
 
        self.addCleanup(cleanup)
48
 
 
49
 
    def test_registered_callback_gets_called(self):
50
 
        calls = []
51
 
 
52
 
        def call_me():
53
 
            calls.append('called')
54
 
        signals.register_on_hangup('myid', call_me)
55
 
        signals._sighup_handler(SIGHUP, None)
56
 
        self.assertEqual(['called'], calls)
57
 
        signals.unregister_on_hangup('myid')
58
 
 
59
 
    def test_unregister_not_present(self):
60
 
        # We don't want unregister to fail, since it is generally run at times
61
 
        # that shouldn't interrupt other flow.
62
 
        signals.unregister_on_hangup('no-such-id')
63
 
        log = self.get_log()
64
 
        self.assertContainsRe(
65
 
            log, 'Error occurred during unregister_on_hangup:')
66
 
        self.assertContainsRe(log, '(?s)Traceback.*KeyError')
67
 
 
68
 
    def test_failing_callback(self):
69
 
        calls = []
70
 
 
71
 
        def call_me():
72
 
            calls.append('called')
73
 
 
74
 
        def fail_me():
75
 
            raise RuntimeError('something bad happened')
76
 
        signals.register_on_hangup('myid', call_me)
77
 
        signals.register_on_hangup('otherid', fail_me)
78
 
        # _sighup_handler should call both, even though it got an exception
79
 
        signals._sighup_handler(SIGHUP, None)
80
 
        signals.unregister_on_hangup('myid')
81
 
        signals.unregister_on_hangup('otherid')
82
 
        log = self.get_log()
83
 
        self.assertContainsRe(log, '(?s)Traceback.*RuntimeError')
84
 
        self.assertEqual(['called'], calls)
85
 
 
86
 
    def test_unregister_during_call(self):
87
 
        # _sighup_handler should handle if some callbacks actually remove
88
 
        # themselves while running.
89
 
        calls = []
90
 
 
91
 
        def call_me_and_unregister():
92
 
            signals.unregister_on_hangup('myid')
93
 
            calls.append('called_and_unregistered')
94
 
 
95
 
        def call_me():
96
 
            calls.append('called')
97
 
        signals.register_on_hangup('myid', call_me_and_unregister)
98
 
        signals.register_on_hangup('other', call_me)
99
 
        signals._sighup_handler(SIGHUP, None)
100
 
 
101
 
    def test_keyboard_interrupt_propagated(self):
102
 
        # In case we get 'stuck' while running a hangup function, we should
103
 
        # not suppress KeyboardInterrupt
104
 
        def call_me_and_raise():
105
 
            raise KeyboardInterrupt()
106
 
        signals.register_on_hangup('myid', call_me_and_raise)
107
 
        self.assertRaises(KeyboardInterrupt,
108
 
                          signals._sighup_handler, SIGHUP, None)
109
 
        signals.unregister_on_hangup('myid')
110
 
 
111
 
    def test_weak_references(self):
112
 
        # TODO: This is probably a very-CPython-specific test
113
 
        # Adding yourself to the callback should not make you immortal
114
 
        # We overrideAttr during the test suite, so that we don't pollute the
115
 
        # original dict. However, we can test that what we override matches
116
 
        # what we are putting there.
117
 
        self.assertIsInstance(signals._on_sighup,
118
 
                              weakref.WeakValueDictionary)
119
 
        calls = []
120
 
 
121
 
        def call_me():
122
 
            calls.append('called')
123
 
        signals.register_on_hangup('myid', call_me)
124
 
        del call_me
125
 
        # Non-CPython might want to do a gc.collect() here
126
 
        signals._sighup_handler(SIGHUP, None)
127
 
        self.assertEqual([], calls)
128
 
 
129
 
    def test_not_installed(self):
130
 
        # If you haven't called breezy.bzr.smart.signals.install_sighup_handler,
131
 
        # then _on_sighup should be None, and all the calls become no-ops.
132
 
        signals._on_sighup = None
133
 
        calls = []
134
 
 
135
 
        def call_me():
136
 
            calls.append('called')
137
 
        signals.register_on_hangup('myid', calls)
138
 
        signals._sighup_handler(SIGHUP, None)
139
 
        signals.unregister_on_hangup('myid')
140
 
        log = self.get_log()
141
 
        self.assertEqual('', log)
142
 
 
143
 
    def test_install_sighup_handler(self):
144
 
        # install_sighup_handler should set up a signal handler for SIGHUP, as
145
 
        # well as the signals._on_sighup dict.
146
 
        signals._on_sighup = None
147
 
        orig = signals.install_sighup_handler()
148
 
        if getattr(signal, 'SIGHUP', None) is not None:
149
 
            cur = signal.getsignal(SIGHUP)
150
 
            self.assertEqual(signals._sighup_handler, cur)
151
 
        self.assertIsNot(None, signals._on_sighup)
152
 
        signals.restore_sighup_handler(orig)
153
 
        self.assertIs(None, signals._on_sighup)
154
 
 
155
 
 
156
 
class TestInetServer(tests.TestCase):
157
 
 
158
 
    def create_file_pipes(self):
159
 
        r, w = os.pipe()
160
 
        rf = os.fdopen(r, 'rb')
161
 
        wf = os.fdopen(w, 'wb')
162
 
        return rf, wf
163
 
 
164
 
    def test_inet_server_responds_to_sighup(self):
165
 
        t = transport.get_transport('memory:///')
166
 
        content = b'a' * 1024 * 1024
167
 
        t.put_bytes('bigfile', content)
168
 
        factory = server.BzrServerFactory()
169
 
        # Override stdin/stdout so that we can inject our own handles
170
 
        client_read, server_write = self.create_file_pipes()
171
 
        server_read, client_write = self.create_file_pipes()
172
 
        factory._get_stdin_stdout = lambda: (server_read, server_write)
173
 
        factory.set_up(t, None, None, inet=True, timeout=4.0)
174
 
        self.addCleanup(factory.tear_down)
175
 
        started = threading.Event()
176
 
        stopped = threading.Event()
177
 
 
178
 
        def serving():
179
 
            started.set()
180
 
            factory.smart_server.serve()
181
 
            stopped.set()
182
 
        server_thread = threading.Thread(target=serving)
183
 
        server_thread.start()
184
 
        started.wait()
185
 
        client_medium = medium.SmartSimplePipesClientMedium(client_read,
186
 
                                                            client_write, 'base')
187
 
        client_client = client._SmartClient(client_medium)
188
 
        resp, response_handler = client_client.call_expecting_body(b'get',
189
 
                                                                   b'bigfile')
190
 
        signals._sighup_handler(SIGHUP, None)
191
 
        self.assertTrue(factory.smart_server.finished)
192
 
        # We can still finish reading the file content, but more than that, and
193
 
        # the file is closed.
194
 
        v = response_handler.read_body_bytes()
195
 
        if v != content:
196
 
            self.fail('Got the wrong content back, expected 1M "a"')
197
 
        stopped.wait()
198
 
        server_thread.join()