~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_http.py

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

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

Show diffs side-by-side

added

removed

bzrlib/tests/test_http.py

128

128

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

129

129

                            _transport=_urllib.HttpTransport_urllib,)),

130

130

        ]

131

 

    if tests.HTTPSServerFeature.available():

 

131

    if features.HTTPSServerFeature.available():

132

132

        activity_scenarios.append(

133

133

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

134

134

                                _transport=_urllib.HttpTransport_urllib,)),)

136

136

        activity_scenarios.append(

137

137

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

138

138

                                _transport=PyCurlTransport,)),)

139

 

        if tests.HTTPSServerFeature.available():

 

139

        if features.HTTPSServerFeature.available():

140

140

            from bzrlib.tests import (

141

141

                ssl_certs,

142

142

                )

533

533

    scenarios = vary_by_http_client_implementation()

534

534

 

535

535

    def test_http_registered(self):

536

 

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

 

536

        t = transport.get_transport_from_url(

 

537

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

537

538

        self.assertIsInstance(t, transport.Transport)

538

539

        self.assertIsInstance(t, self._transport)

539

540

 

551

552

        self.start_server(server)

552

553

        url = server.get_url()

553

554

        # FIXME: needs a cleanup -- vila 20100611

554

 

        http_transport = transport.get_transport(url)

 

555

        http_transport = transport.get_transport_from_url(url)

555

556

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

556

557

        self.assertTrue(

557

558

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

1048

1049

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

1049

1050

 

1050

1051

 

 

1052

class TruncatedBeforeBoundaryRequestHandler(

 

1053

    http_server.TestingHTTPRequestHandler):

 

1054

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

 

1055

 

 

1056

    _truncated_ranges = 1

 

1057

 

 

1058

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

 

1059

        self.send_response(206)

 

1060

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

 

1061

        boundary = 'tagada'

 

1062

        self.send_header('Content-Type',

 

1063

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

 

1064

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

 

1065

        # Calculate the Content-Length

 

1066

        content_length = 0

 

1067

        for (start, end) in ranges:

 

1068

            content_length += len(boundary_line)

 

1069

            content_length += self._header_line_length(

 

1070

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

 

1071

            content_length += self._header_line_length(

 

1072

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

 

1073

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

 

1074

            content_length += end - start # + 1

 

1075

        content_length += len(boundary_line)

 

1076

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

 

1077

        self.end_headers()

 

1078

 

 

1079

        # Send the multipart body

 

1080

        cur = 0

 

1081

        for (start, end) in ranges:

 

1082

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

 

1083

                # Abruptly ends the response and close the connection

 

1084

                self.close_connection = 1

 

1085

                return

 

1086

            self.wfile.write(boundary_line)

 

1087

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

 

1088

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

 

1089

                             % (start, end, file_size))

 

1090

            self.end_headers()

 

1091

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

 

1092

            cur += 1

 

1093

        # Final boundary

 

1094

        self.wfile.write(boundary_line)

 

1095

 

 

1096

 

 

1097

class TestTruncatedBeforeBoundary(TestSpecificRequestHandler):

 

1098

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

 

1099

 

 

1100

    _req_handler_class = TruncatedBeforeBoundaryRequestHandler

 

1101

 

 

1102

    def setUp(self):

 

1103

        super(TestTruncatedBeforeBoundary, self).setUp()

 

1104

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

 

1105

 

 

1106

    def test_readv_with_short_reads(self):

 

1107

        server = self.get_readonly_server()

 

1108

        t = self.get_readonly_transport()

 

1109

        # Force separate ranges for each offset

 

1110

        t._bytes_to_read_before_seek = 0

 

1111

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

 

1112

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

 

1113

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

 

1114

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

 

1115

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

 

1116

 

 

1117

 

1051

1118

class LimitedRangeRequestHandler(http_server.TestingHTTPRequestHandler):

1052

1119

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

1053

1120

 

1569

1636

        return url

1570

1637

 

1571

1638

    def get_user_transport(self, user, password):

1572

 

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

 

1639

        t = transport.get_transport_from_url(

 

1640

            self.get_user_url(user, password))

1573

1641

        return t

1574

1642

 

1575

1643

    def test_no_user(self):

1701

1769

                                     http_utils.ProxyDigestAuthServer):

1702

1770

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

1703

1771

        if self._testing_pycurl():

1704

 

            raise tests.KnownFailure(

 

1772

            self.knownFailure(

1705

1773

                'pycurl does not handle a nonce change')

1706

1774

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

1707

1775

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

1789

1857

        if self._testing_pycurl():

1790

1858

            import pycurl

1791

1859

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

1792

 

                raise tests.KnownFailure(

 

1860

                self.knownFailure(

1793

1861

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

1794

1862

        super(TestProxyAuth, self).test_empty_pass()

1795

1863

 

1849

1917

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

1850

1918

        # bulk data, so we use that.

1851

1919

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

1852

 

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

 

1920

        http_transport = transport.get_transport_from_url(

 

1921

            self.http_server.get_url())

1853

1922

        medium = http_transport.get_smart_medium()

1854

1923

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

1855

1924

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

1863

1932

        post_body = 'hello\n'

1864

1933

        expected_reply_body = 'ok\x012\n'

1865

1934

 

1866

 

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

 

1935

        http_transport = transport.get_transport_from_url(

 

1936

            self.http_server.get_url())

1867

1937

        medium = http_transport.get_smart_medium()

1868

1938

        response = medium.send_http_smart_request(post_body)

1869

1939

        reply_body = response.read()

1927

1997

        self.assertIsInstance(r, type(t))

1928

1998

        # Both transports share the some connection

1929

1999

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

 

2000

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

1930

2001

 

1931

2002

    def test_redirected_to_self_with_slash(self):

1932

2003

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

1943

2014

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

1944

2015

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

1945

2016

        self.assertIsInstance(r, type(t))

 

2017

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

 

2018

            r.external_url())

1946

2019

 

1947

2020

    def test_redirected_to_same_host_sibling_protocol(self):

1948

2021

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

1949

2022

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

1950

2023

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

1951

2024

        self.assertIsInstance(r, type(t))

 

2025

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

 

2026

            r.external_url())

1952

2027

 

1953

2028

    def test_redirected_to_same_host_different_protocol(self):

1954

2029

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

1955

2030

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

1956

2031

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

1957

2032

        self.assertNotEquals(type(r), type(t))

 

2033

        self.assertEquals('ftp://www.example.com/foo/', r.external_url())

 

2034

 

 

2035

    def test_redirected_to_same_host_specific_implementation(self):

 

2036

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

 

2037

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

 

2038

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

 

2039

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

1958

2040

 

1959

2041

    def test_redirected_to_different_host_same_user(self):

1960

2042

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

1961

2043

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

1962

2044

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

1963

2045

        self.assertIsInstance(r, type(t))

1964

 

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

 

2046

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

 

2047

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

1965

2048

 

1966

2049

 

1967

2050

class PredefinedRequestHandler(http_server.TestingHTTPRequestHandler):

2020

2103

    pass

2021

2104

 

2022

2105

 

2023

 

if tests.HTTPSServerFeature.available():

 

2106

if features.HTTPSServerFeature.available():

2024

2107

    from bzrlib.tests import https_server

2025

2108

    class ActivityHTTPSServer(ActivityServerMixin, https_server.HTTPSServer):

2026

2109

        pass

• Older »

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