/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: Breezy landing bot
  • Author(s): Jelmer Vernooij
  • Date: 2018-11-16 18:59:44 UTC
  • mfrom: (7143.15.15 more-cleanups)
  • Revision ID: breezy.the.bot@gmail.com-20181116185944-biefv1sub37qfybm
Sprinkle some PEP8iness.

Merged from https://code.launchpad.net/~jelmer/brz/more-cleanups/+merge/358611

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