← 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: Jelmer Vernooij
  • Date: 2012-01-27 21:28:56 UTC
  • mto: This revision was merged to the branch mainline in revision 6460.
  • Revision ID: jelmer@samba.org-20120127212856-ewnjgn7fyblphcqw
Migrate mail_client to config stacks.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_http.py
32
32
import bzrlib
33
33
from bzrlib import (
34
34
    bzrdir,
35
 
    cethread,
36
35
    config,
37
36
    debug,
38
37
    errors,
128
127
        ('urllib,http', dict(_activity_server=ActivityHTTPServer,
129
128
                            _transport=_urllib.HttpTransport_urllib,)),
130
129
        ]
131
 
    if tests.HTTPSServerFeature.available():
132
 
        activity_scenarios.append(
133
 
            ('urllib,https', dict(_activity_server=ActivityHTTPSServer,
134
 
                                _transport=_urllib.HttpTransport_urllib,)),)
135
130
    if features.pycurl.available():
136
131
        activity_scenarios.append(
137
132
            ('pycurl,http', dict(_activity_server=ActivityHTTPServer,
138
133
                                _transport=PyCurlTransport,)),)
139
 
        if tests.HTTPSServerFeature.available():
140
 
            from bzrlib.tests import (
141
 
                ssl_certs,
142
 
                )
143
 
            # FIXME: Until we have a better way to handle self-signed
144
 
            # certificates (like allowing them in a test specific
145
 
            # authentication.conf for example), we need some specialized pycurl
146
 
            # transport for tests.
 
134
    if features.HTTPSServerFeature.available():
 
135
        # FIXME: Until we have a better way to handle self-signed certificates
 
136
        # (like allowing them in a test specific authentication.conf for
 
137
        # example), we need some specialized pycurl/urllib transport for tests.
 
138
        # -- vila 2012-01-20
 
139
        from bzrlib.tests import (
 
140
            ssl_certs,
 
141
            )
 
142
        class HTTPS_urllib_transport(_urllib.HttpTransport_urllib):
 
143
 
 
144
            def __init__(self, base, _from_transport=None):
 
145
                super(HTTPS_urllib_transport, self).__init__(
 
146
                    base, _from_transport=_from_transport,
 
147
                    ca_certs=ssl_certs.build_path('ca.crt'))
 
148
 
 
149
        activity_scenarios.append(
 
150
            ('urllib,https', dict(_activity_server=ActivityHTTPSServer,
 
151
                                  _transport=HTTPS_urllib_transport,)),)
 
152
        if features.pycurl.available():
147
153
            class HTTPS_pycurl_transport(PyCurlTransport):
148
154
 
149
155
                def __init__(self, base, _from_transport=None):
533
539
    scenarios = vary_by_http_client_implementation()
534
540
 
535
541
    def test_http_registered(self):
536
 
        t = transport.get_transport('%s://foo.com/' % self._url_protocol)
 
542
        t = transport.get_transport_from_url(
 
543
            '%s://foo.com/' % self._url_protocol)
537
544
        self.assertIsInstance(t, transport.Transport)
538
545
        self.assertIsInstance(t, self._transport)
539
546
 
551
558
        self.start_server(server)
552
559
        url = server.get_url()
553
560
        # FIXME: needs a cleanup -- vila 20100611
554
 
        http_transport = transport.get_transport(url)
 
561
        http_transport = transport.get_transport_from_url(url)
555
562
        code, response = http_transport._post('abc def end-of-body')
556
563
        self.assertTrue(
557
564
            server.received_bytes.startswith('POST /.bzr/smart HTTP/1.'))
1635
1642
        return url
1636
1643
 
1637
1644
    def get_user_transport(self, user, password):
1638
 
        t = transport.get_transport(self.get_user_url(user, password))
 
1645
        t = transport.get_transport_from_url(
 
1646
            self.get_user_url(user, password))
1639
1647
        return t
1640
1648
 
1641
1649
    def test_no_user(self):
1767
1775
                                     http_utils.ProxyDigestAuthServer):
1768
1776
            raise tests.TestNotApplicable('HTTP/proxy auth digest only test')
1769
1777
        if self._testing_pycurl():
