~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_http.py

• browse files at revision 6015.59.1
• compare with another revision
• download diff
• download tarball
• view history from revision 6015.59.1

• reverse the comparison (6015.59.1 to 6449.5.3)
• stop comparing with revision 6449.5.3

Show diffs side-by-side

added

removed

bzrlib/tests/test_http.py

32

32

import bzrlib

33

33

from bzrlib import (

34

34

    bzrdir,

 

35

    cethread,

35

36

    config,

36

37

    debug,

37

38

    errors,

127

128

        ('urllib,http', dict(_activity_server=ActivityHTTPServer,

128

129

                            _transport=_urllib.HttpTransport_urllib,)),

129

130

        ]

 

131

    if tests.HTTPSServerFeature.available():

 

132

        activity_scenarios.append(

 

133

            ('urllib,https', dict(_activity_server=ActivityHTTPSServer,

 

134

                                _transport=_urllib.HttpTransport_urllib,)),)

130

135

    if features.pycurl.available():

131

136

        activity_scenarios.append(

132

137

            ('pycurl,http', dict(_activity_server=ActivityHTTPServer,

133

138

                                _transport=PyCurlTransport,)),)

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():

 

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.

153

147

            class HTTPS_pycurl_transport(PyCurlTransport):

154

148

 

155

149

                def __init__(self, base, _from_transport=None):

539

533

    scenarios = vary_by_http_client_implementation()

540

534

 

541

535

    def test_http_registered(self):

542

 

        t = transport.get_transport_from_url(

543

 

            '%s://foo.com/' % self._url_protocol)

 

536

        t = transport.get_transport('%s://foo.com/' % self._url_protocol)

544

537

        self.assertIsInstance(t, transport.Transport)

545

538

        self.assertIsInstance(t, self._transport)

546

539

 

558

551

        self.start_server(server)

559

552

        url = server.get_url()

560

553

        # FIXME: needs a cleanup -- vila 20100611

561

 

        http_transport = transport.get_transport_from_url(url)

 

554

        http_transport = transport.get_transport(url)

562

555

        code, response = http_transport._post('abc def end-of-body')

563

556

        self.assertTrue(

564

557

            server.received_bytes.startswith('POST /.bzr/smart HTTP/1.'))

1642

1635

        return url

1643

1636

 

1644

1637

    def get_user_transport(self, user, password):

1645

 

        t = transport.get_transport_from_url(

1646

 

            self.get_user_url(user, password))

 

1638

        t = transport.get_transport(self.get_user_url(user, password))

1647

1639

        return t

1648

1640

 

1649

1641

    def test_no_user(self):

1775

1767

                                     http_utils.ProxyDigestAuthServer):

1776

1768

            raise tests.TestNotApplicable('HTTP/proxy auth digest only test')

1777

1769

        if self._testing_pycurl():

