13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17
"""Tests for HTTP implementations.
19
This module defines a load_tests() method that parametrize tests classes for
20
transport implementation, http protocol versions and authentication schemes.
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17
# FIXME: This test should be repeated for each available http client
18
# implementation; at the moment we have urllib and pycurl.
23
20
# TODO: Should be renamed to bzrlib.transport.http.tests?
24
21
# TODO: What about renaming to bzrlib.tests.transport.http ?
27
import SimpleHTTPServer
23
from cStringIO import StringIO
33
31
from bzrlib import (
40
remote as _mod_remote,
46
from bzrlib.tests import (
52
from bzrlib.tests.scenarios import (
53
load_tests_apply_scenarios,
39
from bzrlib.tests.HttpServer import (
44
from bzrlib.tests.HTTPTestUtil import (
45
BadProtocolRequestHandler,
46
BadStatusRequestHandler,
47
ForbiddenRequestHandler,
50
HTTPServerRedirecting,
51
InvalidStatusRequestHandler,
52
LimitedRangeHTTPServer,
53
NoRangeRequestHandler,
55
ProxyDigestAuthServer,
57
SingleRangeRequestHandler,
58
SingleOnlyRangeRequestHandler,
59
TestCaseWithRedirectedWebserver,
60
TestCaseWithTwoWebservers,
61
TestCaseWithWebserver,
56
64
from bzrlib.transport import (
66
do_catching_redirections,
60
70
from bzrlib.transport.http import (
66
if features.pycurl.available():
67
from bzrlib.transport.http._pycurl import PyCurlTransport
70
load_tests = load_tests_apply_scenarios
73
def vary_by_http_client_implementation():
74
"""Test the two libraries we can use, pycurl and urllib."""
75
transport_scenarios = [
76
('urllib', dict(_transport=_urllib.HttpTransport_urllib,
77
_server=http_server.HttpServer_urllib,
78
_url_protocol='http+urllib',)),
80
if features.pycurl.available():
81
transport_scenarios.append(
82
('pycurl', dict(_transport=PyCurlTransport,
83
_server=http_server.HttpServer_PyCurl,
84
_url_protocol='http+pycurl',)))
85
return transport_scenarios
88
def vary_by_http_protocol_version():
89
"""Test on http/1.0 and 1.1"""
91
('HTTP/1.0', dict(_protocol_version='HTTP/1.0')),
92
('HTTP/1.1', dict(_protocol_version='HTTP/1.1')),
96
def vary_by_http_auth_scheme():
98
('basic', dict(_auth_server=http_utils.HTTPBasicAuthServer)),
99
('digest', dict(_auth_server=http_utils.HTTPDigestAuthServer)),
101
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
def vary_by_http_activity():
127
activity_scenarios = [
128
('urllib,http', dict(_activity_server=ActivityHTTPServer,
129
_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
if features.pycurl.available():
136
activity_scenarios.append(
137
('pycurl,http', dict(_activity_server=ActivityHTTPServer,
138
_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.
147
class HTTPS_pycurl_transport(PyCurlTransport):
149
def __init__(self, base, _from_transport=None):
150
super(HTTPS_pycurl_transport, self).__init__(
151
base, _from_transport)
152
self.cabundle = str(ssl_certs.build_path('ca.crt'))
154
activity_scenarios.append(
155
('pycurl,https', dict(_activity_server=ActivityHTTPSServer,
156
_transport=HTTPS_pycurl_transport,)),)
157
return activity_scenarios
75
from bzrlib.transport.http._urllib import HttpTransport_urllib
76
from bzrlib.transport.http._urllib2_wrappers import (
160
82
class FakeManager(object):
185
107
self.received_bytes = ''
189
return '%s://%s:%s/' % (self.scheme, self.host, self.port)
191
def start_server(self):
192
110
self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
193
111
self._sock.bind(('127.0.0.1', 0))
194
112
self.host, self.port = self._sock.getsockname()
195
113
self._ready = threading.Event()
196
self._thread = test_server.TestThread(
197
sync_event=self._ready, target=self._accept_read_and_reply)
114
self._thread = threading.Thread(target=self._accept_read_and_reply)
115
self._thread.setDaemon(True)
198
116
self._thread.start()
199
if 'threads' in tests.selftest_debug_flags:
200
sys.stderr.write('Thread started: %s\n' % (self._thread.ident,))
203
119
def _accept_read_and_reply(self):
204
120
self._sock.listen(1)
205
121
self._ready.set()
206
conn, address = self._sock.accept()
207
if self._expect_body_tail is not None:
122
self._sock.settimeout(5)
124
conn, address = self._sock.accept()
125
# On win32, the accepted connection will be non-blocking to start
126
# with because we're using settimeout.
127
conn.setblocking(True)
208
128
while not self.received_bytes.endswith(self._expect_body_tail):
209
129
self.received_bytes += conn.recv(4096)
210
130
conn.sendall('HTTP/1.1 200 OK\r\n')
131
except socket.timeout:
132
# Make sure the client isn't stuck waiting for us to e.g. accept.
212
133
self._sock.close()
213
134
except socket.error:
214
135
# The client may have already closed the socket.
217
def stop_server(self):
219
# Issue a fake connection to wake up the server and allow it to
221
fake_conn = osutils.connect_socket((self.host, self.port))
223
141
except socket.error:
224
142
# We might have already closed it. We don't care.
229
if 'threads' in tests.selftest_debug_flags:
230
sys.stderr.write('Thread joined: %s\n' % (self._thread.ident,))
233
class TestAuthHeader(tests.TestCase):
235
def parse_header(self, header, auth_handler_class=None):
236
if auth_handler_class is None:
237
auth_handler_class = _urllib2_wrappers.AbstractAuthHandler
238
self.auth_handler = auth_handler_class()
239
return self.auth_handler._parse_auth_header(header)
241
def test_empty_header(self):
242
scheme, remainder = self.parse_header('')
243
self.assertEqual('', scheme)
244
self.assertIs(None, remainder)
246
def test_negotiate_header(self):
247
scheme, remainder = self.parse_header('Negotiate')
248
self.assertEqual('negotiate', scheme)
249
self.assertIs(None, remainder)
251
def test_basic_header(self):
252
scheme, remainder = self.parse_header(
253
'Basic realm="Thou should not pass"')
254
self.assertEqual('basic', scheme)
255
self.assertEqual('realm="Thou should not pass"', remainder)
257
def test_basic_extract_realm(self):
258
scheme, remainder = self.parse_header(
259
'Basic realm="Thou should not pass"',
260
_urllib2_wrappers.BasicAuthHandler)
261
match, realm = self.auth_handler.extract_realm(remainder)
262
self.assertTrue(match is not None)
263
self.assertEqual('Thou should not pass', realm)
265
def test_digest_header(self):
266
scheme, remainder = self.parse_header(
267
'Digest realm="Thou should not pass"')
268
self.assertEqual('digest', scheme)
269
self.assertEqual('realm="Thou should not pass"', remainder)
272
class TestHTTPRangeParsing(tests.TestCase):
275
super(TestHTTPRangeParsing, self).setUp()
276
# We focus on range parsing here and ignore everything else
277
class RequestHandler(http_server.TestingHTTPRequestHandler):
278
def setup(self): pass
279
def handle(self): pass
280
def finish(self): pass
282
self.req_handler = RequestHandler(None, None, None)
284
def assertRanges(self, ranges, header, file_size):
285
self.assertEquals(ranges,
286
self.req_handler._parse_ranges(header, file_size))
288
def test_simple_range(self):
289
self.assertRanges([(0,2)], 'bytes=0-2', 12)
292
self.assertRanges([(8, 11)], 'bytes=-4', 12)
294
def test_tail_bigger_than_file(self):
295
self.assertRanges([(0, 11)], 'bytes=-99', 12)
297
def test_range_without_end(self):
298
self.assertRanges([(4, 11)], 'bytes=4-', 12)
300
def test_invalid_ranges(self):
301
self.assertRanges(None, 'bytes=12-22', 12)
302
self.assertRanges(None, 'bytes=1-3,12-22', 12)
303
self.assertRanges(None, 'bytes=-', 12)
306
class TestHTTPServer(tests.TestCase):
307
"""Test the HTTP servers implementations."""
309
def test_invalid_protocol(self):
310
class BogusRequestHandler(http_server.TestingHTTPRequestHandler):
312
protocol_version = 'HTTP/0.1'
314
self.assertRaises(httplib.UnknownProtocol,
315
http_server.HttpServer, BogusRequestHandler)
317
def test_force_invalid_protocol(self):
318
self.assertRaises(httplib.UnknownProtocol,
319
http_server.HttpServer, protocol_version='HTTP/0.1')
321
def test_server_start_and_stop(self):
322
server = http_server.HttpServer()
323
self.addCleanup(server.stop_server)
324
server.start_server()
325
self.assertTrue(server.server is not None)
326
self.assertTrue(server.server.serving is not None)
327
self.assertTrue(server.server.serving)
329
def test_create_http_server_one_zero(self):
330
class RequestHandlerOneZero(http_server.TestingHTTPRequestHandler):
332
protocol_version = 'HTTP/1.0'
334
server = http_server.HttpServer(RequestHandlerOneZero)
335
self.start_server(server)
336
self.assertIsInstance(server.server, http_server.TestingHTTPServer)
338
def test_create_http_server_one_one(self):
339
class RequestHandlerOneOne(http_server.TestingHTTPRequestHandler):
341
protocol_version = 'HTTP/1.1'
343
server = http_server.HttpServer(RequestHandlerOneOne)
344
self.start_server(server)
345
self.assertIsInstance(server.server,
346
http_server.TestingThreadingHTTPServer)
348
def test_create_http_server_force_one_one(self):
349
class RequestHandlerOneZero(http_server.TestingHTTPRequestHandler):
351
protocol_version = 'HTTP/1.0'
353
server = http_server.HttpServer(RequestHandlerOneZero,
354
protocol_version='HTTP/1.1')
355
self.start_server(server)
356
self.assertIsInstance(server.server,
357
http_server.TestingThreadingHTTPServer)
359
def test_create_http_server_force_one_zero(self):
360
class RequestHandlerOneOne(http_server.TestingHTTPRequestHandler):
362
protocol_version = 'HTTP/1.1'
364
server = http_server.HttpServer(RequestHandlerOneOne,
365
protocol_version='HTTP/1.0')
366
self.start_server(server)
367
self.assertIsInstance(server.server,
368
http_server.TestingHTTPServer)
371
148
class TestWithTransport_pycurl(object):
372
149
"""Test case to inherit from if pycurl is present"""
374
151
def _get_pycurl_maybe(self):
375
self.requireFeature(features.pycurl)
376
return PyCurlTransport
153
from bzrlib.transport.http._pycurl import PyCurlTransport
154
return PyCurlTransport
155
except errors.DependencyNotPresent:
156
raise tests.TestSkipped('pycurl not present')
378
158
_transport = property(_get_pycurl_maybe)
596
class TestSpecificRequestHandler(http_utils.TestCaseWithWebserver):
597
"""Tests a specific request handler.
599
Daughter classes are expected to override _req_handler_class
602
scenarios = multiply_scenarios(
603
vary_by_http_client_implementation(),
604
vary_by_http_protocol_version(),
607
# Provide a useful default
608
_req_handler_class = http_server.TestingHTTPRequestHandler
610
def create_transport_readonly_server(self):
611
server = http_server.HttpServer(self._req_handler_class,
612
protocol_version=self._protocol_version)
613
server._url_protocol = self._url_protocol
616
def _testing_pycurl(self):
617
# TODO: This is duplicated for lots of the classes in this file
618
return (features.pycurl.available()
619
and self._transport == PyCurlTransport)
622
class WallRequestHandler(http_server.TestingHTTPRequestHandler):
623
"""Whatever request comes in, close the connection"""
625
def _handle_one_request(self):
626
"""Handle a single HTTP request, by abruptly closing the connection"""
627
self.close_connection = 1
630
class TestWallServer(TestSpecificRequestHandler):
426
class TestWallServer(object):
631
427
"""Tests exceptions during the connection phase"""
633
_req_handler_class = WallRequestHandler
429
def create_transport_readonly_server(self):
430
return HttpServer(WallRequestHandler)
635
432
def test_http_has(self):
636
t = self.get_readonly_transport()
433
server = self.get_readonly_server()
434
t = self._transport(server.get_url())
637
435
# Unfortunately httplib (see HTTPResponse._read_status
638
436
# for details) make no distinction between a closed
639
437
# socket and badly formatted status line, so we can't
640
438
# just test for ConnectionError, we have to test
641
# InvalidHttpResponse too. And pycurl may raise ConnectionReset
642
# instead of ConnectionError too.
643
self.assertRaises(( errors.ConnectionError, errors.ConnectionReset,
644
errors.InvalidHttpResponse),
439
# InvalidHttpResponse too.
440
self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
645
441
t.has, 'foo/bar')
647
443
def test_http_get(self):
648
t = self.get_readonly_transport()
649
self.assertRaises((errors.ConnectionError, errors.ConnectionReset,
650
errors.InvalidHttpResponse),
444
server = self.get_readonly_server()
445
t = self._transport(server.get_url())
446
self.assertRaises((errors.ConnectionError, errors.InvalidHttpResponse),
651
447
t.get, 'foo/bar')
654
class BadStatusRequestHandler(http_server.TestingHTTPRequestHandler):
655
"""Whatever request comes in, returns a bad status"""
657
def parse_request(self):
658
"""Fakes handling a single HTTP request, returns a bad status"""
659
ignored = http_server.TestingHTTPRequestHandler.parse_request(self)
660
self.send_response(0, "Bad status")
661
self.close_connection = 1
665
class TestBadStatusServer(TestSpecificRequestHandler):
450
class TestWallServer_urllib(TestWallServer, TestCaseWithWebserver):
451
"""Tests "wall" server for urllib implementation"""
453
_transport = HttpTransport_urllib
456
class TestWallServer_pycurl(TestWithTransport_pycurl,
458
TestCaseWithWebserver):
459
"""Tests "wall" server for pycurl implementation"""
462
class TestBadStatusServer(object):
666
463
"""Tests bad status from server."""
668
_req_handler_class = BadStatusRequestHandler
465
def create_transport_readonly_server(self):
466
return HttpServer(BadStatusRequestHandler)
670
468
def test_http_has(self):
671
t = self.get_readonly_transport()
469
server = self.get_readonly_server()
470
t = self._transport(server.get_url())
672
471
self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')
674
473
def test_http_get(self):
675
t = self.get_readonly_transport()
474
server = self.get_readonly_server()
475
t = self._transport(server.get_url())
676
476
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
679
class InvalidStatusRequestHandler(http_server.TestingHTTPRequestHandler):
680
"""Whatever request comes in, returns an invalid status"""
682
def parse_request(self):
683
"""Fakes handling a single HTTP request, returns a bad status"""
684
ignored = http_server.TestingHTTPRequestHandler.parse_request(self)
685
self.wfile.write("Invalid status line\r\n")
686
# If we don't close the connection pycurl will hang. Since this is a
687
# stress test we don't *have* to respect the protocol, but we don't
688
# have to sabotage it too much either.
689
self.close_connection = True
479
class TestBadStatusServer_urllib(TestBadStatusServer, TestCaseWithWebserver):
480
"""Tests bad status server for urllib implementation"""
482
_transport = HttpTransport_urllib
485
class TestBadStatusServer_pycurl(TestWithTransport_pycurl,
487
TestCaseWithWebserver):
488
"""Tests bad status server for pycurl implementation"""
693
491
class TestInvalidStatusServer(TestBadStatusServer):
838
660
self.assertEqual(l[2], (3, '34'))
839
661
self.assertEqual(l[3], (9, '9'))
840
662
# The server should have issued 4 requests
841
self.assertEqual(4, server.GET_request_nb)
843
def test_readv_get_max_size(self):
844
server = self.get_readonly_server()
845
t = self.get_readonly_transport()
846
# force transport to issue multiple requests by limiting the number of
847
# bytes by request. Note that this apply to coalesced offsets only, a
848
# single range will keep its size even if bigger than the limit.
850
l = list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
851
self.assertEqual(l[0], (0, '0'))
852
self.assertEqual(l[1], (1, '1'))
853
self.assertEqual(l[2], (2, '2345'))
854
self.assertEqual(l[3], (6, '6789'))
855
# The server should have issued 3 requests
856
self.assertEqual(3, server.GET_request_nb)
858
def test_complete_readv_leave_pipe_clean(self):
859
server = self.get_readonly_server()
860
t = self.get_readonly_transport()
861
# force transport to issue multiple requests
863
list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
864
# The server should have issued 3 requests
865
self.assertEqual(3, server.GET_request_nb)
866
self.assertEqual('0123456789', t.get_bytes('a'))
867
self.assertEqual(4, server.GET_request_nb)
869
def test_incomplete_readv_leave_pipe_clean(self):
870
server = self.get_readonly_server()
871
t = self.get_readonly_transport()
872
# force transport to issue multiple requests
874
# Don't collapse readv results into a list so that we leave unread
875
# bytes on the socket
876
ireadv = iter(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
877
self.assertEqual((0, '0'), ireadv.next())
878
# The server should have issued one request so far
879
self.assertEqual(1, server.GET_request_nb)
880
self.assertEqual('0123456789', t.get_bytes('a'))
881
# get_bytes issued an additional request, the readv pending ones are
883
self.assertEqual(2, server.GET_request_nb)
886
class SingleRangeRequestHandler(http_server.TestingHTTPRequestHandler):
887
"""Always reply to range request as if they were single.
889
Don't be explicit about it, just to annoy the clients.
892
def get_multiple_ranges(self, file, file_size, ranges):
893
"""Answer as if it was a single range request and ignores the rest"""
894
(start, end) = ranges[0]
895
return self.get_single_range(file, file_size, start, end)
663
self.assertEqual(4, self.get_readonly_server().GET_request_nb)
898
666
class TestSingleRangeRequestServer(TestRangeRequestServer):
899
667
"""Test readv against a server which accept only single range requests"""
901
_req_handler_class = SingleRangeRequestHandler
904
class SingleOnlyRangeRequestHandler(http_server.TestingHTTPRequestHandler):
905
"""Only reply to simple range requests, errors out on multiple"""
907
def get_multiple_ranges(self, file, file_size, ranges):
908
"""Refuses the multiple ranges request"""
911
self.send_error(416, "Requested range not satisfiable")
913
(start, end) = ranges[0]
914
return self.get_single_range(file, file_size, start, end)
669
def create_transport_readonly_server(self):
670
return HttpServer(SingleRangeRequestHandler)
673
class TestSingleRangeRequestServer_urllib(TestSingleRangeRequestServer,
674
TestCaseWithWebserver):
675
"""Tests single range requests accepting server for urllib implementation"""
677
_transport = HttpTransport_urllib
680
class TestSingleRangeRequestServer_pycurl(TestWithTransport_pycurl,
681
TestSingleRangeRequestServer,
682
TestCaseWithWebserver):
683
"""Tests single range requests accepting server for pycurl implementation"""
917
686
class TestSingleOnlyRangeRequestServer(TestRangeRequestServer):
918
687
"""Test readv against a server which only accept single range requests"""
920
_req_handler_class = SingleOnlyRangeRequestHandler
923
class NoRangeRequestHandler(http_server.TestingHTTPRequestHandler):
924
"""Ignore range requests without notice"""
927
# Update the statistics
928
self.server.test_case_server.GET_request_nb += 1
929
# Just bypass the range handling done by TestingHTTPRequestHandler
930
return SimpleHTTPServer.SimpleHTTPRequestHandler.do_GET(self)
689
def create_transport_readonly_server(self):
690
return HttpServer(SingleOnlyRangeRequestHandler)
693
class TestSingleOnlyRangeRequestServer_urllib(TestSingleOnlyRangeRequestServer,
694
TestCaseWithWebserver):
695
"""Tests single range requests accepting server for urllib implementation"""
697
_transport = HttpTransport_urllib
700
class TestSingleOnlyRangeRequestServer_pycurl(TestWithTransport_pycurl,
701
TestSingleOnlyRangeRequestServer,
702
TestCaseWithWebserver):
703
"""Tests single range requests accepting server for pycurl implementation"""
933
706
class TestNoRangeRequestServer(TestRangeRequestServer):
934
707
"""Test readv against a server which do not accept range requests"""
936
_req_handler_class = NoRangeRequestHandler
939
class MultipleRangeWithoutContentLengthRequestHandler(
940
http_server.TestingHTTPRequestHandler):
941
"""Reply to multiple range requests without content length header."""
943
def get_multiple_ranges(self, file, file_size, ranges):
944
self.send_response(206)
945
self.send_header('Accept-Ranges', 'bytes')
946
# XXX: this is strange; the 'random' name below seems undefined and
947
# yet the tests pass -- mbp 2010-10-11 bug 658773
948
boundary = "%d" % random.randint(0,0x7FFFFFFF)
949
self.send_header("Content-Type",
950
"multipart/byteranges; boundary=%s" % boundary)
952
for (start, end) in ranges:
953
self.wfile.write("--%s\r\n" % boundary)
954
self.send_header("Content-type", 'application/octet-stream')
955
self.send_header("Content-Range", "bytes %d-%d/%d" % (start,
959
self.send_range_content(file, start, end - start + 1)
961
self.wfile.write("--%s\r\n" % boundary)
964
class TestMultipleRangeWithoutContentLengthServer(TestRangeRequestServer):
966
_req_handler_class = MultipleRangeWithoutContentLengthRequestHandler
969
class TruncatedMultipleRangeRequestHandler(
970
http_server.TestingHTTPRequestHandler):
971
"""Reply to multiple range requests truncating the last ones.
973
This server generates responses whose Content-Length describes all the
974
ranges, but fail to include the last ones leading to client short reads.
975
This has been observed randomly with lighttpd (bug #179368).
709
def create_transport_readonly_server(self):
710
return HttpServer(NoRangeRequestHandler)
713
class TestNoRangeRequestServer_urllib(TestNoRangeRequestServer,
714
TestCaseWithWebserver):
715
"""Tests range requests refusing server for urllib implementation"""
717
_transport = HttpTransport_urllib
720
class TestNoRangeRequestServer_pycurl(TestWithTransport_pycurl,
721
TestNoRangeRequestServer,
722
TestCaseWithWebserver):
723
"""Tests range requests refusing server for pycurl implementation"""
726
class TestLimitedRangeRequestServer(object):
727
"""Tests readv requests against server that errors out on too much ranges.
729
This MUST be used by daughter classes that also inherit from
730
TestCaseWithWebserver.
732
We can't inherit directly from TestCaseWithWebserver or the
733
test framework will try to create an instance which cannot
734
run, its implementation being incomplete.
978
_truncated_ranges = 2
980
def get_multiple_ranges(self, file, file_size, ranges):
981
self.send_response(206)
982
self.send_header('Accept-Ranges', 'bytes')
984
self.send_header('Content-Type',
985
'multipart/byteranges; boundary=%s' % boundary)
986
boundary_line = '--%s\r\n' % boundary
987
# Calculate the Content-Length
989
for (start, end) in ranges:
990
content_length += len(boundary_line)
991
content_length += self._header_line_length(
992
'Content-type', 'application/octet-stream')
993
content_length += self._header_line_length(
994
'Content-Range', 'bytes %d-%d/%d' % (start, end, file_size))
995
content_length += len('\r\n') # end headers
996
content_length += end - start # + 1
997
content_length += len(boundary_line)
998
self.send_header('Content-length', content_length)
1001
# Send the multipart body
1003
for (start, end) in ranges:
1004
self.wfile.write(boundary_line)
1005
self.send_header('Content-type', 'application/octet-stream')
1006
self.send_header('Content-Range', 'bytes %d-%d/%d'
1007
% (start, end, file_size))
1009
if cur + self._truncated_ranges >= len(ranges):
1010
# Abruptly ends the response and close the connection
1011
self.close_connection = 1
1013
self.send_range_content(file, start, end - start + 1)
1016
self.wfile.write(boundary_line)
1019
class TestTruncatedMultipleRangeServer(TestSpecificRequestHandler):
1021
_req_handler_class = TruncatedMultipleRangeRequestHandler
1024
super(TestTruncatedMultipleRangeServer, self).setUp()
1025
self.build_tree_contents([('a', '0123456789')],)
1027
def test_readv_with_short_reads(self):
1028
server = self.get_readonly_server()
1029
t = self.get_readonly_transport()
1030
# Force separate ranges for each offset
1031
t._bytes_to_read_before_seek = 0
1032
ireadv = iter(t.readv('a', ((0, 1), (2, 1), (4, 2), (9, 1))))
1033
self.assertEqual((0, '0'), ireadv.next())
1034
self.assertEqual((2, '2'), ireadv.next())
1035
if not self._testing_pycurl():
1036
# Only one request have been issued so far (except for pycurl that
1037
# try to read the whole response at once)
1038
self.assertEqual(1, server.GET_request_nb)
1039
self.assertEqual((4, '45'), ireadv.next())
1040
self.assertEqual((9, '9'), ireadv.next())
1041
# Both implementations issue 3 requests but:
1042
# - urllib does two multiple (4 ranges, then 2 ranges) then a single
1044
# - pycurl does two multiple (4 ranges, 4 ranges) then a single range
1045
self.assertEqual(3, server.GET_request_nb)
1046
# Finally the client have tried a single range request and stays in
1048
self.assertEqual('single', t._range_hint)
1051
class LimitedRangeRequestHandler(http_server.TestingHTTPRequestHandler):
1052
"""Errors out when range specifiers exceed the limit"""
1054
def get_multiple_ranges(self, file, file_size, ranges):
1055
"""Refuses the multiple ranges request"""
1056
tcs = self.server.test_case_server
1057
if tcs.range_limit is not None and len(ranges) > tcs.range_limit:
1059
# Emulate apache behavior
1060
self.send_error(400, "Bad Request")
1062
return http_server.TestingHTTPRequestHandler.get_multiple_ranges(
1063
self, file, file_size, ranges)
1066
class LimitedRangeHTTPServer(http_server.HttpServer):
1067
"""An HttpServer erroring out on requests with too much range specifiers"""
1069
def __init__(self, request_handler=LimitedRangeRequestHandler,
1070
protocol_version=None,
1072
http_server.HttpServer.__init__(self, request_handler,
1073
protocol_version=protocol_version)
1074
self.range_limit = range_limit
1077
class TestLimitedRangeRequestServer(http_utils.TestCaseWithWebserver):
1078
"""Tests readv requests against a server erroring out on too much ranges."""
1080
scenarios = multiply_scenarios(
1081
vary_by_http_client_implementation(),
1082
vary_by_http_protocol_version(),
1085
# Requests with more range specifiers will error out
1088
739
def create_transport_readonly_server(self):
1089
return LimitedRangeHTTPServer(range_limit=self.range_limit,
1090
protocol_version=self._protocol_version)
740
# Requests with more range specifiers will error out
741
return LimitedRangeHTTPServer(range_limit=self.range_limit)
743
def get_transport(self):
744
return self._transport(self.get_readonly_server().get_url())
1092
746
def setUp(self):
1093
http_utils.TestCaseWithWebserver.setUp(self)
747
TestCaseWithWebserver.setUp(self)
1094
748
# We need to manipulate ranges that correspond to real chunks in the
1095
749
# response, so we build a content appropriately.
1096
750
filler = ''.join(['abcdefghij' for x in range(102)])
1098
752
self.build_tree_contents([('a', content)],)
1100
754
def test_few_ranges(self):
1101
t = self.get_readonly_transport()
755
t = self.get_transport()
1102
756
l = list(t.readv('a', ((0, 4), (1024, 4), )))
1103
757
self.assertEqual(l[0], (0, '0000'))
1104
758
self.assertEqual(l[1], (1024, '0001'))
1105
759
self.assertEqual(1, self.get_readonly_server().GET_request_nb)
1107
761
def test_more_ranges(self):
1108
t = self.get_readonly_transport()
762
t = self.get_transport()
1109
763
l = list(t.readv('a', ((0, 4), (1024, 4), (4096, 4), (8192, 4))))
1110
764
self.assertEqual(l[0], (0, '0000'))
1111
765
self.assertEqual(l[1], (1024, '0001'))
1112
766
self.assertEqual(l[2], (4096, '0004'))
1113
767
self.assertEqual(l[3], (8192, '0008'))
1114
768
# The server will refuse to serve the first request (too much ranges),
1115
# a second request will succeed.
769
# a second request will succeeds.
1116
770
self.assertEqual(2, self.get_readonly_server().GET_request_nb)
773
class TestLimitedRangeRequestServer_urllib(TestLimitedRangeRequestServer,
774
TestCaseWithWebserver):
775
"""Tests limited range requests server for urllib implementation"""
777
_transport = HttpTransport_urllib
780
class TestLimitedRangeRequestServer_pycurl(TestWithTransport_pycurl,
781
TestLimitedRangeRequestServer,
782
TestCaseWithWebserver):
783
"""Tests limited range requests server for pycurl implementation"""
1119
787
class TestHttpProxyWhiteBox(tests.TestCase):
1120
788
"""Whitebox test proxy http authorization.
1122
790
Only the urllib implementation is tested here.
794
tests.TestCase.setUp(self)
800
def _install_env(self, env):
801
for name, value in env.iteritems():
802
self._old_env[name] = osutils.set_or_unset_env(name, value)
804
def _restore_env(self):
805
for name, value in self._old_env.iteritems():
806
osutils.set_or_unset_env(name, value)
1125
808
def _proxied_request(self):
1126
handler = _urllib2_wrappers.ProxyHandler()
1127
request = _urllib2_wrappers.Request('GET', 'http://baz/buzzle')
809
handler = ProxyHandler()
810
request = Request('GET','http://baz/buzzle')
1128
811
handler.set_proxy(request, 'http')
1131
def assertEvaluateProxyBypass(self, expected, host, no_proxy):
1132
handler = _urllib2_wrappers.ProxyHandler()
1133
self.assertEquals(expected,
1134
handler.evaluate_proxy_bypass(host, no_proxy))
1136
814
def test_empty_user(self):
1137
self.overrideEnv('http_proxy', 'http://bar.com')
1138
request = self._proxied_request()
1139
self.assertFalse(request.headers.has_key('Proxy-authorization'))
1141
def test_user_with_at(self):
1142
self.overrideEnv('http_proxy',
1143
'http://username@domain:password@proxy_host:1234')
815
self._install_env({'http_proxy': 'http://bar.com'})
1144
816
request = self._proxied_request()
1145
817
self.assertFalse(request.headers.has_key('Proxy-authorization'))
1147
819
def test_invalid_proxy(self):
1148
820
"""A proxy env variable without scheme"""
1149
self.overrideEnv('http_proxy', 'host:1234')
821
self._install_env({'http_proxy': 'host:1234'})
1150
822
self.assertRaises(errors.InvalidURL, self._proxied_request)
1152
def test_evaluate_proxy_bypass_true(self):
1153
"""The host is not proxied"""
1154
self.assertEvaluateProxyBypass(True, 'example.com', 'example.com')
1155
self.assertEvaluateProxyBypass(True, 'bzr.example.com', '*example.com')
1157
def test_evaluate_proxy_bypass_false(self):
1158
"""The host is proxied"""
1159
self.assertEvaluateProxyBypass(False, 'bzr.example.com', None)
1161
def test_evaluate_proxy_bypass_unknown(self):
1162
"""The host is not explicitly proxied"""
1163
self.assertEvaluateProxyBypass(None, 'example.com', 'not.example.com')
1164
self.assertEvaluateProxyBypass(None, 'bzr.example.com', 'example.com')
1166
def test_evaluate_proxy_bypass_empty_entries(self):
1167
"""Ignore empty entries"""
1168
self.assertEvaluateProxyBypass(None, 'example.com', '')
1169
self.assertEvaluateProxyBypass(None, 'example.com', ',')
1170
self.assertEvaluateProxyBypass(None, 'example.com', 'foo,,bar')
1173
class TestProxyHttpServer(http_utils.TestCaseWithTwoWebservers):
825
class TestProxyHttpServer(object):
1174
826
"""Tests proxy server.
828
This MUST be used by daughter classes that also inherit from
829
TestCaseWithTwoWebservers.
831
We can't inherit directly from TestCaseWithTwoWebservers or
832
the test framework will try to create an instance which
833
cannot run, its implementation being incomplete.
1176
835
Be aware that we do not setup a real proxy here. Instead, we
1177
836
check that the *connection* goes through the proxy by serving
1178
837
different content (the faked proxy server append '-proxied'
1179
838
to the file names).
1182
scenarios = multiply_scenarios(
1183
vary_by_http_client_implementation(),
1184
vary_by_http_protocol_version(),
1187
841
# FIXME: We don't have an https server available, so we don't
1188
# test https connections. --vila toolongago
842
# test https connections.
1190
844
def setUp(self):
1191
super(TestProxyHttpServer, self).setUp()
1192
self.transport_secondary_server = http_utils.ProxyServer
845
TestCaseWithTwoWebservers.setUp(self)
1193
846
self.build_tree_contents([('foo', 'contents of foo\n'),
1194
847
('foo-proxied', 'proxied contents of foo\n')])
1195
848
# Let's setup some attributes for tests
1196
server = self.get_readonly_server()
1197
self.server_host_port = '%s:%d' % (server.host, server.port)
1198
if self._testing_pycurl():
1199
# Oh my ! pycurl does not check for the port as part of
1200
# no_proxy :-( So we just test the host part
1201
self.no_proxy_host = server.host
1203
self.no_proxy_host = self.server_host_port
849
self.server = self.get_readonly_server()
850
self.proxy_address = '%s:%d' % (self.server.host, self.server.port)
851
self.no_proxy_host = self.proxy_address
1204
852
# The secondary server is the proxy
1205
self.proxy_url = self.get_secondary_url()
1207
def _testing_pycurl(self):
1208
# TODO: This is duplicated for lots of the classes in this file
1209
return (features.pycurl.available()
1210
and self._transport == PyCurlTransport)
1212
def assertProxied(self):
1213
t = self.get_readonly_transport()
1214
self.assertEqual('proxied contents of foo\n', t.get('foo').read())
1216
def assertNotProxied(self):
1217
t = self.get_readonly_transport()
1218
self.assertEqual('contents of foo\n', t.get('foo').read())
853
self.proxy = self.get_secondary_server()
854
self.proxy_url = self.proxy.get_url()
857
def create_transport_secondary_server(self):
858
"""Creates an http server that will serve files with
859
'-proxied' appended to their names.
863
def _install_env(self, env):
864
for name, value in env.iteritems():
865
self._old_env[name] = osutils.set_or_unset_env(name, value)
867
def _restore_env(self):
868
for name, value in self._old_env.iteritems():
869
osutils.set_or_unset_env(name, value)
871
def proxied_in_env(self, env):
872
self._install_env(env)
873
url = self.server.get_url()
874
t = self._transport(url)
876
self.assertEqual(t.get('foo').read(), 'proxied contents of foo\n')
880
def not_proxied_in_env(self, env):
881
self._install_env(env)
882
url = self.server.get_url()
883
t = self._transport(url)
885
self.assertEqual(t.get('foo').read(), 'contents of foo\n')
1220
889
def test_http_proxy(self):
1221
self.overrideEnv('http_proxy', self.proxy_url)
1222
self.assertProxied()
890
self.proxied_in_env({'http_proxy': self.proxy_url})
1224
892
def test_HTTP_PROXY(self):
1225
if self._testing_pycurl():
1226
# pycurl does not check HTTP_PROXY for security reasons
1227
# (for use in a CGI context that we do not care
1228
# about. Should we ?)
1229
raise tests.TestNotApplicable(
1230
'pycurl does not check HTTP_PROXY for security reasons')
1231
self.overrideEnv('HTTP_PROXY', self.proxy_url)
1232
self.assertProxied()
893
self.proxied_in_env({'HTTP_PROXY': self.proxy_url})
1234
895
def test_all_proxy(self):
1235
self.overrideEnv('all_proxy', self.proxy_url)
1236
self.assertProxied()
896
self.proxied_in_env({'all_proxy': self.proxy_url})
1238
898
def test_ALL_PROXY(self):
1239
self.overrideEnv('ALL_PROXY', self.proxy_url)
1240
self.assertProxied()
899
self.proxied_in_env({'ALL_PROXY': self.proxy_url})
1242
901
def test_http_proxy_with_no_proxy(self):
1243
self.overrideEnv('no_proxy', self.no_proxy_host)
1244
self.overrideEnv('http_proxy', self.proxy_url)
1245
self.assertNotProxied()
902
self.not_proxied_in_env({'http_proxy': self.proxy_url,
903
'no_proxy': self.no_proxy_host})
1247
905
def test_HTTP_PROXY_with_NO_PROXY(self):
1248
if self._testing_pycurl():
1249
raise tests.TestNotApplicable(
1250
'pycurl does not check HTTP_PROXY for security reasons')
1251
self.overrideEnv('NO_PROXY', self.no_proxy_host)
1252
self.overrideEnv('HTTP_PROXY', self.proxy_url)
1253
self.assertNotProxied()
906
self.not_proxied_in_env({'HTTP_PROXY': self.proxy_url,
907
'NO_PROXY': self.no_proxy_host})
1255
909
def test_all_proxy_with_no_proxy(self):
1256
self.overrideEnv('no_proxy', self.no_proxy_host)
1257
self.overrideEnv('all_proxy', self.proxy_url)
1258
self.assertNotProxied()
910
self.not_proxied_in_env({'all_proxy': self.proxy_url,
911
'no_proxy': self.no_proxy_host})
1260
913
def test_ALL_PROXY_with_NO_PROXY(self):
1261
self.overrideEnv('NO_PROXY', self.no_proxy_host)
1262
self.overrideEnv('ALL_PROXY', self.proxy_url)
1263
self.assertNotProxied()
1265
def test_http_proxy_without_scheme(self):
1266
self.overrideEnv('http_proxy', self.server_host_port)
1267
if self._testing_pycurl():
1268
# pycurl *ignores* invalid proxy env variables. If that ever change
1269
# in the future, this test will fail indicating that pycurl do not
1270
# ignore anymore such variables.
1271
self.assertNotProxied()
1273
self.assertRaises(errors.InvalidURL, self.assertProxied)
1276
class TestRanges(http_utils.TestCaseWithWebserver):
1277
"""Test the Range header in GET methods."""
1279
scenarios = multiply_scenarios(
1280
vary_by_http_client_implementation(),
1281
vary_by_http_protocol_version(),
1285
http_utils.TestCaseWithWebserver.setUp(self)
914
self.not_proxied_in_env({'ALL_PROXY': self.proxy_url,
915
'NO_PROXY': self.no_proxy_host})
917
def test_http_proxy_without_scheme(self):
918
self.assertRaises(errors.InvalidURL,
920
{'http_proxy': self.proxy_address})
923
class TestProxyHttpServer_urllib(TestProxyHttpServer,
924
TestCaseWithTwoWebservers):
925
"""Tests proxy server for urllib implementation"""
927
_transport = HttpTransport_urllib
930
class TestProxyHttpServer_pycurl(TestWithTransport_pycurl,
932
TestCaseWithTwoWebservers):
933
"""Tests proxy server for pycurl implementation"""
936
TestProxyHttpServer.setUp(self)
937
# Oh my ! pycurl does not check for the port as part of
938
# no_proxy :-( So we just test the host part
939
self.no_proxy_host = 'localhost'
941
def test_HTTP_PROXY(self):
942
# pycurl does not check HTTP_PROXY for security reasons
943
# (for use in a CGI context that we do not care
944
# about. Should we ?)
945
raise tests.TestNotApplicable(
946
'pycurl does not check HTTP_PROXY for security reasons')
948
def test_HTTP_PROXY_with_NO_PROXY(self):
949
raise tests.TestNotApplicable(
950
'pycurl does not check HTTP_PROXY for security reasons')
952
def test_http_proxy_without_scheme(self):
953
# pycurl *ignores* invalid proxy env variables. If that
954
# ever change in the future, this test will fail
955
# indicating that pycurl do not ignore anymore such
957
self.not_proxied_in_env({'http_proxy': self.proxy_address})
960
class TestRanges(object):
961
"""Test the Range header in GET methods..
963
This MUST be used by daughter classes that also inherit from
964
TestCaseWithWebserver.
966
We can't inherit directly from TestCaseWithWebserver or the
967
test framework will try to create an instance which cannot
968
run, its implementation being incomplete.
972
TestCaseWithWebserver.setUp(self)
1286
973
self.build_tree_contents([('a', '0123456789')],)
1288
def create_transport_readonly_server(self):
1289
return http_server.HttpServer(protocol_version=self._protocol_version)
974
server = self.get_readonly_server()
975
self.transport = self._transport(server.get_url())
1291
977
def _file_contents(self, relpath, ranges):
1292
t = self.get_readonly_transport()
1293
978
offsets = [ (start, end - start + 1) for start, end in ranges]
1294
coalesce = t._coalesce_offsets
979
coalesce = self.transport._coalesce_offsets
1295
980
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
1296
code, data = t._get(relpath, coalesced)
981
code, data = self.transport._get(relpath, coalesced)
1297
982
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
1298
983
for start, end in ranges:
1299
984
data.seek(start)
1300
985
yield data.read(end - start + 1)
1302
987
def _file_tail(self, relpath, tail_amount):
1303
t = self.get_readonly_transport()
1304
code, data = t._get(relpath, [], tail_amount)
988
code, data = self.transport._get(relpath, [], tail_amount)
1305
989
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
1306
data.seek(-tail_amount, 2)
990
data.seek(-tail_amount + 1, 2)
1307
991
return data.read(tail_amount)
1309
993
def test_range_header(self):
1311
995
map(self.assertEqual,['0', '234'],
1312
996
list(self._file_contents('a', [(0,0), (2,4)])),)
1314
def test_range_header_tail(self):
1315
998
self.assertEqual('789', self._file_tail('a', 3))
1317
def test_syntactically_invalid_range_header(self):
1318
self.assertListRaises(errors.InvalidHttpRange,
999
# Syntactically invalid range
1000
self.assertListRaises(errors.InvalidRange,
1319
1001
self._file_contents, 'a', [(4, 3)])
1321
def test_semantically_invalid_range_header(self):
1322
self.assertListRaises(errors.InvalidHttpRange,
1002
# Semantically invalid range
1003
self.assertListRaises(errors.InvalidRange,
1323
1004
self._file_contents, 'a', [(42, 128)])
1326
class TestHTTPRedirections(http_utils.TestCaseWithRedirectedWebserver):
1327
"""Test redirection between http servers."""
1329
scenarios = multiply_scenarios(
1330
vary_by_http_client_implementation(),
1331
vary_by_http_protocol_version(),
1007
class TestRanges_urllib(TestRanges, TestCaseWithWebserver):
1008
"""Test the Range header in GET methods for urllib implementation"""
1010
_transport = HttpTransport_urllib
1013
class TestRanges_pycurl(TestWithTransport_pycurl,
1015
TestCaseWithWebserver):
1016
"""Test the Range header in GET methods for pycurl implementation"""
1019
class TestHTTPRedirections(object):
1020
"""Test redirection between http servers.
1022
This MUST be used by daughter classes that also inherit from
1023
TestCaseWithRedirectedWebserver.
1025
We can't inherit directly from TestCaseWithTwoWebservers or the
1026
test framework will try to create an instance which cannot
1027
run, its implementation being incomplete.
1030
def create_transport_secondary_server(self):
1031
"""Create the secondary server redirecting to the primary server"""
1032
new = self.get_readonly_server()
1034
redirecting = HTTPServerRedirecting()
1035
redirecting.redirect_to(new.host, new.port)
1334
1038
def setUp(self):
1335
1039
super(TestHTTPRedirections, self).setUp()
1425
1114
('5/a', 'redirected 5 times'),
1117
self.old_transport = self._transport(self.old_server.get_url())
1119
def setup_redirected_request(self):
1120
self.original_class = _urllib2_wrappers.Request
1121
_urllib2_wrappers.Request = RedirectedRequest
1123
def cleanup_redirected_request(self):
1124
_urllib2_wrappers.Request = self.original_class
1126
def create_transport_secondary_server(self):
1127
"""Create the secondary server, redirections are defined in the tests"""
1128
return HTTPServerRedirecting()
1428
1130
def test_one_redirection(self):
1429
t = self.get_old_transport()
1430
req = RedirectedRequest('GET', t._remote_path('a'))
1131
t = self.old_transport
1133
req = RedirectedRequest('GET', t.abspath('a'))
1134
req.follow_redirections = True
1431
1135
new_prefix = 'http://%s:%s' % (self.new_server.host,
1432
1136
self.new_server.port)
1433
1137
self.old_server.redirections = \
1434
1138
[('(.*)', r'%s/1\1' % (new_prefix), 301),]
1435
self.assertEqual('redirected once', t._perform(req).read())
1139
self.assertEquals('redirected once',t._perform(req).read())
1437
1141
def test_five_redirections(self):
1438
t = self.get_old_transport()
1439
req = RedirectedRequest('GET', t._remote_path('a'))
1142
t = self.old_transport
1144
req = RedirectedRequest('GET', t.abspath('a'))
1145
req.follow_redirections = True
1440
1146
old_prefix = 'http://%s:%s' % (self.old_server.host,
1441
1147
self.old_server.port)
1442
1148
new_prefix = 'http://%s:%s' % (self.new_server.host,
1443
1149
self.new_server.port)
1444
self.old_server.redirections = [
1445
('/1(.*)', r'%s/2\1' % (old_prefix), 302),
1446
('/2(.*)', r'%s/3\1' % (old_prefix), 303),
1447
('/3(.*)', r'%s/4\1' % (old_prefix), 307),
1448
('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1449
('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1451
self.assertEqual('redirected 5 times', t._perform(req).read())
1454
class TestDoCatchRedirections(http_utils.TestCaseWithRedirectedWebserver):
1455
"""Test transport.do_catching_redirections."""
1457
scenarios = multiply_scenarios(
1458
vary_by_http_client_implementation(),
1459
vary_by_http_protocol_version(),
1150
self.old_server.redirections = \
1151
[('/1(.*)', r'%s/2\1' % (old_prefix), 302),
1152
('/2(.*)', r'%s/3\1' % (old_prefix), 303),
1153
('/3(.*)', r'%s/4\1' % (old_prefix), 307),
1154
('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1155
('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1157
self.assertEquals('redirected 5 times',t._perform(req).read())
1160
class TestDoCatchRedirections(TestCaseWithRedirectedWebserver):
1161
"""Test transport.do_catching_redirections.
1163
We arbitrarily choose to use urllib transports
1166
_transport = HttpTransport_urllib
1462
1168
def setUp(self):
1463
1169
super(TestDoCatchRedirections, self).setUp()
1464
1170
self.build_tree_contents([('a', '0123456789'),],)
1465
cleanup_http_redirection_connections(self)
1467
self.old_transport = self.get_old_transport()
1172
self.old_transport = self._transport(self.old_server.get_url())
1174
def get_a(self, transport):
1175
return transport.get('a')
1472
1177
def test_no_redirection(self):
1473
t = self.get_new_transport()
1178
t = self._transport(self.new_server.get_url())
1475
1180
# We use None for redirected so that we fail if redirected
1476
self.assertEqual('0123456789',
1477
transport.do_catching_redirections(
1478
self.get_a, t, None).read())
1181
self.assertEquals('0123456789',
1182
do_catching_redirections(self.get_a, t, None).read())
1480
1184
def test_one_redirection(self):
1481
1185
self.redirections = 0
1483
def redirected(t, exception, redirection_notice):
1187
def redirected(transport, exception, redirection_notice):
1484
1188
self.redirections += 1
1485
redirected_t = t._redirected_to(exception.source, exception.target)
1189
dir, file = urlutils.split(exception.target)
1190
return self._transport(dir)
1488
self.assertEqual('0123456789',
1489
transport.do_catching_redirections(
1490
self.get_a, self.old_transport, redirected).read())
1491
self.assertEqual(1, self.redirections)
1192
self.assertEquals('0123456789',
1193
do_catching_redirections(self.get_a,
1197
self.assertEquals(1, self.redirections)
1493
1199
def test_redirection_loop(self):
1696
1326
# Only one 'Authentication Required' error should occur
1697
1327
self.assertEqual(1, self.server.auth_required_errors)
1699
def test_changing_nonce(self):
1700
if self._auth_server not in (http_utils.HTTPDigestAuthServer,
1701
http_utils.ProxyDigestAuthServer):
1702
raise tests.TestNotApplicable('HTTP/proxy auth digest only test')
1703
if self._testing_pycurl():
1704
raise tests.KnownFailure(
1705
'pycurl does not handle a nonce change')
1706
self.server.add_user('joe', 'foo')
1707
t = self.get_user_transport('joe', 'foo')
1708
self.assertEqual('contents of a\n', t.get('a').read())
1709
self.assertEqual('contents of b\n', t.get('b').read())
1710
# Only one 'Authentication Required' error should have
1712
self.assertEqual(1, self.server.auth_required_errors)
1713
# The server invalidates the current nonce
1714
self.server.auth_nonce = self.server.auth_nonce + '. No, now!'
1715
self.assertEqual('contents of a\n', t.get('a').read())
1716
# Two 'Authentication Required' errors should occur (the
1717
# initial 'who are you' and a second 'who are you' with the new nonce)
1718
self.assertEqual(2, self.server.auth_required_errors)
1720
def test_user_from_auth_conf(self):
1721
if self._testing_pycurl():
1722
raise tests.TestNotApplicable(
1723
'pycurl does not support authentication.conf')
1726
self.server.add_user(user, password)
1727
_setup_authentication_config(scheme='http', port=self.server.port,
1728
user=user, password=password)
1729
t = self.get_user_transport(None, None)
1730
# Issue a request to the server to connect
1731
self.assertEqual('contents of a\n', t.get('a').read())
1732
# Only one 'Authentication Required' error should occur
1733
self.assertEqual(1, self.server.auth_required_errors)
1735
def test_no_credential_leaks_in_log(self):
1736
self.overrideAttr(debug, 'debug_flags', set(['http']))
1738
password = 'very-sensitive-password'
1739
self.server.add_user(user, password)
1740
t = self.get_user_transport(user, password)
1741
# Capture the debug calls to mutter
1744
lines = args[0] % args[1:]
1745
# Some calls output multiple lines, just split them now since we
1746
# care about a single one later.
1747
self.mutters.extend(lines.splitlines())
1748
self.overrideAttr(trace, 'mutter', mutter)
1749
# Issue a request to the server to connect
1750
self.assertEqual(True, t.has('a'))
1751
# Only one 'Authentication Required' error should occur
1752
self.assertEqual(1, self.server.auth_required_errors)
1753
# Since the authentification succeeded, there should be a corresponding
1755
sent_auth_headers = [line for line in self.mutters
1756
if line.startswith('> %s' % (self._auth_header,))]
1757
self.assertLength(1, sent_auth_headers)
1758
self.assertStartsWith(sent_auth_headers[0],
1759
'> %s: <masked>' % (self._auth_header,))
1331
class TestHTTPAuth(TestAuth):
1332
"""Test HTTP authentication schemes.
1334
Daughter classes MUST inherit from TestCaseWithWebserver too.
1337
_auth_header = 'Authorization'
1340
TestCaseWithWebserver.setUp(self)
1341
self.server = self.get_readonly_server()
1342
TestAuth.setUp(self)
1344
def get_user_transport(self, user=None, password=None):
1345
return self._transport(self.get_user_url(user, password))
1762
1348
class TestProxyAuth(TestAuth):
1763
1349
"""Test proxy authentication schemes.
1765
This inherits from TestAuth to tweak the setUp and filter some failing
1351
Daughter classes MUST also inherit from TestCaseWithWebserver.
1353
_auth_header = 'Proxy-authorization'
1354
_password_prompt_prefix = 'Proxy '
1769
scenarios = multiply_scenarios(
1770
vary_by_http_client_implementation(),
1771
vary_by_http_protocol_version(),
1772
vary_by_http_proxy_auth_scheme(),
1775
1357
def setUp(self):
1776
super(TestProxyAuth, self).setUp()
1358
TestCaseWithWebserver.setUp(self)
1359
self.server = self.get_readonly_server()
1361
self.addCleanup(self._restore_env)
1362
TestAuth.setUp(self)
1777
1363
# Override the contents to avoid false positives
1778
1364
self.build_tree_contents([('a', 'not proxied contents of a\n'),
1779
1365
('b', 'not proxied contents of b\n'),
1781
1367
('b-proxied', 'contents of b\n'),
1784
def get_user_transport(self, user, password):
1785
self.overrideEnv('all_proxy', self.get_user_url(user, password))
1786
return TestAuth.get_user_transport(self, user, password)
1788
def test_empty_pass(self):
1789
if self._testing_pycurl():
1791
if pycurl.version_info()[1] < '7.16.0':
1792
raise tests.KnownFailure(
1793
'pycurl < 7.16.0 does not handle empty proxy passwords')
1794
super(TestProxyAuth, self).test_empty_pass()
1797
class SampleSocket(object):
1798
"""A socket-like object for use in testing the HTTP request handler."""
1800
def __init__(self, socket_read_content):
1801
"""Constructs a sample socket.
1803
:param socket_read_content: a byte sequence
1805
# Use plain python StringIO so we can monkey-patch the close method to
1806
# not discard the contents.
1807
from StringIO import StringIO
1808
self.readfile = StringIO(socket_read_content)
1809
self.writefile = StringIO()
1810
self.writefile.close = lambda: None
1811
self.close = lambda: None
1813
def makefile(self, mode='r', bufsize=None):
1815
return self.readfile
1817
return self.writefile
1820
class SmartHTTPTunnellingTest(tests.TestCaseWithTransport):
1822
scenarios = multiply_scenarios(
1823
vary_by_http_client_implementation(),
1824
vary_by_http_protocol_version(),
1828
super(SmartHTTPTunnellingTest, self).setUp()
1829
# We use the VFS layer as part of HTTP tunnelling tests.
1830
self.overrideEnv('BZR_NO_SMART_VFS', None)
1831
self.transport_readonly_server = http_utils.HTTPServerWithSmarts
1832
self.http_server = self.get_readonly_server()
1834
def create_transport_readonly_server(self):
1835
server = http_utils.HTTPServerWithSmarts(
1836
protocol_version=self._protocol_version)
1837
server._url_protocol = self._url_protocol
1840
def test_open_bzrdir(self):
1841
branch = self.make_branch('relpath')
1842
url = self.http_server.get_url() + 'relpath'
1843
bd = bzrdir.BzrDir.open(url)
1844
self.addCleanup(bd.transport.disconnect)
1845
self.assertIsInstance(bd, _mod_remote.RemoteBzrDir)
1847
def test_bulk_data(self):
1848
# We should be able to send and receive bulk data in a single message.
1849
# The 'readv' command in the smart protocol both sends and receives
1850
# bulk data, so we use that.
1851
self.build_tree(['data-file'])
1852
http_transport = transport.get_transport(self.http_server.get_url())
1853
medium = http_transport.get_smart_medium()
1854
# Since we provide the medium, the url below will be mostly ignored
1855
# during the test, as long as the path is '/'.
1856
remote_transport = remote.RemoteTransport('bzr://fake_host/',
1859
[(0, "c")], list(remote_transport.readv("data-file", [(0,1)])))
1861
def test_http_send_smart_request(self):
1863
post_body = 'hello\n'
1864
expected_reply_body = 'ok\x012\n'
1866
http_transport = transport.get_transport(self.http_server.get_url())
1867
medium = http_transport.get_smart_medium()
1868
response = medium.send_http_smart_request(post_body)
1869
reply_body = response.read()
1870
self.assertEqual(expected_reply_body, reply_body)
1872
def test_smart_http_server_post_request_handler(self):
1873
httpd = self.http_server.server
1875
socket = SampleSocket(
1876
'POST /.bzr/smart %s \r\n' % self._protocol_version
1877
# HTTP/1.1 posts must have a Content-Length (but it doesn't hurt
1879
+ 'Content-Length: 6\r\n'
1882
# Beware: the ('localhost', 80) below is the
1883
# client_address parameter, but we don't have one because
1884
# we have defined a socket which is not bound to an
1885
# address. The test framework never uses this client
1886
# address, so far...
1887
request_handler = http_utils.SmartRequestHandler(socket,
1890
response = socket.writefile.getvalue()
1891
self.assertStartsWith(response, '%s 200 ' % self._protocol_version)
1892
# This includes the end of the HTTP headers, and all the body.
1893
expected_end_of_response = '\r\n\r\nok\x012\n'
1894
self.assertEndsWith(response, expected_end_of_response)
1897
class ForbiddenRequestHandler(http_server.TestingHTTPRequestHandler):
1898
"""No smart server here request handler."""
1901
self.send_error(403, "Forbidden")
1904
class SmartClientAgainstNotSmartServer(TestSpecificRequestHandler):
1905
"""Test smart client behaviour against an http server without smarts."""
1907
_req_handler_class = ForbiddenRequestHandler
1909
def test_probe_smart_server(self):
1910
"""Test error handling against server refusing smart requests."""
1911
t = self.get_readonly_transport()
1912
# No need to build a valid smart request here, the server will not even
1913
# try to interpret it.
1914
self.assertRaises(errors.SmartProtocolError,
1915
t.get_smart_medium().send_http_smart_request,
1919
class Test_redirected_to(tests.TestCase):
1921
scenarios = vary_by_http_client_implementation()
1923
def test_redirected_to_subdir(self):
1924
t = self._transport('http://www.example.com/foo')
1925
r = t._redirected_to('http://www.example.com/foo',
1926
'http://www.example.com/foo/subdir')
1927
self.assertIsInstance(r, type(t))
1928
# Both transports share the some connection
1929
self.assertEqual(t._get_connection(), r._get_connection())
1931
def test_redirected_to_self_with_slash(self):
1932
t = self._transport('http://www.example.com/foo')
1933
r = t._redirected_to('http://www.example.com/foo',
1934
'http://www.example.com/foo/')
1935
self.assertIsInstance(r, type(t))
1936
# Both transports share the some connection (one can argue that we
1937
# should return the exact same transport here, but that seems
1939
self.assertEqual(t._get_connection(), r._get_connection())
1941
def test_redirected_to_host(self):
1942
t = self._transport('http://www.example.com/foo')
1943
r = t._redirected_to('http://www.example.com/foo',
1944
'http://foo.example.com/foo/subdir')
1945
self.assertIsInstance(r, type(t))
1947
def test_redirected_to_same_host_sibling_protocol(self):
1948
t = self._transport('http://www.example.com/foo')
1949
r = t._redirected_to('http://www.example.com/foo',
1950
'https://www.example.com/foo')
1951
self.assertIsInstance(r, type(t))
1953
def test_redirected_to_same_host_different_protocol(self):
1954
t = self._transport('http://www.example.com/foo')
1955
r = t._redirected_to('http://www.example.com/foo',
1956
'ftp://www.example.com/foo')
1957
self.assertNotEquals(type(r), type(t))
1959
def test_redirected_to_different_host_same_user(self):
1960
t = self._transport('http://joe@www.example.com/foo')
1961
r = t._redirected_to('http://www.example.com/foo',
1962
'https://foo.example.com/foo')
1963
self.assertIsInstance(r, type(t))
1964
self.assertEqual(t._user, r._user)
1967
class PredefinedRequestHandler(http_server.TestingHTTPRequestHandler):
1968
"""Request handler for a unique and pre-defined request.
1970
The only thing we care about here is how many bytes travel on the wire. But
1971
since we want to measure it for a real http client, we have to send it
1974
We expect to receive a *single* request nothing more (and we won't even
1975
check what request it is, we just measure the bytes read until an empty
1979
def _handle_one_request(self):
1980
tcs = self.server.test_case_server
1981
requestline = self.rfile.readline()
1982
headers = self.MessageClass(self.rfile, 0)
1983
# We just read: the request, the headers, an empty line indicating the
1984
# end of the headers.
1985
bytes_read = len(requestline)
1986
for line in headers.headers:
1987
bytes_read += len(line)
1988
bytes_read += len('\r\n')
1989
if requestline.startswith('POST'):
1990
# The body should be a single line (or we don't know where it ends
1991
# and we don't want to issue a blocking read)
1992
body = self.rfile.readline()
1993
bytes_read += len(body)
1994
tcs.bytes_read = bytes_read
1996
# We set the bytes written *before* issuing the write, the client is
1997
# supposed to consume every produced byte *before* checking that value.
1999
# Doing the oppposite may lead to test failure: we may be interrupted
2000
# after the write but before updating the value. The client can then
2001
# continue and read the value *before* we can update it. And yes,
2002
# this has been observed -- vila 20090129
2003
tcs.bytes_written = len(tcs.canned_response)
2004
self.wfile.write(tcs.canned_response)
2007
class ActivityServerMixin(object):
2009
def __init__(self, protocol_version):
2010
super(ActivityServerMixin, self).__init__(
2011
request_handler=PredefinedRequestHandler,
2012
protocol_version=protocol_version)
2013
# Bytes read and written by the server
2015
self.bytes_written = 0
2016
self.canned_response = None
2019
class ActivityHTTPServer(ActivityServerMixin, http_server.HttpServer):
2023
if tests.HTTPSServerFeature.available():
2024
from bzrlib.tests import https_server
2025
class ActivityHTTPSServer(ActivityServerMixin, https_server.HTTPSServer):
2029
class TestActivityMixin(object):
2030
"""Test socket activity reporting.
2032
We use a special purpose server to control the bytes sent and received and
2033
be able to predict the activity on the client socket.
2037
tests.TestCase.setUp(self)
2038
self.server = self._activity_server(self._protocol_version)
2039
self.server.start_server()
2040
_activities = {} # Don't close over self and create a cycle
2041
def report_activity(t, bytes, direction):
2042
count = _activities.get(direction, 0)
2044
_activities[direction] = count
2045
self.activities = _activities
2047
# We override at class level because constructors may propagate the
2048
# bound method and render instance overriding ineffective (an
2049
# alternative would be to define a specific ui factory instead...)
2050
self.overrideAttr(self._transport, '_report_activity', report_activity)
2051
self.addCleanup(self.server.stop_server)
2053
def get_transport(self):
2054
t = self._transport(self.server.get_url())
2055
# FIXME: Needs cleanup -- vila 20100611
2058
def assertActivitiesMatch(self):
2059
self.assertEqual(self.server.bytes_read,
2060
self.activities.get('write', 0), 'written bytes')
2061
self.assertEqual(self.server.bytes_written,
2062
self.activities.get('read', 0), 'read bytes')
2065
self.server.canned_response = '''HTTP/1.1 200 OK\r
2066
Date: Tue, 11 Jul 2006 04:32:56 GMT\r
2067
Server: Apache/2.0.54 (Fedora)\r
2068
Last-Modified: Sun, 23 Apr 2006 19:35:20 GMT\r
2069
ETag: "56691-23-38e9ae00"\r
2070
Accept-Ranges: bytes\r
2071
Content-Length: 35\r
2073
Content-Type: text/plain; charset=UTF-8\r
2075
Bazaar-NG meta directory, format 1
2077
t = self.get_transport()
2078
self.assertEqual('Bazaar-NG meta directory, format 1\n',
2079
t.get('foo/bar').read())
2080
self.assertActivitiesMatch()
2083
self.server.canned_response = '''HTTP/1.1 200 OK\r
2084
Server: SimpleHTTP/0.6 Python/2.5.2\r
2085
Date: Thu, 29 Jan 2009 20:21:47 GMT\r
2086
Content-type: application/octet-stream\r
2087
Content-Length: 20\r
2088
Last-Modified: Thu, 29 Jan 2009 20:21:47 GMT\r
2091
t = self.get_transport()
2092
self.assertTrue(t.has('foo/bar'))
2093
self.assertActivitiesMatch()
2095
def test_readv(self):
2096
self.server.canned_response = '''HTTP/1.1 206 Partial Content\r
2097
Date: Tue, 11 Jul 2006 04:49:48 GMT\r
2098
Server: Apache/2.0.54 (Fedora)\r
2099
Last-Modified: Thu, 06 Jul 2006 20:22:05 GMT\r
2100
ETag: "238a3c-16ec2-805c5540"\r
2101
Accept-Ranges: bytes\r
2102
Content-Length: 1534\r
2104
Content-Type: multipart/byteranges; boundary=418470f848b63279b\r
2107
--418470f848b63279b\r
2108
Content-type: text/plain; charset=UTF-8\r
2109
Content-range: bytes 0-254/93890\r
2111
mbp@sourcefrog.net-20050309040815-13242001617e4a06
2112
mbp@sourcefrog.net-20050309040929-eee0eb3e6d1e7627
2113
mbp@sourcefrog.net-20050309040957-6cad07f466bb0bb8
2114
mbp@sourcefrog.net-20050309041501-c840e09071de3b67
2115
mbp@sourcefrog.net-20050309044615-c24a3250be83220a
2117
--418470f848b63279b\r
2118
Content-type: text/plain; charset=UTF-8\r
2119
Content-range: bytes 1000-2049/93890\r
2122
mbp@sourcefrog.net-20050311063625-07858525021f270b
2123
mbp@sourcefrog.net-20050311231934-aa3776aff5200bb9
2124
mbp@sourcefrog.net-20050311231953-73aeb3a131c3699a
2125
mbp@sourcefrog.net-20050311232353-f5e33da490872c6a
2126
mbp@sourcefrog.net-20050312071639-0a8f59a34a024ff0
2127
mbp@sourcefrog.net-20050312073432-b2c16a55e0d6e9fb
2128
mbp@sourcefrog.net-20050312073831-a47c3335ece1920f
2129
mbp@sourcefrog.net-20050312085412-13373aa129ccbad3
2130
mbp@sourcefrog.net-20050313052251-2bf004cb96b39933
2131
mbp@sourcefrog.net-20050313052856-3edd84094687cb11
2132
mbp@sourcefrog.net-20050313053233-e30a4f28aef48f9d
2133
mbp@sourcefrog.net-20050313053853-7c64085594ff3072
2134
mbp@sourcefrog.net-20050313054757-a86c3f5871069e22
2135
mbp@sourcefrog.net-20050313061422-418f1f73b94879b9
2136
mbp@sourcefrog.net-20050313120651-497bd231b19df600
2137
mbp@sourcefrog.net-20050314024931-eae0170ef25a5d1a
2138
mbp@sourcefrog.net-20050314025438-d52099f915fe65fc
2139
mbp@sourcefrog.net-20050314025539-637a636692c055cf
2140
mbp@sourcefrog.net-20050314025737-55eb441f430ab4ba
2141
mbp@sourcefrog.net-20050314025901-d74aa93bb7ee8f62
2143
--418470f848b63279b--\r
2145
t = self.get_transport()
2146
# Remember that the request is ignored and that the ranges below
2147
# doesn't have to match the canned response.
2148
l = list(t.readv('/foo/bar', ((0, 255), (1000, 1050))))
2149
self.assertEqual(2, len(l))
2150
self.assertActivitiesMatch()
2152
def test_post(self):
2153
self.server.canned_response = '''HTTP/1.1 200 OK\r
2154
Date: Tue, 11 Jul 2006 04:32:56 GMT\r
2155
Server: Apache/2.0.54 (Fedora)\r
2156
Last-Modified: Sun, 23 Apr 2006 19:35:20 GMT\r
2157
ETag: "56691-23-38e9ae00"\r
2158
Accept-Ranges: bytes\r
2159
Content-Length: 35\r
2161
Content-Type: text/plain; charset=UTF-8\r
2163
lalala whatever as long as itsssss
2165
t = self.get_transport()
2166
# We must send a single line of body bytes, see
2167
# PredefinedRequestHandler._handle_one_request
2168
code, f = t._post('abc def end-of-body\n')
2169
self.assertEqual('lalala whatever as long as itsssss\n', f.read())
2170
self.assertActivitiesMatch()
2173
class TestActivity(tests.TestCase, TestActivityMixin):
2175
scenarios = multiply_scenarios(
2176
vary_by_http_activity(),
2177
vary_by_http_protocol_version(),
2181
TestActivityMixin.setUp(self)
2184
class TestNoReportActivity(tests.TestCase, TestActivityMixin):
2186
# Unlike TestActivity, we are really testing ReportingFileSocket and
2187
# ReportingSocket, so we don't need all the parametrization. Since
2188
# ReportingFileSocket and ReportingSocket are wrappers, it's easier to
2189
# test them through their use by the transport than directly (that's a
2190
# bit less clean but far more simpler and effective).
2191
_activity_server = ActivityHTTPServer
2192
_protocol_version = 'HTTP/1.1'
2195
self._transport =_urllib.HttpTransport_urllib
2196
TestActivityMixin.setUp(self)
2198
def assertActivitiesMatch(self):
2199
# Nothing to check here
2203
class TestAuthOnRedirected(http_utils.TestCaseWithRedirectedWebserver):
2204
"""Test authentication on the redirected http server."""
2206
scenarios = vary_by_http_protocol_version()
2208
_auth_header = 'Authorization'
2209
_password_prompt_prefix = ''
2210
_username_prompt_prefix = ''
2211
_auth_server = http_utils.HTTPBasicAuthServer
2212
_transport = _urllib.HttpTransport_urllib
2215
super(TestAuthOnRedirected, self).setUp()
2216
self.build_tree_contents([('a','a'),
2218
('1/a', 'redirected once'),
2220
new_prefix = 'http://%s:%s' % (self.new_server.host,
2221
self.new_server.port)
2222
self.old_server.redirections = [
2223
('(.*)', r'%s/1\1' % (new_prefix), 301),]
2224
self.old_transport = self.get_old_transport()
2225
self.new_server.add_user('joe', 'foo')
2226
cleanup_http_redirection_connections(self)
2228
def create_transport_readonly_server(self):
2229
server = self._auth_server(protocol_version=self._protocol_version)
2230
server._url_protocol = self._url_protocol
2236
def test_auth_on_redirected_via_do_catching_redirections(self):
2237
self.redirections = 0
2239
def redirected(t, exception, redirection_notice):
2240
self.redirections += 1
2241
redirected_t = t._redirected_to(exception.source, exception.target)
2242
self.addCleanup(redirected_t.disconnect)
2245
stdout = tests.StringIOWrapper()
2246
stderr = tests.StringIOWrapper()
2247
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',
2248
stdout=stdout, stderr=stderr)
2249
self.assertEqual('redirected once',
2250
transport.do_catching_redirections(
2251
self.get_a, self.old_transport, redirected).read())
2252
self.assertEqual(1, self.redirections)
2253
# stdin should be empty
2254
self.assertEqual('', ui.ui_factory.stdin.readline())
2255
# stdout should be empty, stderr will contains the prompts
2256
self.assertEqual('', stdout.getvalue())
2258
def test_auth_on_redirected_via_following_redirections(self):
2259
self.new_server.add_user('joe', 'foo')
2260
stdout = tests.StringIOWrapper()
2261
stderr = tests.StringIOWrapper()
2262
ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',
2263
stdout=stdout, stderr=stderr)
2264
t = self.old_transport
2265
req = RedirectedRequest('GET', t.abspath('a'))
2266
new_prefix = 'http://%s:%s' % (self.new_server.host,
2267
self.new_server.port)
2268
self.old_server.redirections = [
2269
('(.*)', r'%s/1\1' % (new_prefix), 301),]
2270
self.assertEqual('redirected once', t._perform(req).read())
2271
# stdin should be empty
2272
self.assertEqual('', ui.ui_factory.stdin.readline())
2273
# stdout should be empty, stderr will contains the prompts
2274
self.assertEqual('', stdout.getvalue())
1370
def get_user_transport(self, user=None, password=None):
1371
self._install_env({'all_proxy': self.get_user_url(user, password)})
1372
return self._transport(self.server.get_url())
1374
def _install_env(self, env):
1375
for name, value in env.iteritems():
1376
self._old_env[name] = osutils.set_or_unset_env(name, value)
1378
def _restore_env(self):
1379
for name, value in self._old_env.iteritems():
1380
osutils.set_or_unset_env(name, value)
1383
class TestHTTPBasicAuth(TestHTTPAuth, TestCaseWithWebserver):
1384
"""Test http basic authentication scheme"""
1386
_transport = HttpTransport_urllib
1388
def create_transport_readonly_server(self):
1389
return HTTPBasicAuthServer()
1392
class TestHTTPProxyBasicAuth(TestProxyAuth, TestCaseWithWebserver):
1393
"""Test proxy basic authentication scheme"""
1395
_transport = HttpTransport_urllib
1397
def create_transport_readonly_server(self):
1398
return ProxyBasicAuthServer()
1401
class TestDigestAuth(object):
1402
"""Digest Authentication specific tests"""
1404
def test_changing_nonce(self):
1405
self.server.add_user('joe', 'foo')
1406
t = self.get_user_transport('joe', 'foo')
1407
self.assertEqual('contents of a\n', t.get('a').read())
1408
self.assertEqual('contents of b\n', t.get('b').read())
1409
# Only one 'Authentication Required' error should have
1411
self.assertEqual(1, self.server.auth_required_errors)
1412
# The server invalidates the current nonce
1413
self.server.auth_nonce = self.server.auth_nonce + '. No, now!'
1414
self.assertEqual('contents of a\n', t.get('a').read())
1415
# Two 'Authentication Required' errors should occur (the
1416
# initial 'who are you' and a second 'who are you' with the new nonce)
1417
self.assertEqual(2, self.server.auth_required_errors)
1420
class TestHTTPDigestAuth(TestHTTPAuth, TestDigestAuth, TestCaseWithWebserver):
1421
"""Test http digest authentication scheme"""
1423
_transport = HttpTransport_urllib
1425
def create_transport_readonly_server(self):
1426
return HTTPDigestAuthServer()
1429
class TestHTTPProxyDigestAuth(TestProxyAuth, TestDigestAuth,
1430
TestCaseWithWebserver):
1431
"""Test proxy digest authentication scheme"""
1433
_transport = HttpTransport_urllib
1435
def create_transport_readonly_server(self):
1436
return ProxyDigestAuthServer()
added
removed
bzrlib/tests/test_http.py
