/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: 2018-05-06 11:48:54 UTC
  • mto: This revision was merged to the branch mainline in revision 6960.
  • Revision ID: jelmer@jelmer.uk-20180506114854-h4qd9ojaqy8wxjsd
Move .mailmap to root.

Show diffs side-by-side

added added

removed removed

Lines of Context:
breezy/tests/test_smart_signals.py
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)
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