← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/http_utils.py

  • Committer: Jelmer Vernooij
  • Date: 2012-01-24 13:14:06 UTC
  • mto: (6445.4.5 nested-trees-spec)
  • mto: This revision was merged to the branch mainline in revision 6518.
  • Revision ID: jelmer@samba.org-20120124131406-wedftkorbpv37bm0
Import nested tree doc from devnotes.

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 (
32
27
    )
33
28
from bzrlib.smart import (
34
29
    medium,
35
 
    protocol,
36
30
    )
37
31
from bzrlib.tests import http_server
38
32
from bzrlib.transport import chroot
57
51
 
58
52
    def do_POST(self):
59
53
        """Hand the request off to a smart server instance."""
60
 
        backing = transport.get_transport(
 
54
        backing = transport.get_transport_from_path(
61
55
            self.server.test_case_server._home_dir)
62
56
        chroot_server = chroot.ChrootServer(backing)
63
57
        chroot_server.start_server()
64
58
        try:
65
 
            t = transport.get_transport(chroot_server.get_url())
 
59
            t = transport.get_transport_from_url(chroot_server.get_url())
66
60
            self.do_POST_inner(t)
67
61
        finally:
68
62
            chroot_server.stop_server()
159
153
        return self._adjust_url(base, relpath)
160
154
 
161
155
    def get_secondary_transport(self, relpath=None):
162
 
        t = transport.get_transport(self.get_secondary_url(relpath))
 
156
        t = transport.get_transport_from_url(self.get_secondary_url(relpath))
163
157
        self.assertTrue(t.is_readonly())
164
158
        return t
165
159
 
263
257
        return self._adjust_url(base, relpath)
264
258
 
265
259
   def get_old_transport(self, relpath=None):
266
 
        t = transport.get_transport(self.get_old_url(relpath))
 
260
        t = transport.get_transport_from_url(self.get_old_url(relpath))
267
261
        self.assertTrue(t.is_readonly())
268
262
        return t
269
263
 
272
266
        return self._adjust_url(base, relpath)
273
267
 
274
268
   def get_new_transport(self, relpath=None):
275
 
        t = transport.get_transport(self.get_new_url(relpath))
 
269
        t = transport.get_transport_from_url(self.get_new_url(relpath))
276
270
        self.assertTrue(t.is_readonly())
277
271
        return t
278
272
 
290
284
    # - auth_header_recv: the header received containing auth
291
285
    # - auth_error_code: the error code to indicate auth required
292
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
 
293
300
    def do_GET(self):
294
301
        if self.authorized():
295
302
            return http_server.TestingHTTPRequestHandler.do_GET(self)
296
303
        else:
297
 
            # Note that we must update test_case_server *before*
298
 
            # sending the error or the client may try to read it
299
 
            # before we have sent the whole error back.
300
 
            tcs = self.server.test_case_server
301
 
            tcs.auth_required_errors += 1
302
 
            self.send_response(tcs.auth_error_code)
303
 
            self.send_header_auth_reqed()
304
 
            # We do not send a body
305
 
            self.send_header('Content-Length', '0')
306
 
            self.end_headers()
307
 
            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()
308
311
 
309
312
 
310
313
class BasicAuthRequestHandler(AuthRequestHandler):