/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 breezy/tests/http_utils.py

  • Committer: Jelmer Vernooij
  • Date: 2017-06-04 19:17:13 UTC
  • mfrom: (0.193.10 trunk)
  • mto: This revision was merged to the branch mainline in revision 6778.
  • Revision ID: jelmer@jelmer.uk-20170604191713-hau7dfsqsl035slm
Bundle the cvs plugin.

Show diffs side-by-side

added added

removed removed

Lines of Context:
breezy/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
14
14
# along with this program; if not, write to the Free Software
15
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
16
 
17
 
from cStringIO import StringIO
18
 
import errno
19
17
import re
20
 
import socket
21
 
import threading
22
 
import time
23
18
import urllib2
24
 
import urlparse
25
 
 
26
 
 
27
 
from bzrlib import (
 
19
 
 
20
 
 
21
from .. import (
28
22
    errors,
29
23
    osutils,
30
24
    tests,
31
 
    )
32
 
from bzrlib.smart import medium, protocol
33
 
from bzrlib.tests import http_server
34
 
from bzrlib.transport import (
35
 
    chroot,
36
 
    get_transport,
37
 
    )
 
25
    transport,
 
26
    )
 
27
from ..sixish import (
 
28
    BytesIO,
 
29
    )
 
30
from ..smart import (
 
31
    medium,
 
32
    )
 
33
from . import http_server
 
34
from ..transport import chroot
38
35
 
39
36
 
40
37
class HTTPServerWithSmarts(http_server.HttpServer):
51
48
class SmartRequestHandler(http_server.TestingHTTPRequestHandler):
52
49
    """Extend TestingHTTPRequestHandler to support smart client POSTs.
53
50
 
54
 
    XXX: This duplicates a fair bit of the logic in bzrlib.transport.http.wsgi.
 
51
    XXX: This duplicates a fair bit of the logic in breezy.transport.http.wsgi.
55
52
    """
56
53
 
57
54
    def do_POST(self):
58
55
        """Hand the request off to a smart server instance."""
59
 
        backing = get_transport(self.server.test_case_server._home_dir)
 
56
        backing = transport.get_transport_from_path(
 
57
            self.server.test_case_server._home_dir)
60
58
        chroot_server = chroot.ChrootServer(backing)
61
59
        chroot_server.start_server()
62
60
        try:
63
 
            t = get_transport(chroot_server.get_url())
 
61
            t = transport.get_transport_from_url(chroot_server.get_url())
64
62
            self.do_POST_inner(t)
65
63
        finally:
66
64
            chroot_server.stop_server()
85
83
        request_bytes = self.rfile.read(data_length)
86
84
        protocol_factory, unused_bytes = medium._get_protocol_factory_for_bytes(
87
85
            request_bytes)
88
 
        out_buffer = StringIO()
 
86
        out_buffer = BytesIO()
89
87
        smart_protocol_request = protocol_factory(t, out_buffer.write, '/')
90
88
        # Perhaps there should be a SmartServerHTTPMedium that takes care of
91
89
        # feeding the bytes in the http request to the smart_protocol_request,
106
104
    one. This will currently fail if the primary transport is not
107
105
    backed by regular disk files.
108
106
    """
 
107
 
 
108
    # These attributes can be overriden or parametrized by daughter clasess if
 
109
    # needed, but must exist so that the create_transport_readonly_server()
 
110
    # method (or any method creating an http(s) server) can propagate it.
 
111
    _protocol_version = None
 
112
    _url_protocol = 'http'
 
113
 
109
114
    def setUp(self):
110
115
        super(TestCaseWithWebserver, self).setUp()
111
116
        self.transport_readonly_server = http_server.HttpServer
112
117
 
 
118
    def create_transport_readonly_server(self):
 
119
        server = self.transport_readonly_server(
 
120
            protocol_version=self._protocol_version)
 
121
        server._url_protocol = self._url_protocol
 
122
        return server
 
123
 
113
124
 
114
125
class TestCaseWithTwoWebservers(TestCaseWithWebserver):
115
126
    """A support class providing readonly urls on two servers that are http://.
127
138
 
128
139
        This is mostly a hook for daughter classes.
