~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_http.py

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

• reverse the comparison (5972.1.1 to 6217)
• stop comparing with revision 6217

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 features.HTTPSServerFeature.available():

 

131

    if tests.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 features.HTTPSServerFeature.available():

 

139

        if tests.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_from_url(

537

 

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

 

536

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

538

537

        self.assertIsInstance(t, transport.Transport)

539

538

        self.assertIsInstance(t, self._transport)

540

539

 

552

551

        self.start_server(server)

553

552

        url = server.get_url()

554

553

        # FIXME: needs a cleanup -- vila 20100611

555

 

        http_transport = transport.get_transport_from_url(url)

 

554

        http_transport = transport.get_transport(url)

556

555

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

557

556

        self.assertTrue(

558

557

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

1049

1048

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

1050

1049

 

1051

1050

 

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

 

 

1118

1051

class LimitedRangeRequestHandler(http_server.TestingHTTPRequestHandler):

1119

1052

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

1120

1053

 

1636

1569

        return url

1637

1570

 

1638

1571

    def get_user_transport(self, user, password):

1639

 

        t = transport.get_transport_from_url(

1640

 

            self.get_user_url(user, password))

 

1572

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

1641

1573

        return t

1642

1574

 

1643

1575

    def test_no_user(self):

1769

1701

                                     http_utils.ProxyDigestAuthServer):

1770

1702

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

1771

1703

        if self._testing_pycurl():

1772

 

            self.knownFailure(

 

1704

            raise tests.KnownFailure(

1773

1705

                'pycurl does not handle a nonce change')

1774

1706

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

1775

1707

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

1857

1789

        if self._testing_pycurl():

1858

1790

            import pycurl

1859

1791

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

1860

 

                self.knownFailure(

 

1792

                raise tests.KnownFailure(

1861

1793

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

1862

1794

        super(TestProxyAuth, self).test_empty_pass()

1863

1795

 

1917

1849

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

1918

1850

        # bulk data, so we use that.

1919

1851

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

1920

 

        http_transport = transport.get_transport_from_url(

1921

 

            self.http_server.get_url())

 

1852

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

1922

1853

        medium = http_transport.get_smart_medium()

1923

1854

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

1924

1855

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

1932

1863

        post_body = 'hello\n'

1933

1864

        expected_reply_body = 'ok\x012\n'

1934

1865

 

1935

 

        http_transport = transport.get_transport_from_url(

1936

 

            self.http_server.get_url())

 

1866

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

1937

1867

        medium = http_transport.get_smart_medium()

1938

1868

        response = medium.send_http_smart_request(post_body)

1939

1869

        reply_body = response.read()

1997

1927

        self.assertIsInstance(r, type(t))

1998

1928

        # Both transports share the some connection

1999

1929

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

2000

 

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

2001

1930

 

2002

1931

    def test_redirected_to_self_with_slash(self):

2003

1932

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

2014

1943

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

2015

1944

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

2016

1945

        self.assertIsInstance(r, type(t))

2017

 

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

2018

 

            r.external_url())

2019

1946

 

2020

1947

    def test_redirected_to_same_host_sibling_protocol(self):

2021

1948

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

2022

1949

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

2023

1950

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

2024

1951

        self.assertIsInstance(r, type(t))

2025

 

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

2026

 

            r.external_url())

2027

1952

 

2028

1953

    def test_redirected_to_same_host_different_protocol(self):

2029

1954

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

2030

1955

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

2031

1956

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

2032

1957

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

2040

1958

 

2041

1959

    def test_redirected_to_different_host_same_user(self):

2042

1960

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

2043

1961

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

2044

1962

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

2045

1963

        self.assertIsInstance(r, type(t))

2046

 

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

2047

 

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

 

1964

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

2048

1965

 

2049

1966

 

2050

1967

class PredefinedRequestHandler(http_server.TestingHTTPRequestHandler):

2103

2020

    pass

2104

2021

 

2105

2022

 

2106

 

if features.HTTPSServerFeature.available():

 

2023

if tests.HTTPSServerFeature.available():

2107

2024

    from bzrlib.tests import https_server

2108

2025

    class ActivityHTTPSServer(ActivityServerMixin, https_server.HTTPSServer):

2109

2026

        pass

• Older »

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