~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_http.py

  • Committer: Martin Pool
  • Date: 2007-04-04 01:22:11 UTC
  • mfrom: (2393.1.1 bzr.docs)
  • mto: This revision was merged to the branch mainline in revision 2397.
  • Revision ID: mbp@sourcefrog.net-20070404012211-sq269me6bai9m6xk
merge trunk and doc fix from elliot

Show diffs side-by-side

added added

removed removed

Lines of Context:
26
26
import threading
27
27
 
28
28
import bzrlib
29
 
from bzrlib import errors
30
 
from bzrlib import osutils
 
29
from bzrlib import (
 
30
    errors,
 
31
    osutils,
 
32
    urlutils,
 
33
    )
31
34
from bzrlib.tests import (
32
35
    TestCase,
33
36
    TestSkipped,
42
45
    BadStatusRequestHandler,
43
46
    FakeProxyRequestHandler,
44
47
    ForbiddenRequestHandler,
 
48
    HTTPServerRedirecting,
45
49
    InvalidStatusRequestHandler,
46
50
    NoRangeRequestHandler,
47
51
    SingleRangeRequestHandler,
 
52
    TestCaseWithRedirectedWebserver,
48
53
    TestCaseWithTwoWebservers,
49
54
    TestCaseWithWebserver,
50
55
    WallRequestHandler,
51
56
    )
52
57
from bzrlib.transport import (
 
58
    do_catching_redirections,
53
59
    get_transport,
54
60
    Transport,
55
61
    )
56
62
from bzrlib.transport.http import (
57
63
    extract_auth,
58
64
    HttpTransportBase,
 
65
    _urllib2_wrappers,
59
66
    )
60
67
from bzrlib.transport.http._urllib import HttpTransport_urllib
61
68
 
946
953
                        TestRanges,
947
954
                        TestCaseWithWebserver):
948
955
    """Test the Range header in GET methods for pycurl implementation"""
 
956
 
 
957
 
 
958
class TestHTTPRedirections(object):
 
959
    """Test redirection between http servers.
 
960
 
 
961
    This MUST be used by daughter classes that also inherit from
 
962
    TestCaseWithRedirectedWebserver.
 
963
 
 
964
    We can't inherit directly from TestCaseWithTwoWebservers or the
 
965
    test framework will try to create an instance which cannot
 
966
    run, its implementation being incomplete. 
 
967
    """
 
968
 
 
969
    def create_transport_secondary_server(self):
 
970
        """Create the secondary server redirecting to the primary server"""
 
971
        new = self.get_readonly_server()
 
972
 
 
973
        redirecting = HTTPServerRedirecting()
 
974
        redirecting.redirect_to(new.host, new.port)
 
975
        return redirecting
 
976
 
 
977
    def setUp(self):
 
978
        super(TestHTTPRedirections, self).setUp()
 
979
        self.build_tree_contents([('a', '0123456789'),
 
980
                                  ('bundle',
 
981
                                  '# Bazaar revision bundle v0.9\n#\n')
 
982
                                  ],)
 
983
 
 
984
        self.old_transport = self._transport(self.old_server.get_url())
 
985
 
 
986
    def test_redirected(self):
 
987
        self.assertRaises(errors.RedirectRequested, self.old_transport.get, 'a')
 
988
        t = self._transport(self.new_server.get_url())
 
989
        self.assertEqual('0123456789', t.get('a').read())
 
990
 
 
991
    def test_read_redirected_bundle_from_url(self):
 
992
        from bzrlib.bundle import read_bundle_from_url
 
993
        url = self.old_transport.abspath('bundle')
 
994
        bundle = read_bundle_from_url(url)
 
995
        # If read_bundle_from_url was successful we get an empty bundle
 
996
        self.assertEqual([], bundle.revisions)
 
997
 
 
998
 
 
999
class TestHTTPRedirections_urllib(TestHTTPRedirections,
 
1000
                                  TestCaseWithRedirectedWebserver):
 
1001
    """Tests redirections for urllib implementation"""
 
1002
 
 
1003
    _transport = HttpTransport_urllib
 
1004
 
 
1005
 
 
1006
 
 
1007
class TestHTTPRedirections_pycurl(TestWithTransport_pycurl,
 
1008
                                  TestHTTPRedirections,
 
1009
                                  TestCaseWithRedirectedWebserver):
 
1010
    """Tests redirections for pycurl implementation"""
 
1011
 
 
1012
 
 
1013
class RedirectedRequest(_urllib2_wrappers.Request):
 
1014
    """Request following redirections"""
 
1015
 
 
1016
    init_orig = _urllib2_wrappers.Request.__init__
 
1017
 
 
1018
    def __init__(self, method, url, *args, **kwargs):
 
1019
        RedirectedRequest.init_orig(self, method, url, args, kwargs)
 
1020
        self.follow_redirections = True
 
1021
 
 
1022
 
 
1023
class TestHTTPSilentRedirections_urllib(TestCaseWithRedirectedWebserver):
 
1024
    """Test redirections provided by urllib.
 
1025
 
 
1026
    http implementations do not redirect silently anymore (they
 
1027
    do not redirect at all in fact). The mechanism is still in
 
1028
    place at the _urllib2_wrappers.Request level and these tests
 
1029
    exercise it.
 
1030
 
 
1031
    For the pycurl implementation
 
1032
    the redirection have been deleted as we may deprecate pycurl
 
1033
    and I have no place to keep a working implementation.
 
1034
    -- vila 20070212
 
1035
    """
 
