← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_http.py

  • Committer: John Arbash Meinel
  • Date: 2007-04-28 15:04:17 UTC
  • mfrom: (2466 +trunk)
  • mto: This revision was merged to the branch mainline in revision 2566.
  • Revision ID: john@arbash-meinel.com-20070428150417-trp3pi0pzd411pu4
[merge] bzr.dev 2466

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_http.py
20
20
# TODO: Should be renamed to bzrlib.transport.http.tests?
21
21
# TODO: What about renaming to bzrlib.tests.transport.http ?
22
22
 
 
23
from cStringIO import StringIO
23
24
import os
24
25
import select
25
26
import socket
 
27
import sys
26
28
import threading
27
29
 
28
30
import bzrlib
29
31
from bzrlib import (
30
32
    errors,
31
33
    osutils,
 
34
    ui,
32
35
    urlutils,
33
36
    )
34
37
from bzrlib.tests import (
35
38
    TestCase,
 
39
    TestUIFactory,
36
40
    TestSkipped,
 
41
    StringIOWrapper,
37
42
    )
38
43
from bzrlib.tests.HttpServer import (
39
44
    HttpServer,
43
48
from bzrlib.tests.HTTPTestUtil import (
44
49
    BadProtocolRequestHandler,
45
50
    BadStatusRequestHandler,
46
 
    FakeProxyRequestHandler,
47
51
    ForbiddenRequestHandler,
 
52
    HTTPBasicAuthServer,
 
53
    HTTPDigestAuthServer,
48
54
    HTTPServerRedirecting,
49
55
    InvalidStatusRequestHandler,
50
56
    NoRangeRequestHandler,
 
57
    ProxyBasicAuthServer,
 
58
    ProxyDigestAuthServer,
 
59
    ProxyServer,
51
60
    SingleRangeRequestHandler,
52
61
    TestCaseWithRedirectedWebserver,
53
62
    TestCaseWithTwoWebservers,
65
74
    _urllib2_wrappers,
66
75
    )
67
76
from bzrlib.transport.http._urllib import HttpTransport_urllib
 
77
from bzrlib.transport.http._urllib2_wrappers import (
 
78
    PasswordManager,
 
79
    ProxyHandler,
 
80
    Request,
 
81
    )
68
82
 
69
83
 
70
84
class FakeManager(object):
702
716
class TestHttpProxyWhiteBox(TestCase):
703
717
    """Whitebox test proxy http authorization.
704
718
 
705
 
    These tests concern urllib implementation only.
 
719
    Only the urllib implementation is tested here.
706
720
    """
707
721
 
708
722
    def setUp(self):
712
726
    def tearDown(self):
713
727
        self._restore_env()
714
728
 
715
 
    def _set_and_capture_env_var(self, name, new_value):
716
 
        """Set an environment variable, and reset it when finished."""
717
 
        self._old_env[name] = osutils.set_or_unset_env(name, new_value)
718
 
 
719
729
    def _install_env(self, env):
720
730
        for name, value in env.iteritems():
721
 
            self._set_and_capture_env_var(name, value)
 
731
            self._old_env[name] = osutils.set_or_unset_env(name, value)
722
732
 
723
733
    def _restore_env(self):
724
734
        for name, value in self._old_env.iteritems():
725
735
            osutils.set_or_unset_env(name, value)
726
736
 
727
737
    def _proxied_request(self):
728
 
        from bzrlib.transport.http._urllib2_wrappers import (
729
 
            ProxyHandler,
730
 
            Request,
731
 
            )
732
 
 
733
 
        handler = ProxyHandler()
 
738
        handler = ProxyHandler(PasswordManager())
734
739
        request = Request('GET','http://baz/buzzle')
735
740
        handler.set_proxy(request, 'http')
736
741
        return request
740
745
        request = self._proxied_request()
741
746
        self.assertFalse(request.headers.has_key('Proxy-authorization'))
742
747
 
743
 
    def test_empty_pass(self):
744
 
        self._install_env({'http_proxy': 'http://joe@bar.com'})
745
 
        request = self._proxied_request()
746
 
        self.assertEqual('Basic ' + 'joe:'.encode('base64').strip(),
747
 
                         request.headers['Proxy-authorization'])
748
 
    def test_user_pass(self):
749
 
        self._install_env({'http_proxy': 'http://joe:foo@bar.com'})
750
 
        request = self._proxied_request()
751
 
        self.assertEqual('Basic ' + 'joe:foo'.encode('base64').strip(),
752
 
                         request.headers['Proxy-authorization'])
753
 
 
754
748
    def test_invalid_proxy(self):
755
749
        """A proxy env variable without scheme"""
756
750
        self._install_env({'http_proxy': 'host:1234'})
786
780
                                  ('foo-proxied', 'proxied contents of foo\n')])
