1
# Copyright (C) 2005-2011 Canonical Ltd
1
# Copyright (C) 2005-2010 Canonical Ltd
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
15
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17
17
from cStringIO import StringIO
22
27
from bzrlib import (
28
from bzrlib.smart import (
32
from bzrlib.smart import medium, protocol
31
33
from bzrlib.tests import http_server
32
from bzrlib.transport import chroot
34
from bzrlib.transport import (
35
40
class HTTPServerWithSmarts(http_server.HttpServer):
53
58
"""Hand the request off to a smart server instance."""
54
backing = transport.get_transport_from_path(
55
self.server.test_case_server._home_dir)
59
backing = get_transport(self.server.test_case_server._home_dir)
56
60
chroot_server = chroot.ChrootServer(backing)
57
61
chroot_server.start_server()
59
t = transport.get_transport_from_url(chroot_server.get_url())
63
t = get_transport(chroot_server.get_url())
60
64
self.do_POST_inner(t)
62
66
chroot_server.stop_server()
102
106
one. This will currently fail if the primary transport is not
103
107
backed by regular disk files.
106
# These attributes can be overriden or parametrized by daughter clasess if
107
# needed, but must exist so that the create_transport_readonly_server()
108
# method (or any method creating an http(s) server) can propagate it.
109
_protocol_version = None
110
_url_protocol = 'http'
113
110
super(TestCaseWithWebserver, self).setUp()
114
111
self.transport_readonly_server = http_server.HttpServer
116
def create_transport_readonly_server(self):
117
server = self.transport_readonly_server(
118
protocol_version=self._protocol_version)
119
server._url_protocol = self._url_protocol
123
114
class TestCaseWithTwoWebservers(TestCaseWithWebserver):
124
115
"""A support class providing readonly urls on two servers that are http://.
137
128
This is mostly a hook for daughter classes.
139
server = self.transport_secondary_server(
140
protocol_version=self._protocol_version)
141
server._url_protocol = self._url_protocol
130
return self.transport_secondary_server()
144
132
def get_secondary_server(self):
145
133
"""Get the server instance for the secondary transport."""
148
136
self.start_server(self.__secondary_server)
149
137
return self.__secondary_server
151
def get_secondary_url(self, relpath=None):
152
base = self.get_secondary_server().get_url()
153
return self._adjust_url(base, relpath)
155
def get_secondary_transport(self, relpath=None):
156
t = transport.get_transport_from_url(self.get_secondary_url(relpath))
157
self.assertTrue(t.is_readonly())
161
140
class ProxyServer(http_server.HttpServer):
162
141
"""A proxy test server for http transports."""
236
215
The 'old' server is redirected to the 'new' server.
218
def create_transport_secondary_server(self):
219
"""Create the secondary server redirecting to the primary server"""
220
new = self.get_readonly_server()
221
redirecting = HTTPServerRedirecting()
222
redirecting.redirect_to(new.host, new.port)
240
226
super(TestCaseWithRedirectedWebserver, self).setUp()
241
227
# The redirections will point to the new server
242
228
self.new_server = self.get_readonly_server()
243
# The requests to the old server will be redirected to the new server
229
# The requests to the old server will be redirected
244
230
self.old_server = self.get_secondary_server()
246
def create_transport_secondary_server(self):
247
"""Create the secondary server redirecting to the primary server"""
248
new = self.get_readonly_server()
249
redirecting = HTTPServerRedirecting(
250
protocol_version=self._protocol_version)
251
redirecting.redirect_to(new.host, new.port)
252
redirecting._url_protocol = self._url_protocol
255
def get_old_url(self, relpath=None):
256
base = self.old_server.get_url()
257
return self._adjust_url(base, relpath)
259
def get_old_transport(self, relpath=None):
260
t = transport.get_transport_from_url(self.get_old_url(relpath))
261
self.assertTrue(t.is_readonly())
264
def get_new_url(self, relpath=None):
265
base = self.new_server.get_url()
266
return self._adjust_url(base, relpath)
268
def get_new_transport(self, relpath=None):
269
t = transport.get_transport_from_url(self.get_new_url(relpath))
270
self.assertTrue(t.is_readonly())
274
233
class AuthRequestHandler(http_server.TestingHTTPRequestHandler):
275
234
"""Requires an authentication to process requests.
284
243
# - auth_header_recv: the header received containing auth
285
244
# - auth_error_code: the error code to indicate auth required
287
def _require_authentication(self):
288
# Note that we must update test_case_server *before*
289
# sending the error or the client may try to read it
290
# before we have sent the whole error back.
291
tcs = self.server.test_case_server
292
tcs.auth_required_errors += 1
293
self.send_response(tcs.auth_error_code)
294
self.send_header_auth_reqed()
295
# We do not send a body
296
self.send_header('Content-Length', '0')
300
246
def do_GET(self):
301
247
if self.authorized():
302
248
return http_server.TestingHTTPRequestHandler.do_GET(self)
304
return self._require_authentication()
307
if self.authorized():
308
return http_server.TestingHTTPRequestHandler.do_HEAD(self)
310
return self._require_authentication()
250
# Note that we must update test_case_server *before*
251
# sending the error or the client may try to read it
252
# before we have sent the whole error back.
253
tcs = self.server.test_case_server
254
tcs.auth_required_errors += 1
255
self.send_response(tcs.auth_error_code)
256
self.send_header_auth_reqed()
257
# We do not send a body
258
self.send_header('Content-Length', '0')
313
263
class BasicAuthRequestHandler(AuthRequestHandler):
added
removed
bzrlib/tests/http_utils.py
