1
# Copyright (C) 2005-2010 Canonical Ltd
1
# Copyright (C) 2005-2011 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
27
22
from bzrlib import (
32
from bzrlib.smart import medium, protocol
28
from bzrlib.smart import (
33
31
from bzrlib.tests import http_server
34
from bzrlib.transport import (
32
from bzrlib.transport import chroot
40
35
class HTTPServerWithSmarts(http_server.HttpServer):
58
53
"""Hand the request off to a smart server instance."""
59
backing = get_transport(self.server.test_case_server._home_dir)
54
backing = transport.get_transport_from_path(
55
self.server.test_case_server._home_dir)
60
56
chroot_server = chroot.ChrootServer(backing)
61
57
chroot_server.start_server()
63
t = get_transport(chroot_server.get_url())
59
t = transport.get_transport_from_url(chroot_server.get_url())
64
60
self.do_POST_inner(t)
66
62
chroot_server.stop_server()
106
102
one. This will currently fail if the primary transport is not
107
103
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'
110
113
super(TestCaseWithWebserver, self).setUp()
111
114
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
114
123
class TestCaseWithTwoWebservers(TestCaseWithWebserver):
115
124
"""A support class providing readonly urls on two servers that are http://.
128
137
This is mostly a hook for daughter classes.
130
return self.transport_secondary_server()
139
server = self.transport_secondary_server(
140
protocol_version=self._protocol_version)
141
server._url_protocol = self._url_protocol
132
144
def get_secondary_server(self):
133
145
"""Get the server instance for the secondary transport."""
136
148
self.start_server(self.__secondary_server)
137
149
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())
140
161
class ProxyServer(http_server.HttpServer):
141
162
"""A proxy test server for http transports."""
215
236
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()
218
246
def create_transport_secondary_server(self):
219
247
"""Create the secondary server redirecting to the primary server"""
220
248
new = self.get_readonly_server()
221
redirecting = HTTPServerRedirecting()
249
redirecting = HTTPServerRedirecting(
250
protocol_version=self._protocol_version)
222
251
redirecting.redirect_to(new.host, new.port)
252
redirecting._url_protocol = self._url_protocol
223
253
return redirecting
226
super(TestCaseWithRedirectedWebserver, self).setUp()
227
# The redirections will point to the new server
228
self.new_server = self.get_readonly_server()
229
# The requests to the old server will be redirected
230
self.old_server = self.get_secondary_server()
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())
233
274
class AuthRequestHandler(http_server.TestingHTTPRequestHandler):
243
284
# - auth_header_recv: the header received containing auth
244
285
# - 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')
246
300
def do_GET(self):
247
301
if self.authorized():
248
302
return http_server.TestingHTTPRequestHandler.do_GET(self)
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')
304
return self._require_authentication()
307
if self.authorized():
308
return http_server.TestingHTTPRequestHandler.do_HEAD(self)
310
return self._require_authentication()
263
313
class BasicAuthRequestHandler(AuthRequestHandler):
added
removed
bzrlib/tests/http_utils.py
