/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/HTTPTestUtil.py

Merge 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
 
121
122
        """Hand the request off to a smart server instance."""
122
123
        self.send_response(200)
123
124
        self.send_header("Content-type", "application/octet-stream")
124
 
        transport = get_transport(self.server.test_case._home_dir)
 
125
        transport = get_transport(self.server.test_case_server._home_dir)
125
126
        # TODO: We might like to support streaming responses.  1.0 allows no
126
127
        # Content-length in this case, so for integrity we should perform our
127
128
        # own chunking within the stream.
178
179
 
179
180
 
180
181
class TestCaseWithTwoWebservers(TestCaseWithWebserver):
181
 
    """A support class providinf readonly urls (on two servers) that are http://.
 
182
    """A support class providing readonly urls on two servers that are http://.
182
183
 
183
 
    We setup two webservers to allows various tests involving
 
184
    We set up two webservers to allows various tests involving
184
185
    proxies or redirections from one server to the other.
185
186
    """
186
187
    def setUp(self):
226
227
        return TestingHTTPRequestHandler.translate_path(self, path)
227
228
 
228
229
 
 
230
class RedirectRequestHandler(TestingHTTPRequestHandler):
 
231
    """Redirect all request to the specified server"""
 
232
 
 
233
    def parse_request(self):
 
234
        """Redirect a single HTTP request to another host"""
 
235
        valid = TestingHTTPRequestHandler.parse_request(self)
 
236
        if valid:
 
237
            tcs = self.server.test_case_server
 
238
            code, target = tcs.is_redirected(self.path)
 
239
            if code is not None and target is not None:
 
240
                # Redirect as instructed
 
241
                self.send_response(code)
 
242
                self.send_header('Location', target)
 
243
                self.end_headers()
 
244
                return False # The job is done
 
245
            else:
 
246
                # We leave the parent class serve the request
 
247
                pass
 
248
        return valid
 
249
 
 
250
 
 
251
class HTTPServerRedirecting(HttpServer):
 
252
    """An HttpServer redirecting to another server """
 
253
 
 
254
    def __init__(self, request_handler=RedirectRequestHandler):
 
255
        HttpServer.__init__(self, request_handler)
 
256
        # redirections is a list of tuples (source, target, code)
 
257
        # - source is a regexp for the paths requested
 
258
        # - target is a replacement for re.sub describing where
 
259
        #   the request will be redirected
 
260
        # - code is the http error code associated to the
 
261
        #   redirection (301 permanent, 302 temporarry, etc
 
262
        self.redirections = []
 
263
 
 
264
    def redirect_to(self, host, port):
 
265
        """Redirect all requests to a specific host:port"""
 
266
        self.redirections = [('(.*)',
 
267
                              r'http://%s:%s\1' % (host, port) ,
 
268
                              301)]
 
269
 
 
270
    def is_redirected(self, path):
 
271
        """Is the path redirected by this server.
 
272
 
 
273
        :param path: the requested relative path
 
274
 
 
275
        :returns: a tuple (code, target) if a matching
 
276
             redirection is found, (None, None) otherwise.
 
277
        """
 
278
        code = None
 
279
        target = None
 
280
        for (rsource, rtarget, rcode) in self.redirections:
 
281
            target, match = re.subn(rsource, rtarget, path)
 
282
            if match:
 
283
                code = rcode
 
284
                break # The first match wins
 
285
            else:
 
286
                target = None
 
287
        return code, target
 
288
 
 
289
 
 
290
class TestCaseWithRedirectedWebserver(TestCaseWithTwoWebservers):
 
291
   """A support class providing redirections from one server to another.
 
292
 
 
293
   We set up two webservers to allows various tests involving
 
294
   redirections.
 
295
   The 'old' server is redirected to the 'new' server.
 
296
   """
 
297
 
 
298
   def create_transport_secondary_server(self):
 
299
       """Create the secondary server redirecting to the primary server"""
 
300
       new = self.get_readonly_server()
 
301
       redirecting = HTTPServerRedirecting()
 
302
       redirecting.redirect_to(new.host, new.port)
 
303
       return redirecting
 
304
 
 
305
   def setUp(self):
 
306
       super(TestCaseWithRedirectedWebserver, self).setUp()
 
307
       # The redirections will point to the new server
 
308
       self.new_server = self.get_readonly_server()
 
309
       # The requests to the old server will be redirected
 
310
       self.old_server = self.get_secondary_server()
 
311
 
229
312