787
781
        # Let's setup some attributes for tests
788
782
        self.server = self.get_readonly_server()
789
 
        # FIXME: We should not rely on 'localhost' being the hostname
790
 
        self.proxy_address = 'localhost:%d' % self.server.port
 
783
        self.proxy_address = '%s:%d' % (self.server.host, self.server.port)
791
784
        self.no_proxy_host = self.proxy_address
792
785
        # The secondary server is the proxy
793
786
        self.proxy = self.get_secondary_server()
798
791
        """Creates an http server that will serve files with
799
792
        '-proxied' appended to their names.
800
793
        """
801
 
        return HttpServer(FakeProxyRequestHandler)
802
 
 
803
 
    def _set_and_capture_env_var(self, name, new_value):
804
 
        """Set an environment variable, and reset it when finished."""
805
 
        self._old_env[name] = osutils.set_or_unset_env(name, new_value)
 
794
        return ProxyServer()
806
795
 
807
796
    def _install_env(self, env):
808
797
        for name, value in env.iteritems():
809
 
            self._set_and_capture_env_var(name, value)
 
798
            self._old_env[name] = osutils.set_or_unset_env(name, value)
810
799
 
811
800
    def _restore_env(self):
812
801
        for name, value in self._old_env.iteritems():
1010
999
    """Tests redirections for pycurl implementation"""
1011
1000
 
1012
1001
 
1013
 
class RedirectedRequest(_urllib2_wrappers.Request):
 
1002
class RedirectedRequest(Request):
1014
1003
    """Request following redirections"""
1015
1004
 
1016
 
    init_orig = _urllib2_wrappers.Request.__init__
 
1005
    init_orig = Request.__init__
1017
1006
 
1018
1007
    def __init__(self, method, url, *args, **kwargs):
1019
1008
        RedirectedRequest.init_orig(self, method, url, args, kwargs)
1145
1134
 
1146
1135
        self.assertRaises(errors.TooManyRedirections, do_catching_redirections,
1147
1136
                          self.get_a, self.old_transport, redirected)
 
1137
 
 
1138
 
 
1139
class TestAuth(object):
 
1140
    """Test some authentication scheme specified by daughter class.
 
1141
 
 
1142
    This MUST be used by daughter classes that also inherit from
 
1143
    either TestCaseWithWebserver or TestCaseWithTwoWebservers.
 
1144
    """
 
1145
 
 
1146
    def setUp(self):
 
1147
        """Set up the test environment
 
1148
 
 
1149
        Daughter classes should set up their own environment
 
1150
        (including self.server) and explicitely call this
 
1151
        method. This is needed because we want to reuse the same
 
1152
        tests for proxy and no-proxy accesses which have
 
1153
        different ways of setting self.server.
 
1154
        """
 
1155
        self.build_tree_contents([('a', 'contents of a\n'),
 
1156
                                  ('b', 'contents of b\n'),])
 
1157
        self.old_factory = ui.ui_factory
 
1158
        self.old_stdout = sys.stdout
 
1159
        sys.stdout = StringIOWrapper()
 
1160
        self.addCleanup(self.restoreUIFactory)
 
1161
 
 
1162
    def restoreUIFactory(self):
 
1163
        ui.ui_factory = self.old_factory
 
1164
        sys.stdout = self.old_stdout
 
1165
 
 
1166
    def get_user_url(self, user=None, password=None):
 
