32
from bzrlib.smart import medium, protocol
33
from bzrlib.smart import (
33
37
from bzrlib.tests import http_server
34
from bzrlib.transport import (
38
from bzrlib.transport import chroot
40
41
class HTTPServerWithSmarts(http_server.HttpServer):
58
59
"""Hand the request off to a smart server instance."""
59
backing = get_transport(self.server.test_case_server._home_dir)
60
backing = transport.get_transport(
61
self.server.test_case_server._home_dir)
60
62
chroot_server = chroot.ChrootServer(backing)
61
63
chroot_server.start_server()
63
t = get_transport(chroot_server.get_url())
65
t = transport.get_transport(chroot_server.get_url())
64
66
self.do_POST_inner(t)
66
68
chroot_server.stop_server()
107
109
backed by regular disk files.
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
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.
113
115
_protocol_version = None
116
_url_protocol = 'http'
116
119
super(TestCaseWithWebserver, self).setUp()
117
120
self.transport_readonly_server = http_server.HttpServer
119
122
def create_transport_readonly_server(self):
120
return self.transport_readonly_server(
123
server = self.transport_readonly_server(
121
124
protocol_version=self._protocol_version)
125
server._url_protocol = self._url_protocol
124
129
class TestCaseWithTwoWebservers(TestCaseWithWebserver):
138
143
This is mostly a hook for daughter classes.
140
return self.transport_secondary_server(
145
server = self.transport_secondary_server(
141
146
protocol_version=self._protocol_version)
147
server._url_protocol = self._url_protocol
143
150
def get_secondary_server(self):
144
151
"""Get the server instance for the secondary transport."""
147
154
self.start_server(self.__secondary_server)
148
155
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())
151
167
class ProxyServer(http_server.HttpServer):
152
168
"""A proxy test server for http transports."""
226
242
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()
229
252
def create_transport_secondary_server(self):
230
253
"""Create the secondary server redirecting to the primary server"""
231
254
new = self.get_readonly_server()
232
255
redirecting = HTTPServerRedirecting(
233
256
protocol_version=self._protocol_version)
234
257
redirecting.redirect_to(new.host, new.port)
258
redirecting._url_protocol = self._url_protocol
235
259
return redirecting
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()
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())
245
280
class AuthRequestHandler(http_server.TestingHTTPRequestHandler):