← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_http_implementations.py

  • Committer: Vincent Ladeuil
  • Date: 2007-12-20 09:18:04 UTC
  • mto: (3146.3.1 179368) (3156.2.1 trunk)
  • mto: This revision was merged to the branch mainline in revision 3158.
  • Revision ID: v.ladeuil+lp@free.fr-20071220091804-8by0r5bwsw3qfg1q
Finish http parameterization, 24 auth tests failing for pycurl (not
tested before).

* bzrlib/tests/test_http_implementations.py:
(AuthenticationTestProviderAdapter): Multiply tests against the
authentication schemes supported.
(TestRanges, TestHTTPSilentRedirections, TestDoCatchRedirections,
TestAuth, TestProxyAuth): Transferred from test_http.py.

* bzrlib/tests/test_http.py: 
Delete the FIXME about testing against all http client
implementations, hurrah ! Fix some more imports. Transfer all
authentication, redirection, ranges tests to http_implementations.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_http_implementations.py
25
25
protocol implementation to guarantee the robustness of our transports.
26
26
"""
27
27
 
 
28
import errno
28
29
import SimpleHTTPServer
29
30
import socket
30
31
 
31
32
import bzrlib
32
33
from bzrlib import (
 
34
    config,
33
35
    errors,
 
36
    osutils,
34
37
    tests,
35
38
    transport,
 
39
    ui,
 
40
    urlutils,
36
41
    )
37
42
from bzrlib.tests import (
38
43
    http_server,
39
44
    http_utils,
40
45
    )
41
 
from bzrlib.transport.http._urllib import HttpTransport_urllib
 
46
from bzrlib.transport.http import (
 
47
    _urllib,
 
48
    _urllib2_wrappers,
 
49
    )
42
50
 
43
51
 
44
52
try:
47
55
except errors.DependencyNotPresent:
48
56
    pycurl_present = False
49
57
 
 
58
 
50
59
class HTTPImplementationsTestProviderAdapter(tests.TestScenarioApplier):
51
60
 
52
61
    def __init__(self):
53
 
        transport_scenarios = [('urllib',
54
 
                                dict(_transport=HttpTransport_urllib,
55
 
                                     _server=http_server.HttpServer_urllib,
56
 
                                     _qualified_prefix='http+urllib',
57
 
                                     )),]
 
62
        transport_scenarios = [
 
63
            ('urllib',
 
64
             dict(_transport=_urllib.HttpTransport_urllib,
 
65
                  _server=http_server.HttpServer_urllib,
 
66
                  _qualified_prefix='http+urllib',
 
67
                  )),]
58
68
        if pycurl_present:
59
69
            transport_scenarios.append(
60
70
                ('pycurl', dict(_transport=PyCurlTransport,
64
74
        self.scenarios = transport_scenarios
65
75
 
66
76
 
 
77
class AuthenticationTestProviderAdapter(HTTPImplementationsTestProviderAdapter):
 
78
 
 
79
    def __init__(self):
 
80
        super(AuthenticationTestProviderAdapter, self).__init__()
 
81
        auth_scheme_scenarios = [
 
82
            ('basic', dict(_auth_scheme='basic')),
 
83
            ('digest', dict(_auth_scheme='digest')),
 
84
            ]
 
85
 
 
86
        self.scenarios = tests.multiply_scenarios(self.scenarios,
 
87
                                                  auth_scheme_scenarios)
 
88
 
 
89
 
67
90
def load_tests(standard_tests, module, loader):
68
91
    """Multiply tests for http clients and protocol versions."""
69
 
    adapter = HTTPImplementationsTestProviderAdapter()
 
92
    http_adapter = HTTPImplementationsTestProviderAdapter()
 
93
    auth_adapter = AuthenticationTestProviderAdapter()
70
94
    result = loader.suiteClass()
71
95
    for test in tests.iter_suite_tests(standard_tests):
72
 
        result.addTests(adapter.adapt(test))
 
96
        if isinstance(test, TestAuth):
 
97
            result.addTests(auth_adapter.adapt(test))
 
98
        else:
 
99
            result.addTests(http_adapter.adapt(test))
73
100
    return result
74
101
 
75
102
 
372
399
        self.assertRaises(errors.TransportError, t.get, 'foo/bar')
373
400
 
374
401
 
 
402
class TestRanges(http_utils.TestCaseWithWebserver):
 
403
    """Test the Range header in GET methods."""
 
404
 
 
405
    def setUp(self):
 
406
        http_utils.TestCaseWithWebserver.setUp(self)
 
407
        self.build_tree_contents([('a', '0123456789')],)
 
408
        server = self.get_readonly_server()
 
409
        self.transport = self._transport(server.get_url())
 
410
 
 
411
    def _file_contents(self, relpath, ranges):
 
412
        offsets = [ (start, end - start + 1) for start, end in ranges]
 
413
        coalesce = self.transport._coalesce_offsets
 
414
        coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
 
415
        code, data = self.transport._get(relpath, coalesced)
 
416
        self.assertTrue(code in (200, 206),'_get returns: %d' % code)
 
417
        for start, end in ranges:
 
418
            data.seek(start)
 
419
            yield data.read(end - start + 1)
 
420
 
 
421
    def _file_tail(self, relpath, tail_amount):
 
422
        code, data = self.transport._get(relpath, [], tail_amount)
 
423
        self.assertTrue(code in (200, 206),'_get returns: %d' % code)
 
424
        data.seek(-tail_amount, 2)
 
425
        return data.read(tail_amount)
 
426
 
 
427
    def test_range_header(self):
 
428
        # Valid ranges
 
429
        map(self.assertEqual,['0', '234'],
 
430
            list(self._file_contents('a', [(0,0), (2,4)])),)
 
431
 
 
432
    def test_range_header_tail(self):
 
433
        self.assertEqual('789', self._file_tail('a', 3))
 
434
 
 
435
    def test_syntactically_invalid_range_header(self):
 
436
        self.assertListRaises(errors.InvalidHttpRange,
 
437
                          self._file_contents, 'a', [(4, 3)])
 
438
 
 
439
    def test_semantically_invalid_range_header(self):
 
440
        self.assertListRaises(errors.InvalidHttpRange,
 
441
                          self._file_contents, 'a', [(42, 128)])
 
442
 
 
443
 
375
444
class TestRangeRequestServer(TestSpecificRequestHandler):
376
445
    """Tests readv requests against server.
