← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_wsgi.py

  • Committer: Martin Pool
  • Date: 2007-04-04 06:17:31 UTC
  • mto: This revision was merged to the branch mainline in revision 2397.
  • Revision ID: mbp@sourcefrog.net-20070404061731-tt2xrzllqhbodn83
Contents of TODO file moved into bug tracker

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_wsgi.py
12
12
#
13
13
# You should have received a copy of the GNU General Public License
14
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
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
16
 
17
17
"""Tests for WSGI application"""
18
18
 
19
19
from cStringIO import StringIO
20
20
 
21
21
from bzrlib import tests
22
 
from bzrlib.smart import protocol
23
22
from bzrlib.transport.http import wsgi
24
23
from bzrlib.transport import chroot, memory
25
24
 
33
32
 
34
33
    def build_environ(self, updates=None):
35
34
        """Builds an environ dict with all fields required by PEP 333.
36
 
 
 
35
        
37
36
        :param updates: a dict to that will be incorporated into the returned
38
37
            dict using dict.update(updates).
39
38
        """
58
57
        if updates is not None:
59
58
            environ.update(updates)
60
59
        return environ
61
 
 
 
60
        
62
61
    def read_response(self, iterable):
63
62
        response = ''
64
63
        for string in iterable:
82
81
        self.read_response(iterable)
83
82
        self.assertEqual('405 Method not allowed', self.status)
84
83
        self.assertTrue(('Allow', 'POST') in self.headers)
85
 
 
86
 
    def _fake_make_request(self, transport, write_func, bytes, rcp):
87
 
        request = FakeRequest(transport, write_func)
88
 
        request.accept_bytes(bytes)
89
 
        self.request = request
90
 
        return request
91
 
 
 
84
        
92
85
    def test_smart_wsgi_app_uses_given_relpath(self):
93
86
        # The SmartWSGIApp should use the "bzrlib.relpath" field from the
94
87
        # WSGI environ to clone from its backing transport to get a specific
96
89
        transport = FakeTransport()
97
90
        wsgi_app = wsgi.SmartWSGIApp(transport)
98
91
        wsgi_app.backing_transport = transport
99
 
        wsgi_app.make_request = self._fake_make_request
 
92
        def make_request(transport, write_func):
 
93
            request = FakeRequest(transport, write_func)
 
94
            self.request = request
 
95
            return request
 
96
        wsgi_app.make_request = make_request
100
97
        fake_input = StringIO('fake request')
101
98
        environ = self.build_environ({
102
99
            'REQUEST_METHOD': 'POST',
106
103
        })
107
104
        iterable = wsgi_app(environ, self.start_response)
108
105
        response = self.read_response(iterable)
109
 
        self.assertEqual([('clone', 'foo/bar/')] , transport.calls)
 
106
        self.assertEqual([('clone', 'foo/bar')] , transport.calls)
110
107
 
111
108
    def test_smart_wsgi_app_request_and_response(self):
112
109
        # SmartWSGIApp reads the smart request from the 'wsgi.input' file-like
115
112
        transport = memory.MemoryTransport()
116
113
        transport.put_bytes('foo', 'some bytes')
117
114
        wsgi_app = wsgi.SmartWSGIApp(transport)
118
 
        wsgi_app.make_request = self._fake_make_request
 
115
        def make_request(transport, write_func):
 
116
            request = FakeRequest(transport, write_func)
 
117
            self.request = request
 
118
            return request
 
119
        wsgi_app.make_request = make_request
119
120
        fake_input = StringIO('fake request')
120
121
        environ = self.build_environ({
121
122
            'REQUEST_METHOD': 'POST',
138
139
            fake_app, prefix='/abc/', path_var='FOO')
139
140
        wrapped_app({'FOO': '/abc/xyz/.bzr/smart'}, None)
140
141
        self.assertEqual(['xyz'], calls)
141
 
 
 
142
       
142
143
    def test_relpath_setter_bad_path_prefix(self):
143
144
        # wsgi.RelpathSetter will reject paths with that don't match the prefix
144
145
        # with a 404.  This is probably a sign of misconfiguration; a server
151
152
            {'FOO': 'AAA/abc/xyz/.bzr/smart'}, self.start_response)
