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

  • Committer: Jelmer Vernooij
  • Date: 2017-05-21 12:41:27 UTC
  • mto: This revision was merged to the branch mainline in revision 6623.
  • Revision ID: jelmer@jelmer.uk-20170521124127-iv8etg0vwymyai6y
s/bzr/brz/ in apport config.

Show diffs side-by-side

added added

removed removed

Lines of Context:
brzlib/tests/test_smart_signals.py
20
20
import threading
21
21
import weakref
22
22
 
23
 
from breezy import tests, transport
24
 
from breezy.bzr.smart import client, medium, server, signals
 
23
from brzlib import tests, transport
 
24
from brzlib.smart import client, medium, server, signals
25
25
 
26
26
# Windows doesn't define SIGHUP. And while we could just skip a lot of these
27
27
# tests, we often don't actually care about interaction with 'signal', so we
41
41
        #       that we need it for signal handling.
42
42
        orig = signals._setup_on_hangup_dict()
43
43
        self.assertIs(None, orig)
44
 
 
45
44
        def cleanup():
46
45
            signals._on_sighup = None
47
46
        self.addCleanup(cleanup)
48
47
 
49
48
    def test_registered_callback_gets_called(self):
50
49
        calls = []
51
 
 
52
50
        def call_me():
53
51
            calls.append('called')
54
52
        signals.register_on_hangup('myid', call_me)
61
59
        # that shouldn't interrupt other flow.
62
60
        signals.unregister_on_hangup('no-such-id')
63
61
        log = self.get_log()
64
 
        self.assertContainsRe(
65
 
            log, 'Error occurred during unregister_on_hangup:')
 
62
        self.assertContainsRe(log, 'Error occurred during unregister_on_hangup:')
66
63
        self.assertContainsRe(log, '(?s)Traceback.*KeyError')
67
64
 
68
65
    def test_failing_callback(self):
69
66
        calls = []
70
 
 
71
67
        def call_me():
72
68
            calls.append('called')
73
 
 
74
69
        def fail_me():
75
70
            raise RuntimeError('something bad happened')
76
71
        signals.register_on_hangup('myid', call_me)
87
82
        # _sighup_handler should handle if some callbacks actually remove
88
83
        # themselves while running.
89
84
        calls = []
90
 
 
91
85
        def call_me_and_unregister():
92
86
            signals.unregister_on_hangup('myid')
93
87
            calls.append('called_and_unregistered')
94
 
 
95
88
        def call_me():
96
89
            calls.append('called')
97
90
        signals.register_on_hangup('myid', call_me_and_unregister)
117
110
        self.assertIsInstance(signals._on_sighup,
118
111
                              weakref.WeakValueDictionary)
119
112
        calls = []
120
 
 
121
113
        def call_me():
122
114
            calls.append('called')
123
115
        signals.register_on_hangup('myid', call_me)
127
119
        self.assertEqual([], calls)
128
120
 
129
121
    def test_not_installed(self):
130
 
        # If you haven't called breezy.bzr.smart.signals.install_sighup_handler,
 
122
        # If you haven't called brzlib.smart.signals.install_sighup_handler,
131
123
        # then _on_sighup should be None, and all the calls become no-ops.
132
124
        signals._on_sighup = None
133
125
        calls = []
134
 
 
135
126
        def call_me():
136
127
            calls.append('called')
137
128
        signals.register_on_hangup('myid', calls)
163
154
 
164
155
    def test_inet_server_responds_to_sighup(self):
165
156
        t = transport.get_transport('memory:///')
166
 
        content = b'a' * 1024 * 1024
 
157
        content = 'a'*1024*1024
167
158
        t.put_bytes('bigfile', content)
168
159
        factory = server.BzrServerFactory()
169
160
        # Override stdin/stdout so that we can inject our own handles
174
165
        self.addCleanup(factory.tear_down)
175
166
        started = threading.Event()
176
167
        stopped = threading.Event()
177
 
 
178
168
        def serving():
179
169
            started.set()
180
170
            factory.smart_server.serve()
183
173
        server_thread.start()
184
174
        started.wait()
185
175
        client_medium = medium.SmartSimplePipesClientMedium(client_read,
186
 
                                                            client_write, 'base')
 
176
                            client_write, 'base')
187
177
        client_client = client._SmartClient(client_medium)
188
 
        resp, response_handler = client_client.call_expecting_body(b'get',
189
 
                                                                   b'bigfile')
 
178
        resp, response_handler = client_client.call_expecting_body('get',
 
179
            'bigfile')
190
180
        signals._sighup_handler(SIGHUP, None)
191
181
        self.assertTrue(factory.smart_server.finished)
192
182
        # We can still finish reading the file content, but more than that, and
196
186
            self.fail('Got the wrong content back, expected 1M "a"')
197
187
        stopped.wait()
198
188
        server_thread.join()
 
189