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."""
668
657
_req_handler_class = BadStatusRequestHandler
660
super(TestBadStatusServer, self).setUp()
661
# See https://bugs.launchpad.net/bzr/+bug/1451448 for details.
662
# TD;LR: Running both a TCP client and server in the same process and
663
# thread uncovers a race in python. The fix is to run the server in a
664
# different process. Trying to fix yet another race here is not worth
665
# the effort. -- vila 2015-09-06
666
if 'HTTP/1.0' in self.id():
667
raise tests.TestSkipped(
668
'Client/Server in the same process and thread can hang')
670
670
def test_http_has(self):
671
671
t = self.get_readonly_transport()
672
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
672
self.assertRaises((errors.ConnectionError, errors.ConnectionReset,
673
errors.InvalidHttpResponse),
674
676
def test_http_get(self):
675
677
t = self.get_readonly_transport()
676
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
678
self.assertRaises((errors.ConnectionError, errors.ConnectionReset,
679
errors.InvalidHttpResponse),
679
683
class InvalidStatusRequestHandler(http_server.TestingHTTPRequestHandler):
1048
1052
self.assertEqual('single', t._range_hint)
1055
class TruncatedBeforeBoundaryRequestHandler(
1056
http_server.TestingHTTPRequestHandler):
1057
"""Truncation before a boundary, like in bug 198646"""
1059
_truncated_ranges = 1
1061
def get_multiple_ranges(self, file, file_size, ranges):
1062
self.send_response(206)
1063
self.send_header('Accept-Ranges', 'bytes')
1065
self.send_header('Content-Type',
1066
'multipart/byteranges; boundary=%s' % boundary)
1067
boundary_line = '--%s\r\n' % boundary
1068
# Calculate the Content-Length
1070
for (start, end) in ranges:
1071
content_length += len(boundary_line)
1072
content_length += self._header_line_length(
1073
'Content-type', 'application/octet-stream')
1074
content_length += self._header_line_length(
1075
'Content-Range', 'bytes %d-%d/%d' % (start, end, file_size))
1076
content_length += len('\r\n') # end headers
1077
content_length += end - start # + 1
1078
content_length += len(boundary_line)
1079
self.send_header('Content-length', content_length)
1082
# Send the multipart body
1084
for (start, end) in ranges:
1085
if cur + self._truncated_ranges >= len(ranges):
1086
# Abruptly ends the response and close the connection
1087
self.close_connection = 1
1089
self.wfile.write(boundary_line)
1090
self.send_header('Content-type', 'application/octet-stream')
1091
self.send_header('Content-Range', 'bytes %d-%d/%d'
1092
% (start, end, file_size))
1094
self.send_range_content(file, start, end - start + 1)
1097
self.wfile.write(boundary_line)
1100
class TestTruncatedBeforeBoundary(TestSpecificRequestHandler):
1101
"""Tests the case of bug 198646, disconnecting before a boundary."""
1103
_req_handler_class = TruncatedBeforeBoundaryRequestHandler
1106
super(TestTruncatedBeforeBoundary, self).setUp()
1107
self.build_tree_contents([('a', '0123456789')],)
1109
def test_readv_with_short_reads(self):
1110
server = self.get_readonly_server()
1111
t = self.get_readonly_transport()
1112
# Force separate ranges for each offset
1113
t._bytes_to_read_before_seek = 0
1114
ireadv = iter(t.readv('a', ((0, 1), (2, 1), (4, 2), (9, 1))))
1115
self.assertEqual((0, '0'), ireadv.next())
1116
self.assertEqual((2, '2'), ireadv.next())
1117
self.assertEqual((4, '45'), ireadv.next())
1118
self.assertEqual((9, '9'), ireadv.next())
1051
1121
class LimitedRangeRequestHandler(http_server.TestingHTTPRequestHandler):
1052
1122
"""Errors out when range specifiers exceed the limit"""
1849
1920
# The 'readv' command in the smart protocol both sends and receives
1850
1921
# bulk data, so we use that.
1851
1922
self.build_tree(['data-file'])
1852
http_transport = transport.get_transport(self.http_server.get_url())
1923
http_transport = transport.get_transport_from_url(
1924
self.http_server.get_url())
1853
1925
medium = http_transport.get_smart_medium()
1854
1926
# Since we provide the medium, the url below will be mostly ignored
1855
1927
# during the test, as long as the path is '/'.
1943
2017
r = t._redirected_to('http://www.example.com/foo',
1944
2018
'http://foo.example.com/foo/subdir')
1945
2019
self.assertIsInstance(r, type(t))
2020
self.assertEquals('http://foo.example.com/foo/subdir/',
1947
2023
def test_redirected_to_same_host_sibling_protocol(self):
1948
2024
t = self._transport('http://www.example.com/foo')
1949
2025
r = t._redirected_to('http://www.example.com/foo',
1950
2026
'https://www.example.com/foo')
1951
2027
self.assertIsInstance(r, type(t))
2028
self.assertEquals('https://www.example.com/foo/',
1953
2031
def test_redirected_to_same_host_different_protocol(self):
1954
2032
t = self._transport('http://www.example.com/foo')
1955
2033
r = t._redirected_to('http://www.example.com/foo',
1956
2034
'ftp://www.example.com/foo')
1957
2035
self.assertNotEquals(type(r), type(t))
2036
self.assertEquals('ftp://www.example.com/foo/', r.external_url())
2038
def test_redirected_to_same_host_specific_implementation(self):
2039
t = self._transport('http://www.example.com/foo')
2040
r = t._redirected_to('http://www.example.com/foo',
2041
'https+urllib://www.example.com/foo')
2042
self.assertEquals('https://www.example.com/foo/', r.external_url())
1959
2044
def test_redirected_to_different_host_same_user(self):
1960
2045
t = self._transport('http://joe@www.example.com/foo')
1961
2046
r = t._redirected_to('http://www.example.com/foo',
1962
2047
'https://foo.example.com/foo')
1963
2048
self.assertIsInstance(r, type(t))
1964
self.assertEqual(t._user, r._user)
2049
self.assertEqual(t._parsed_url.user, r._parsed_url.user)
2050
self.assertEquals('https://joe@foo.example.com/foo/', r.external_url())
1967
2053
class PredefinedRequestHandler(http_server.TestingHTTPRequestHandler):
2036
2122
def setUp(self):
2037
tests.TestCase.setUp(self)
2038
2123
self.server = self._activity_server(self._protocol_version)
2039
2124
self.server.start_server()
2125
self.addCleanup(self.server.stop_server)
2040
2126
_activities = {} # Don't close over self and create a cycle
2041
2127
def report_activity(t, bytes, direction):
2042
2128
count = _activities.get(direction, 0)
2044
2130
_activities[direction] = count
2045
2131
self.activities = _activities
2047
2132
# We override at class level because constructors may propagate the
2048
2133
# bound method and render instance overriding ineffective (an
2049
2134
# alternative would be to define a specific ui factory instead...)
2050
2135
self.overrideAttr(self._transport, '_report_activity', report_activity)
2051
self.addCleanup(self.server.stop_server)
2053
2137
def get_transport(self):
2054
2138
t = self._transport(self.server.get_url())
added
removed
bzrlib/tests/test_http.py