377
446
 
560
629
        self.assertEqual(2, self.get_readonly_server().GET_request_nb)
561
630
 
562
631
 
 
632
class TestProxyHttpServer(http_utils.TestCaseWithTwoWebservers):
 
633
    """Tests proxy server.
 
634
 
 
635
    Be aware that we do not setup a real proxy here. Instead, we
 
636
    check that the *connection* goes through the proxy by serving
 
637
    different content (the faked proxy server append '-proxied'
 
638
    to the file names).
 
639
    """
 
640
 
 
641
    # FIXME: We don't have an https server available, so we don't
 
642
    # test https connections.
 
643
 
 
644
    def setUp(self):
 
645
        super(TestProxyHttpServer, self).setUp()
 
646
        self.build_tree_contents([('foo', 'contents of foo\n'),
 
647
                                  ('foo-proxied', 'proxied contents of foo\n')])
 
648
        # Let's setup some attributes for tests
 
649
        self.server = self.get_readonly_server()
 
650
        self.proxy_address = '%s:%d' % (self.server.host, self.server.port)
 
651
        if self._testing_pycurl():
 
652
            # Oh my ! pycurl does not check for the port as part of
 
653
            # no_proxy :-( So we just test the host part
 
654
            self.no_proxy_host = 'localhost'
 
655
        else:
 
656
            self.no_proxy_host = self.proxy_address
 
657
        # The secondary server is the proxy
 
658
        self.proxy = self.get_secondary_server()
 
659
        self.proxy_url = self.proxy.get_url()
 
660
        self._old_env = {}
 
