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

1st cut merge of bzr.dev r3907

Show diffs side-by-side

added added

removed removed

Lines of Context:
16
16
 
17
17
from cStringIO import StringIO
18
18
import errno
19
 
import md5
20
19
import re
21
 
import sha
22
20
import socket
23
21
import threading
24
22
import time
25
23
import urllib2
26
24
import urlparse
27
25
 
 
26
 
28
27
from bzrlib import (
29
28
    errors,
 
29
    osutils,
30
30
    tests,
31
 
    transport,
32
31
    )
33
32
from bzrlib.smart import medium, protocol
34
33
from bzrlib.tests import http_server
 
34
from bzrlib.transport import (
 
35
    chroot,
 
36
    get_transport,
 
37
    )
35
38
 
36
39
 
37
40
class HTTPServerWithSmarts(http_server.HttpServer):
46
49
 
47
50
 
48
51
class SmartRequestHandler(http_server.TestingHTTPRequestHandler):
49
 
    """Extend TestingHTTPRequestHandler to support smart client POSTs."""
 
52
    """Extend TestingHTTPRequestHandler to support smart client POSTs.
 
53
    
 
54
    XXX: This duplicates a fair bit of the logic in bzrlib.transport.http.wsgi.
 
55
    """
50
56
 
51
57
    def do_POST(self):
52
58
        """Hand the request off to a smart server instance."""
 
59
        backing = get_transport(self.server.test_case_server._home_dir)
 
60
        chroot_server = chroot.ChrootServer(backing)
 
61
        chroot_server.setUp()
 
62
        try:
 
63
            t = get_transport(chroot_server.get_url())
 
64
            self.do_POST_inner(t)
 
65
        finally:
 
66
            chroot_server.tearDown()
 
67
 
 
68
    def do_POST_inner(self, chrooted_transport):
53
69
        self.send_response(200)
54
70
        self.send_header("Content-type", "application/octet-stream")
55
 
        t = transport.get_transport(self.server.test_case_server._home_dir)
 
71
        if not self.path.endswith('.bzr/smart'):
 
72
            raise AssertionError(
 
73
                'POST to path not ending in .bzr/smart: %r' % (self.path,))
 
74
        t = chrooted_transport.clone(self.path[:-len('.bzr/smart')])
56
75
        # if this fails, we should return 400 bad request, but failure is
57
76
        # failure for now - RBC 20060919
58
77
        data_length = int(self.headers['Content-Length'])
378
397
        A1 = '%s:%s:%s' % (user, realm, password)
379
398
        A2 = '%s:%s' % (command, auth['uri'])
380
399
 
381
 
        H = lambda x: md5.new(x).hexdigest()
 
400
        H = lambda x: osutils.md5(x).hexdigest()
382
401
        KD = lambda secret, data: H("%s:%s" % (secret, data))
383
402
 
384
403
        nonce_count = int(auth['nc'], 16)