← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/http_utils.py

(vila) Fix test failures blocking package builds. (Vincent Ladeuil)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/http_utils.py
1
 
# Copyright (C) 2005 Canonical Ltd
 
1
# Copyright (C) 2005-2011 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
15
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
16
 
17
17
from cStringIO import StringIO
18
 
import errno
19
18
import re
20
 
import socket
21
 
import threading
22
 
import time
23
19
import urllib2
24
 
import urlparse
25
20
 
26
21
 
27
22
from bzrlib import (
28
23
    errors,
29
24
    osutils,
30
25
    tests,
31
 
    )
32
 
from bzrlib.smart import medium, protocol
 
26
    transport,
 
27
    )
 
28
from bzrlib.smart import (
 
29
    medium,
 
30
    )
33
31
from bzrlib.tests import http_server
34
 
from bzrlib.transport import (
35
 
    chroot,
36
 
    get_transport,
37
 
    )
 
32
from bzrlib.transport import chroot
38
33
 
39
34
 
40
35
class HTTPServerWithSmarts(http_server.HttpServer):
56
51
 
57
52
    def do_POST(self):
58
53
        """Hand the request off to a smart server instance."""
59
 
        backing = get_transport(self.server.test_case_server._home_dir)
 
54
        backing = transport.get_transport_from_path(
 
55
            self.server.test_case_server._home_dir)
60
56
        chroot_server = chroot.ChrootServer(backing)
61
 
        chroot_server.setUp()
 
57
        chroot_server.start_server()
62
58
        try:
63
 
            t = get_transport(chroot_server.get_url())
 
59
            t = transport.get_transport_from_url(chroot_server.get_url())
64
60
            self.do_POST_inner(t)
65
61
        finally:
66
 
            chroot_server.tearDown()
 
62
            chroot_server.stop_server()
67
63
 
68
64
    def do_POST_inner(self, chrooted_transport):
69
65
        self.send_response(200)
106
102
    one. This will currently fail if the primary transport is not
107
103
    backed by regular disk files.
108
104
    """
 
105
 
 
106
    # These attributes can be overriden or parametrized by daughter clasess if
 
107
    # needed, but must exist so that the create_transport_readonly_server()
 
108
    # method (or any method creating an http(s) server) can propagate it.
 
109
    _protocol_version = None
 
110
    _url_protocol = 'http'
 
111
 
109
112
    def setUp(self):
110
113
        super(TestCaseWithWebserver, self).setUp()
111
114
        self.transport_readonly_server = http_server.HttpServer
112
115
 
 
116
    def create_transport_readonly_server(self):
 
117
        server = self.transport_readonly_server(
 
118
            protocol_version=self._protocol_version)
 
119
        server._url_protocol = self._url_protocol
 
120
        return server
 
121
 
113
122
 
114
123
class TestCaseWithTwoWebservers(TestCaseWithWebserver):
115
124
    """A support class providing readonly urls on two servers that are http://.
127
136
 
128
137
        This is mostly a hook for daughter classes.
129
138
        """
130
 
        return self.transport_secondary_server()
 
139
        server = self.transport_secondary_server(
 
140
            protocol_version=self._protocol_version)
 
141
        server._url_protocol = self._url_protocol
 
142
        return server
131
143
 
132
144
    def get_secondary_server(self):
133
145
        """Get the server instance for the secondary transport."""
134
146
        if self.__secondary_server is None:
135
147
            self.__secondary_server = self.create_transport_secondary_server()
136
 
            self.__secondary_server.setUp()
137
 
            self.addCleanup(self.__secondary_server.tearDown)
 
148
            self.start_server(self.__secondary_server)
138
149
        return self.__secondary_server
139
150
 
 
151
    def get_secondary_url(self, relpath=None):
 
152
        base = self.get_secondary_server().get_url()
 
153
        return self._adjust_url(base, relpath)
 
154
 
 
155
    def get_secondary_transport(self, relpath=None):
 
156
        t = transport.get_transport_from_url(self.get_secondary_url(relpath))
 
157
        self.assertTrue(t.is_readonly())
 
158
        return t
 
159
 
140
160
 
141
161
class ProxyServer(http_server.HttpServer):
142
162
    """A proxy test server for http transports."""
216
236
   The 'old' server is redirected to the 'new' server.
217
237
   """
