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)),
95
103
def vary_by_http_auth_scheme():
97
105
('basic', dict(_auth_server=http_utils.HTTPBasicAuthServer)),
98
106
('digest', dict(_auth_server=http_utils.HTTPDigestAuthServer)),
100
108
dict(_auth_server=http_utils.HTTPBasicAndDigestAuthServer)),
102
# Add some attributes common to all scenarios
103
for scenario_id, scenario_dict in scenarios:
104
scenario_dict.update(_auth_header='Authorization',
105
_username_prompt_prefix='',
106
_password_prompt_prefix='')
110
def vary_by_http_proxy_auth_scheme():
112
('proxy-basic', dict(_auth_server=http_utils.ProxyBasicAuthServer)),
113
('proxy-digest', dict(_auth_server=http_utils.ProxyDigestAuthServer)),
114
('proxy-basicdigest',
115
dict(_auth_server=http_utils.ProxyBasicAndDigestAuthServer)),
117
# Add some attributes common to all scenarios
118
for scenario_id, scenario_dict in scenarios:
119
scenario_dict.update(_auth_header='Proxy-Authorization',
120
_username_prompt_prefix='Proxy ',
121
_password_prompt_prefix='Proxy ')
125
112
def vary_by_http_activity():
127
114
('urllib,http', dict(_activity_server=ActivityHTTPServer,
128
115
_transport=_urllib.HttpTransport_urllib,)),
117
if tests.HTTPSServerFeature.available():
118
activity_scenarios.append(
119
('urllib,https', dict(_activity_server=ActivityHTTPSServer,
120
_transport=_urllib.HttpTransport_urllib,)),)
130
121
if features.pycurl.available():
131
122
activity_scenarios.append(
132
123
('pycurl,http', dict(_activity_server=ActivityHTTPServer,
133
124
_transport=PyCurlTransport,)),)
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():
125
if tests.HTTPSServerFeature.available():
126
from bzrlib.tests import (
129
# FIXME: Until we have a better way to handle self-signed
130
# certificates (like allowing them in a test specific
131
# authentication.conf for example), we need some specialized pycurl
132
# transport for tests.
153
133
class HTTPS_pycurl_transport(PyCurlTransport):
155
135
def __init__(self, base, _from_transport=None):
384
364
_transport = property(_get_pycurl_maybe)
367
class TestHttpUrls(tests.TestCase):
369
# TODO: This should be moved to authorization tests once they
372
def test_url_parsing(self):
374
url = http.extract_auth('http://example.com', f)
375
self.assertEqual('http://example.com', url)
376
self.assertEqual(0, len(f.credentials))
377
url = http.extract_auth(
378
'http://user:pass@example.com/bzr/bzr.dev', f)
379
self.assertEqual('http://example.com/bzr/bzr.dev', url)
380
self.assertEqual(1, len(f.credentials))
381
self.assertEqual([None, 'example.com', 'user', 'pass'],
387
385
class TestHttpTransportUrls(tests.TestCase):
388
386
"""Test the http urls."""
657
654
_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
656
def test_http_has(self):
671
657
t = self.get_readonly_transport()
672
self.assertRaises((errors.ConnectionError, errors.ConnectionReset,
673
errors.InvalidHttpResponse),
658
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
676
660
def test_http_get(self):
677
661
t = self.get_readonly_transport()
678
self.assertRaises((errors.ConnectionError, errors.ConnectionReset,
679
errors.InvalidHttpResponse),
662
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
683
665
class InvalidStatusRequestHandler(http_server.TestingHTTPRequestHandler):
1052
1034
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())
1121
1037
class LimitedRangeRequestHandler(http_server.TestingHTTPRequestHandler):
1122
1038
"""Errors out when range specifiers exceed the limit"""
1573
1489
self.get_a, self.old_transport, redirected)
1576
def _setup_authentication_config(**kwargs):
1577
conf = config.AuthenticationConfig()
1578
conf._get_config().update({'httptest': kwargs})
1582
class TestUrllib2AuthHandler(tests.TestCaseWithTransport):
1583
"""Unit tests for glue by which urllib2 asks us for authentication"""
1585
def test_get_user_password_without_port(self):
1586
"""We cope if urllib2 doesn't tell us the port.
1588
See https://bugs.launchpad.net/bzr/+bug/654684
1592
_setup_authentication_config(scheme='http', host='localhost',
1593
user=user, password=password)
1594
handler = _urllib2_wrappers.HTTPAuthHandler()
1595
got_pass = handler.get_user_password(dict(
1602
self.assertEqual((user, password), got_pass)
1605
1492
class TestAuth(http_utils.TestCaseWithWebserver):
1606
1493
"""Test authentication scheme"""
1758
1650
ui.ui_factory = tests.TestUIFactory(stdin=stdin_content,
1759
1651
stderr=tests.StringIOWrapper())
1760
1652
# Create a minimal config file with the right password
1761
_setup_authentication_config(scheme='http', port=self.server.port,
1762
user=user, password=password)
1653
_setup_authentication_config(
1655
port=self.server.port,
1763
1658
# Issue a request to the server to connect
1764
1659
self.assertEqual('contents of a\n',t.get('a').read())
1765
1660
# stdin should have been left untouched
1796
1691
password = 'foo'
1797
1692
self.server.add_user(user, password)
1798
_setup_authentication_config(scheme='http', port=self.server.port,
1799
user=user, password=password)
1693
_setup_authentication_config(
1695
port=self.server.port,
1800
1698
t = self.get_user_transport(None, None)
1801
1699
# Issue a request to the server to connect
1802
1700
self.assertEqual('contents of a\n', t.get('a').read())
1803
1701
# Only one 'Authentication Required' error should occur
1804
1702
self.assertEqual(1, self.server.auth_required_errors)
1806
def test_no_credential_leaks_in_log(self):
1807
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
1809
password = 'very-sensitive-password'
1810
self.server.add_user(user, password)
1811
t = self.get_user_transport(user, password)
1812
# Capture the debug calls to mutter
1815
lines = args[0] % args[1:]
1816
# Some calls output multiple lines, just split them now since we
1817
# care about a single one later.
1818
self.mutters.extend(lines.splitlines())
1819
self.overrideAttr(trace, 'mutter', mutter)
1820
# Issue a request to the server to connect
1821
self.assertEqual(True, t.has('a'))
1822
# Only one 'Authentication Required' error should occur
1823
self.assertEqual(1, self.server.auth_required_errors)
1824
# Since the authentification succeeded, there should be a corresponding
1826
sent_auth_headers = [line for line in self.mutters
1827
if line.startswith('> %s' % (self._auth_header,))]
1828
self.assertLength(1, sent_auth_headers)
1829
self.assertStartsWith(sent_auth_headers[0],
1830
'> %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)
1833
1738
class TestProxyAuth(TestAuth):
1834
"""Test proxy authentication schemes.
1836
This inherits from TestAuth to tweak the setUp and filter some failing
1739
"""Test proxy authentication schemes."""
1840
1741
scenarios = multiply_scenarios(
1841
1742
vary_by_http_client_implementation(),
1920
1825
# The 'readv' command in the smart protocol both sends and receives
1921
1826
# bulk data, so we use that.
1922
1827
self.build_tree(['data-file'])
1923
http_transport = transport.get_transport_from_url(
1924
self.http_server.get_url())
1828
http_transport = transport.get_transport(self.http_server.get_url())
1925
1829
medium = http_transport.get_smart_medium()
1926
1830
# Since we provide the medium, the url below will be mostly ignored
1927
1831
# during the test, as long as the path is '/'.
2017
1919
r = t._redirected_to('http://www.example.com/foo',
2018
1920
'http://foo.example.com/foo/subdir')
2019
1921
self.assertIsInstance(r, type(t))
2020
self.assertEqual('http://foo.example.com/foo/subdir/',
2023
1923
def test_redirected_to_same_host_sibling_protocol(self):
2024
1924
t = self._transport('http://www.example.com/foo')
2025
1925
r = t._redirected_to('http://www.example.com/foo',
2026
1926
'https://www.example.com/foo')
2027
1927
self.assertIsInstance(r, type(t))
2028
self.assertEqual('https://www.example.com/foo/',
2031
1929
def test_redirected_to_same_host_different_protocol(self):
2032
1930
t = self._transport('http://www.example.com/foo')
2033
1931
r = t._redirected_to('http://www.example.com/foo',
2034
1932
'ftp://www.example.com/foo')
2035
self.assertNotEqual(type(r), type(t))
2036
self.assertEqual('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.assertEqual('https://www.example.com/foo/', r.external_url())
1933
self.assertNotEquals(type(r), type(t))
2044
1935
def test_redirected_to_different_host_same_user(self):
2045
1936
t = self._transport('http://joe@www.example.com/foo')
2046
1937
r = t._redirected_to('http://www.example.com/foo',
2047
1938
'https://foo.example.com/foo')
2048
1939
self.assertIsInstance(r, type(t))
2049
self.assertEqual(t._parsed_url.user, r._parsed_url.user)
2050
self.assertEqual('https://joe@foo.example.com/foo/', r.external_url())
1940
self.assertEqual(t._user, r._user)
2053
1943
class PredefinedRequestHandler(http_server.TestingHTTPRequestHandler):
2122
2012
def setUp(self):
2013
tests.TestCase.setUp(self)
2123
2014
self.server = self._activity_server(self._protocol_version)
2124
2015
self.server.start_server()
2125
self.addCleanup(self.server.stop_server)
2126
_activities = {} # Don't close over self and create a cycle
2016
self.activities = {}
2127
2017
def report_activity(t, bytes, direction):
2128
count = _activities.get(direction, 0)
2018
count = self.activities.get(direction, 0)
2130
_activities[direction] = count
2131
self.activities = _activities
2020
self.activities[direction] = count
2132
2022
# We override at class level because constructors may propagate the
2133
2023
# bound method and render instance overriding ineffective (an
2134
2024
# alternative would be to define a specific ui factory instead...)
2135
2025
self.overrideAttr(self._transport, '_report_activity', report_activity)
2026
self.addCleanup(self.server.stop_server)
2137
2028
def get_transport(self):
2138
2029
t = self._transport(self.server.get_url())
added
removed
bzrlib/tests/test_http.py
