/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: Jelmer Vernooij
  • Date: 2019-10-13 22:53:02 UTC
  • mfrom: (7290.1.35 work)
  • mto: This revision was merged to the branch mainline in revision 7405.
  • Revision ID: jelmer@jelmer.uk-20191013225302-vg88ztajzq05hkas
Merge lp:brz/3.0.

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()