661
 
 
662
    def _testing_pycurl(self):
 
663
        return pycurl_present and self._transport == PyCurlTransport
 
664
 
 
665
    def create_transport_secondary_server(self):
 
666
        """Creates an http server that will serve files with
 
667
        '-proxied' appended to their names.
 
668
        """
 
669
        return http_utils.ProxyServer()
 
670
 
 
671
    def _install_env(self, env):
 
672
        for name, value in env.iteritems():
 
673
            self._old_env[name] = osutils.set_or_unset_env(name, value)
 
674
 
 
675
    def _restore_env(self):
 
676
        for name, value in self._old_env.iteritems():
 
677
            osutils.set_or_unset_env(name, value)
 
678
 
 
679
    def proxied_in_env(self, env):
 
680
        self._install_env(env)
 
681
        url = self.server.get_url()
 
682
        t = self._transport(url)
 
683
        try:
 
684
            self.assertEqual(t.get('foo').read(), 'proxied contents of foo\n')
 
685
        finally:
 
686
            self._restore_env()
 
687
 
 
688
    def not_proxied_in_env(self, env):
 
689
        self._install_env(env)
 
690
        url = self.server.get_url()
 
691
        t = self._transport(url)
 
692
        try:
 
693
            self.assertEqual(t.get('foo').read(), 'contents of foo\n')
 
694
        finally:
 
695
            self._restore_env()
 
696
 
 
697
    def test_http_proxy(self):
 
698
        self.proxied_in_env({'http_proxy': self.proxy_url})
 
699
 
 
700
    def test_HTTP_PROXY(self):
 
701
        if self._testing_pycurl():
 
702
            # pycurl does not check HTTP_PROXY for security reasons
 
703
            # (for use in a CGI context that we do not care
 
704
            # about. Should we ?)
 
705
            raise tests.TestNotApplicable(
 
706
                'pycurl does not check HTTP_PROXY for security reasons')
 
707
        self.proxied_in_env({'HTTP_PROXY': self.proxy_url})
 
708
 
 
709
    def test_all_proxy(self):
 
710
        self.proxied_in_env({'all_proxy': self.proxy_url})
 
711
 
 
712
    def test_ALL_PROXY(self):
 
713
        self.proxied_in_env({'ALL_PROXY': self.proxy_url})
 
714
 
 
715
    def test_http_proxy_with_no_proxy(self):
 
716
        self.not_proxied_in_env({'http_proxy': self.proxy_url,
 
717
                                 'no_proxy': self.no_proxy_host})
 
718
 
 
719
    def test_HTTP_PROXY_with_NO_PROXY(self):
 
720
        if self._testing_pycurl():
 
721
            raise tests.TestNotApplicable(
 
722
                'pycurl does not check HTTP_PROXY for security reasons')
 
723
        self.not_proxied_in_env({'HTTP_PROXY': self.proxy_url,
 
724
                                 'NO_PROXY': self.no_proxy_host})
 
725
 
 
726
    def test_all_proxy_with_no_proxy(self):
 
727
        self.not_proxied_in_env({'all_proxy': self.proxy_url,
 
728
                                 'no_proxy': self.no_proxy_host})
 
729
 
 
730
    def test_ALL_PROXY_with_NO_PROXY(self):
 
731
        self.not_proxied_in_env({'ALL_PROXY': self.proxy_url,
 
732
                                 'NO_PROXY': self.no_proxy_host})
 
733
 
 
734
    def test_http_proxy_without_scheme(self):
 
735
        if self._testing_pycurl():
 
736
            # pycurl *ignores* invalid proxy env variables. If that ever change
 
737
            # in the future, this test will fail indicating that pycurl do not
 
738
            # ignore anymore such variables.
 
739
            self.not_proxied_in_env({'http_proxy': self.proxy_address})
 
740
        else:
 
741
            self.assertRaises(errors.InvalidURL,
 
742
                              self.proxied_in_env,
 
743
                              {'http_proxy': self.proxy_address})
 