129
140
        """
130
 
        return self.transport_secondary_server()
 
141
        server = self.transport_secondary_server(
 
142
            protocol_version=self._protocol_version)
 
143
        server._url_protocol = self._url_protocol
 
144
        return server
131
145
 
132
146
    def get_secondary_server(self):
133
147
        """Get the server instance for the secondary transport."""
136
150
            self.start_server(self.__secondary_server)
137
151
        return self.__secondary_server
138
152
 
 
153
    def get_secondary_url(self, relpath=None):
 
154
        base = self.get_secondary_server().get_url()
 
155
        return self._adjust_url(base, relpath)
 
156
 
 
157
    def get_secondary_transport(self, relpath=None):
 
158
        t = transport.get_transport_from_url(self.get_secondary_url(relpath))
 
159
        self.assertTrue(t.is_readonly())
 
160
        return t
 
161
 
139
162
 
140
163
class ProxyServer(http_server.HttpServer):
141
164
    """A proxy test server for http transports."""
215
238
   The 'old' server is redirected to the 'new' server.
216
239
   """
217
240
 
 
241
   def setUp(self):
 
242
       super(TestCaseWithRedirectedWebserver, self).setUp()
 
243
       # The redirections will point to the new server
 
244
       self.new_server = self.get_readonly_server()
 
245
       # The requests to the old server will be redirected to the new server
 
246
       self.old_server = self.get_secondary_server()
 
247
 
218
248
   def create_transport_secondary_server(self):
219
249
       """Create the secondary server redirecting to the primary server"""
220
250
       new = self.get_readonly_server()
221
 
       redirecting = HTTPServerRedirecting()
 
251
       redirecting = HTTPServerRedirecting(
 
252
           protocol_version=self._protocol_version)
222
253
       redirecting.redirect_to(new.host, new.port)
 
254
       redirecting._url_protocol = self._url_protocol
223
255
       return redirecting
224
256
 
225
 
   def setUp(self):
226
 
       super(TestCaseWithRedirectedWebserver, self).setUp()
227
 
       # The redirections will point to the new server
228
 
       self.new_server = self.get_readonly_server()
229
 
       # The requests to the old server will be redirected
230
 
       self.old_server = self.get_secondary_server()
 
257
   def get_old_url(self, relpath=None):
 
258
        base = self.old_server.get_url()
 
259
        return self._adjust_url(base, relpath)
 
260
 
 
261
   def get_old_transport(self, relpath=None):
 
262
        t = transport.get_transport_from_url(self.get_old_url(relpath))
 
263
        self.assertTrue(t.is_readonly())
 
264
        return t
 
265
 
 
266
   def get_new_url(self, relpath=None):
 
267
        base = self.new_server.get_url()
 
268
        return self._adjust_url(base, relpath)
 
269
 
 
270
   def get_new_transport(self, relpath=None):
 
271
        t = transport.get_transport_from_url(self.get_new_url(relpath))
 
272
        self.assertTrue(t.is_readonly())
 
273
        return t
231
274
 
232
275
 
233
276
class AuthRequestHandler(http_server.TestingHTTPRequestHandler):
243
286
    # - auth_header_recv: the header received containing auth
244
287
    # - auth_error_code: the error code to indicate auth required
245
288
 
 
289
    def _require_authentication(self):
 
290
        # Note that we must update test_case_server *before*
 
291
        # sending the error or the client may try to read it
 
292
        # before we have sent the whole error back.
 
293
        tcs = self.server.test_case_server
 
294
        tcs.auth_required_errors += 1
 
295
        self.send_response(tcs.auth_error_code)
 
296
        self.send_header_auth_reqed()
 
297
        # We do not send a body
 
298
        self.send_header('Content-Length', '0')
 
299
        self.end_headers()
 
300
        return
 
301
 
246
302
    def do_GET(self):
247
303
        if self.authorized():
248
304
            return http_server.TestingHTTPRequestHandler.do_GET(self)
249
305
        else:
250
 
            # Note that we must update test_case_server *before*
251
 
            # sending the error or the client may try to read it
252
 
            # before we have sent the whole error back.
253
 
            tcs = self.server.test_case_server
254
 
            tcs.auth_required_errors += 1
255
 
            self.send_response(tcs.auth_error_code)
256
 
            self.send_header_auth_reqed()
257
 
            # We do not send a body
258
 
            self.send_header('Content-Length', '0')
259
 
            self.end_headers()
260
 
            return
 
306
            return self._require_authentication()
 
307
 
 
308
    def do_HEAD(self):
 
309
        if self.authorized():
 
310
            return http_server.TestingHTTPRequestHandler.do_HEAD(self)
 
311
        else:
 
312
            return self._require_authentication()
261
313
 
262
314
 
263
315
class BasicAuthRequestHandler(AuthRequestHandler):
396
448
        if realm != self.auth_realm:
397
449
            return False
398
450
        user = auth['username']
399
 
        if not self.password_of.has_key(user):
 
451
        if user not in self.password_of:
400
452
            return False
401
453
        algorithm= auth['algorithm']
402
454
        if algorithm != 'MD5':