~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_http.py

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

• reverse the comparison (6499.3.1 to 6005)
• stop comparing with revision 6005

Show diffs side-by-side

added

removed

bzrlib/tests/test_http.py

31

31

 

32

32

import bzrlib

33

33

from bzrlib import (

34

 

    bzrdir,

35

 

    cethread,

36

34

    config,

 

35

    controldir,

37

36

    debug,

38

37

    errors,

39

38

    osutils,

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.'))

1048

1055

        self.assertEqual('single', t._range_hint)

1049

1056

 

1050

1057

 

 

1058

class TruncatedBeforeBoundaryRequestHandler(

 

1059

    http_server.TestingHTTPRequestHandler):

 

1060

    """Truncation before a boundary, like in bug 198646"""

 

1061

 

 

1062

    _truncated_ranges = 1

 

1063

 

 

1064

    def get_multiple_ranges(self, file, file_size, ranges):

 

1065

        self.send_response(206)

 

1066

        self.send_header('Accept-Ranges', 'bytes')

 

1067

        boundary = 'tagada'

 

1068

        self.send_header('Content-Type',

 

1069

                         'multipart/byteranges; boundary=%s' % boundary)

 

1070

        boundary_line = '--%s\r\n' % boundary

 

1071

        # Calculate the Content-Length

 

1072

        content_length = 0

 

1073

        for (start, end) in ranges:

 

1074

            content_length += len(boundary_line)

 

1075

            content_length += self._header_line_length(

 

1076

                'Content-type', 'application/octet-stream')

 

1077

            content_length += self._header_line_length(

 

1078

                'Content-Range', 'bytes %d-%d/%d' % (start, end, file_size))

 

1079

            content_length += len('\r\n') # end headers

 

1080

            content_length += end - start # + 1

 

1081

        content_length += len(boundary_line)

 

1082

        self.send_header('Content-length', content_length)

 

1083

        self.end_headers()

 

1084

 

 

1085

        # Send the multipart body

 

1086

        cur = 0

 

1087

        for (start, end) in ranges:

 

1088

            if cur + self._truncated_ranges >= len(ranges):

 

1089

                # Abruptly ends the response and close the connection

 

1090

                self.close_connection = 1

 

1091

                return

 

1092

            self.wfile.write(boundary_line)

 

1093

            self.send_header('Content-type', 'application/octet-stream')

 

1094

            self.send_header('Content-Range', 'bytes %d-%d/%d'

 

1095

                             % (start, end, file_size))

 

1096

            self.end_headers()

 

1097

            self.send_range_content(file, start, end - start + 1)

 

1098

            cur += 1

 

1099

        # Final boundary

 

1100

        self.wfile.write(boundary_line)

 

1101

 

 

1102

 

 

1103

class TestTruncatedBeforeBoundary(TestSpecificRequestHandler):

 

1104

    """Tests the case of bug 198646, disconnecting before a boundary."""

 

1105

 

 

1106

    _req_handler_class = TruncatedBeforeBoundaryRequestHandler

 

1107

 

 

1108

    def setUp(self):

 

1109

        super(TestTruncatedBeforeBoundary, self).setUp()

 

1110

        self.build_tree_contents([('a', '0123456789')],)

 

1111

 

 

1112

    def test_readv_with_short_reads(self):

 

1113

        server = self.get_readonly_server()

 

1114

        t = self.get_readonly_transport()

 

1115

        # Force separate ranges for each offset

 

1116

        t._bytes_to_read_before_seek = 0

 

1117

        ireadv = iter(t.readv('a', ((0, 1), (2, 1), (4, 2), (9, 1))))

 

1118

        self.assertEqual((0, '0'), ireadv.next())

 

1119

        self.assertEqual((2, '2'), ireadv.next())

 

1120

        self.assertEqual((4, '45'), ireadv.next())

 

1121

        self.assertEqual((9, '9'), ireadv.next())

 

1122

 

 

1123

 

1051

1124

class LimitedRangeRequestHandler(http_server.TestingHTTPRequestHandler):

1052

1125

    """Errors out when range specifiers exceed the limit"""

1053

1126

 

1569

1642

        return url

1570

1643

 

1571

1644

    def get_user_transport(self, user, password):

1572

 

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

 

1645

        t = transport.get_transport_from_url(

 

1646

            self.get_user_url(user, password))

1573

1647

        return t

1574

1648

 

1575

1649

    def test_no_user(self):

1701

1775

                                     http_utils.ProxyDigestAuthServer):