1778

 

            self.knownFailure(

 

1770

            raise tests.KnownFailure(

1779

1771

                'pycurl does not handle a nonce change')

1780

1772

        self.server.add_user('joe', 'foo')

1781

1773

        t = self.get_user_transport('joe', 'foo')

1863

1855

        if self._testing_pycurl():

1864

1856

            import pycurl

1865

1857

            if pycurl.version_info()[1] < '7.16.0':

1866

 

                self.knownFailure(

 

1858

                raise tests.KnownFailure(

1867

1859

                    'pycurl < 7.16.0 does not handle empty proxy passwords')

1868

1860

        super(TestProxyAuth, self).test_empty_pass()

1869

1861

 

1923

1915

        # The 'readv' command in the smart protocol both sends and receives

1924

1916

        # bulk data, so we use that.

1925

1917

        self.build_tree(['data-file'])

1926

 

        http_transport = transport.get_transport_from_url(

1927

 

            self.http_server.get_url())

 

1918

        http_transport = transport.get_transport(self.http_server.get_url())

1928

1919

        medium = http_transport.get_smart_medium()

1929

1920

        # Since we provide the medium, the url below will be mostly ignored

1930

1921

        # during the test, as long as the path is '/'.

1938

1929

        post_body = 'hello\n'

1939

1930

        expected_reply_body = 'ok\x012\n'

1940

1931

 

1941

 

        http_transport = transport.get_transport_from_url(

1942

 

            self.http_server.get_url())

 

1932

        http_transport = transport.get_transport(self.http_server.get_url())

1943

1933

        medium = http_transport.get_smart_medium()

1944

1934

        response = medium.send_http_smart_request(post_body)

1945

1935

        reply_body = response.read()

2003

1993

        self.assertIsInstance(r, type(t))

2004

1994

        # Both transports share the some connection

2005

1995

        self.assertEqual(t._get_connection(), r._get_connection())

2006

 

        self.assertEquals('http://www.example.com/foo/subdir/', r.base)

2007

1996

 

2008

1997

    def test_redirected_to_self_with_slash(self):

2009

1998

        t = self._transport('http://www.example.com/foo')

2020

2009

        r = t._redirected_to('http://www.example.com/foo',

2021

2010

                             'http://foo.example.com/foo/subdir')

2022

2011

        self.assertIsInstance(r, type(t))

2023

 

        self.assertEquals('http://foo.example.com/foo/subdir/',

2024

 

            r.external_url())

2025

2012

 

2026

2013

    def test_redirected_to_same_host_sibling_protocol(self):

2027

2014

        t = self._transport('http://www.example.com/foo')

2028

2015

        r = t._redirected_to('http://www.example.com/foo',

2029

2016

                             'https://www.example.com/foo')

2030

2017

        self.assertIsInstance(r, type(t))

2031

 

        self.assertEquals('https://www.example.com/foo/',

2032

 

            r.external_url())

2033

2018

 

2034

2019

    def test_redirected_to_same_host_different_protocol(self):

2035

2020

        t = self._transport('http://www.example.com/foo')

2036

2021

        r = t._redirected_to('http://www.example.com/foo',

2037

2022

                             'ftp://www.example.com/foo')

2038

2023

        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())

2046

2024

 

2047

2025

    def test_redirected_to_different_host_same_user(self):

2048

2026

        t = self._transport('http://joe@www.example.com/foo')

2049

2027

        r = t._redirected_to('http://www.example.com/foo',

2050

2028

                             'https://foo.example.com/foo')

2051

2029

        self.assertIsInstance(r, type(t))

2052

 

        self.assertEqual(t._parsed_url.user, r._parsed_url.user)

2053

 

        self.assertEquals('https://joe@foo.example.com/foo/', r.external_url())

 

2030

        self.assertEqual(t._user, r._user)

2054

2031

 

2055

2032

 

2056

2033

class PredefinedRequestHandler(http_server.TestingHTTPRequestHandler):

2109

2086

    pass

2110

2087

 

2111

2088

 

2112

 

if features.HTTPSServerFeature.available():

 

2089

if tests.HTTPSServerFeature.available():

2113

2090

    from bzrlib.tests import https_server

2114

2091

    class ActivityHTTPSServer(ActivityServerMixin, https_server.HTTPSServer):

2115

2092

        pass

2126

2103

        tests.TestCase.setUp(self)

2127

2104

        self.server = self._activity_server(self._protocol_version)

2128

2105

        self.server.start_server()

2129

 

        self.addCleanup(self.server.stop_server)

2130

2106

        _activities = {} # Don't close over self and create a cycle

2131

2107

        def report_activity(t, bytes, direction):

2132

2108

            count = _activities.get(direction, 0)

2133

2109

            count += bytes

2134

2110

            _activities[direction] = count

2135

2111

        self.activities = _activities

 

2112

 

2136

2113

        # We override at class level because constructors may propagate the

2137

2114

        # bound method and render instance overriding ineffective (an

2138

2115

        # alternative would be to define a specific ui factory instead...)

2139

2116

        self.overrideAttr(self._transport, '_report_activity', report_activity)

 

2117

        self.addCleanup(self.server.stop_server)

2140

2118

 

2141

2119

    def get_transport(self):

2142

2120

        t = self._transport(self.server.get_url())

• Older »

Loggerhead is a web-based interface for Breezy
Version: 2.0.1