152
153
        self.read_response(iterable)
153
154
        self.assertTrue(self.status.startswith('404'))
154
 
 
 
155
        
155
156
    def test_relpath_setter_bad_path_suffix(self):
156
157
        # Similar to test_relpath_setter_bad_path_prefix: wsgi.RelpathSetter
157
158
        # will reject paths with that don't match the suffix '.bzr/smart' with a
165
166
            {'FOO': '/abc/xyz/.bzr/AAA'}, self.start_response)
166
167
        self.read_response(iterable)
167
168
        self.assertTrue(self.status.startswith('404'))
168
 
 
 
169
        
169
170
    def test_make_app(self):
170
171
        # The make_app helper constructs a SmartWSGIApp wrapped in a
171
172
        # RelpathSetter.
179
180
        backing_transport = app.app.backing_transport
180
181
        chroot_backing_transport = backing_transport.server.backing_transport
181
182
        self.assertEndsWith(chroot_backing_transport.base, 'a%20root/')
182
 
        self.assertEqual(app.app.root_client_path, 'a prefix')
 
183
        self.assertEqual(app.prefix, 'a prefix')
183
184
        self.assertEqual(app.path_var, 'a path_var')
184
185
 
185
186
    def test_incomplete_request(self):
186
187
        transport = FakeTransport()
187
188
        wsgi_app = wsgi.SmartWSGIApp(transport)
188
 
        def make_request(transport, write_func, bytes, root_client_path):
 
189
        def make_request(transport, write_func):
189
190
            request = IncompleteRequest(transport, write_func)
190
 
            request.accept_bytes(bytes)
191
191
            self.request = request
192
192
            return request
193
193
        wsgi_app.make_request = make_request
204
204
        self.assertEqual('200 OK', self.status)
205
205
        self.assertEqual('error\x01incomplete request\n', response)
206
206
 
207
 
    def test_protocol_version_detection_one(self):
208
 
        # SmartWSGIApp detects requests that don't start with
209
 
        # REQUEST_VERSION_TWO as version one.
210
 
        transport = memory.MemoryTransport()
211
 
        wsgi_app = wsgi.SmartWSGIApp(transport)
212
 
        fake_input = StringIO('hello\n')
213
 
        environ = self.build_environ({
214
 
            'REQUEST_METHOD': 'POST',
215
 
            'CONTENT_LENGTH': len(fake_input.getvalue()),
216
 
            'wsgi.input': fake_input,
217
 
            'bzrlib.relpath': 'foo',
218
 
        })
219
 
        iterable = wsgi_app(environ, self.start_response)
220
 
        response = self.read_response(iterable)
221
 
        self.assertEqual('200 OK', self.status)
222
 
        # Expect a version 1-encoded response.
223
 
        self.assertEqual('ok\x012\n', response)
224
 
 
225
 
    def test_protocol_version_detection_two(self):
226
 
        # SmartWSGIApp detects requests that start with REQUEST_VERSION_TWO
227
 
        # as version two.
228
 
        transport = memory.MemoryTransport()
229
 
        wsgi_app = wsgi.SmartWSGIApp(transport)
230
 
        fake_input = StringIO(protocol.REQUEST_VERSION_TWO + 'hello\n')
231
 
        environ = self.build_environ({
232
 
            'REQUEST_METHOD': 'POST',
233
 
            'CONTENT_LENGTH': len(fake_input.getvalue()),
234
 
            'wsgi.input': fake_input,
235
 
            'bzrlib.relpath': 'foo',
236
 
        })
237
 
        iterable = wsgi_app(environ, self.start_response)
238
 
        response = self.read_response(iterable)
239
 
        self.assertEqual('200 OK', self.status)
240
 
        # Expect a version 2-encoded response.
241
 
        self.assertEqual(
242
 
            protocol.RESPONSE_VERSION_TWO + 'success\nok\x012\n', response)
243
 
 
244
207
 
245
208
class FakeRequest(object):
246
 
 
 
209
    
247
210
    def __init__(self, transport, write_func):
248
211
        self.transport = transport
249
212
        self.write_func = write_func