744
 
 
745
 
 
746
class TestHTTPRedirections(http_utils.TestCaseWithRedirectedWebserver):
 
747
    """Test redirection between http servers."""
 
748
 
 
749
    def create_transport_secondary_server(self):
 
750
        """Create the secondary server redirecting to the primary server"""
 
751
        new = self.get_readonly_server()
 
752
 
 
753
        redirecting = http_utils.HTTPServerRedirecting()
 
754
        redirecting.redirect_to(new.host, new.port)
 
755
        return redirecting
 
756
 
 
757
    def setUp(self):
 
758
        super(TestHTTPRedirections, self).setUp()
 
759
        self.build_tree_contents([('a', '0123456789'),
 
760
                                  ('bundle',
 
761
                                  '# Bazaar revision bundle v0.9\n#\n')
 
762
                                  ],)
 
763
 
 
764
        self.old_transport = self._transport(self.old_server.get_url())
 
765
 
 
766
    def test_redirected(self):
 
767
        self.assertRaises(errors.RedirectRequested, self.old_transport.get, 'a')
 
768
        t = self._transport(self.new_server.get_url())
 
769
        self.assertEqual('0123456789', t.get('a').read())
 
770
 
 
771
    def test_read_redirected_bundle_from_url(self):
 
772
        from bzrlib.bundle import read_bundle_from_url
 
773
        url = self.old_transport.abspath('bundle')
 
774
        bundle = read_bundle_from_url(url)
 
775
        # If read_bundle_from_url was successful we get an empty bundle
 
776
        self.assertEqual([], bundle.revisions)
 
777
 
 
778
 
 
779
class RedirectedRequest(_urllib2_wrappers.Request):
 
780
    """Request following redirections. """
 
781
 
 
782
    init_orig = _urllib2_wrappers.Request.__init__
 
783
 
 
784
    def __init__(self, method, url, *args, **kwargs):
 
785
        """Constructor.
 
786
 
 
787
        """
 
788
        # Since the tests using this class will replace
 
789
        # _urllib2_wrappers.Request, we can't just call the base class __init__
 
790
        # or we'll loop.
 
791
        RedirectedRequest.init_orig(self, method, url, args, kwargs)
 
792
        self.follow_redirections = True
 
793
 
 
794
 
 
795
class TestHTTPSilentRedirections(http_utils.TestCaseWithRedirectedWebserver):
 
796
    """Test redirections.
 
797
 
 
798
    http implementations do not redirect silently anymore (they
 
799
    do not redirect at all in fact). The mechanism is still in
 
800
    place at the _urllib2_wrappers.Request level and these tests
 
801
    exercise it.
 
802
 
 
803
    For the pycurl implementation
 
804
    the redirection have been deleted as we may deprecate pycurl
 
805
    and I have no place to keep a working implementation.
 
806
    -- vila 20070212
 
807
    """
 
808
 
 
809
    def setUp(self):
 
810
        if pycurl_present and self._transport == PyCurlTransport:
 
811
            raise tests.TestNotApplicable(
 
812
                "pycurl doesn't redirect silently annymore")
 
813
        super(TestHTTPSilentRedirections, self).setUp()
 
814
        self.setup_redirected_request()
 
815
        self.addCleanup(self.cleanup_redirected_request)
 
816
        self.build_tree_contents([('a','a'),
 
817
                                  ('1/',),
 
818
                                  ('1/a', 'redirected once'),
 
819
                                  ('2/',),
 
820
                                  ('2/a', 'redirected twice'),
 
821
                                  ('3/',),
 
822
                                  ('3/a', 'redirected thrice'),
 
823
                                  ('4/',),
 
824
                                  ('4/a', 'redirected 4 times'),
 
825
                                  ('5/',),
 
826
                                  ('5/a', 'redirected 5 times'),
 
827
                                  ],)
 
828
 
 
829
        self.old_transport = self._transport(self.old_server.get_url())
 
830
 
 
831
    def setup_redirected_request(self):
 
832
        self.original_class = _urllib2_wrappers.Request
 