1702

1776

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

1703

1777

        if self._testing_pycurl():

1704

 

            raise tests.KnownFailure(

 

1778

            self.knownFailure(

1705

1779

                'pycurl does not handle a nonce change')

1706

1780

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

1707

1781

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

1789

1863

        if self._testing_pycurl():

1790

1864

            import pycurl

1791

1865

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

1792

 

                raise tests.KnownFailure(

 

1866

                self.knownFailure(

1793

1867

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

1794

1868

        super(TestProxyAuth, self).test_empty_pass()

1795

1869

 

1837

1911

        server._url_protocol = self._url_protocol

1838

1912

        return server

1839

1913

 

1840

 

    def test_open_bzrdir(self):

 

1914

    def test_open_controldir(self):

1841

1915

        branch = self.make_branch('relpath')

1842

1916

        url = self.http_server.get_url() + 'relpath'

1843

 

        bd = bzrdir.BzrDir.open(url)

 

1917

        bd = controldir.ControlDir.open(url)

1844

1918

        self.addCleanup(bd.transport.disconnect)

1845

1919

        self.assertIsInstance(bd, _mod_remote.RemoteBzrDir)

1846

1920

 

1849

1923

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

1850

1924

        # bulk data, so we use that.

1851

1925

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

1852

 

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

 

1926

        http_transport = transport.get_transport_from_url(

 

1927

            self.http_server.get_url())

1853

1928

        medium = http_transport.get_smart_medium()

1854

1929

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

1855

1930

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

1863

1938

        post_body = 'hello\n'

1864

1939

        expected_reply_body = 'ok\x012\n'

1865

1940

 

1866

 

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

 

1941

        http_transport = transport.get_transport_from_url(

 

1942

            self.http_server.get_url())

1867

1943

        medium = http_transport.get_smart_medium()

1868

1944

        response = medium.send_http_smart_request(post_body)

1869

1945

        reply_body = response.read()

1927

2003

        self.assertIsInstance(r, type(t))

1928

2004

        # Both transports share the some connection

1929

2005

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

 

2006

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

1930

2007

 

1931

2008

    def test_redirected_to_self_with_slash(self):

1932

2009

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

1943

2020

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

1944

2021

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

1945

2022

        self.assertIsInstance(r, type(t))

 

2023

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

 

2024

            r.external_url())

1946

2025

 

1947

2026

    def test_redirected_to_same_host_sibling_protocol(self):

1948

2027

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

1949

2028

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

1950

2029

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

1951

2030

        self.assertIsInstance(r, type(t))

 

2031

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

 

2032

            r.external_url())

1952

2033

 

1953

2034

    def test_redirected_to_same_host_different_protocol(self):

1954

2035

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

1955

2036

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

1956

2037

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

1957

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

1958

2046

 

1959

2047

    def test_redirected_to_different_host_same_user(self):

1960

2048

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

1961

2049

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

1962

2050

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

1963

2051

        self.assertIsInstance(r, type(t))

1964

 

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

1965

2054

 

1966

2055

 

1967

2056

class PredefinedRequestHandler(http_server.TestingHTTPRequestHandler):

2020

2109

    pass

2021

2110

 

2022

2111

 

2023

 

if tests.HTTPSServerFeature.available():

 

2112

if features.HTTPSServerFeature.available():

2024

2113

    from bzrlib.tests import https_server

2025

2114

    class ActivityHTTPSServer(ActivityServerMixin, https_server.HTTPSServer):

2026

2115

        pass

2037

2126

        tests.TestCase.setUp(self)

2038

2127

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

2039

2128

        self.server.start_server()

 

2129

        self.addCleanup(self.server.stop_server)

2040

2130

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

2041

2131

        def report_activity(t, bytes, direction):

2042

2132

            count = _activities.get(direction, 0)

2043

2133

            count += bytes

2044

2134

            _activities[direction] = count

2045

2135

        self.activities = _activities

2046

 

 

2047

2136

        # We override at class level because constructors may propagate the

2048

2137

        # bound method and render instance overriding ineffective (an

2049

2138

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

2050

2139

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

2051

 

        self.addCleanup(self.server.stop_server)

2052

2140

 

2053

2141

    def get_transport(self):

2054

2142

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

• Older »

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