1036
 
 
1037
    _transport = HttpTransport_urllib
 
1038
 
 
1039
    def setUp(self):
 
1040
        super(TestHTTPSilentRedirections_urllib, self).setUp()
 
1041
        self.setup_redirected_request()
 
1042
        self.addCleanup(self.cleanup_redirected_request)
 
1043
        self.build_tree_contents([('a','a'),
 
1044
                                  ('1/',),
 
1045
                                  ('1/a', 'redirected once'),
 
1046
                                  ('2/',),
 
1047
                                  ('2/a', 'redirected twice'),
 
1048
                                  ('3/',),
 
1049
                                  ('3/a', 'redirected thrice'),
 
1050
                                  ('4/',),
 
1051
                                  ('4/a', 'redirected 4 times'),
 
1052
                                  ('5/',),
 
1053
                                  ('5/a', 'redirected 5 times'),
 
1054
                                  ],)
 
1055
 
 
1056
        self.old_transport = self._transport(self.old_server.get_url())
 
1057
 
 
1058
    def setup_redirected_request(self):
 
1059
        self.original_class = _urllib2_wrappers.Request
 
1060
        _urllib2_wrappers.Request = RedirectedRequest
 
1061
 
 
1062
    def cleanup_redirected_request(self):
 
1063
        _urllib2_wrappers.Request = self.original_class
 
1064
 
 
1065
    def create_transport_secondary_server(self):
 
1066
        """Create the secondary server, redirections are defined in the tests"""
 
1067
        return HTTPServerRedirecting()
 
1068
 
 
1069
    def test_one_redirection(self):
 
1070
        t = self.old_transport
 
1071
 
 
1072
        req = RedirectedRequest('GET', t.abspath('a'))
 
1073
        req.follow_redirections = True
 
1074
        new_prefix = 'http://%s:%s' % (self.new_server.host,
 
1075
                                       self.new_server.port)
 
1076
        self.old_server.redirections = \
 
1077
            [('(.*)', r'%s/1\1' % (new_prefix), 301),]
 
1078
        self.assertEquals('redirected once',t._perform(req).read())
 
1079
 
 
1080
    def test_five_redirections(self):
 
1081
        t = self.old_transport
 
1082
 
 
1083
        req = RedirectedRequest('GET', t.abspath('a'))
 
1084
        req.follow_redirections = True
 
1085
        old_prefix = 'http://%s:%s' % (self.old_server.host,
 
1086
                                       self.old_server.port)
 
1087
        new_prefix = 'http://%s:%s' % (self.new_server.host,
 
1088
                                       self.new_server.port)
 
1089
        self.old_server.redirections = \
 
1090
            [('/1(.*)', r'%s/2\1' % (old_prefix), 302),
 
1091
             ('/2(.*)', r'%s/3\1' % (old_prefix), 303),
 
1092
             ('/3(.*)', r'%s/4\1' % (old_prefix), 307),
 
1093
             ('/4(.*)', r'%s/5\1' % (new_prefix), 301),
 
1094
             ('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
 
1095
             ]
 
1096
        self.assertEquals('redirected 5 times',t._perform(req).read())
 
1097
 
 
1098
 
 
1099
class TestDoCatchRedirections(TestCaseWithRedirectedWebserver):
 
1100
    """Test transport.do_catching_redirections.
 
1101
 
 
1102
    We arbitrarily choose to use urllib transports
 
1103
    """
 
1104
 
 
1105
    _transport = HttpTransport_urllib
 
1106
 
 
1107
    def setUp(self):
 
1108
        super(TestDoCatchRedirections, self).setUp()
 
1109
        self.build_tree_contents([('a', '0123456789'),],)
 
1110
 
 
1111
        self.old_transport = self._transport(self.old_server.get_url())
 
1112
 
 
1113
    def get_a(self, transport):
 
1114
        return transport.get('a')
 
1115
 
 
1116
    def test_no_redirection(self):
 
1117
        t = self._transport(self.new_server.get_url())
 
1118
 
 
1119
        # We use None for redirected so that we fail if redirected
 
1120
        self.assertEquals('0123456789',
 
1121
                          do_catching_redirections(self.get_a, t, None).read())
 
1122
 
 
1123
    def test_one_redirection(self):
 
1124
        self.redirections = 0
 
1125
 
 
1126
        def redirected(transport, exception, redirection_notice):
 
1127
            self.redirections += 1
 
1128
            dir, file = urlutils.split(exception.target)
 
1129
            return self._transport(dir)
 
1130
 
 
1131
        self.assertEquals('0123456789',
 
1132
                          do_catching_redirections(self.get_a,
 
1133
                                                   self.old_transport,
 
1134
                                                   redirected
 
1135
                                                   ).read())
 
1136
        self.assertEquals(1, self.redirections)
 
1137
 
 
1138
    def test_redirection_loop(self):
 
1139
 
 
1140
        def redirected(transport, exception, redirection_notice):
 
1141
            # By using the redirected url as a base dir for the
 
1142
            # *old* transport, we create a loop: a => a/a =>
 
1143
            # a/a/a
 
1144
            return self.old_transport.clone(exception.target)
 
1145
 
 
1146
        self.assertRaises(errors.TooManyRedirections, do_catching_redirections,
 
1147
                          self.get_a, self.old_transport, redirected)