833
        _urllib2_wrappers.Request = RedirectedRequest
 
834
 
 
835
    def cleanup_redirected_request(self):
 
836
        _urllib2_wrappers.Request = self.original_class
 
837
 
 
838
    def create_transport_secondary_server(self):
 
839
        """Create the secondary server, redirections are defined in the tests"""
 
840
        return http_utils.HTTPServerRedirecting()
 
841
 
 
842
    def test_one_redirection(self):
 
843
        t = self.old_transport
 
844
 
 
845
        req = RedirectedRequest('GET', t.abspath('a'))
 
846
        req.follow_redirections = True
 
847
        new_prefix = 'http://%s:%s' % (self.new_server.host,
 
848
                                       self.new_server.port)
 
849
        self.old_server.redirections = \
 
850
            [('(.*)', r'%s/1\1' % (new_prefix), 301),]
 
851
        self.assertEquals('redirected once',t._perform(req).read())
 
852
 
 
853
    def test_five_redirections(self):
 
854
        t = self.old_transport
 
855
 
 
856
        req = RedirectedRequest('GET', t.abspath('a'))
 
857
        req.follow_redirections = True
 
858
        old_prefix = 'http://%s:%s' % (self.old_server.host,
 
859
                                       self.old_server.port)
 
860
        new_prefix = 'http://%s:%s' % (self.new_server.host,
 
861
                                       self.new_server.port)
 
862
        self.old_server.redirections = \
 