1167
        """Build an url embedding user and password"""
 
1168
        url = '%s://' % self.server._url_protocol
 
1169
        if user is not None:
 
1170
            url += user
 
1171
            if password is not None:
 
1172
                url += ':' + password
 
1173
            url += '@'
 
1174
        url += '%s:%s/' % (self.server.host, self.server.port)
 
1175
        return url
 
1176
 
 
1177
    def test_no_user(self):
 
1178
        self.server.add_user('joe', 'foo')
 
1179
        t = self.get_user_transport()
 
1180
        self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
 
1181
        # Only one 'Authentication Required' error should occur
 
1182
        self.assertEqual(1, self.server.auth_required_errors)
 
1183
 
 
1184
    def test_empty_pass(self):
 
1185
        self.server.add_user('joe', '')
 
1186
        t = self.get_user_transport('joe', '')
 
1187
        self.assertEqual('contents of a\n', t.get('a').read())
 
1188
        # Only one 'Authentication Required' error should occur
 
1189
        self.assertEqual(1, self.server.auth_required_errors)
 
1190
 
 
1191
    def test_user_pass(self):
 
1192
        self.server.add_user('joe', 'foo')
 
1193
        t = self.get_user_transport('joe', 'foo')
 
1194
        self.assertEqual('contents of a\n', t.get('a').read())
 
1195
        # Only one 'Authentication Required' error should occur
 
1196
        self.assertEqual(1, self.server.auth_required_errors)
 
1197
 
 
1198
    def test_unknown_user(self):
 
1199
        self.server.add_user('joe', 'foo')
 
1200
        t = self.get_user_transport('bill', 'foo')
 
1201
        self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
 
1202
        # Two 'Authentication Required' errors should occur (the
 
1203
        # initial 'who are you' and 'I don't know you, who are
 
1204
        # you').
 
1205
        self.assertEqual(2, self.server.auth_required_errors)
 
1206
 
 
1207
    def test_wrong_pass(self):
 
1208
        self.server.add_user('joe', 'foo')
 
1209
        t = self.get_user_transport('joe', 'bar')
 
1210
        self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
 
1211
        # Two 'Authentication Required' errors should occur (the
 
1212
        # initial 'who are you' and 'this is not you, who are you')
 
1213
        self.assertEqual(2, self.server.auth_required_errors)
 
1214
 
 
1215
    def test_prompt_for_password(self):
 
1216
        self.server.add_user('joe', 'foo')
 
1217
        t = self.get_user_transport('joe', None)
 
1218
        ui.ui_factory = TestUIFactory(stdin='foo\n')
 
1219
        self.assertEqual('contents of a\n',t.get('a').read())
 
1220
        # stdin should be empty
 
1221
        self.assertEqual('', ui.ui_factory.stdin.readline())
 
1222
        # And we shouldn't prompt again for a different request
 
1223
        # against the same transport.
 
1224
        self.assertEqual('contents of b\n',t.get('b').read())
 
1225
        t2 = t.clone()
 
1226
        # And neither against a clone
 
1227
        self.assertEqual('contents of b\n',t2.get('b').read())
 
1228
        # Only one 'Authentication Required' error should occur
 
1229
        self.assertEqual(1, self.server.auth_required_errors)
 
1230
 
 
1231
 
 
1232
class TestHTTPAuth(TestAuth):
 
1233
    """Test HTTP authentication schemes.
 
1234
 
 
1235
    Daughter classes MUST inherit from TestCaseWithWebserver too.
 
1236
    """
 
1237
 
 
1238
    _auth_header = 'Authorization'
 
1239
 
 
1240
    def setUp(self):
 
1241
        TestCaseWithWebserver.setUp(self)
 
1242
        self.server = self.get_readonly_server()
 
1243
        TestAuth.setUp(self)
 
1244
 
 
1245
    def get_user_transport(self, user=None, password=None):
 
1246
        return self._transport(self.get_user_url(user, password))
 
1247
 
 
1248
 
 
1249
class TestProxyAuth(TestAuth):
 
1250
    """Test proxy authentication schemes.
 
1251
 
 
1252
    Daughter classes MUST also inherit from TestCaseWithWebserver.
 
1253
    """
 