1770
 
            raise tests.KnownFailure(
 
1778
            self.knownFailure(
1771
1779
                'pycurl does not handle a nonce change')
1772
1780
        self.server.add_user('joe', 'foo')
1773
1781
        t = self.get_user_transport('joe', 'foo')
1855
1863
        if self._testing_pycurl():
1856
1864
            import pycurl
1857
1865
            if pycurl.version_info()[1] < '7.16.0':
1858
 
                raise tests.KnownFailure(
 
1866
                self.knownFailure(
1859
1867
                    'pycurl < 7.16.0 does not handle empty proxy passwords')
1860
1868
        super(TestProxyAuth, self).test_empty_pass()
1861
1869
 
1915
1923
        # The 'readv' command in the smart protocol both sends and receives
1916
1924
        # bulk data, so we use that.
1917
1925
        self.build_tree(['data-file'])
1918
 
        http_transport = transport.get_transport(self.http_server.get_url())
 
1926
        http_transport = transport.get_transport_from_url(
 
1927
            self.http_server.get_url())
1919
1928
        medium = http_transport.get_smart_medium()
1920
1929
        # Since we provide the medium, the url below will be mostly ignored
1921
1930
        # during the test, as long as the path is '/'.
1929
1938
        post_body = 'hello\n'
1930
1939
        expected_reply_body = 'ok\x012\n'
1931
1940
 
1932
 
        http_transport = transport.get_transport(self.http_server.get_url())
 
1941
        http_transport = transport.get_transport_from_url(
 
1942
            self.http_server.get_url())
1933
1943
        medium = http_transport.get_smart_medium()
1934
1944
        response = medium.send_http_smart_request(post_body)
1935
1945
        reply_body = response.read()
1993
2003
        self.assertIsInstance(r, type(t))
1994
2004
        # Both transports share the some connection
1995
2005
        self.assertEqual(t._get_connection(), r._get_connection())
 
2006
        self.assertEquals('http://www.example.com/foo/subdir/', r.base)
1996
2007
 
1997
2008
    def test_redirected_to_self_with_slash(self):
1998
2009
        t = self._transport('http://www.example.com/foo')
2009
2020
        r = t._redirected_to('http://www.example.com/foo',
2010
2021
                             'http://foo.example.com/foo/subdir')
2011
2022
        self.assertIsInstance(r, type(t))
 
2023
        self.assertEquals('http://foo.example.com/foo/subdir/',
 
2024
            r.external_url())
2012
2025
 
2013
2026
    def test_redirected_to_same_host_sibling_protocol(self):
2014
2027
        t = self._transport('http://www.example.com/foo')
2015
2028
        r = t._redirected_to('http://www.example.com/foo',
2016
2029
                             'https://www.example.com/foo')
2017
2030
        self.assertIsInstance(r, type(t))
 
2031
        self.assertEquals('https://www.example.com/foo/',
 
2032
            r.external_url())
2018
2033
 
2019
2034
    def test_redirected_to_same_host_different_protocol(self):
2020
2035
        t = self._transport('http://www.example.com/foo')
2021
2036
        r = t._redirected_to('http://www.example.com/foo',
2022
2037
                             'ftp://www.example.com/foo')
2023
2038
        self.assertNotEquals(type(r), type(t))
 
2039
        self.assertEquals('ftp://www.example.com/foo/', r.external_url())
 
2040
 
 
2041
    def test_redirected_to_same_host_specific_implementation(self):
 
2042
        t = self._transport('http://www.example.com/foo')
 
2043
        r = t._redirected_to('http://www.example.com/foo',
 
2044
                             'https+urllib://www.example.com/foo')
 
2045
        self.assertEquals('https://www.example.com/foo/', r.external_url())
2024
2046
 
2025
2047
    def test_redirected_to_different_host_same_user(self):
2026
2048
        t = self._transport('http://joe@www.example.com/foo')
2027
2049
        r = t._redirected_to('http://www.example.com/foo',
2028
2050
                             'https://foo.example.com/foo')
2029
2051
        self.assertIsInstance(r, type(t))
2030
 
        self.assertEqual(t._user, r._user)
 
2052
        self.assertEqual(t._parsed_url.user, r._parsed_url.user)
 
2053
        self.assertEquals('https://joe@foo.example.com/foo/', r.external_url())
2031
2054
 
2032
2055
 
2033
2056
class PredefinedRequestHandler(http_server.TestingHTTPRequestHandler):
2086
2109
    pass
2087
2110
 
2088
2111
 
2089
 
if tests.HTTPSServerFeature.available():
 
2112
if features.HTTPSServerFeature.available():
2090
2113
    from bzrlib.tests import https_server
2091
2114
    class ActivityHTTPSServer(ActivityServerMixin, https_server.HTTPSServer):
2092
2115
        pass
2103
2126
        tests.TestCase.setUp(self)
2104
2127
        self.server = self._activity_server(self._protocol_version)
2105
2128
        self.server.start_server()
 
2129
        self.addCleanup(self.server.stop_server)
2106
2130
        _activities = {} # Don't close over self and create a cycle
2107
2131
        def report_activity(t, bytes, direction):
2108
2132
            count = _activities.get(direction, 0)
2109
2133
            count += bytes
2110
2134
            _activities[direction] = count
2111
2135
        self.activities = _activities
2112
 
 
2113
2136
        # We override at class level because constructors may propagate the
2114
2137
        # bound method and render instance overriding ineffective (an
2115
2138
        # alternative would be to define a specific ui factory instead...)
2116
2139
        self.overrideAttr(self._transport, '_report_activity', report_activity)
2117
 
        self.addCleanup(self.server.stop_server)
2118
2140
 
2119
2141
    def get_transport(self):
2120
2142
        t = self._transport(self.server.get_url())