91
91
# we have to stop early due to error, but we would also have to use the
92
92
# HTTP trailer facility which may not be widely available.
93
93
request_bytes = self.rfile.read(data_length)
94
protocol_factory, unused_bytes = medium._get_protocol_factory_for_bytes(
94
protocol_factory, unused_bytes = (
95
medium._get_protocol_factory_for_bytes(request_bytes))
96
96
out_buffer = BytesIO()
97
97
smart_protocol_request = protocol_factory(t, out_buffer.write, '/')
98
98
# Perhaps there should be a SmartServerHTTPMedium that takes care of
233
234
for (rsource, rtarget, rcode) in self.redirections:
234
target, match = re.subn(rsource, rtarget, path)
235
target, match = re.subn(rsource, rtarget, path, count=1)
237
break # The first match wins
238
break # The first match wins
240
241
return code, target
243
244
class TestCaseWithRedirectedWebserver(TestCaseWithTwoWebservers):
244
"""A support class providing redirections from one server to another.
246
We set up two webservers to allows various tests involving
248
The 'old' server is redirected to the 'new' server.
252
super(TestCaseWithRedirectedWebserver, self).setUp()
253
# The redirections will point to the new server
254
self.new_server = self.get_readonly_server()
255
# The requests to the old server will be redirected to the new server
256
self.old_server = self.get_secondary_server()
258
def create_transport_secondary_server(self):
259
"""Create the secondary server redirecting to the primary server"""
260
new = self.get_readonly_server()
261
redirecting = HTTPServerRedirecting(
262
protocol_version=self._protocol_version)
263
redirecting.redirect_to(new.host, new.port)
264
redirecting._url_protocol = self._url_protocol
267
def get_old_url(self, relpath=None):
245
"""A support class providing redirections from one server to another.
247
We set up two webservers to allows various tests involving
249
The 'old' server is redirected to the 'new' server.
253
super(TestCaseWithRedirectedWebserver, self).setUp()
254
# The redirections will point to the new server
255
self.new_server = self.get_readonly_server()
256
# The requests to the old server will be redirected to the new server
257
self.old_server = self.get_secondary_server()
259
def create_transport_secondary_server(self):
260
"""Create the secondary server redirecting to the primary server"""
261
new = self.get_readonly_server()
262
redirecting = HTTPServerRedirecting(
263
protocol_version=self._protocol_version)
264
redirecting.redirect_to(new.host, new.port)
265
redirecting._url_protocol = self._url_protocol
268
def get_old_url(self, relpath=None):
268
269
base = self.old_server.get_url()
269
270
return self._adjust_url(base, relpath)
271
def get_old_transport(self, relpath=None):
272
def get_old_transport(self, relpath=None):
272
273
t = transport.get_transport_from_url(self.get_old_url(relpath))
273
274
self.assertTrue(t.is_readonly())
276
def get_new_url(self, relpath=None):
277
def get_new_url(self, relpath=None):
277
278
base = self.new_server.get_url()
278
279
return self._adjust_url(base, relpath)
280
def get_new_transport(self, relpath=None):
281
def get_new_transport(self, relpath=None):
281
282
t = transport.get_transport_from_url(self.get_new_url(relpath))
282
283
self.assertTrue(t.is_readonly())
475
476
A1 = ('%s:%s:%s' % (user, realm, password)).encode('utf-8')
476
477
A2 = ('%s:%s' % (command, auth['uri'])).encode('utf-8')
478
H = lambda x: osutils.md5(x).hexdigest()
479
KD = lambda secret, data: H(("%s:%s" % (secret, data)).encode('utf-8'))
480
return osutils.md5(x).hexdigest()
482
def KD(secret, data):
483
return H(("%s:%s" % (secret, data)).encode('utf-8'))
481
485
nonce_count = int(auth['nc'], 16)
added
removed
breezy/tests/http_utils.py
