← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/HTTPTestUtil.py

Upgraded to the latest bzr.dev

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/HTTPTestUtil.py
17
17
from cStringIO import StringIO
18
18
import errno
19
19
from SimpleHTTPServer import SimpleHTTPRequestHandler
 
20
import re
20
21
import socket
21
22
import urlparse
22
23
 
 
24
from bzrlib.smart import protocol
23
25
from bzrlib.tests import TestCaseWithTransport
24
26
from bzrlib.tests.HttpServer import (
25
27
    HttpServer,
27
29
    )
28
30
from bzrlib.transport import (
29
31
    get_transport,
30
 
    smart,
31
32
    )
32
33
 
33
34
 
120
121
        """Hand the request off to a smart server instance."""
121
122
        self.send_response(200)
122
123
        self.send_header("Content-type", "application/octet-stream")
123
 
        transport = get_transport(self.server.test_case._home_dir)
 
124
        transport = get_transport(self.server.test_case_server._home_dir)
124
125
        # TODO: We might like to support streaming responses.  1.0 allows no
125
126
        # Content-length in this case, so for integrity we should perform our
126
127
        # own chunking within the stream.
129
130
        # we have to stop early due to error, but we would also have to use the
130
131
        # HTTP trailer facility which may not be widely available.
131
132
        out_buffer = StringIO()
132
 
        smart_protocol_request = smart.SmartServerRequestProtocolOne(
 
133
        smart_protocol_request = protocol.SmartServerRequestProtocolOne(
133
134
                transport, out_buffer.write)
134
135
        # if this fails, we should return 400 bad request, but failure is
135
136
        # failure for now - RBC 20060919
177
178
 
178
179
 
179
180
class TestCaseWithTwoWebservers(TestCaseWithWebserver):
180
 
    """A support class providinf readonly urls (on two servers) that are http://.
 
181
    """A support class providing readonly urls on two servers that are http://.
181
182
 
182
 
    We setup two webservers to allows various tests involving
 
183
    We set up two webservers to allows various tests involving
183
184
    proxies or redirections from one server to the other.
184
185
    """
185
186
    def setUp(self):
225
226
        return TestingHTTPRequestHandler.translate_path(self, path)
226
227
 
227
228
 
 
229
class RedirectRequestHandler(TestingHTTPRequestHandler):
 
230
    """Redirect all request to the specified server"""
 
231
 
 
232
    def parse_request(self):
 
233
        """Redirect a single HTTP request to another host"""
 
234
        valid = TestingHTTPRequestHandler.parse_request(self)
 
235
        if valid:
 
236
            tcs = self.server.test_case_server
 
237
            code, target = tcs.is_redirected(self.path)
 
238
            if code is not None and target is not None:
 
239
                # Redirect as instructed
 
240
                self.send_response(code)
 
241
                self.send_header('Location', target)
 
242
                self.end_headers()
 
243
                return False # The job is done
 
244
            else:
 
245
                # We leave the parent class serve the request
 
246
                pass
 
247
        return valid
 
248
 
 
249
 
 
250
class HTTPServerRedirecting(HttpServer):
 
251
    """An HttpServer redirecting to another server """
 
252
 
 
253
    def __init__(self, request_handler=RedirectRequestHandler):
 
254
        HttpServer.__init__(self, request_handler)
 
255
        # redirections is a list of tuples (source, target, code)
 
256
        # - source is a regexp for the paths requested
 
257
        # - target is a replacement for re.sub describing where
 
258
        #   the request will be redirected
 
259
        # - code is the http error code associated to the
 
260
        #   redirection (301 permanent, 302 temporarry, etc
 
261
        self.redirections = []
 
262
 
 
263
    def redirect_to(self, host, port):
 
264
        """Redirect all requests to a specific host:port"""
 
265
        self.redirections = [('(.*)',
 
266
                              r'http://%s:%s\1' % (host, port) ,
 
267
                              301)]
 
268
 
 
269
    def is_redirected(self, path):
 
270
        """Is the path redirected by this server.
 
271
 
 
272
        :param path: the requested relative path
 
273
 
 
274
        :returns: a tuple (code, target) if a matching
 
275
             redirection is found, (None, None) otherwise.
 
276
        """
 
277
        code = None
 
278
        target = None
 
279
        for (rsource, rtarget, rcode) in self.redirections:
 
280
            target, match = re.subn(rsource, rtarget, path)
 
281
            if match:
 
282
                code = rcode
 
283
                break # The first match wins
 
284
            else:
 
285
                target = None
 
286
        return code, target
 
287
 
 
288
 
 
289
class TestCaseWithRedirectedWebserver(TestCaseWithTwoWebservers):
 
290
   """A support class providing redirections from one server to another.
 
291
 
 
292
   We set up two webservers to allows various tests involving
 
293
   redirections.
 
294
   The 'old' server is redirected to the 'new' server.
 
295
   """
 
296
 
 
297
   def create_transport_secondary_server(self):
 
298
       """Create the secondary server redirecting to the primary server"""
 
299
       new = self.get_readonly_server()
 
300
       redirecting = HTTPServerRedirecting()
 
301
       redirecting.redirect_to(new.host, new.port)
 
302
       return redirecting
 
303
 
 
304
   def setUp(self):
 
305
       super(TestCaseWithRedirectedWebserver, self).setUp()
 
306
       # The redirections will point to the new server
 
307
       self.new_server = self.get_readonly_server()
 
308
       # The requests to the old server will be redirected
 
309
       self.old_server = self.get_secondary_server()
 
310
 
228
311