863
            [('/1(.*)', r'%s/2\1' % (old_prefix), 302),
 
864
             ('/2(.*)', r'%s/3\1' % (old_prefix), 303),
 
865
             ('/3(.*)', r'%s/4\1' % (old_prefix), 307),
 
866
             ('/4(.*)', r'%s/5\1' % (new_prefix), 301),
 
867
             ('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
 
868
             ]
 
869
        self.assertEquals('redirected 5 times',t._perform(req).read())
 
870
 
 
871
 
 
872
class TestDoCatchRedirections(http_utils.TestCaseWithRedirectedWebserver):
 
873
    """Test transport.do_catching_redirections."""
 
874
 
 
875
    def setUp(self):
 
876
        super(TestDoCatchRedirections, self).setUp()
 
877
        self.build_tree_contents([('a', '0123456789'),],)
 
878
 
 
879
        self.old_transport = self._transport(self.old_server.get_url())
 
880
 
 
881
    def get_a(self, transport):
 
882
        return transport.get('a')
 
883
 
 
884
    def test_no_redirection(self):
 
885
        t = self._transport(self.new_server.get_url())
 
886
 
 
887
        # We use None for redirected so that we fail if redirected
 
888
        self.assertEquals('0123456789',
 
889
                          transport.do_catching_redirections(
 
890
                self.get_a, t, None).read())
 
891
 
 
892
    def test_one_redirection(self):
 
893
        self.redirections = 0
 
894
 
 
895
        def redirected(transport, exception, redirection_notice):
 
896
            self.redirections += 1
 
897
            dir, file = urlutils.split(exception.target)
 
898
            return self._transport(dir)
 
899
 
 
900
        self.assertEquals('0123456789',
 
901
                          transport.do_catching_redirections(
 
902
                self.get_a, self.old_transport, redirected).read())
 
903
        self.assertEquals(1, self.redirections)
 
904
 
 
905
    def test_redirection_loop(self):
 
906
 
 
907
        def redirected(transport, exception, redirection_notice):
 
908
            # By using the redirected url as a base dir for the
 
909
            # *old* transport, we create a loop: a => a/a =>
 
910
            # a/a/a
 
911
            return self.old_transport.clone(exception.target)
 
912
 
 
913
        self.assertRaises(errors.TooManyRedirections,
 
914
                          transport.do_catching_redirections,
 
915
                          self.get_a, self.old_transport, redirected)
 
916
 
 
917
 
 
918
class TestAuth(http_utils.TestCaseWithWebserver):
 
919
    """Test authentication scheme"""
 
920
 
 
921
    _auth_header = 'Authorization'
 
922
    _password_prompt_prefix = ''
 
923
 
 
924
    def setUp(self):
 
925
        super(TestAuth, self).setUp()
 
926
        self.server = self.get_readonly_server()
 
927
        self.build_tree_contents([('a', 'contents of a\n'),
 
928
                                  ('b', 'contents of b\n'),])
 
929
 
 
930
    def create_transport_readonly_server(self):
 
931
        if self._auth_scheme == 'basic':
 
932
            server = http_utils.HTTPBasicAuthServer()
 
933
        else:
 
934
            if self._auth_scheme != 'digest':
 
935
                raise AssertionError('Unknown auth scheme: %r'
 
936
                                     % self._auth_scheme)
 
937
            server = http_utils.HTTPDigestAuthServer()
 
938
        return server
 
939
 
 
940
    def get_user_url(self, user=None, password=None):
 
941
        """Build an url embedding user and password"""
 
942
        url = '%s://' % self.server._url_protocol
 
943
        if user is not None:
 
944
            url += user
 
945
            if password is not None:
 
946
                url += ':' + password
 
947
            url += '@'
 
948
        url += '%s:%s/' % (self.server.host, self.server.port)
 
949
        return url
 
950
 
 
951
    def get_user_transport(self, user=None, password=None):
 
952
        return self._transport(self.get_user_url(user, password))
 
953
 
 
954
    def test_no_user(self):
 
955
        self.server.add_user('joe', 'foo')
 
956
        t = self.get_user_transport()
 
957
        self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
 
958
        # Only one 'Authentication Required' error should occur
 
959
        self.assertEqual(1, self.server.auth_required_errors)
 
960
 
 
961
    def test_empty_pass(self):
 
962
        self.server.add_user('joe', '')
 
963
        t = self.get_user_transport('joe', '')
 
964
        self.assertEqual('contents of a\n', t.get('a').read())
 
965
        # Only one 'Authentication Required' error should occur
 
966
        self.assertEqual(1, self.server.auth_required_errors)
 
967
 
 
968
    def test_user_pass(self):
 
969
        self.server.add_user('joe', 'foo')
 
970
        t = self.get_user_transport('joe', 'foo')
 
971
        self.assertEqual('contents of a\n', t.get('a').read())
 
972
        # Only one 'Authentication Required' error should occur
 
973
        self.assertEqual(1, self.server.auth_required_errors)
 
974
 
 
975
    def test_unknown_user(self):
 
976
        self.server.add_user('joe', 'foo')
 
977
        t = self.get_user_transport('bill', 'foo')
 
978
        self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
 
979
        # Two 'Authentication Required' errors should occur (the
 
980
        # initial 'who are you' and 'I don't know you, who are
 
981
        # you').
 
982
        self.assertEqual(2, self.server.auth_required_errors)
 
983
 
 
984
    def test_wrong_pass(self):
 
985
        self.server.add_user('joe', 'foo')
 
986
        t = self.get_user_transport('joe', 'bar')
 
987
        self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
 
988
        # Two 'Authentication Required' errors should occur (the
 
989
        # initial 'who are you' and 'this is not you, who are you')
 
990
        self.assertEqual(2, self.server.auth_required_errors)
 
991
 
 
992
    def test_prompt_for_password(self):
 
993
        self.server.add_user('joe', 'foo')
 
994
        t = self.get_user_transport('joe', None)
 
995
        stdout = tests.StringIOWrapper()
 
996
        ui.ui_factory = tests.TestUIFactory(stdin='foo\n', stdout=stdout)
 
997
        self.assertEqual('contents of a\n',t.get('a').read())
 
998
        # stdin should be empty
 
999
        self.assertEqual('', ui.ui_factory.stdin.readline())
 
1000
        self._check_password_prompt(t._unqualified_scheme, 'joe',
 
1001
                                    stdout.getvalue())
 
1002
        # And we shouldn't prompt again for a different request
 
1003
        # against the same transport.
 
1004
        self.assertEqual('contents of b\n',t.get('b').read())
 
1005
        t2 = t.clone()
 
1006
        # And neither against a clone
 
1007
        self.assertEqual('contents of b\n',t2.get('b').read())
 
1008
        # Only one 'Authentication Required' error should occur
 
1009
        self.assertEqual(1, self.server.auth_required_errors)
 
1010
 
 
1011
    def _check_password_prompt(self, scheme, user, actual_prompt):
 
1012
        expected_prompt = (self._password_prompt_prefix
 
1013
                           + ("%s %s@%s:%d, Realm: '%s' password: "
 
1014
                              % (scheme.upper(),
 
1015
                                 user, self.server.host, self.server.port,
 
1016
                                 self.server.auth_realm)))
 
1017
        self.assertEquals(expected_prompt, actual_prompt)
 
1018
 
 
1019
    def test_no_prompt_for_password_when_using_auth_config(self):
 
1020
        user =' joe'
 
1021
        password = 'foo'
 
1022
        stdin_content = 'bar\n'  # Not the right password
 
1023
        self.server.add_user(user, password)
 
1024
        t = self.get_user_transport(user, None)
 
1025
        ui.ui_factory = tests.TestUIFactory(stdin=stdin_content,
 
1026
                                            stdout=tests.StringIOWrapper())
 
1027
        # Create a minimal config file with the right password
 
1028
        conf = config.AuthenticationConfig()
 
1029
        conf._get_config().update(
 
1030
            {'httptest': {'scheme': 'http', 'port': self.server.port,
 
1031
                          'user': user, 'password': password}})
 
1032
        conf._save()
 
1033
        # Issue a request to the server to connect
 
1034
        self.assertEqual('contents of a\n',t.get('a').read())
 
1035
        # stdin should have  been left untouched
 
1036
        self.assertEqual(stdin_content, ui.ui_factory.stdin.readline())
 
1037
        # Only one 'Authentication Required' error should occur
 
1038
        self.assertEqual(1, self.server.auth_required_errors)
 
1039
 
 
1040
 
 
1041
 
 
1042
class TestProxyAuth(TestAuth):
 
1043
    """Test proxy authentication schemes."""
 
1044
 
 
1045
    _auth_header = 'Proxy-authorization'
 
1046
    _password_prompt_prefix='Proxy '
 
1047
 
 
1048
    def setUp(self):
 
1049
        super(TestProxyAuth, self).setUp()
 
1050
        self._old_env = {}
 
1051
        self.addCleanup(self._restore_env)
 
1052
        # Override the contents to avoid false positives
 
1053
        self.build_tree_contents([('a', 'not proxied contents of a\n'),
 
1054
                                  ('b', 'not proxied contents of b\n'),
 
1055
                                  ('a-proxied', 'contents of a\n'),
 
1056
                                  ('b-proxied', 'contents of b\n'),
 
1057
                                  ])
 
1058
 
 
1059
    def create_transport_readonly_server(self):
 
1060
        if self._auth_scheme == 'basic':
 
1061
            server = http_utils.ProxyBasicAuthServer()
 
1062
        else:
 
1063
            if self._auth_scheme != 'digest':
 
1064
                raise AssertionError('Unknown auth scheme: %r'
 
1065
                                     % self._auth_scheme)
 
1066
            server = http_utils.ProxyDigestAuthServer()
 
1067
        return server
 
1068
 
 
1069
    def get_user_transport(self, user=None, password=None):
 
1070
        self._install_env({'all_proxy': self.get_user_url(user, password)})
 
1071
        return self._transport(self.server.get_url())
 
1072
 
 
1073
    def _install_env(self, env):
 
1074
        for name, value in env.iteritems():
 
1075
            self._old_env[name] = osutils.set_or_unset_env(name, value)
 
1076
 
 
1077
    def _restore_env(self):
 
1078
        for name, value in self._old_env.iteritems():
 
1079
            osutils.set_or_unset_env(name, value)
 
1080
 
 
1081