1254
    _auth_header = 'Proxy-authorization'
 
1255
 
 
1256
    def setUp(self):
 
1257
        TestCaseWithWebserver.setUp(self)
 
1258
        self.server = self.get_readonly_server()
 
1259
        self._old_env = {}
 
1260
        self.addCleanup(self._restore_env)
 
1261
        TestAuth.setUp(self)
 
1262
        # Override the contents to avoid false positives
 
1263
        self.build_tree_contents([('a', 'not proxied contents of a\n'),
 
1264
                                  ('b', 'not proxied contents of b\n'),
 
1265
                                  ('a-proxied', 'contents of a\n'),
 
1266
                                  ('b-proxied', 'contents of b\n'),
 
1267
                                  ])
 
1268
 
 
1269
    def get_user_transport(self, user=None, password=None):
 
1270
        self._install_env({'all_proxy': self.get_user_url(user, password)})
 
1271
        return self._transport(self.server.get_url())
 
1272
 
 
1273
    def _install_env(self, env):
 
1274
        for name, value in env.iteritems():
 
1275
            self._old_env[name] = osutils.set_or_unset_env(name, value)
 
1276
 
 
1277
    def _restore_env(self):
 
1278
        for name, value in self._old_env.iteritems():
 
1279
            osutils.set_or_unset_env(name, value)
 
1280
 
 
1281
 
 
1282
class TestHTTPBasicAuth(TestHTTPAuth, TestCaseWithWebserver):
 
1283
    """Test http basic authentication scheme"""
 
1284
 
 
1285
    _transport = HttpTransport_urllib
 
1286
 
 
1287
    def create_transport_readonly_server(self):
 
1288
        return HTTPBasicAuthServer()
 
1289
 
 
1290
 
 
1291
class TestHTTPProxyBasicAuth(TestProxyAuth, TestCaseWithWebserver):
 
1292
    """Test proxy basic authentication scheme"""
 
1293
 
 
1294
    _transport = HttpTransport_urllib
 
1295
 
 
1296
    def create_transport_readonly_server(self):
 
1297
        return ProxyBasicAuthServer()
 
1298
 
 
1299
 
 
1300
class TestDigestAuth(object):
 
1301
    """Digest Authentication specific tests"""
 
1302
 
 
1303
    def test_changing_nonce(self):
 
1304
        self.server.add_user('joe', 'foo')
 
1305
        t = self.get_user_transport('joe', 'foo')
 
1306
        self.assertEqual('contents of a\n', t.get('a').read())
 
1307
        self.assertEqual('contents of b\n', t.get('b').read())
 
1308
        # Only one 'Authentication Required' error should have
 
1309
        # occured so far
 
1310
        self.assertEqual(1, self.server.auth_required_errors)
 
1311
        # The server invalidates the current nonce
 
1312
        self.server.auth_nonce = self.server.auth_nonce + '. No, now!'
 
1313
        self.assertEqual('contents of a\n', t.get('a').read())
 
1314
        # Two 'Authentication Required' errors should occur (the
 
1315
        # initial 'who are you' and a second 'who are you' with the new nonce)
 
1316
        self.assertEqual(2, self.server.auth_required_errors)
 
1317
 
 
1318
 
 
1319
class TestHTTPDigestAuth(TestHTTPAuth, TestDigestAuth, TestCaseWithWebserver):
 
1320
    """Test http digest authentication scheme"""
 
1321
 
 
1322
    _transport = HttpTransport_urllib
 
1323
 
 
1324
    def create_transport_readonly_server(self):
 
1325
        return HTTPDigestAuthServer()
 
1326
 
 
1327
 
 
1328
class TestHTTPProxyDigestAuth(TestProxyAuth, TestDigestAuth,
 
1329
                              TestCaseWithWebserver):
 
1330
    """Test proxy digest authentication scheme"""
 
1331
 
 
1332
    _transport = HttpTransport_urllib
 
1333
 
 
1334
    def create_transport_readonly_server(self):
 
1335
        return ProxyDigestAuthServer()
 
1336