« back to all changes in this revision
Viewing changes to bzrlib/tests/test_http.py
- browse files at revision 2457
- compare with another revision
- download diff
- download tarball
- view history from revision 2457
- Committer: Tobias Toedter
- Date: 2008-02-10 08:01:48 UTC
- mfrom: (2438.1.783 +trunk)
- Revision ID: t.toedter@gmx.net-20080210080148-bg5rh61oq2zk2xw3
Merge trunk
- files removed:
- bzrlib/plugins/multiparent.py
- files renamed:
- bzrlib/tests/HttpServer.py => bzrlib/tests/http_server.py
- bzrlib/tests/HTTPTestUtil.py => bzrlib/tests/http_utils.py
- files modified:
- NEWS
added
removed
bzrlib/tests/test_http.py
14
14
# along with this program; if not, write to the Free Software
15
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
16
16
17
# FIXME: This test should be repeated for each available http client
18
# implementation; at the moment we have urllib and pycurl.
17
"""Tests for HTTP implementations.
18
19
This module defines a load_tests() method that parametrize tests classes for
20
transport implementation, http protocol versions and authentication schemes.
21
"""
19
22
20
23
# TODO: Should be renamed to bzrlib.transport.http.tests?
21
24
# TODO: What about renaming to bzrlib.tests.transport.http ?
22
25
23
26
from cStringIO import StringIO
27
import httplib
24
28
import os
25
29
import select
30
import SimpleHTTPServer
26
31
import socket
27
32
import sys
28
33
import threading
33
38
errors,
34
39
osutils,
35
40
tests,
41
transport,
36
42
ui,
37
43
urlutils,
38
44
)
39
from bzrlib.tests.HttpServer import (
40
HttpServer,
41
HttpServer_PyCurl,
42
HttpServer_urllib,
43
)
44
from bzrlib.tests.HTTPTestUtil import (
45
BadProtocolRequestHandler,
46
BadStatusRequestHandler,
47
ForbiddenRequestHandler,
48
HTTPBasicAuthServer,
49
HTTPDigestAuthServer,
50
HTTPServerRedirecting,
51
InvalidStatusRequestHandler,
52
LimitedRangeHTTPServer,
53
NoRangeRequestHandler,
54
ProxyBasicAuthServer,
55
ProxyDigestAuthServer,
56
ProxyServer,
57
SingleRangeRequestHandler,
58
SingleOnlyRangeRequestHandler,
59
TestCaseWithRedirectedWebserver,
60
TestCaseWithTwoWebservers,
61
TestCaseWithWebserver,
62
WallRequestHandler,
45
from bzrlib.tests import (
46
http_server,
47
http_utils,
63
48
)
64
49
from bzrlib.transport import (
65
_CoalescedOffset,
66
do_catching_redirections,
67
get_transport,
68
Transport,
50
http,
51
remote,
69
52
)
70
53
from bzrlib.transport.http import (
71
extract_auth,
72
HttpTransportBase,
54
_urllib,
73
55
_urllib2_wrappers,
74
56
)
75
from bzrlib.transport.http._urllib import HttpTransport_urllib
76
from bzrlib.transport.http._urllib2_wrappers import (
77
ProxyHandler,
78
Request,
79
)
57
58
59
try:
60
from bzrlib.transport.http._pycurl import PyCurlTransport
61
pycurl_present = True
62
except errors.DependencyNotPresent:
63
pycurl_present = False
64
65
66
class TransportAdapter(tests.TestScenarioApplier):
67
"""Generate the same test for each transport implementation."""
68
69
def __init__(self):
70
transport_scenarios = [
71
('urllib', dict(_transport=_urllib.HttpTransport_urllib,
72
_server=http_server.HttpServer_urllib,
73
_qualified_prefix='http+urllib',)),
74
]
75
if pycurl_present:
76
transport_scenarios.append(
77
('pycurl', dict(_transport=PyCurlTransport,
78
_server=http_server.HttpServer_PyCurl,
79
_qualified_prefix='http+pycurl',)))
80
self.scenarios = transport_scenarios
81
82
83
class TransportProtocolAdapter(TransportAdapter):
84
"""Generate the same test for each protocol implementation.
85
86
In addition to the transport adaptatation that we inherit from.
87
"""
88
89
def __init__(self):
90
super(TransportProtocolAdapter, self).__init__()
91
protocol_scenarios = [
92
('HTTP/1.0', dict(_protocol_version='HTTP/1.0')),
93
('HTTP/1.1', dict(_protocol_version='HTTP/1.1')),
94
]
95
self.scenarios = tests.multiply_scenarios(self.scenarios,
96
protocol_scenarios)
97
98
99
class TransportProtocolAuthenticationAdapter(TransportProtocolAdapter):
100
"""Generate the same test for each authentication scheme implementation.
101
102
In addition to the protocol adaptatation that we inherit from.
103
"""
104
105
def __init__(self):
106
super(TransportProtocolAuthenticationAdapter, self).__init__()
107
auth_scheme_scenarios = [
108
('basic', dict(_auth_scheme='basic')),
109
('digest', dict(_auth_scheme='digest')),
110
]
111
112
self.scenarios = tests.multiply_scenarios(self.scenarios,
113
auth_scheme_scenarios)
114
115
def load_tests(standard_tests, module, loader):
116
"""Multiply tests for http clients and protocol versions."""
117
# one for each transport
118
t_adapter = TransportAdapter()
119
t_classes= (TestHttpTransportRegistration,
120
TestHttpTransportUrls,
121
)
122
is_testing_for_transports = tests.condition_isinstance(t_classes)
123
124
# multiplied by one for each protocol version
125
tp_adapter = TransportProtocolAdapter()
126
tp_classes= (SmartHTTPTunnellingTest,
127
TestDoCatchRedirections,
128
TestHTTPConnections,
129
TestHTTPRedirections,
130
TestHTTPSilentRedirections,
131
TestLimitedRangeRequestServer,
132
TestPost,
133
TestProxyHttpServer,
134
TestRanges,
135
TestSpecificRequestHandler,
136
)
137
is_also_testing_for_protocols = tests.condition_isinstance(tp_classes)
138
139
# multiplied by one for each authentication scheme
140
tpa_adapter = TransportProtocolAuthenticationAdapter()
141
tpa_classes = (TestAuth,
142
)
143
is_also_testing_for_authentication = tests.condition_isinstance(
144
tpa_classes)
145
146
result = loader.suiteClass()
147
for test_class in tests.iter_suite_tests(standard_tests):
148
# Each test class is either standalone or testing for some combination
149
# of transport, protocol version, authentication scheme. Use the right
150
# adpater (or none) depending on the class.
151
if is_testing_for_transports(test_class):
152
result.addTests(t_adapter.adapt(test_class))
153
elif is_also_testing_for_protocols(test_class):
154
result.addTests(tp_adapter.adapt(test_class))
155
elif is_also_testing_for_authentication(test_class):
156
result.addTests(tpa_adapter.adapt(test_class))
157
else:
158
result.addTest(test_class)
159
return result
80
160
81
161
82
162
class FakeManager(object):
145
225
self.port = None
146
226
147
227
228
class TestHTTPServer(tests.TestCase):
229
"""Test the HTTP servers implementations."""
230
231
def test_invalid_protocol(self):
232
class BogusRequestHandler(http_server.TestingHTTPRequestHandler):
233
234
protocol_version = 'HTTP/0.1'
235
236
server = http_server.HttpServer(BogusRequestHandler)
237
try:
238
self.assertRaises(httplib.UnknownProtocol,server.setUp)
239
except:
240
server.tearDown()
241
self.fail('HTTP Server creation did not raise UnknownProtocol')
242
243
def test_force_invalid_protocol(self):
244
server = http_server.HttpServer(protocol_version='HTTP/0.1')
245
try:
246
self.assertRaises(httplib.UnknownProtocol,server.setUp)
247
except:
248
server.tearDown()
249
self.fail('HTTP Server creation did not raise UnknownProtocol')
250
251
def test_server_start_and_stop(self):
252
server = http_server.HttpServer()
253
server.setUp()
254
self.assertTrue(server._http_running)
255
server.tearDown()
256
self.assertFalse(server._http_running)
257
258
def test_create_http_server_one_zero(self):
259
class RequestHandlerOneZero(http_server.TestingHTTPRequestHandler):
260
261
protocol_version = 'HTTP/1.0'
262
263
server = http_server.HttpServer(RequestHandlerOneZero)
264
server.setUp()
265
self.addCleanup(server.tearDown)
266
self.assertIsInstance(server._httpd, http_server.TestingHTTPServer)
267
268
def test_create_http_server_one_one(self):
269
class RequestHandlerOneOne(http_server.TestingHTTPRequestHandler):
270
271
protocol_version = 'HTTP/1.1'
272
273
server = http_server.HttpServer(RequestHandlerOneOne)
274
server.setUp()
275
self.addCleanup(server.tearDown)
276
self.assertIsInstance(server._httpd,
277
http_server.TestingThreadingHTTPServer)
278
279
def test_create_http_server_force_one_one(self):
280
class RequestHandlerOneZero(http_server.TestingHTTPRequestHandler):
281
282
protocol_version = 'HTTP/1.0'
283
284
server = http_server.HttpServer(RequestHandlerOneZero,
285
protocol_version='HTTP/1.1')
286
server.setUp()
287
self.addCleanup(server.tearDown)
288
self.assertIsInstance(server._httpd,
289
http_server.TestingThreadingHTTPServer)
290
291
def test_create_http_server_force_one_zero(self):
292
class RequestHandlerOneOne(http_server.TestingHTTPRequestHandler):
293
294
protocol_version = 'HTTP/1.1'
295
296
server = http_server.HttpServer(RequestHandlerOneOne,
297
protocol_version='HTTP/1.0')
298
server.setUp()
299
self.addCleanup(server.tearDown)
300
self.assertIsInstance(server._httpd,
301
http_server.TestingHTTPServer)
302
303
148
304
class TestWithTransport_pycurl(object):
149
305
"""Test case to inherit from if pycurl is present"""
150
306
165
321
166
322
def test_url_parsing(self):
167
323
f = FakeManager()
168
url = extract_auth('http://example.com', f)
324
url = http.extract_auth('http://example.com', f)
169
325
self.assertEquals('http://example.com', url)
170
326
self.assertEquals(0, len(f.credentials))
171
url = extract_auth('http://user:pass@www.bazaar-vcs.org/bzr/bzr.dev', f)
327
url = http.extract_auth(
328
'http://user:pass@www.bazaar-vcs.org/bzr/bzr.dev', f)
172
329
self.assertEquals('http://www.bazaar-vcs.org/bzr/bzr.dev', url)
173
330
self.assertEquals(1, len(f.credentials))
174
331
self.assertEquals([None, 'www.bazaar-vcs.org', 'user', 'pass'],
175
332
f.credentials[0])
176
333
177
334
178
class TestHttpTransportUrls(object):
179
"""Test the http urls.
180
181
This MUST be used by daughter classes that also inherit from
182
TestCase.
183
184
We can't inherit directly from TestCase or the
185
test framework will try to create an instance which cannot
186
run, its implementation being incomplete.
187
"""
335
class TestHttpTransportUrls(tests.TestCase):
336
"""Test the http urls."""
188
337
189
338
def test_abs_url(self):
190
339
"""Construction of absolute http URLs"""
221
370
server.tearDown()
222
371
223
372
224
class TestHttpUrls_urllib(TestHttpTransportUrls, tests.TestCase):
225
"""Test http urls with urllib"""
226
227
_transport = HttpTransport_urllib
228
_server = HttpServer_urllib
229
_qualified_prefix = 'http+urllib'
230
231
232
class TestHttpUrls_pycurl(TestWithTransport_pycurl, TestHttpTransportUrls,
233
tests.TestCase):
234
"""Test http urls with pycurl"""
235
236
_server = HttpServer_PyCurl
237
_qualified_prefix = 'http+pycurl'
373
class TestHttps_pycurl(TestWithTransport_pycurl, tests.TestCase):
238
374
239
375
# TODO: This should really be moved into another pycurl
240
376
# specific test. When https tests will be implemented, take
250
386
import pycurl
251
387
except ImportError:
252
388
raise tests.TestSkipped('pycurl not present')
253
# Now that we have pycurl imported, we can fake its version_info
254
# This was taken from a windows pycurl without SSL
255
# (thanks to bialix)
256
pycurl.version_info = lambda : (2,
257
'7.13.2',
258
462082,
259
'i386-pc-win32',
260
2576,
261
None,
262
0,
263
None,
264
('ftp', 'gopher', 'telnet',
265
'dict', 'ldap', 'http', 'file'),
266
None,
267
0,
268
None)
269
self.assertRaises(errors.DependencyNotPresent, self._transport,
270
'https://launchpad.net')
271
272
class TestHttpConnections(object):
273
"""Test the http connections.
274
275
This MUST be used by daughter classes that also inherit from
276
TestCaseWithWebserver.
277
278
We can't inherit directly from TestCaseWithWebserver or the
279
test framework will try to create an instance which cannot
280
run, its implementation being incomplete.
281
"""
389
390
version_info_orig = pycurl.version_info
391
try:
392
# Now that we have pycurl imported, we can fake its version_info
393
# This was taken from a windows pycurl without SSL
394
# (thanks to bialix)
395
pycurl.version_info = lambda : (2,
396
'7.13.2',
397
462082,
398
'i386-pc-win32',
399
2576,
400
None,
401
0,
402
None,
403
('ftp', 'gopher', 'telnet',
404
'dict', 'ldap', 'http', 'file'),
405
None,
406
0,
407
None)
408
self.assertRaises(errors.DependencyNotPresent, self._transport,
409
'https://launchpad.net')
410
finally:
411
# Restore the right function
412
pycurl.version_info = version_info_orig
413
414
415
class TestHTTPConnections(http_utils.TestCaseWithWebserver):
416
"""Test the http connections."""
282
417
283
418
def setUp(self):
284
TestCaseWithWebserver.setUp(self)
285
self.build_tree(['xxx', 'foo/', 'foo/bar'], line_endings='binary',
419
http_utils.TestCaseWithWebserver.setUp(self)
420
self.build_tree(['foo/', 'foo/bar'], line_endings='binary',
286
421
transport=self.get_transport())
287
422
288
423
def test_http_has(self):
334
469
socket.setdefaulttimeout(default_timeout)
335
470
336
471
337
class TestHttpConnections_urllib(TestHttpConnections, TestCaseWithWebserver):
338
"""Test http connections with urllib"""
339
340
_transport = HttpTransport_urllib
341
342
343
344
class TestHttpConnections_pycurl(TestWithTransport_pycurl,
345
TestHttpConnections,
346
TestCaseWithWebserver):
347
"""Test http connections with pycurl"""
348
349
350
472
class TestHttpTransportRegistration(tests.TestCase):
351
473
"""Test registrations of various http implementations"""
352
474
353
475
def test_http_registered(self):
354
# urlllib should always be present
355
t = get_transport('http+urllib://bzr.google.com/')
356
self.assertIsInstance(t, Transport)
357
self.assertIsInstance(t, HttpTransport_urllib)
358
359
360
class TestPost(object):
361
362
def _test_post_body_is_received(self, scheme):
476
t = transport.get_transport('%s://foo.com/' % self._qualified_prefix)
477
self.assertIsInstance(t, transport.Transport)
478
self.assertIsInstance(t, self._transport)
479
480
481
class TestPost(tests.TestCase):
482
483
def test_post_body_is_received(self):
363
484
server = RecordingServer(expect_body_tail='end-of-body')
364
485
server.setUp()
365
486
self.addCleanup(server.tearDown)
487
scheme = self._qualified_prefix
366
488
url = '%s://%s:%s/' % (scheme, server.host, server.port)
367
try:
368
http_transport = get_transport(url)
369
except errors.UnsupportedProtocol:
370
raise tests.TestSkipped('%s not available' % scheme)
489
http_transport = self._transport(url)
371
490
code, response = http_transport._post('abc def end-of-body')
372
491
self.assertTrue(
373
492
server.received_bytes.startswith('POST /.bzr/smart HTTP/1.'))
379
498
server.received_bytes.endswith('\r\n\r\nabc def end-of-body'))
380
499
381
500
382
class TestPost_urllib(tests.TestCase, TestPost):
383
"""TestPost for urllib implementation"""
384
385
_transport = HttpTransport_urllib
386
387
def test_post_body_is_received_urllib(self):
388
self._test_post_body_is_received('http+urllib')
389
390
391
class TestPost_pycurl(TestWithTransport_pycurl, tests.TestCase, TestPost):
392
"""TestPost for pycurl implementation"""
393
394
def test_post_body_is_received_pycurl(self):
395
self._test_post_body_is_received('http+pycurl')
396
397
398
501
class TestRangeHeader(tests.TestCase):
399
502
"""Test range_header method"""
400
503
401
504
def check_header(self, value, ranges=[], tail=0):
402
505
offsets = [ (start, end - start + 1) for start, end in ranges]
403
coalesce = Transport._coalesce_offsets
506
coalesce = transport.Transport._coalesce_offsets
404
507
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
405
range_header = HttpTransportBase._range_header
508
range_header = http.HttpTransportBase._range_header
406
509
self.assertEqual(value, range_header(coalesced, tail))
407
510
408
511
def test_range_header_single(self):
423
526
tail=50)
424
527
425
528
426
class TestWallServer(object):
529
class TestSpecificRequestHandler(http_utils.TestCaseWithWebserver):
530
"""Tests a specific request handler.
531
532
Daughter classes are expected to override _req_handler_class
533
"""
534
535
# Provide a useful default
536
_req_handler_class = http_server.TestingHTTPRequestHandler
537
538
def create_transport_readonly_server(self):
539
return http_server.HttpServer(self._req_handler_class,
540
protocol_version=self._protocol_version)
541
542
def _testing_pycurl(self):
543
return pycurl_present and self._transport == PyCurlTransport
544
545
546
class WallRequestHandler(http_server.TestingHTTPRequestHandler):
547
"""Whatever request comes in, close the connection"""
548
549
def handle_one_request(self):
550
"""Handle a single HTTP request, by abruptly closing the connection"""
551
self.close_connection = 1
552
553
554
class TestWallServer(TestSpecificRequestHandler):
427
555
"""Tests exceptions during the connection phase"""
428
556
429
def create_transport_readonly_server(self):
430
return HttpServer(WallRequestHandler)
557
_req_handler_class = WallRequestHandler
431
558
432
559
def test_http_has(self):
433
560
server = self.get_readonly_server()
447
574
t.get, 'foo/bar')
448
575
449
576
450
class TestWallServer_urllib(TestWallServer, TestCaseWithWebserver):
451
"""Tests "wall" server for urllib implementation"""
452
453
_transport = HttpTransport_urllib
454
455
456
class TestWallServer_pycurl(TestWithTransport_pycurl,
457
TestWallServer,
458
TestCaseWithWebserver):
459
"""Tests "wall" server for pycurl implementation"""
460
461
462
class TestBadStatusServer(object):
577
class BadStatusRequestHandler(http_server.TestingHTTPRequestHandler):
578
"""Whatever request comes in, returns a bad status"""
579
580
def parse_request(self):
581
"""Fakes handling a single HTTP request, returns a bad status"""
582
ignored = http_server.TestingHTTPRequestHandler.parse_request(self)
583
self.send_response(0, "Bad status")
584
self.close_connection = 1
585
return False
586
587
588
class TestBadStatusServer(TestSpecificRequestHandler):
463
589
"""Tests bad status from server."""
464
590
465
def create_transport_readonly_server(self):
466
return HttpServer(BadStatusRequestHandler)
591
_req_handler_class = BadStatusRequestHandler
467
592
468
593
def test_http_has(self):
469
594
server = self.get_readonly_server()
476
601
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
477
602
478
603
479
class TestBadStatusServer_urllib(TestBadStatusServer, TestCaseWithWebserver):
480
"""Tests bad status server for urllib implementation"""
481
482
_transport = HttpTransport_urllib
483
484
485
class TestBadStatusServer_pycurl(TestWithTransport_pycurl,
486
TestBadStatusServer,
487
TestCaseWithWebserver):
488
"""Tests bad status server for pycurl implementation"""
604
class InvalidStatusRequestHandler(http_server.TestingHTTPRequestHandler):
605
"""Whatever request comes in, returns an invalid status"""
606
607
def parse_request(self):
608
"""Fakes handling a single HTTP request, returns a bad status"""
609
ignored = http_server.TestingHTTPRequestHandler.parse_request(self)
610
self.wfile.write("Invalid status line\r\n")
611
return False
489
612
490
613
491
614
class TestInvalidStatusServer(TestBadStatusServer):
494
617
Both implementations raises the same error as for a bad status.
495
618
"""
496
619
497
def create_transport_readonly_server(self):
498
return HttpServer(InvalidStatusRequestHandler)
499
500
501
class TestInvalidStatusServer_urllib(TestInvalidStatusServer,
502
TestCaseWithWebserver):
503
"""Tests invalid status server for urllib implementation"""
504
505
_transport = HttpTransport_urllib
506
507
508
class TestInvalidStatusServer_pycurl(TestWithTransport_pycurl,
509
TestInvalidStatusServer,
510
TestCaseWithWebserver):
511
"""Tests invalid status server for pycurl implementation"""
512
513
514
class TestBadProtocolServer(object):
620
_req_handler_class = InvalidStatusRequestHandler
621
622
def test_http_has(self):
623
if self._testing_pycurl() and self._protocol_version == 'HTTP/1.1':
624
raise tests.KnownFailure(
625
'pycurl hangs if the server send back garbage')
626
super(TestInvalidStatusServer, self).test_http_has()
627
628
def test_http_get(self):
629
if self._testing_pycurl() and self._protocol_version == 'HTTP/1.1':
630
raise tests.KnownFailure(
631
'pycurl hangs if the server send back garbage')
632
super(TestInvalidStatusServer, self).test_http_get()
633
634
635
class BadProtocolRequestHandler(http_server.TestingHTTPRequestHandler):
636
"""Whatever request comes in, returns a bad protocol version"""
637
638
def parse_request(self):
639
"""Fakes handling a single HTTP request, returns a bad status"""
640
ignored = http_server.TestingHTTPRequestHandler.parse_request(self)
641
# Returns an invalid protocol version, but curl just
642
# ignores it and those cannot be tested.
643
self.wfile.write("%s %d %s\r\n" % ('HTTP/0.0',
644
404,
645
'Look at my protocol version'))
646
return False
647
648
649
class TestBadProtocolServer(TestSpecificRequestHandler):
515
650
"""Tests bad protocol from server."""
516
651
517
def create_transport_readonly_server(self):
518
return HttpServer(BadProtocolRequestHandler)
652
_req_handler_class = BadProtocolRequestHandler
653
654
def setUp(self):
655
if pycurl_present and self._transport == PyCurlTransport:
656
raise tests.TestNotApplicable(
657
"pycurl doesn't check the protocol version")
658
super(TestBadProtocolServer, self).setUp()
519
659
520
660
def test_http_has(self):
521
661
server = self.get_readonly_server()
528
668
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
529
669
530
670
531
class TestBadProtocolServer_urllib(TestBadProtocolServer,
532
TestCaseWithWebserver):
533
"""Tests bad protocol server for urllib implementation"""
534
535
_transport = HttpTransport_urllib
536
537
# curl don't check the protocol version
538
#class TestBadProtocolServer_pycurl(TestWithTransport_pycurl,
539
# TestBadProtocolServer,
540
# TestCaseWithWebserver):
541
# """Tests bad protocol server for pycurl implementation"""
542
543
544
class TestForbiddenServer(object):
671
class ForbiddenRequestHandler(http_server.TestingHTTPRequestHandler):
672
"""Whatever request comes in, returns a 403 code"""
673
674
def parse_request(self):
675
"""Handle a single HTTP request, by replying we cannot handle it"""
676
ignored = http_server.TestingHTTPRequestHandler.parse_request(self)
677
self.send_error(403)
678
return False
679
680
681
class TestForbiddenServer(TestSpecificRequestHandler):
545
682
"""Tests forbidden server"""
546
683
547
def create_transport_readonly_server(self):
548
return HttpServer(ForbiddenRequestHandler)
684
_req_handler_class = ForbiddenRequestHandler
549
685
550
686
def test_http_has(self):
551
687
server = self.get_readonly_server()
558
694
self.assertRaises(errors.TransportError, t.get, 'foo/bar')
559
695
560
696
561
class TestForbiddenServer_urllib(TestForbiddenServer, TestCaseWithWebserver):
562
"""Tests forbidden server for urllib implementation"""
563
564
_transport = HttpTransport_urllib
565
566
567
class TestForbiddenServer_pycurl(TestWithTransport_pycurl,
568
TestForbiddenServer,
569
TestCaseWithWebserver):
570
"""Tests forbidden server for pycurl implementation"""
571
572
573
697
class TestRecordingServer(tests.TestCase):
574
698
575
699
def test_create(self):
601
725
self.assertEqual('abc', server.received_bytes)
602
726
603
727
604
class TestRangeRequestServer(object):
728
class TestRangeRequestServer(TestSpecificRequestHandler):
605
729
"""Tests readv requests against server.
606
730
607
This MUST be used by daughter classes that also inherit from
608
TestCaseWithWebserver.
609
610
We can't inherit directly from TestCaseWithWebserver or the
611
test framework will try to create an instance which cannot
612
run, its implementation being incomplete.
731
We test against default "normal" server.
613
732
"""
614
733
615
734
def setUp(self):
616
TestCaseWithWebserver.setUp(self)
735
super(TestRangeRequestServer, self).setUp()
617
736
self.build_tree_contents([('a', '0123456789')],)
618
737
619
738
def test_readv(self):
667
786
t = self._transport(server.get_url())
668
787
# force transport to issue multiple requests by limiting the number of
669
788
# bytes by request. Note that this apply to coalesced offsets only, a
670
# single range ill keep its size even if bigger than the limit.
789
# single range will keep its size even if bigger than the limit.
671
790
t._get_max_size = 2
672
791
l = list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
673
792
self.assertEqual(l[0], (0, '0'))
677
796
# The server should have issued 3 requests
678
797
self.assertEqual(3, server.GET_request_nb)
679
798
799
def test_complete_readv_leave_pipe_clean(self):
800
server = self.get_readonly_server()
801
t = self._transport(server.get_url())
802
# force transport to issue multiple requests
803
t._get_max_size = 2
804
l = list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
805
# The server should have issued 3 requests
806
self.assertEqual(3, server.GET_request_nb)
807
self.assertEqual('0123456789', t.get_bytes('a'))
808
self.assertEqual(4, server.GET_request_nb)
809
810
def test_incomplete_readv_leave_pipe_clean(self):
811
server = self.get_readonly_server()
812
t = self._transport(server.get_url())
813
# force transport to issue multiple requests
814
t._get_max_size = 2
815
# Don't collapse readv results into a list so that we leave unread
816
# bytes on the socket
817
ireadv = iter(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
818
self.assertEqual((0, '0'), ireadv.next())
819
# The server should have issued one request so far
820
self.assertEqual(1, server.GET_request_nb)
821
self.assertEqual('0123456789', t.get_bytes('a'))
822
# get_bytes issued an additional request, the readv pending ones are
823
# lost
824
self.assertEqual(2, server.GET_request_nb)
825
826
827
class SingleRangeRequestHandler(http_server.TestingHTTPRequestHandler):
828
"""Always reply to range request as if they were single.
829
830
Don't be explicit about it, just to annoy the clients.
831
"""
832
833
def get_multiple_ranges(self, file, file_size, ranges):
834
"""Answer as if it was a single range request and ignores the rest"""
835
(start, end) = ranges[0]
836
return self.get_single_range(file, file_size, start, end)
837
680
838
681
839
class TestSingleRangeRequestServer(TestRangeRequestServer):
682
840
"""Test readv against a server which accept only single range requests"""
683
841
684
def create_transport_readonly_server(self):
685
return HttpServer(SingleRangeRequestHandler)
686
687
688
class TestSingleRangeRequestServer_urllib(TestSingleRangeRequestServer,
689
TestCaseWithWebserver):
690
"""Tests single range requests accepting server for urllib implementation"""
691
692
_transport = HttpTransport_urllib
693
694
695
class TestSingleRangeRequestServer_pycurl(TestWithTransport_pycurl,
696
TestSingleRangeRequestServer,
697
TestCaseWithWebserver):
698
"""Tests single range requests accepting server for pycurl implementation"""
842
_req_handler_class = SingleRangeRequestHandler
843
844
845
class SingleOnlyRangeRequestHandler(http_server.TestingHTTPRequestHandler):
846
"""Only reply to simple range requests, errors out on multiple"""
847
848
def get_multiple_ranges(self, file, file_size, ranges):
849
"""Refuses the multiple ranges request"""
850
if len(ranges) > 1:
851
file.close()
852
self.send_error(416, "Requested range not satisfiable")
853
return
854
(start, end) = ranges[0]
855
return self.get_single_range(file, file_size, start, end)
699
856
700
857
701
858
class TestSingleOnlyRangeRequestServer(TestRangeRequestServer):
702
859
"""Test readv against a server which only accept single range requests"""
703
860
704
def create_transport_readonly_server(self):
705
return HttpServer(SingleOnlyRangeRequestHandler)
706
707
708
class TestSingleOnlyRangeRequestServer_urllib(TestSingleOnlyRangeRequestServer,
709
TestCaseWithWebserver):
710
"""Tests single range requests accepting server for urllib implementation"""
711
712
_transport = HttpTransport_urllib
713
714
715
class TestSingleOnlyRangeRequestServer_pycurl(TestWithTransport_pycurl,
716
TestSingleOnlyRangeRequestServer,
717
TestCaseWithWebserver):
718
"""Tests single range requests accepting server for pycurl implementation"""
861
_req_handler_class = SingleOnlyRangeRequestHandler
862
863
864
class NoRangeRequestHandler(http_server.TestingHTTPRequestHandler):
865
"""Ignore range requests without notice"""
866
867
def do_GET(self):
868
# Update the statistics
869
self.server.test_case_server.GET_request_nb += 1
870
# Just bypass the range handling done by TestingHTTPRequestHandler
871
return SimpleHTTPServer.SimpleHTTPRequestHandler.do_GET(self)
719
872
720
873
721
874
class TestNoRangeRequestServer(TestRangeRequestServer):
722
875
"""Test readv against a server which do not accept range requests"""
723
876
724
def create_transport_readonly_server(self):
725
return HttpServer(NoRangeRequestHandler)
726
727
728
class TestNoRangeRequestServer_urllib(TestNoRangeRequestServer,
729
TestCaseWithWebserver):
730
"""Tests range requests refusing server for urllib implementation"""
731
732
_transport = HttpTransport_urllib
733
734
735
class TestNoRangeRequestServer_pycurl(TestWithTransport_pycurl,
736
TestNoRangeRequestServer,
737
TestCaseWithWebserver):
738
"""Tests range requests refusing server for pycurl implementation"""
739
740
741
class TestLimitedRangeRequestServer(object):
742
"""Tests readv requests against server that errors out on too much ranges.
743
744
This MUST be used by daughter classes that also inherit from
745
TestCaseWithWebserver.
746
747
We can't inherit directly from TestCaseWithWebserver or the
748
test framework will try to create an instance which cannot
749
run, its implementation being incomplete.
877
_req_handler_class = NoRangeRequestHandler
878
879
880
class MultipleRangeWithoutContentLengthRequestHandler(
881
http_server.TestingHTTPRequestHandler):
882
"""Reply to multiple range requests without content length header."""
883
884
def get_multiple_ranges(self, file, file_size, ranges):
885
self.send_response(206)
886
self.send_header('Accept-Ranges', 'bytes')
887
boundary = "%d" % random.randint(0,0x7FFFFFFF)
888
self.send_header("Content-Type",
889
"multipart/byteranges; boundary=%s" % boundary)
890
self.end_headers()
891
for (start, end) in ranges:
892
self.wfile.write("--%s\r\n" % boundary)
893
self.send_header("Content-type", 'application/octet-stream')
894
self.send_header("Content-Range", "bytes %d-%d/%d" % (start,
895
end,
896
file_size))
897
self.end_headers()
898
self.send_range_content(file, start, end - start + 1)
899
# Final boundary
900
self.wfile.write("--%s\r\n" % boundary)
901
902
903
class TestMultipleRangeWithoutContentLengthServer(TestRangeRequestServer):
904
905
_req_handler_class = MultipleRangeWithoutContentLengthRequestHandler
906
907
908
class TruncatedMultipleRangeRequestHandler(
909
http_server.TestingHTTPRequestHandler):
910
"""Reply to multiple range requests truncating the last ones.
911
912
This server generates responses whose Content-Length describes all the
913
ranges, but fail to include the last ones leading to client short reads.
914
This has been observed randomly with lighttpd (bug #179368).
750
915
"""
751
916
917
_truncated_ranges = 2
918
919
def get_multiple_ranges(self, file, file_size, ranges):
920
self.send_response(206)
921
self.send_header('Accept-Ranges', 'bytes')
922
boundary = 'tagada'
923
self.send_header('Content-Type',
924
'multipart/byteranges; boundary=%s' % boundary)
925
boundary_line = '--%s\r\n' % boundary
926
# Calculate the Content-Length
927
content_length = 0
928
for (start, end) in ranges:
929
content_length += len(boundary_line)
930
content_length += self._header_line_length(
931
'Content-type', 'application/octet-stream')
932
content_length += self._header_line_length(
933
'Content-Range', 'bytes %d-%d/%d' % (start, end, file_size))
934
content_length += len('\r\n') # end headers
935
content_length += end - start # + 1
936
content_length += len(boundary_line)
937
self.send_header('Content-length', content_length)
938
self.end_headers()
939
940
# Send the multipart body
941
cur = 0
942
for (start, end) in ranges:
943
self.wfile.write(boundary_line)
944
self.send_header('Content-type', 'application/octet-stream')
945
self.send_header('Content-Range', 'bytes %d-%d/%d'
946
% (start, end, file_size))
947
self.end_headers()
948
if cur + self._truncated_ranges >= len(ranges):
949
# Abruptly ends the response and close the connection
950
self.close_connection = 1
951
return
952
self.send_range_content(file, start, end - start + 1)
953
cur += 1
954
# No final boundary
955
self.wfile.write(boundary_line)
956
957
958
class TestTruncatedMultipleRangeServer(TestSpecificRequestHandler):
959
960
_req_handler_class = TruncatedMultipleRangeRequestHandler
961
962
def setUp(self):
963
super(TestTruncatedMultipleRangeServer, self).setUp()
964
self.build_tree_contents([('a', '0123456789')],)
965
966
def test_readv_with_short_reads(self):
967
server = self.get_readonly_server()
968
t = self._transport(server.get_url())
969
# Force separate ranges for each offset
970
t._bytes_to_read_before_seek = 0
971
ireadv = iter(t.readv('a', ((0, 1), (2, 1), (4, 2), (9, 1))))
972
self.assertEqual((0, '0'), ireadv.next())
973
self.assertEqual((2, '2'), ireadv.next())
974
if not self._testing_pycurl():
975
# Only one request have been issued so far (except for pycurl that
976
# try to read the whole response at once)
977
self.assertEqual(1, server.GET_request_nb)
978
self.assertEqual((4, '45'), ireadv.next())
979
self.assertEqual((9, '9'), ireadv.next())
980
# Both implementations issue 3 requests but:
981
# - urllib does two multiple (4 ranges, then 2 ranges) then a single
982
# range,
983
# - pycurl does two multiple (4 ranges, 4 ranges) then a single range
984
self.assertEqual(3, server.GET_request_nb)
985
# Finally the client have tried a single range request and stays in
986
# that mode
987
self.assertEqual('single', t._range_hint)
988
989
class LimitedRangeRequestHandler(http_server.TestingHTTPRequestHandler):
990
"""Errors out when range specifiers exceed the limit"""
991
992
def get_multiple_ranges(self, file, file_size, ranges):
993
"""Refuses the multiple ranges request"""
994
tcs = self.server.test_case_server
995
if tcs.range_limit is not None and len(ranges) > tcs.range_limit:
996
file.close()
997
# Emulate apache behavior
998
self.send_error(400, "Bad Request")
999
return
1000
return http_server.TestingHTTPRequestHandler.get_multiple_ranges(
1001
self, file, file_size, ranges)
1002
1003
1004
class LimitedRangeHTTPServer(http_server.HttpServer):
1005
"""An HttpServer erroring out on requests with too much range specifiers"""
1006
1007
def __init__(self, request_handler=LimitedRangeRequestHandler,
1008
protocol_version=None,
1009
range_limit=None):
1010
http_server.HttpServer.__init__(self, request_handler,
1011
protocol_version=protocol_version)
1012
self.range_limit = range_limit
1013
1014
1015
class TestLimitedRangeRequestServer(http_utils.TestCaseWithWebserver):
1016
"""Tests readv requests against a server erroring out on too much ranges."""
1017
1018
# Requests with more range specifiers will error out
752
1019
range_limit = 3
753
1020
754
1021
def create_transport_readonly_server(self):
755
# Requests with more range specifiers will error out
756
return LimitedRangeHTTPServer(range_limit=self.range_limit)
1022
return LimitedRangeHTTPServer(range_limit=self.range_limit,
1023
protocol_version=self._protocol_version)
757
1024
758
1025
def get_transport(self):
759
1026
return self._transport(self.get_readonly_server().get_url())
760
1027
761
1028
def setUp(self):
762
TestCaseWithWebserver.setUp(self)
1029
http_utils.TestCaseWithWebserver.setUp(self)
763
1030
# We need to manipulate ranges that correspond to real chunks in the
764
1031
# response, so we build a content appropriately.
765
1032
filler = ''.join(['abcdefghij' for x in range(102)])
781
1048
self.assertEqual(l[2], (4096, '0004'))
782
1049
self.assertEqual(l[3], (8192, '0008'))
783
1050
# The server will refuse to serve the first request (too much ranges),
784
# a second request will succeeds.
1051
# a second request will succeed.
785
1052
self.assertEqual(2, self.get_readonly_server().GET_request_nb)
786
1053
787
1054
788
class TestLimitedRangeRequestServer_urllib(TestLimitedRangeRequestServer,
789
TestCaseWithWebserver):
790
"""Tests limited range requests server for urllib implementation"""
791
792
_transport = HttpTransport_urllib
793
794
795
class TestLimitedRangeRequestServer_pycurl(TestWithTransport_pycurl,
796
TestLimitedRangeRequestServer,
797
TestCaseWithWebserver):
798
"""Tests limited range requests server for pycurl implementation"""
799
800
801
802
1055
class TestHttpProxyWhiteBox(tests.TestCase):
803
1056
"""Whitebox test proxy http authorization.
804
1057
811
1064
812
1065
def tearDown(self):
813
1066
self._restore_env()
1067
tests.TestCase.tearDown(self)
814
1068
815
1069
def _install_env(self, env):
816
1070
for name, value in env.iteritems():
821
1075
osutils.set_or_unset_env(name, value)
822
1076
823
1077
def _proxied_request(self):
824
handler = ProxyHandler()
825
request = Request('GET','http://baz/buzzle')
1078
handler = _urllib2_wrappers.ProxyHandler()
1079
request = _urllib2_wrappers.Request('GET','http://baz/buzzle')
826
1080
handler.set_proxy(request, 'http')
827
1081
return request
828
1082
837
1091
self.assertRaises(errors.InvalidURL, self._proxied_request)
838
1092
839
1093
840
class TestProxyHttpServer(object):
1094
class TestProxyHttpServer(http_utils.TestCaseWithTwoWebservers):
841
1095
"""Tests proxy server.
842
1096
843
This MUST be used by daughter classes that also inherit from
844
TestCaseWithTwoWebservers.
845
846
We can't inherit directly from TestCaseWithTwoWebservers or
847
the test framework will try to create an instance which
848
cannot run, its implementation being incomplete.
849
850
1097
Be aware that we do not setup a real proxy here. Instead, we
851
1098
check that the *connection* goes through the proxy by serving
852
1099
different content (the faked proxy server append '-proxied'
857
1104
# test https connections.
858
1105
859
1106
def setUp(self):
860
TestCaseWithTwoWebservers.setUp(self)
1107
super(TestProxyHttpServer, self).setUp()
861
1108
self.build_tree_contents([('foo', 'contents of foo\n'),
862
1109
('foo-proxied', 'proxied contents of foo\n')])
863
1110
# Let's setup some attributes for tests
864
1111
self.server = self.get_readonly_server()
865
1112
self.proxy_address = '%s:%d' % (self.server.host, self.server.port)
866
self.no_proxy_host = self.proxy_address
1113
if self._testing_pycurl():
1114
# Oh my ! pycurl does not check for the port as part of
1115
# no_proxy :-( So we just test the host part
1116
self.no_proxy_host = 'localhost'
1117
else:
1118
self.no_proxy_host = self.proxy_address
867
1119
# The secondary server is the proxy
868
1120
self.proxy = self.get_secondary_server()
869
1121
self.proxy_url = self.proxy.get_url()
870
1122
self._old_env = {}
871
1123
1124
def _testing_pycurl(self):
1125
return pycurl_present and self._transport == PyCurlTransport
1126
872
1127
def create_transport_secondary_server(self):
873
1128
"""Creates an http server that will serve files with
874
1129
'-proxied' appended to their names.
875
1130
"""
876
return ProxyServer()
1131
return http_utils.ProxyServer(protocol_version=self._protocol_version)
877
1132
878
1133
def _install_env(self, env):
879
1134
for name, value in env.iteritems():
905
1160
self.proxied_in_env({'http_proxy': self.proxy_url})
906
1161
907
1162
def test_HTTP_PROXY(self):
1163
if self._testing_pycurl():
1164
# pycurl does not check HTTP_PROXY for security reasons
1165
# (for use in a CGI context that we do not care
1166
# about. Should we ?)
1167
raise tests.TestNotApplicable(
1168
'pycurl does not check HTTP_PROXY for security reasons')
908
1169
self.proxied_in_env({'HTTP_PROXY': self.proxy_url})
909
1170
910
1171
def test_all_proxy(self):
918
1179
'no_proxy': self.no_proxy_host})
919
1180
920
1181
def test_HTTP_PROXY_with_NO_PROXY(self):
1182
if self._testing_pycurl():
1183
raise tests.TestNotApplicable(
1184
'pycurl does not check HTTP_PROXY for security reasons')
921
1185
self.not_proxied_in_env({'HTTP_PROXY': self.proxy_url,
922
1186
'NO_PROXY': self.no_proxy_host})
923
1187
930
1194
'NO_PROXY': self.no_proxy_host})
931
1195
932
1196
def test_http_proxy_without_scheme(self):
933
self.assertRaises(errors.InvalidURL,
934
self.proxied_in_env,
935
{'http_proxy': self.proxy_address})
936
937
938
class TestProxyHttpServer_urllib(TestProxyHttpServer,
939
TestCaseWithTwoWebservers):
940
"""Tests proxy server for urllib implementation"""
941
942
_transport = HttpTransport_urllib
943
944
945
class TestProxyHttpServer_pycurl(TestWithTransport_pycurl,
946
TestProxyHttpServer,
947
TestCaseWithTwoWebservers):
948
"""Tests proxy server for pycurl implementation"""
949
950
def setUp(self):
951
TestProxyHttpServer.setUp(self)
952
# Oh my ! pycurl does not check for the port as part of
953
# no_proxy :-( So we just test the host part
954
self.no_proxy_host = 'localhost'
955
956
def test_HTTP_PROXY(self):
957
# pycurl does not check HTTP_PROXY for security reasons
958
# (for use in a CGI context that we do not care
959
# about. Should we ?)
960
raise tests.TestNotApplicable(
961
'pycurl does not check HTTP_PROXY for security reasons')
962
963
def test_HTTP_PROXY_with_NO_PROXY(self):
964
raise tests.TestNotApplicable(
965
'pycurl does not check HTTP_PROXY for security reasons')
966
967
def test_http_proxy_without_scheme(self):
968
# pycurl *ignores* invalid proxy env variables. If that
969
# ever change in the future, this test will fail
970
# indicating that pycurl do not ignore anymore such
971
# variables.
972
self.not_proxied_in_env({'http_proxy': self.proxy_address})
973
974
975
class TestRanges(object):
976
"""Test the Range header in GET methods..
977
978
This MUST be used by daughter classes that also inherit from
979
TestCaseWithWebserver.
980
981
We can't inherit directly from TestCaseWithWebserver or the
982
test framework will try to create an instance which cannot
983
run, its implementation being incomplete.
984
"""
985
986
def setUp(self):
987
TestCaseWithWebserver.setUp(self)
1197
if self._testing_pycurl():
1198
# pycurl *ignores* invalid proxy env variables. If that ever change
1199
# in the future, this test will fail indicating that pycurl do not
1200
# ignore anymore such variables.
1201
self.not_proxied_in_env({'http_proxy': self.proxy_address})
1202
else:
1203
self.assertRaises(errors.InvalidURL,
1204
self.proxied_in_env,
1205
{'http_proxy': self.proxy_address})
1206
1207
1208
class TestRanges(http_utils.TestCaseWithWebserver):
1209
"""Test the Range header in GET methods."""
1210
1211
def setUp(self):
1212
http_utils.TestCaseWithWebserver.setUp(self)
988
1213
self.build_tree_contents([('a', '0123456789')],)
989
1214
server = self.get_readonly_server()
990
1215
self.transport = self._transport(server.get_url())
991
1216
1217
def create_transport_readonly_server(self):
1218
return http_server.HttpServer(protocol_version=self._protocol_version)
1219
992
1220
def _file_contents(self, relpath, ranges):
993
1221
offsets = [ (start, end - start + 1) for start, end in ranges]
994
1222
coalesce = self.transport._coalesce_offsets
1009
1237
# Valid ranges
1010
1238
map(self.assertEqual,['0', '234'],
1011
1239
list(self._file_contents('a', [(0,0), (2,4)])),)
1012
# Tail
1240
1241
def test_range_header_tail(self):
1013
1242
self.assertEqual('789', self._file_tail('a', 3))
1014
# Syntactically invalid range
1243
1244
def test_syntactically_invalid_range_header(self):
1015
1245
self.assertListRaises(errors.InvalidHttpRange,
1016
1246
self._file_contents, 'a', [(4, 3)])
1017
# Semantically invalid range
1247
1248
def test_semantically_invalid_range_header(self):
1018
1249
self.assertListRaises(errors.InvalidHttpRange,
1019
1250
self._file_contents, 'a', [(42, 128)])
1020
1251
1021
1252
1022
class TestRanges_urllib(TestRanges, TestCaseWithWebserver):
1023
"""Test the Range header in GET methods for urllib implementation"""
1024
1025
_transport = HttpTransport_urllib
1026
1027
1028
class TestRanges_pycurl(TestWithTransport_pycurl,
1029
TestRanges,
1030
TestCaseWithWebserver):
1031
"""Test the Range header in GET methods for pycurl implementation"""
1032
1033
1034
class TestHTTPRedirections(object):
1035
"""Test redirection between http servers.
1036
1037
This MUST be used by daughter classes that also inherit from
1038
TestCaseWithRedirectedWebserver.
1039
1040
We can't inherit directly from TestCaseWithTwoWebservers or the
1041
test framework will try to create an instance which cannot
1042
run, its implementation being incomplete.
1043
"""
1253
class TestHTTPRedirections(http_utils.TestCaseWithRedirectedWebserver):
1254
"""Test redirection between http servers."""
1044
1255
1045
1256
def create_transport_secondary_server(self):
1046
1257
"""Create the secondary server redirecting to the primary server"""
1047
1258
new = self.get_readonly_server()
1048
1259
1049
redirecting = HTTPServerRedirecting()
1260
redirecting = http_utils.HTTPServerRedirecting(
1261
protocol_version=self._protocol_version)
1050
1262
redirecting.redirect_to(new.host, new.port)
1051
1263
return redirecting
1052
1264
1072
1284
self.assertEqual([], bundle.revisions)
1073
1285
1074
1286
1075
class TestHTTPRedirections_urllib(TestHTTPRedirections,
1076
TestCaseWithRedirectedWebserver):
1077
"""Tests redirections for urllib implementation"""
1078
1079
_transport = HttpTransport_urllib
1080
1081
1082
1083
class TestHTTPRedirections_pycurl(TestWithTransport_pycurl,
1084
TestHTTPRedirections,
1085
TestCaseWithRedirectedWebserver):
1086
"""Tests redirections for pycurl implementation"""
1087
1088
1089
class RedirectedRequest(Request):
1090
"""Request following redirections"""
1091
1092
init_orig = Request.__init__
1287
class RedirectedRequest(_urllib2_wrappers.Request):
1288
"""Request following redirections. """
1289
1290
init_orig = _urllib2_wrappers.Request.__init__
1093
1291
1094
1292
def __init__(self, method, url, *args, **kwargs):
1293
"""Constructor.
1294
1295
"""
1296
# Since the tests using this class will replace
1297
# _urllib2_wrappers.Request, we can't just call the base class __init__
1298
# or we'll loop.
1095
1299
RedirectedRequest.init_orig(self, method, url, args, kwargs)
1096
1300
self.follow_redirections = True
1097
1301
1098
1302
1099
class TestHTTPSilentRedirections_urllib(TestCaseWithRedirectedWebserver):
1100
"""Test redirections provided by urllib.
1303
class TestHTTPSilentRedirections(http_utils.TestCaseWithRedirectedWebserver):
1304
"""Test redirections.
1101
1305
1102
1306
http implementations do not redirect silently anymore (they
1103
1307
do not redirect at all in fact). The mechanism is still in
1110
1314
-- vila 20070212
1111
1315
"""
1112
1316
1113
_transport = HttpTransport_urllib
1114
1115
1317
def setUp(self):
1116
super(TestHTTPSilentRedirections_urllib, self).setUp()
1318
if pycurl_present and self._transport == PyCurlTransport:
1319
raise tests.TestNotApplicable(
1320
"pycurl doesn't redirect silently annymore")
1321
super(TestHTTPSilentRedirections, self).setUp()
1117
1322
self.setup_redirected_request()
1118
1323
self.addCleanup(self.cleanup_redirected_request)
1119
1324
self.build_tree_contents([('a','a'),
1140
1345
1141
1346
def create_transport_secondary_server(self):
1142
1347
"""Create the secondary server, redirections are defined in the tests"""
1143
return HTTPServerRedirecting()
1348
return http_utils.HTTPServerRedirecting(
1349
protocol_version=self._protocol_version)
1144
1350
1145
1351
def test_one_redirection(self):
1146
1352
t = self.old_transport
1162
1368
self.old_server.port)
1163
1369
new_prefix = 'http://%s:%s' % (self.new_server.host,
1164
1370
self.new_server.port)
1165
self.old_server.redirections = \
1166
[('/1(.*)', r'%s/2\1' % (old_prefix), 302),
1167
('/2(.*)', r'%s/3\1' % (old_prefix), 303),
1168
('/3(.*)', r'%s/4\1' % (old_prefix), 307),
1169
('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1170
('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1171
]
1371
self.old_server.redirections = [
1372
('/1(.*)', r'%s/2\1' % (old_prefix), 302),
1373
('/2(.*)', r'%s/3\1' % (old_prefix), 303),
1374
('/3(.*)', r'%s/4\1' % (old_prefix), 307),
1375
('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1376
('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1377
]
1172
1378
self.assertEquals('redirected 5 times',t._perform(req).read())
1173
1379
1174
1380
1175
class TestDoCatchRedirections(TestCaseWithRedirectedWebserver):
1176
"""Test transport.do_catching_redirections.
1177
1178
We arbitrarily choose to use urllib transports
1179
"""
1180
1181
_transport = HttpTransport_urllib
1381
class TestDoCatchRedirections(http_utils.TestCaseWithRedirectedWebserver):
1382
"""Test transport.do_catching_redirections."""
1182
1383
1183
1384
def setUp(self):
1184
1385
super(TestDoCatchRedirections, self).setUp()
1194
1395
1195
1396
# We use None for redirected so that we fail if redirected
1196
1397
self.assertEquals('0123456789',
1197
do_catching_redirections(self.get_a, t, None).read())
1398
transport.do_catching_redirections(
1399
self.get_a, t, None).read())
1198
1400
1199
1401
def test_one_redirection(self):
1200
1402
self.redirections = 0
1205
1407
return self._transport(dir)
1206
1408
1207
1409
self.assertEquals('0123456789',
1208
do_catching_redirections(self.get_a,
1209
self.old_transport,
1210
redirected
1211
).read())
1410
transport.do_catching_redirections(
1411
self.get_a, self.old_transport, redirected).read())
1212
1412
self.assertEquals(1, self.redirections)
1213
1413
1214
1414
def test_redirection_loop(self):
1219
1419
# a/a/a
1220
1420
return self.old_transport.clone(exception.target)
1221
1421
1222
self.assertRaises(errors.TooManyRedirections, do_catching_redirections,
1422
self.assertRaises(errors.TooManyRedirections,
1423
transport.do_catching_redirections,
1223
1424
self.get_a, self.old_transport, redirected)
1224
1425
1225
1426
1226
class TestAuth(object):
1227
"""Test some authentication scheme specified by daughter class.
1228
1229
This MUST be used by daughter classes that also inherit from
1230
either TestCaseWithWebserver or TestCaseWithTwoWebservers.
1231
"""
1232
1427
class TestAuth(http_utils.TestCaseWithWebserver):
1428
"""Test authentication scheme"""
1429
1430
_auth_header = 'Authorization'
1233
1431
_password_prompt_prefix = ''
1234
1432
1235
1433
def setUp(self):
1236
"""Set up the test environment
1237
1238
Daughter classes should set up their own environment
1239
(including self.server) and explicitely call this
1240
method. This is needed because we want to reuse the same
1241
tests for proxy and no-proxy accesses which have
1242
different ways of setting self.server.
1243
"""
1434
super(TestAuth, self).setUp()
1435
self.server = self.get_readonly_server()
1244
1436
self.build_tree_contents([('a', 'contents of a\n'),
1245
1437
('b', 'contents of b\n'),])
1246
1438
1439
def create_transport_readonly_server(self):
1440
if self._auth_scheme == 'basic':
1441
server = http_utils.HTTPBasicAuthServer(
1442
protocol_version=self._protocol_version)
1443
else:
1444
if self._auth_scheme != 'digest':
1445
raise AssertionError('Unknown auth scheme: %r'
1446
% self._auth_scheme)
1447
server = http_utils.HTTPDigestAuthServer(
1448
protocol_version=self._protocol_version)
1449
return server
1450
1451
def _testing_pycurl(self):
1452
return pycurl_present and self._transport == PyCurlTransport
1453
1247
1454
def get_user_url(self, user=None, password=None):
1248
1455
"""Build an url embedding user and password"""
1249
1456
url = '%s://' % self.server._url_protocol
1255
1462
url += '%s:%s/' % (self.server.host, self.server.port)
1256
1463
return url
1257
1464
1465
def get_user_transport(self, user=None, password=None):
1466
return self._transport(self.get_user_url(user, password))
1467
1258
1468
def test_no_user(self):
1259
1469
self.server.add_user('joe', 'foo')
1260
1470
t = self.get_user_transport()
1294
1504
self.assertEqual(2, self.server.auth_required_errors)
1295
1505
1296
1506
def test_prompt_for_password(self):
1507
if self._testing_pycurl():
1508
raise tests.TestNotApplicable(
1509
'pycurl cannot prompt, it handles auth by embedding'
1510
' user:pass in urls only')
1511
1297
1512
self.server.add_user('joe', 'foo')
1298
1513
t = self.get_user_transport('joe', None)
1299
1514
stdout = tests.StringIOWrapper()
1321
1536
self.assertEquals(expected_prompt, actual_prompt)
1322
1537
1323
1538
def test_no_prompt_for_password_when_using_auth_config(self):
1324
user ='joe'
1539
if self._testing_pycurl():
1540
raise tests.TestNotApplicable(
1541
'pycurl does not support authentication.conf'
1542
' since it cannot prompt')
1543
1544
user =' joe'
1325
1545
password = 'foo'
1326
1546
stdin_content = 'bar\n' # Not the right password
1327
1547
self.server.add_user(user, password)
1341
1561
# Only one 'Authentication Required' error should occur
1342
1562
self.assertEqual(1, self.server.auth_required_errors)
1343
1563
1344
1345
1346
class TestHTTPAuth(TestAuth):
1347
"""Test HTTP authentication schemes.
1348
1349
Daughter classes MUST inherit from TestCaseWithWebserver too.
1350
"""
1351
1352
_auth_header = 'Authorization'
1353
1354
def setUp(self):
1355
TestCaseWithWebserver.setUp(self)
1356
self.server = self.get_readonly_server()
1357
TestAuth.setUp(self)
1358
1359
def get_user_transport(self, user=None, password=None):
1360
return self._transport(self.get_user_url(user, password))
1564
def test_changing_nonce(self):
1565
if self._auth_scheme != 'digest':
1566
raise tests.TestNotApplicable('HTTP auth digest only test')
1567
if self._testing_pycurl():
1568
raise tests.KnownFailure(
1569
'pycurl does not handle a nonce change')
1570
self.server.add_user('joe', 'foo')
1571
t = self.get_user_transport('joe', 'foo')
1572
self.assertEqual('contents of a\n', t.get('a').read())
1573
self.assertEqual('contents of b\n', t.get('b').read())
1574
# Only one 'Authentication Required' error should have
1575
# occured so far
1576
self.assertEqual(1, self.server.auth_required_errors)
1577
# The server invalidates the current nonce
1578
self.server.auth_nonce = self.server.auth_nonce + '. No, now!'
1579
self.assertEqual('contents of a\n', t.get('a').read())
1580
# Two 'Authentication Required' errors should occur (the
1581
# initial 'who are you' and a second 'who are you' with the new nonce)
1582
self.assertEqual(2, self.server.auth_required_errors)
1583
1361
1584
1362
1585
1363
1586
class TestProxyAuth(TestAuth):
1364
"""Test proxy authentication schemes.
1587
"""Test proxy authentication schemes."""
1365
1588
1366
Daughter classes MUST also inherit from TestCaseWithWebserver.
1367
"""
1368
1589
_auth_header = 'Proxy-authorization'
1369
_password_prompt_prefix = 'Proxy '
1370
1590
_password_prompt_prefix='Proxy '
1371
1591
1372
1592
def setUp(self):
1373
TestCaseWithWebserver.setUp(self)
1374
self.server = self.get_readonly_server()
1593
super(TestProxyAuth, self).setUp()
1375
1594
self._old_env = {}
1376
1595
self.addCleanup(self._restore_env)
1377
TestAuth.setUp(self)
1378
1596
# Override the contents to avoid false positives
1379
1597
self.build_tree_contents([('a', 'not proxied contents of a\n'),
1380
1598
('b', 'not proxied contents of b\n'),
1382
1600
('b-proxied', 'contents of b\n'),
1383
1601
])
1384
1602
1603
def create_transport_readonly_server(self):
1604
if self._auth_scheme == 'basic':
1605
server = http_utils.ProxyBasicAuthServer(
1606
protocol_version=self._protocol_version)
1607
else:
1608
if self._auth_scheme != 'digest':
1609
raise AssertionError('Unknown auth scheme: %r'
1610
% self._auth_scheme)
1611
server = http_utils.ProxyDigestAuthServer(
1612
protocol_version=self._protocol_version)
1613
return server
1614
1385
1615
def get_user_transport(self, user=None, password=None):
1386
1616
self._install_env({'all_proxy': self.get_user_url(user, password)})
1387
1617
return self._transport(self.server.get_url())
1394
1624
for name, value in self._old_env.iteritems():
1395
1625
osutils.set_or_unset_env(name, value)
1396
1626
1397
1398
class TestHTTPBasicAuth(TestHTTPAuth, TestCaseWithWebserver):
1399
"""Test http basic authentication scheme"""
1400
1401
_transport = HttpTransport_urllib
1402
1403
def create_transport_readonly_server(self):
1404
return HTTPBasicAuthServer()
1405
1406
1407
class TestHTTPProxyBasicAuth(TestProxyAuth, TestCaseWithWebserver):
1408
"""Test proxy basic authentication scheme"""
1409
1410
_transport = HttpTransport_urllib
1411
1412
def create_transport_readonly_server(self):
1413
return ProxyBasicAuthServer()
1414
1415
1416
class TestDigestAuth(object):
1417
"""Digest Authentication specific tests"""
1418
1419
def test_changing_nonce(self):
1420
self.server.add_user('joe', 'foo')
1421
t = self.get_user_transport('joe', 'foo')
1422
self.assertEqual('contents of a\n', t.get('a').read())
1423
self.assertEqual('contents of b\n', t.get('b').read())
1424
# Only one 'Authentication Required' error should have
1425
# occured so far
1426
self.assertEqual(1, self.server.auth_required_errors)
1427
# The server invalidates the current nonce
1428
self.server.auth_nonce = self.server.auth_nonce + '. No, now!'
1429
self.assertEqual('contents of a\n', t.get('a').read())
1430
# Two 'Authentication Required' errors should occur (the
1431
# initial 'who are you' and a second 'who are you' with the new nonce)
1432
self.assertEqual(2, self.server.auth_required_errors)
1433
1434
1435
class TestHTTPDigestAuth(TestHTTPAuth, TestDigestAuth, TestCaseWithWebserver):
1436
"""Test http digest authentication scheme"""
1437
1438
_transport = HttpTransport_urllib
1439
1440
def create_transport_readonly_server(self):
1441
return HTTPDigestAuthServer()
1442
1443
1444
class TestHTTPProxyDigestAuth(TestProxyAuth, TestDigestAuth,
1445
TestCaseWithWebserver):
1446
"""Test proxy digest authentication scheme"""
1447
1448
_transport = HttpTransport_urllib
1449
1450
def create_transport_readonly_server(self):
1451
return ProxyDigestAuthServer()
1452
1453
1454
class TestAuth_pycurl(object):
1455
"Tests that can't be applied to pycurl."""
1456
1457
def test_prompt_for_password(self):
1458
raise tests.TestNotApplicable(
1459
'pycurl cannot prompt, it handles auth by embedding'
1460
' user:pass in urls only')
1461
1462
def test_no_prompt_for_password_when_using_auth_config(self):
1463
raise tests.TestNotApplicable(
1464
'pycurl does not support authentication.conf'
1465
' since it cannot prompt')
1466
1467
1468
class TestHTTPBasicAuth_pycurl(TestWithTransport_pycurl, TestAuth_pycurl,
1469
TestHTTPBasicAuth):
1470
"""Test http basic authentication scheme for pycurl"""
1471
1472
1473
class TestHTTPProxyBasicAuth_pycurl(TestWithTransport_pycurl, TestAuth_pycurl,
1474
TestHTTPProxyBasicAuth):
1475
"""Test proxy basic authentication scheme for pycurl"""
1476
1477
def test_empty_pass(self):
1478
raise tests.KnownFailure(
1479
'some versions of pycurl does not handle empty proxy passwords')
1480
1481
1482
class TestHTTPDigestAuth_pycurl(TestWithTransport_pycurl, TestAuth_pycurl,
1483
TestHTTPDigestAuth):
1484
"""Test http digest authentication scheme for pycurl"""
1485
1486
def test_changing_nonce(self):
1487
raise tests.KnownFailure(
1488
'pycurl does not handle a nonce change')
1489
1490
1491
class TestHTTPProxyDigestAuth_pycurl(TestWithTransport_pycurl, TestAuth_pycurl,
1492
TestHTTPProxyDigestAuth):
1493
"""Test http digest authentication scheme for pycurl"""
1494
1495
def test_empty_pass(self):
1496
raise tests.KnownFailure(
1497
'some versions of pycurl does not handle empty proxy passwords')
1498
1499
def test_changing_nonce(self):
1500
raise tests.KnownFailure(
1501
'pycurl does not handle a nonce change')
1627
def test_empty_pass(self):
1628
if self._testing_pycurl():
1629
import pycurl
1630
if pycurl.version_info()[1] < '7.16.0':
1631
raise tests.KnownFailure(
1632
'pycurl < 7.16.0 does not handle empty proxy passwords')
1633
super(TestProxyAuth, self).test_empty_pass()
1634
1635
1636
class SampleSocket(object):
1637
"""A socket-like object for use in testing the HTTP request handler."""
1638
1639
def __init__(self, socket_read_content):
1640
"""Constructs a sample socket.
1641
1642
:param socket_read_content: a byte sequence
1643
"""
1644
# Use plain python StringIO so we can monkey-patch the close method to
1645
# not discard the contents.
1646
from StringIO import StringIO
1647
self.readfile = StringIO(socket_read_content)
1648
self.writefile = StringIO()
1649
self.writefile.close = lambda: None
1650
1651
def makefile(self, mode='r', bufsize=None):
1652
if 'r' in mode:
1653
return self.readfile
1654
else:
1655
return self.writefile
1656
1657
1658
class SmartHTTPTunnellingTest(tests.TestCaseWithTransport):
1659
1660
def setUp(self):
1661
super(SmartHTTPTunnellingTest, self).setUp()
1662
# We use the VFS layer as part of HTTP tunnelling tests.
1663
self._captureVar('BZR_NO_SMART_VFS', None)
1664
self.transport_readonly_server = http_utils.HTTPServerWithSmarts
1665
1666
def create_transport_readonly_server(self):
1667
return http_utils.HTTPServerWithSmarts(
1668
protocol_version=self._protocol_version)
1669
1670
def test_bulk_data(self):
1671
# We should be able to send and receive bulk data in a single message.
1672
# The 'readv' command in the smart protocol both sends and receives
1673
# bulk data, so we use that.
1674
self.build_tree(['data-file'])
1675
http_server = self.get_readonly_server()
1676
http_transport = self._transport(http_server.get_url())
1677
medium = http_transport.get_smart_medium()
1678
# Since we provide the medium, the url below will be mostly ignored
1679
# during the test, as long as the path is '/'.
1680
remote_transport = remote.RemoteTransport('bzr://fake_host/',
1681
medium=medium)
1682
self.assertEqual(
1683
[(0, "c")], list(remote_transport.readv("data-file", [(0,1)])))
1684
1685
def test_http_send_smart_request(self):
1686
1687
post_body = 'hello\n'
1688
expected_reply_body = 'ok\x012\n'
1689
1690
http_server = self.get_readonly_server()
1691
http_transport = self._transport(http_server.get_url())
1692
medium = http_transport.get_smart_medium()
1693
response = medium.send_http_smart_request(post_body)
1694
reply_body = response.read()
1695
self.assertEqual(expected_reply_body, reply_body)
1696
1697
def test_smart_http_server_post_request_handler(self):
1698
httpd = self.get_readonly_server()._get_httpd()
1699
1700
socket = SampleSocket(
1701
'POST /.bzr/smart %s \r\n' % self._protocol_version
1702
# HTTP/1.1 posts must have a Content-Length (but it doesn't hurt
1703
# for 1.0)
1704
+ 'Content-Length: 6\r\n'
1705
'\r\n'
1706
'hello\n')
1707
# Beware: the ('localhost', 80) below is the
1708
# client_address parameter, but we don't have one because
1709
# we have defined a socket which is not bound to an
1710
# address. The test framework never uses this client
1711
# address, so far...
1712
request_handler = http_utils.SmartRequestHandler(socket,
1713
('localhost', 80),
1714
httpd)
1715
response = socket.writefile.getvalue()
1716
self.assertStartsWith(response, '%s 200 ' % self._protocol_version)
1717
# This includes the end of the HTTP headers, and all the body.
1718
expected_end_of_response = '\r\n\r\nok\x012\n'
1719
self.assertEndsWith(response, expected_end_of_response)
1720
1502
1721
Loggerhead is a web-based interface for Breezy
Version: 2.0.1