128
127
('urllib,http', dict(_activity_server=ActivityHTTPServer,
129
128
_transport=_urllib.HttpTransport_urllib,)),
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 (
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.
139
from bzrlib.tests import (
142
class HTTPS_urllib_transport(_urllib.HttpTransport_urllib):
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'))
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):
149
155
def __init__(self, base, _from_transport=None):
378
384
_transport = property(_get_pycurl_maybe)
381
class TestHttpUrls(tests.TestCase):
383
# TODO: This should be moved to authorization tests once they
386
def test_url_parsing(self):
388
url = http.extract_auth('http://example.com', f)
389
self.assertEqual('http://example.com', url)
390
self.assertEqual(0, len(f.credentials))
391
url = http.extract_auth(
392
'http://user:pass@example.com/bzr/bzr.dev', f)
393
self.assertEqual('http://example.com/bzr/bzr.dev', url)
394
self.assertEqual(1, len(f.credentials))
395
self.assertEqual([None, 'example.com', 'user', 'pass'],
399
387
class TestHttpTransportUrls(tests.TestCase):
400
388
"""Test the http urls."""
1048
1037
self.assertEqual('single', t._range_hint)
1040
class TruncatedBeforeBoundaryRequestHandler(
1041
http_server.TestingHTTPRequestHandler):
1042
"""Truncation before a boundary, like in bug 198646"""
1044
_truncated_ranges = 1
1046
def get_multiple_ranges(self, file, file_size, ranges):
1047
self.send_response(206)
1048
self.send_header('Accept-Ranges', 'bytes')
1050
self.send_header('Content-Type',
1051
'multipart/byteranges; boundary=%s' % boundary)
1052
boundary_line = '--%s\r\n' % boundary
1053
# Calculate the Content-Length
1055
for (start, end) in ranges:
1056
content_length += len(boundary_line)
1057
content_length += self._header_line_length(
1058
'Content-type', 'application/octet-stream')
1059
content_length += self._header_line_length(
1060
'Content-Range', 'bytes %d-%d/%d' % (start, end, file_size))
1061
content_length += len('\r\n') # end headers
1062
content_length += end - start # + 1
1063
content_length += len(boundary_line)
1064
self.send_header('Content-length', content_length)
1067
# Send the multipart body
1069
for (start, end) in ranges:
1070
if cur + self._truncated_ranges >= len(ranges):
1071
# Abruptly ends the response and close the connection
1072
self.close_connection = 1
1074
self.wfile.write(boundary_line)
1075
self.send_header('Content-type', 'application/octet-stream')
1076
self.send_header('Content-Range', 'bytes %d-%d/%d'
1077
% (start, end, file_size))
1079
self.send_range_content(file, start, end - start + 1)
1082
self.wfile.write(boundary_line)
1085
class TestTruncatedBeforeBoundary(TestSpecificRequestHandler):
1086
"""Tests the case of bug 198646, disconnecting before a boundary."""
1088
_req_handler_class = TruncatedBeforeBoundaryRequestHandler
1091
super(TestTruncatedBeforeBoundary, self).setUp()
1092
self.build_tree_contents([('a', '0123456789')],)
1094
def test_readv_with_short_reads(self):
1095
server = self.get_readonly_server()
1096
t = self.get_readonly_transport()
1097
# Force separate ranges for each offset
1098
t._bytes_to_read_before_seek = 0
1099
ireadv = iter(t.readv('a', ((0, 1), (2, 1), (4, 2), (9, 1))))
1100
self.assertEqual((0, '0'), ireadv.next())
1101
self.assertEqual((2, '2'), ireadv.next())
1102
self.assertEqual((4, '45'), ireadv.next())
1103
self.assertEqual((9, '9'), ireadv.next())
1051
1106
class LimitedRangeRequestHandler(http_server.TestingHTTPRequestHandler):
1052
1107
"""Errors out when range specifiers exceed the limit"""
1849
1905
# The 'readv' command in the smart protocol both sends and receives
1850
1906
# bulk data, so we use that.
1851
1907
self.build_tree(['data-file'])
1852
http_transport = transport.get_transport(self.http_server.get_url())
1908
http_transport = transport.get_transport_from_url(
1909
self.http_server.get_url())
1853
1910
medium = http_transport.get_smart_medium()
1854
1911
# Since we provide the medium, the url below will be mostly ignored
1855
1912
# during the test, as long as the path is '/'.
1943
2002
r = t._redirected_to('http://www.example.com/foo',
1944
2003
'http://foo.example.com/foo/subdir')
1945
2004
self.assertIsInstance(r, type(t))
2005
self.assertEquals('http://foo.example.com/foo/subdir/',
1947
2008
def test_redirected_to_same_host_sibling_protocol(self):
1948
2009
t = self._transport('http://www.example.com/foo')
1949
2010
r = t._redirected_to('http://www.example.com/foo',
1950
2011
'https://www.example.com/foo')
1951
2012
self.assertIsInstance(r, type(t))
2013
self.assertEquals('https://www.example.com/foo/',
1953
2016
def test_redirected_to_same_host_different_protocol(self):
1954
2017
t = self._transport('http://www.example.com/foo')
1955
2018
r = t._redirected_to('http://www.example.com/foo',
1956
2019
'ftp://www.example.com/foo')
1957
2020
self.assertNotEquals(type(r), type(t))
2021
self.assertEquals('ftp://www.example.com/foo/', r.external_url())
2023
def test_redirected_to_same_host_specific_implementation(self):
2024
t = self._transport('http://www.example.com/foo')
2025
r = t._redirected_to('http://www.example.com/foo',
2026
'https+urllib://www.example.com/foo')
2027
self.assertEquals('https://www.example.com/foo/', r.external_url())
1959
2029
def test_redirected_to_different_host_same_user(self):
1960
2030
t = self._transport('http://joe@www.example.com/foo')
1961
2031
r = t._redirected_to('http://www.example.com/foo',
1962
2032
'https://foo.example.com/foo')
1963
2033
self.assertIsInstance(r, type(t))
1964
self.assertEqual(t._user, r._user)
2034
self.assertEqual(t._parsed_url.user, r._parsed_url.user)
2035
self.assertEquals('https://joe@foo.example.com/foo/', r.external_url())
1967
2038
class PredefinedRequestHandler(http_server.TestingHTTPRequestHandler):
2036
2107
def setUp(self):
2037
tests.TestCase.setUp(self)
2038
2108
self.server = self._activity_server(self._protocol_version)
2039
2109
self.server.start_server()
2110
self.addCleanup(self.server.stop_server)
2040
2111
_activities = {} # Don't close over self and create a cycle
2041
2112
def report_activity(t, bytes, direction):
2042
2113
count = _activities.get(direction, 0)
2044
2115
_activities[direction] = count
2045
2116
self.activities = _activities
2047
2117
# We override at class level because constructors may propagate the
2048
2118
# bound method and render instance overriding ineffective (an
2049
2119
# alternative would be to define a specific ui factory instead...)
2050
2120
self.overrideAttr(self._transport, '_report_activity', report_activity)
2051
self.addCleanup(self.server.stop_server)
2053
2122
def get_transport(self):
2054
2123
t = self._transport(self.server.get_url())
added
removed
bzrlib/tests/test_http.py
