← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_smart_request.py

  • Committer: Martin Pool
  • Date: 2005-09-19 08:02:41 UTC
  • Revision ID: mbp@sourcefrog.net-20050919080241-d9460223db8f7d90
- rename to Revision.parent_ids to avoid confusion with old usage
  where Revision.parents held objects rather than strings\t

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_smart_request.py
1
 
# Copyright (C) 2009, 2010 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
 
"""Tests for smart server request infrastructure (bzrlib.smart.request)."""
18
 
 
19
 
import threading
20
 
 
21
 
from bzrlib import (
22
 
    errors,
23
 
    transport,
24
 
    )
25
 
from bzrlib.bzrdir import BzrDir
26
 
from bzrlib.smart import request
27
 
from bzrlib.tests import TestCase, TestCaseWithMemoryTransport
28
 
 
29
 
 
30
 
class NoBodyRequest(request.SmartServerRequest):
31
 
    """A request that does not implement do_body."""
32
 
 
33
 
    def do(self):
34
 
        return request.SuccessfulSmartServerResponse(('ok',))
35
 
 
36
 
 
37
 
class DoErrorRequest(request.SmartServerRequest):
38
 
    """A request that raises an error from self.do()."""
39
 
 
40
 
    def do(self):
41
 
        raise errors.NoSuchFile('xyzzy')
42
 
 
43
 
 
44
 
class ChunkErrorRequest(request.SmartServerRequest):
45
 
    """A request that raises an error from self.do_chunk()."""
46
 
    
47
 
    def do(self):
48
 
        """No-op."""
49
 
        pass
50
 
 
51
 
    def do_chunk(self, bytes):
52
 
        raise errors.NoSuchFile('xyzzy')
53
 
 
54
 
 
55
 
class EndErrorRequest(request.SmartServerRequest):
56
 
    """A request that raises an error from self.do_end()."""
57
 
    
58
 
    def do(self):
59
 
        """No-op."""
60
 
        pass
61
 
 
62
 
    def do_chunk(self, bytes):
63
 
        """No-op."""
64
 
        pass
65
 
        
66
 
    def do_end(self):
67
 
        raise errors.NoSuchFile('xyzzy')
68
 
 
69
 
 
70
 
class CheckJailRequest(request.SmartServerRequest):
71
 
 
72
 
    def __init__(self, *args):
73
 
        request.SmartServerRequest.__init__(self, *args)
74
 
        self.jail_transports_log = []
75
 
 
76
 
    def do(self):
77
 
        self.jail_transports_log.append(request.jail_info.transports)
78
 
 
79
 
    def do_chunk(self, bytes):
80
 
        self.jail_transports_log.append(request.jail_info.transports)
81
 
 
82
 
    def do_end(self):
83
 
        self.jail_transports_log.append(request.jail_info.transports)
84
 
 
85
 
 
86
 
class TestSmartRequest(TestCase):
87
 
 
88
 
    def test_request_class_without_do_body(self):
89
 
        """If a request has no body data, and the request's implementation does
90
 
        not override do_body, then no exception is raised.
91
 
        """
92
 
        # Create a SmartServerRequestHandler with a SmartServerRequest subclass
93
 
        # that does not implement do_body.
94
 
        handler = request.SmartServerRequestHandler(
95
 
            None, {'foo': NoBodyRequest}, '/')
96
 
        # Emulate a request with no body (i.e. just args).
97
 
        handler.args_received(('foo',))
98
 
        handler.end_received()
99
 
        # Request done, no exception was raised.
100
 
 
101
 
    def test_only_request_code_is_jailed(self):
102
 
        transport = 'dummy transport'
103
 
        handler = request.SmartServerRequestHandler(
104
 
            transport, {'foo': CheckJailRequest}, '/')
105
 
        handler.args_received(('foo',))
106
 
        self.assertEqual(None, request.jail_info.transports)
107
 
        handler.accept_body('bytes')
108
 
        self.assertEqual(None, request.jail_info.transports)
109
 
        handler.end_received()
110
 
        self.assertEqual(None, request.jail_info.transports)
111
 
        self.assertEqual(
112
 
            [[transport]] * 3, handler._command.jail_transports_log)
113
 
 
114
 
 
115
 
 
116
 
class TestSmartRequestHandlerErrorTranslation(TestCase):
117
 
    """Tests that SmartServerRequestHandler will translate exceptions raised by
118
 
    a SmartServerRequest into FailedSmartServerResponses.
119
 
    """
120
 
 
121
 
    def assertNoResponse(self, handler):
122
 
        self.assertEqual(None, handler.response)
123
 
 
124
 
    def assertResponseIsTranslatedError(self, handler):
125
 
        expected_translation = ('NoSuchFile', 'xyzzy')
126
 
        self.assertEqual(
127
 
            request.FailedSmartServerResponse(expected_translation),
128
 
            handler.response)
