← 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) Revise legal option names to be less drastic. (Vincent Ladeuil)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/http_utils.py
1
 
# Copyright (C) 2005-2010 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
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
62
            chroot_server.stop_server()
107
103
    backed by regular disk files.
108
104
    """
109
105
 
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
 
    # propagate it.
 
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.
113
109
    _protocol_version = None
 
110
    _url_protocol = 'http'
114
111
 
115
112
    def setUp(self):
116
113
        super(TestCaseWithWebserver, self).setUp()
117
114
        self.transport_readonly_server = http_server.HttpServer
118
115
 
119
116
    def create_transport_readonly_server(self):
120
 
        return self.transport_readonly_server(
 
117
        server = self.transport_readonly_server(
121
118
            protocol_version=self._protocol_version)
 
119
        server._url_protocol = self._url_protocol
 
120
        return server
122
121
 
123
122
 
124
123
class TestCaseWithTwoWebservers(TestCaseWithWebserver):
137
136
 
138
137
        This is mostly a hook for daughter classes.
139
138
        """
140
 
        return self.transport_secondary_server(
 
139
        server = self.transport_secondary_server(
141
140
            protocol_version=self._protocol_version)
 
141
        server._url_protocol = self._url_protocol
 
142
        return server
142
143
 
143
144
    def get_secondary_server(self):
144
145
        """Get the server instance for the secondary transport."""
147
148
            self.start_server(self.__secondary_server)
148
149
        return self.__secondary_server
149
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
 
150
160
 
151
161
class ProxyServer(http_server.HttpServer):
152
162
    """A proxy test server for http transports."""
226
236
   The 'old' server is redirected to the 'new' server.
227
237
   """
228
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
 
229
246
   def create_transport_secondary_server(self):
230
247
       """Create the secondary server redirecting to the primary server"""
231
248
       new = self.get_readonly_server()
232
249
       redirecting = HTTPServerRedirecting(
233
250
           protocol_version=self._protocol_version)
234
251
       redirecting.redirect_to(new.host, new.port)
 
252
       redirecting._url_protocol = self._url_protocol
235
253
       return redirecting
236
254
 
237
 
   def setUp(self):
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()
 
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
243
272
 
244
273
 
245
274
class AuthRequestHandler(http_server.TestingHTTPRequestHandler):
255
284
    # - auth_header_recv: the header received containing auth
256
285
    # - auth_error_code: the error code to indicate auth required
257
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
 
258
300
    def do_GET(self):
259
301
        if self.authorized():
260
302
            return http_server.TestingHTTPRequestHandler.do_GET(self)
261
303
        else:
262
 
            # Note that we must update test_case_server *before*
263
 
            # sending the error or the client may try to read it
264
 
            # before we have sent the whole error back.
265
 
            tcs = self.server.test_case_server
266
 
            tcs.auth_required_errors += 1
267
 
            self.send_response(tcs.auth_error_code)
268
 
            self.send_header_auth_reqed()
269
 
            # We do not send a body
270
 
            self.send_header('Content-Length', '0')
271
 
            self.end_headers()
272
 
            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()
273
311
 
274
312
 
275
313
class BasicAuthRequestHandler(AuthRequestHandler):