94
def vary_by_http_proxy_auth_scheme():
96
('basic', dict(_auth_server=http_utils.ProxyBasicAuthServer)),
97
('digest', dict(_auth_server=http_utils.ProxyDigestAuthServer)),
99
dict(_auth_server=http_utils.ProxyBasicAndDigestAuthServer)),
96
103
def vary_by_http_auth_scheme():
98
105
('basic', dict(_auth_server=http_utils.HTTPBasicAuthServer)),
99
106
('digest', dict(_auth_server=http_utils.HTTPDigestAuthServer)),
101
108
dict(_auth_server=http_utils.HTTPBasicAndDigestAuthServer)),
103
# Add some attributes common to all scenarios
104
for scenario_id, scenario_dict in scenarios:
105
scenario_dict.update(_auth_header='Authorization',
106
_username_prompt_prefix='',
107
_password_prompt_prefix='')
111
def vary_by_http_proxy_auth_scheme():
113
('proxy-basic', dict(_auth_server=http_utils.ProxyBasicAuthServer)),
114
('proxy-digest', dict(_auth_server=http_utils.ProxyDigestAuthServer)),
115
('proxy-basicdigest',
116
dict(_auth_server=http_utils.ProxyBasicAndDigestAuthServer)),
118
# Add some attributes common to all scenarios
119
for scenario_id, scenario_dict in scenarios:
120
scenario_dict.update(_auth_header='Proxy-Authorization',
121
_username_prompt_prefix='Proxy ',
122
_password_prompt_prefix='Proxy ')
126
112
def vary_by_http_activity():
128
114
('urllib,http', dict(_activity_server=ActivityHTTPServer,
129
115
_transport=_urllib.HttpTransport_urllib,)),
131
if features.HTTPSServerFeature.available():
117
if tests.HTTPSServerFeature.available():
132
118
activity_scenarios.append(
133
119
('urllib,https', dict(_activity_server=ActivityHTTPSServer,
134
120
_transport=_urllib.HttpTransport_urllib,)),)
1049
1034
self.assertEqual('single', t._range_hint)
1052
class TruncatedBeforeBoundaryRequestHandler(
1053
http_server.TestingHTTPRequestHandler):
1054
"""Truncation before a boundary, like in bug 198646"""
1056
_truncated_ranges = 1
1058
def get_multiple_ranges(self, file, file_size, ranges):
1059
self.send_response(206)
1060
self.send_header('Accept-Ranges', 'bytes')
1062
self.send_header('Content-Type',
1063
'multipart/byteranges; boundary=%s' % boundary)
1064
boundary_line = '--%s\r\n' % boundary
1065
# Calculate the Content-Length
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)
1079
# Send the multipart body
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
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))
1091
self.send_range_content(file, start, end - start + 1)
1094
self.wfile.write(boundary_line)
1097
class TestTruncatedBeforeBoundary(TestSpecificRequestHandler):
1098
"""Tests the case of bug 198646, disconnecting before a boundary."""
1100
_req_handler_class = TruncatedBeforeBoundaryRequestHandler
1103
super(TestTruncatedBeforeBoundary, self).setUp()
1104
self.build_tree_contents([('a', '0123456789')],)
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())
1118
1037
class LimitedRangeRequestHandler(http_server.TestingHTTPRequestHandler):
1119
1038
"""Errors out when range specifiers exceed the limit"""
1570
1489
self.get_a, self.old_transport, redirected)
1573
def _setup_authentication_config(**kwargs):
1574
conf = config.AuthenticationConfig()
1575
conf._get_config().update({'httptest': kwargs})
1579
class TestUrllib2AuthHandler(tests.TestCaseWithTransport):
1580
"""Unit tests for glue by which urllib2 asks us for authentication"""
1582
def test_get_user_password_without_port(self):
1583
"""We cope if urllib2 doesn't tell us the port.
1585
See https://bugs.launchpad.net/bzr/+bug/654684
1589
_setup_authentication_config(scheme='http', host='localhost',
1590
user=user, password=password)
1591
handler = _urllib2_wrappers.HTTPAuthHandler()
1592
got_pass = handler.get_user_password(dict(
1599
self.assertEquals((user, password), got_pass)
1602
1492
class TestAuth(http_utils.TestCaseWithWebserver):
1603
1493
"""Test authentication scheme"""
1755
1650
ui.ui_factory = tests.TestUIFactory(stdin=stdin_content,
1756
1651
stderr=tests.StringIOWrapper())
1757
1652
# Create a minimal config file with the right password
1758
_setup_authentication_config(scheme='http', port=self.server.port,
1759
user=user, password=password)
1653
_setup_authentication_config(
1655
port=self.server.port,
1760
1658
# Issue a request to the server to connect
1761
1659
self.assertEqual('contents of a\n',t.get('a').read())
1762
1660
# stdin should have been left untouched
1793
1691
password = 'foo'
1794
1692
self.server.add_user(user, password)
1795
_setup_authentication_config(scheme='http', port=self.server.port,
1796
user=user, password=password)
1693
_setup_authentication_config(
1695
port=self.server.port,
1797
1698
t = self.get_user_transport(None, None)
1798
1699
# Issue a request to the server to connect
1799
1700
self.assertEqual('contents of a\n', t.get('a').read())
1800
1701
# Only one 'Authentication Required' error should occur
1801
1702
self.assertEqual(1, self.server.auth_required_errors)
1803
def test_no_credential_leaks_in_log(self):
1804
self.overrideAttr(debug, 'debug_flags', set(['http']))
1705
def _setup_authentication_config(**kwargs):
1706
conf = config.AuthenticationConfig()
1707
conf._get_config().update({'httptest': kwargs})
1712
class TestUrllib2AuthHandler(tests.TestCaseWithTransport):
1713
"""Unit tests for glue by which urllib2 asks us for authentication"""
1715
def test_get_user_password_without_port(self):
1716
"""We cope if urllib2 doesn't tell us the port.
1718
See https://bugs.launchpad.net/bzr/+bug/654684
1806
password = 'very-sensitive-password'
1807
self.server.add_user(user, password)
1808
t = self.get_user_transport(user, password)
1809
# Capture the debug calls to mutter
1812
lines = args[0] % args[1:]
1813
# Some calls output multiple lines, just split them now since we
1814
# care about a single one later.
1815
self.mutters.extend(lines.splitlines())
1816
self.overrideAttr(trace, 'mutter', mutter)
1817
# Issue a request to the server to connect
1818
self.assertEqual(True, t.has('a'))
1819
# Only one 'Authentication Required' error should occur
1820
self.assertEqual(1, self.server.auth_required_errors)
1821
# Since the authentification succeeded, there should be a corresponding
1823
sent_auth_headers = [line for line in self.mutters
1824
if line.startswith('> %s' % (self._auth_header,))]
1825
self.assertLength(1, sent_auth_headers)
1826
self.assertStartsWith(sent_auth_headers[0],
1827
'> %s: <masked>' % (self._auth_header,))
1722
_setup_authentication_config(
1727
handler = _urllib2_wrappers.HTTPAuthHandler()
1728
got_pass = handler.get_user_password(dict(
1735
self.assertEquals((user, password), got_pass)
1830
1738
class TestProxyAuth(TestAuth):
1831
"""Test proxy authentication schemes.
1833
This inherits from TestAuth to tweak the setUp and filter some failing
1739
"""Test proxy authentication schemes."""
1837
1741
scenarios = multiply_scenarios(
1838
1742
vary_by_http_client_implementation(),
1917
1825
# The 'readv' command in the smart protocol both sends and receives
1918
1826
# bulk data, so we use that.
1919
1827
self.build_tree(['data-file'])
1920
http_transport = transport.get_transport_from_url(
1921
self.http_server.get_url())
1828
http_transport = transport.get_transport(self.http_server.get_url())
1922
1829
medium = http_transport.get_smart_medium()
1923
1830
# Since we provide the medium, the url below will be mostly ignored
1924
1831
# during the test, as long as the path is '/'.
2014
1919
r = t._redirected_to('http://www.example.com/foo',
2015
1920
'http://foo.example.com/foo/subdir')
2016
1921
self.assertIsInstance(r, type(t))
2017
self.assertEquals('http://foo.example.com/foo/subdir/',
2020
1923
def test_redirected_to_same_host_sibling_protocol(self):
2021
1924
t = self._transport('http://www.example.com/foo')
2022
1925
r = t._redirected_to('http://www.example.com/foo',
2023
1926
'https://www.example.com/foo')
2024
1927
self.assertIsInstance(r, type(t))
2025
self.assertEquals('https://www.example.com/foo/',
2028
1929
def test_redirected_to_same_host_different_protocol(self):
2029
1930
t = self._transport('http://www.example.com/foo')
2030
1931
r = t._redirected_to('http://www.example.com/foo',
2031
1932
'ftp://www.example.com/foo')
2032
1933
self.assertNotEquals(type(r), type(t))
2033
self.assertEquals('ftp://www.example.com/foo/', r.external_url())
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())
2041
1935
def test_redirected_to_different_host_same_user(self):
2042
1936
t = self._transport('http://joe@www.example.com/foo')
2043
1937
r = t._redirected_to('http://www.example.com/foo',
2044
1938
'https://foo.example.com/foo')
2045
1939
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())
1940
self.assertEqual(t._user, r._user)
2050
1943
class PredefinedRequestHandler(http_server.TestingHTTPRequestHandler):
added
removed
bzrlib/tests/test_http.py