129
 
 
130
 
    def test_error_translation_from_args_received(self):
131
 
        handler = request.SmartServerRequestHandler(
132
 
            None, {'foo': DoErrorRequest}, '/')
133
 
        handler.args_received(('foo',))
134
 
        self.assertResponseIsTranslatedError(handler)
135
 
 
136
 
    def test_error_translation_from_chunk_received(self):
137
 
        handler = request.SmartServerRequestHandler(
138
 
            None, {'foo': ChunkErrorRequest}, '/')
139
 
        handler.args_received(('foo',))
140
 
        self.assertNoResponse(handler)
141
 
        handler.accept_body('bytes')
142
 
        self.assertResponseIsTranslatedError(handler)
143
 
 
144
 
    def test_error_translation_from_end_received(self):
145
 
        handler = request.SmartServerRequestHandler(
146
 
            None, {'foo': EndErrorRequest}, '/')
147
 
        handler.args_received(('foo',))
148
 
        self.assertNoResponse(handler)
149
 
        handler.end_received()
150
 
        self.assertResponseIsTranslatedError(handler)
151
 
 
152
 
 
153
 
class TestRequestHanderErrorTranslation(TestCase):
154
 
    """Tests for bzrlib.smart.request._translate_error."""
155
 
 
156
 
    def assertTranslationEqual(self, expected_tuple, error):
157
 
        self.assertEqual(expected_tuple, request._translate_error(error))
158
 
 
159
 
    def test_NoSuchFile(self):
160
 
        self.assertTranslationEqual(
161
 
            ('NoSuchFile', 'path'), errors.NoSuchFile('path'))
162
 
 
163
 
    def test_LockContention(self):
164
 
        # For now, LockContentions are always transmitted with no details.
165
 
        # Eventually they should include a relpath or url or something else to
166
 
        # identify which lock is busy.
167
 
        self.assertTranslationEqual(
168
 
            ('LockContention',), errors.LockContention('lock', 'msg'))
169
 
 
170
 
    def test_TokenMismatch(self):
171
 
        self.assertTranslationEqual(
172
 
            ('TokenMismatch', 'some-token', 'actual-token'),
173
 
            errors.TokenMismatch('some-token', 'actual-token'))
174
 
 
175
 
 
176
 
class TestRequestJail(TestCaseWithMemoryTransport):
177
 
 
178
 
    def test_jail(self):
179
 
        transport = self.get_transport('blah')
180
 
        req = request.SmartServerRequest(transport)
181
 
        self.assertEqual(None, request.jail_info.transports)
182
 
        req.setup_jail()
183
 
        self.assertEqual([transport], request.jail_info.transports)
184
 
        req.teardown_jail()
185
 
        self.assertEqual(None, request.jail_info.transports)
186
 
 
187
 
 
188
 
class TestJailHook(TestCaseWithMemoryTransport):
189
 
 
190
 
    def setUp(self):
191
 
        super(TestJailHook, self).setUp()
192
 
        def clear_jail_info():
193
 
            request.jail_info.transports = None
194
 
        self.addCleanup(clear_jail_info)
195
 
 
196
 
    def test_jail_hook(self):
197
 
        request.jail_info.transports = None
198
 
        _pre_open_hook = request._pre_open_hook
199
 
        # Any transport is fine if jail_info.transports is None
200
 
        t = self.get_transport('foo')
201
 
        _pre_open_hook(t)
202
 
        # A transport in jail_info.transports is allowed
203
 
        request.jail_info.transports = [t]
204
 
        _pre_open_hook(t)
205
 
        # A child of a transport in jail_info is allowed
206
 
        _pre_open_hook(t.clone('child'))
207
 
        # A parent is not allowed
208
 
        self.assertRaises(errors.JailBreak, _pre_open_hook, t.clone('..'))
209
 
        # A completely unrelated transport is not allowed
210
 
        self.assertRaises(errors.JailBreak, _pre_open_hook,
211
 
                          transport.get_transport('http://host/'))
212
 
 
213
 
    def test_open_bzrdir_in_non_main_thread(self):
214
 
        """Opening a bzrdir in a non-main thread should work ok.
215
 
        
216
 
        This makes sure that the globally-installed
217
 
        bzrlib.smart.request._pre_open_hook, which uses a threading.local(),
218
 
        works in a newly created thread.
219
 
        """
220
 
        bzrdir = self.make_bzrdir('.')
221
 
        transport = bzrdir.root_transport
222
 
        thread_result = []
223
 
        def t():
224
 
            BzrDir.open_from_transport(transport)
225
 
            thread_result.append('ok')
226
 
        thread = threading.Thread(target=t)
227
 
        thread.start()
228
 
        thread.join()
229
 
        self.assertEqual(['ok'], thread_result)
230