~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_http.py

mergeĀ basicĀ auth

Show diffs side-by-side

added added

removed removed

Lines of Context:
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
29
30
from bzrlib import (
30
31
    errors,
31
32
    osutils,
 
33
    ui,
32
34
    urlutils,
33
35
    )
34
36
from bzrlib.tests import (
35
37
    TestCase,
 
38
    TestUIFactory,
36
39
    TestSkipped,
 
40
    StringIOWrapper,
37
41
    )
38
42
from bzrlib.tests.HttpServer import (
39
43
    HttpServer,
43
47
from bzrlib.tests.HTTPTestUtil import (
44
48
    BadProtocolRequestHandler,
45
49
    BadStatusRequestHandler,
 
50
    BasicAuthHTTPServer,
46
51
    FakeProxyRequestHandler,
47
52
    ForbiddenRequestHandler,
48
53
    HTTPServerRedirecting,
725
730
            osutils.set_or_unset_env(name, value)
726
731
 
727
732
    def _proxied_request(self):
728
 
        from bzrlib.transport.http._urllib2_wrappers import (
729
 
            ProxyHandler,
730
 
            Request,
731
 
            )
732
 
 
733
 
        handler = ProxyHandler()
734
 
        request = Request('GET','http://baz/buzzle')
 
733
        handler = _urllib2_wrappers.ProxyHandler()
 
734
        request = _urllib2_wrappers.Request('GET','http://baz/buzzle')
735
735
        handler.set_proxy(request, 'http')
736
736
        return request
737
737
 
1145
1145
 
1146
1146
        self.assertRaises(errors.TooManyRedirections, do_catching_redirections,
1147
1147
                          self.get_a, self.old_transport, redirected)
 
1148
 
 
1149
 
 
1150
class TestHTTPBasicAuth(TestCaseWithWebserver):
 
1151
    """Test basic authentication scheme"""
 
1152
 
 
1153
    _transport = HttpTransport_urllib
 
1154
    _auth_header = 'Authorization'
 
1155
 
 
1156
    def create_transport_readonly_server(self):
 
1157
        return BasicAuthHTTPServer()
 
1158
 
 
1159
    def setUp(self):
 
1160
        super(TestHTTPBasicAuth, self).setUp()
 
1161
        self.build_tree_contents([('a', 'contents of a\n'),
 
1162
                                  ('b', 'contents of b\n'),])
 
1163
        self.server = self.get_readonly_server()
 
1164
 
 
1165
        self.old_factory = ui.ui_factory
 
1166
        self.addCleanup(self.restoreUIFactory)
 
1167
 
 
1168
    def restoreUIFactory(self):
 
1169
        ui.ui_factory = self.old_factory
 
1170
 
 
1171
    def get_user_url(self, user=None, password=None):
 
1172
        """Build an url embedding user and password"""
 
1173
        url = '%s://' % self.server._url_protocol
 
1174
        if user is not None:
 
1175
            url += user
 
1176
            if password is not None:
 
1177
                url += ':' + password
 
1178
            url += '@'
 
1179
        url += '%s:%s/' % (self.server.host, self.server.port)
 
1180
        return url
 
1181
 
 
1182
    def test_empty_pass(self):
 
1183
        self.server.add_user('joe', '')
 
1184
        t = self._transport(self.get_user_url('joe', ''))
 
1185
        self.assertEqual('contents of a\n', t.get('a').read())
 
1186
 
 
1187
    def test_user_pass(self):
 
1188
        self.server.add_user('joe', 'foo')
 
1189
        t = self._transport(self.get_user_url('joe', 'foo'))
 
1190
        self.assertEqual('contents of a\n', t.get('a').read())
 
1191
 
 
1192
    def test_unknown_user(self):
 
1193
        self.server.add_user('joe', 'foo')
 
1194
        t = self._transport(self.get_user_url('bill', 'foo'))
 
1195
        self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
 
1196
 
 
1197
    def test_wrong_pass(self):
 
1198
        self.server.add_user('joe', 'foo')
 
1199
        t = self._transport(self.get_user_url('joe', 'bar'))
 
1200
        self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
 
1201
 
 
1202
    def test_prompt_for_password(self):
 
1203
        self.server.add_user('joe', 'foo')
 
1204
        t = self._transport(self.get_user_url('joe'))
 
1205
        ui.ui_factory = TestUIFactory(stdin='foo\n', stdout=StringIOWrapper())
 
1206
        self.assertEqual('contents of a\n',t.get('a').read())
 
1207
        # stdin should be empty
 
1208
        self.assertEqual('', ui.ui_factory.stdin.readline())
 
1209
        # And we shouldn't prompt again for a different request
 
1210
        # against the same transport.
 
1211
        self.assertEqual('contents of b\n',t.get('b').read())
 
1212
        t2 = t.clone()
 
1213
        # And neither against a clone
 
1214
        self.assertEqual('contents of b\n',t2.get('b').read())