218
238
 
 
239
   def setUp(self):
 
240
       super(TestCaseWithRedirectedWebserver, self).setUp()
 
241
       # The redirections will point to the new server
 
242
       self.new_server = self.get_readonly_server()
 
243
       # The requests to the old server will be redirected to the new server
 
244
       self.old_server = self.get_secondary_server()
 
245
 
219
246
   def create_transport_secondary_server(self):
220
247
       """Create the secondary server redirecting to the primary server"""
221
248
       new = self.get_readonly_server()
222
 
       redirecting = HTTPServerRedirecting()
 
249
       redirecting = HTTPServerRedirecting(
 
250
           protocol_version=self._protocol_version)
223
251
       redirecting.redirect_to(new.host, new.port)
 
252
       redirecting._url_protocol = self._url_protocol
224
253
       return redirecting
225
254
 
226
 
   def setUp(self):
227
 
       super(TestCaseWithRedirectedWebserver, self).setUp()
228
 
       # The redirections will point to the new server
229
 
       self.new_server = self.get_readonly_server()
230
 
       # The requests to the old server will be redirected
231
 
       self.old_server = self.get_secondary_server()
 
255
   def get_old_url(self, relpath=None):
 
256
        base = self.old_server.get_url()
 
257
        return self._adjust_url(base, relpath)
 
258
 
 
259
   def get_old_transport(self, relpath=None):
 
260
        t = transport.get_transport_from_url(self.get_old_url(relpath))
 
261
        self.assertTrue(t.is_readonly())
 
262
        return t
 
263
 
 
264
   def get_new_url(self, relpath=None):
 
265
        base = self.new_server.get_url()
 
266
        return self._adjust_url(base, relpath)
 
267
 
 
268
   def get_new_transport(self, relpath=None):
 
269
        t = transport.get_transport_from_url(self.get_new_url(relpath))
 
270
        self.assertTrue(t.is_readonly())
 
271
        return t
232
272
 
233
273
 
234
274
class AuthRequestHandler(http_server.TestingHTTPRequestHandler):
244
284
    # - auth_header_recv: the header received containing auth
245
285
    # - auth_error_code: the error code to indicate auth required
246
286
 
 
287
    def _require_authentication(self):
 
288
        # Note that we must update test_case_server *before*
 
289
        # sending the error or the client may try to read it
 
290
        # before we have sent the whole error back.
 
291
        tcs = self.server.test_case_server
 
292
        tcs.auth_required_errors += 1
 
293
        self.send_response(tcs.auth_error_code)
 
294
        self.send_header_auth_reqed()
 
295
        # We do not send a body
 
296
        self.send_header('Content-Length', '0')
 
297
        self.end_headers()
 
298
        return
 
299
 
247
300
    def do_GET(self):
248
301
        if self.authorized():
249
302
            return http_server.TestingHTTPRequestHandler.do_GET(self)
250
303
        else:
251
 
            # Note that we must update test_case_server *before*
252
 
            # sending the error or the client may try to read it
253
 
            # before we have sent the whole error back.
254
 
            tcs = self.server.test_case_server
255
 
            tcs.auth_required_errors += 1
256
 
            self.send_response(tcs.auth_error_code)
257
 
            self.send_header_auth_reqed()
258
 
            # We do not send a body
259
 
            self.send_header('Content-Length', '0')
260
 
            self.end_headers()
261
 
            return
 
304
            return self._require_authentication()
 
305
 
 
306
    def do_HEAD(self):
 
307
        if self.authorized():
 
308
            return http_server.TestingHTTPRequestHandler.do_HEAD(self)
 
309
        else:
 
310
            return self._require_authentication()
262
311
 
263
312
 
264
313
class BasicAuthRequestHandler(AuthRequestHandler):