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()
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.
110
# This can be overriden or parametrized by daughter clasess if needed, but
111
# it must exist so that the create_transport_readonly_server() method can
109
113
_protocol_version = None
110
_url_protocol = 'http'
113
116
super(TestCaseWithWebserver, self).setUp()
114
117
self.transport_readonly_server = http_server.HttpServer
116
119
def create_transport_readonly_server(self):
117
server = self.transport_readonly_server(
120
return self.transport_readonly_server(
118
121
protocol_version=self._protocol_version)
119
server._url_protocol = self._url_protocol
123
124
class TestCaseWithTwoWebservers(TestCaseWithWebserver):
137
138
This is mostly a hook for daughter classes.
139
server = self.transport_secondary_server(
140
return self.transport_secondary_server(
140
141
protocol_version=self._protocol_version)
141
server._url_protocol = self._url_protocol
144
143
def get_secondary_server(self):
145
144
"""Get the server instance for the secondary transport."""
148
147
self.start_server(self.__secondary_server)
149
148
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
151
class ProxyServer(http_server.HttpServer):
162
152
"""A proxy test server for http transports."""
236
226
The 'old' server is redirected to the 'new' server.
240
super(TestCaseWithRedirectedWebserver, self).setUp()
241
# The redirections will point to the new server
242
self.new_server = self.get_readonly_server()
243
# The requests to the old server will be redirected to the new server
244
self.old_server = self.get_secondary_server()
246
229
def create_transport_secondary_server(self):
247
230
"""Create the secondary server redirecting to the primary server"""
248
231
new = self.get_readonly_server()
249
232
redirecting = HTTPServerRedirecting(
250
233
protocol_version=self._protocol_version)
251
234
redirecting.redirect_to(new.host, new.port)
252
redirecting._url_protocol = self._url_protocol
253
235
return redirecting
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())
238
super(TestCaseWithRedirectedWebserver, self).setUp()
239
# The redirections will point to the new server
240
self.new_server = self.get_readonly_server()
241
# The requests to the old server will be redirected
242
self.old_server = self.get_secondary_server()
274
245
class AuthRequestHandler(http_server.TestingHTTPRequestHandler):
284
255
# - auth_header_recv: the header received containing auth
285
256
# - 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
258
def do_GET(self):
301
259
if self.authorized():
302
260
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()
262
# Note that we must update test_case_server *before*
263
# sending the error or the client may try to read it
264
# before we have sent the whole error back.
265
tcs = self.server.test_case_server
266
tcs.auth_required_errors += 1
267
self.send_response(tcs.auth_error_code)
268
self.send_header_auth_reqed()
269
# We do not send a body
270
self.send_header('Content-Length', '0')
313
275
class BasicAuthRequestHandler(AuthRequestHandler):
added
removed
bzrlib/tests/http_utils.py
