33
from bzrlib.smart import (
32
from bzrlib.smart import medium, protocol
37
33
from bzrlib.tests import http_server
38
from bzrlib.transport import chroot
34
from bzrlib.transport import (
41
40
class HTTPServerWithSmarts(http_server.HttpServer):
59
58
"""Hand the request off to a smart server instance."""
60
backing = transport.get_transport(
61
self.server.test_case_server._home_dir)
59
backing = get_transport(self.server.test_case_server._home_dir)
62
60
chroot_server = chroot.ChrootServer(backing)
63
61
chroot_server.start_server()
65
t = transport.get_transport(chroot_server.get_url())
63
t = get_transport(chroot_server.get_url())
66
64
self.do_POST_inner(t)
68
66
chroot_server.stop_server()
109
107
backed by regular disk files.
112
# These attributes can be overriden or parametrized by daughter clasess if
113
# needed, but must exist so that the create_transport_readonly_server()
114
# 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
115
113
_protocol_version = None
116
_url_protocol = 'http'
119
116
super(TestCaseWithWebserver, self).setUp()
120
117
self.transport_readonly_server = http_server.HttpServer
122
119
def create_transport_readonly_server(self):
123
server = self.transport_readonly_server(
120
return self.transport_readonly_server(
124
121
protocol_version=self._protocol_version)
125
server._url_protocol = self._url_protocol
129
124
class TestCaseWithTwoWebservers(TestCaseWithWebserver):
143
138
This is mostly a hook for daughter classes.
145
server = self.transport_secondary_server(
140
return self.transport_secondary_server(
146
141
protocol_version=self._protocol_version)
147
server._url_protocol = self._url_protocol
150
143
def get_secondary_server(self):
151
144
"""Get the server instance for the secondary transport."""
154
147
self.start_server(self.__secondary_server)
155
148
return self.__secondary_server
157
def get_secondary_url(self, relpath=None):
158
base = self.get_secondary_server().get_url()
159
return self._adjust_url(base, relpath)
161
def get_secondary_transport(self, relpath=None):
162
t = transport.get_transport(self.get_secondary_url(relpath))
163
self.assertTrue(t.is_readonly())
167
151
class ProxyServer(http_server.HttpServer):
168
152
"""A proxy test server for http transports."""
242
226
The 'old' server is redirected to the 'new' server.
246
super(TestCaseWithRedirectedWebserver, self).setUp()
247
# The redirections will point to the new server
248
self.new_server = self.get_readonly_server()
249
# The requests to the old server will be redirected to the new server
250
self.old_server = self.get_secondary_server()
252
229
def create_transport_secondary_server(self):
253
230
"""Create the secondary server redirecting to the primary server"""
254
231
new = self.get_readonly_server()
255
232
redirecting = HTTPServerRedirecting(
256
233
protocol_version=self._protocol_version)
257
234
redirecting.redirect_to(new.host, new.port)
258
redirecting._url_protocol = self._url_protocol
259
235
return redirecting
261
def get_old_url(self, relpath=None):
262
base = self.old_server.get_url()
263
return self._adjust_url(base, relpath)
265
def get_old_transport(self, relpath=None):
266
t = transport.get_transport(self.get_old_url(relpath))
267
self.assertTrue(t.is_readonly())
270
def get_new_url(self, relpath=None):
271
base = self.new_server.get_url()
272
return self._adjust_url(base, relpath)
274
def get_new_transport(self, relpath=None):
275
t = transport.get_transport(self.get_new_url(relpath))
276
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()
280
245
class AuthRequestHandler(http_server.TestingHTTPRequestHandler):