To get this branch, use:
bzr branch
http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar
« back to all changes in this revision
Viewing changes to bzrlib/tests/test_http.py
- browse files at revision 3031.3.2
- compare with another revision
- compare with revision 1663
- download diff
- download tarball
- view history from revision 3031.3.2
- Committer: Robert Collins
- Date: 2008-02-06 04:06:42 UTC
- mfrom: (3216 +trunk)
- mto: This revision was merged to the branch mainline in revision 3217.
- Revision ID: robertc@robertcollins.net-20080206040642-2efx3l4iv5f95lxp
Merge up with bzr.dev.
- files added:
- bzrlib/help_topics
- bzrlib/help_topics/en
- doc/en/tutorials
- doc/en/user-guide/images
- files removed:
- bzrlib/plugins/multiparent.py
- files renamed:
- bzrlib/help_topics.py => bzrlib/help_topics/__init__.py
- doc/en/user-guide/authentication_conf.txt => bzrlib/help_topics/en/authentication.txt
- doc/en/user-guide/configuration.txt => bzrlib/help_topics/en/configuration.txt
- doc/en/user-guide/conflicts.txt => bzrlib/help_topics/en/conflicts.txt
- doc/en/user-reference/hooks.txt => bzrlib/help_topics/en/hooks.txt
- bzrlib/tests/HttpServer.py => bzrlib/tests/http_server.py
- bzrlib/tests/HTTPTestUtil.py => bzrlib/tests/http_utils.py
- doc/en/user-guide/centralized_workflow.txt => doc/en/tutorials/centralized_workflow.txt
- doc/en/user-guide/tutorial.txt => doc/en/tutorials/tutorial.txt
- files modified:
- .bzrignore
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
32
37
config,
33
38
errors,
34
39
osutils,
40
tests,
41
transport,
35
42
ui,
36
43
urlutils,
37
44
)
38
45
from bzrlib.tests import (
39
TestCase,
40
TestUIFactory,
41
TestSkipped,
42
StringIOWrapper,
43
)
44
from bzrlib.tests.HttpServer import (
45
HttpServer,
46
HttpServer_PyCurl,
47
HttpServer_urllib,
48
)
49
from bzrlib.tests.HTTPTestUtil import (
50
BadProtocolRequestHandler,
51
BadStatusRequestHandler,
52
ForbiddenRequestHandler,
53
HTTPBasicAuthServer,
54
HTTPDigestAuthServer,
55
HTTPServerRedirecting,
56
InvalidStatusRequestHandler,
57
LimitedRangeHTTPServer,
58
NoRangeRequestHandler,
59
ProxyBasicAuthServer,
60
ProxyDigestAuthServer,
61
ProxyServer,
62
SingleRangeRequestHandler,
63
SingleOnlyRangeRequestHandler,
64
TestCaseWithRedirectedWebserver,
65
TestCaseWithTwoWebservers,
66
TestCaseWithWebserver,
67
WallRequestHandler,
46
http_server,
47
http_utils,
68
48
)
69
49
from bzrlib.transport import (
70
_CoalescedOffset,
71
do_catching_redirections,
72
get_transport,
73
Transport,
50
http,
51
remote,
74
52
)
75
53
from bzrlib.transport.http import (
76
extract_auth,
77
HttpTransportBase,
54
_urllib,
78
55
_urllib2_wrappers,
79
56
)
80
from bzrlib.transport.http._urllib import HttpTransport_urllib
81
from bzrlib.transport.http._urllib2_wrappers import (
82
ProxyHandler,
83
Request,
84
)
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
85
160
86
161
87
162
class FakeManager(object):
150
225
self.port = None
151
226
152
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
153
304
class TestWithTransport_pycurl(object):
154
305
"""Test case to inherit from if pycurl is present"""
155
306
158
309
from bzrlib.transport.http._pycurl import PyCurlTransport
159
310
return PyCurlTransport
160
311
except errors.DependencyNotPresent:
161
raise TestSkipped('pycurl not present')
312
raise tests.TestSkipped('pycurl not present')
162
313
163
314
_transport = property(_get_pycurl_maybe)
164
315
165
316
166
class TestHttpUrls(TestCase):
317
class TestHttpUrls(tests.TestCase):
167
318
168
319
# TODO: This should be moved to authorization tests once they
169
320
# are written.
170
321
171
322
def test_url_parsing(self):
172
323
f = FakeManager()
173
url = extract_auth('http://example.com', f)
324
url = http.extract_auth('http://example.com', f)
174
325
self.assertEquals('http://example.com', url)
175
326
self.assertEquals(0, len(f.credentials))
176
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)
177
329
self.assertEquals('http://www.bazaar-vcs.org/bzr/bzr.dev', url)
178
330
self.assertEquals(1, len(f.credentials))
179
331
self.assertEquals([None, 'www.bazaar-vcs.org', 'user', 'pass'],
180
332
f.credentials[0])
181
333
182
334
183
class TestHttpTransportUrls(object):
184
"""Test the http urls.
185
186
This MUST be used by daughter classes that also inherit from
187
TestCase.
188
189
We can't inherit directly from TestCase or the
190
test framework will try to create an instance which cannot
191
run, its implementation being incomplete.
192
"""
335
class TestHttpTransportUrls(tests.TestCase):
336
"""Test the http urls."""
193
337
194
338
def test_abs_url(self):
195
339
"""Construction of absolute http URLs"""
226
370
server.tearDown()
227
371
228
372
229
class TestHttpUrls_urllib(TestHttpTransportUrls, TestCase):
230
"""Test http urls with urllib"""
231
232
_transport = HttpTransport_urllib
233
_server = HttpServer_urllib
234
_qualified_prefix = 'http+urllib'
235
236
237
class TestHttpUrls_pycurl(TestWithTransport_pycurl, TestHttpTransportUrls,
238
TestCase):
239
"""Test http urls with pycurl"""
240
241
_server = HttpServer_PyCurl
242
_qualified_prefix = 'http+pycurl'
373
class TestHttps_pycurl(TestWithTransport_pycurl, tests.TestCase):
243
374
244
375
# TODO: This should really be moved into another pycurl
245
376
# specific test. When https tests will be implemented, take
254
385
try:
255
386
import pycurl
256
387
except ImportError:
257
raise TestSkipped('pycurl not present')
258
# Now that we have pycurl imported, we can fake its version_info
259
# This was taken from a windows pycurl without SSL
260
# (thanks to bialix)
261
pycurl.version_info = lambda : (2,
262
'7.13.2',
263
462082,
264
'i386-pc-win32',
265
2576,
266
None,
267
0,
268
None,
269
('ftp', 'gopher', 'telnet',
270
'dict', 'ldap', 'http', 'file'),
271
None,
272
0,
273
None)
274
self.assertRaises(errors.DependencyNotPresent, self._transport,
275
'https://launchpad.net')
276
277
class TestHttpConnections(object):
278
"""Test the http connections.
279
280
This MUST be used by daughter classes that also inherit from
281
TestCaseWithWebserver.
282
283
We can't inherit directly from TestCaseWithWebserver or the
284
test framework will try to create an instance which cannot
285
run, its implementation being incomplete.
286
"""
388
raise tests.TestSkipped('pycurl not present')
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."""
287
417
288
418
def setUp(self):
289
TestCaseWithWebserver.setUp(self)
290
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',
291
421
transport=self.get_transport())
292
422
293
423
def test_http_has(self):
339
469
socket.setdefaulttimeout(default_timeout)
340
470
341
471
342
class TestHttpConnections_urllib(TestHttpConnections, TestCaseWithWebserver):
343
"""Test http connections with urllib"""
344
345
_transport = HttpTransport_urllib
346
347
348
349
class TestHttpConnections_pycurl(TestWithTransport_pycurl,
350
TestHttpConnections,
351
TestCaseWithWebserver):
352
"""Test http connections with pycurl"""
353
354
355
class TestHttpTransportRegistration(TestCase):
472
class TestHttpTransportRegistration(tests.TestCase):
356
473
"""Test registrations of various http implementations"""
357
474
358
475
def test_http_registered(self):
359
# urlllib should always be present
360
t = get_transport('http+urllib://bzr.google.com/')
361
self.assertIsInstance(t, Transport)
362
self.assertIsInstance(t, HttpTransport_urllib)
363
364
365
class TestPost(object):
366
367
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):
368
484
server = RecordingServer(expect_body_tail='end-of-body')
369
485
server.setUp()
370
486
self.addCleanup(server.tearDown)
487
scheme = self._qualified_prefix
371
488
url = '%s://%s:%s/' % (scheme, server.host, server.port)
372
try:
373
http_transport = get_transport(url)
374
except errors.UnsupportedProtocol:
375
raise TestSkipped('%s not available' % scheme)
489
http_transport = self._transport(url)
376
490
code, response = http_transport._post('abc def end-of-body')
377
491
self.assertTrue(
378
492
server.received_bytes.startswith('POST /.bzr/smart HTTP/1.'))
384
498
server.received_bytes.endswith('\r\n\r\nabc def end-of-body'))
385
499
386
500
387
class TestPost_urllib(TestCase, TestPost):
388
"""TestPost for urllib implementation"""
389
390
_transport = HttpTransport_urllib
391
392
def test_post_body_is_received_urllib(self):
393
self._test_post_body_is_received('http+urllib')
394
395
396
class TestPost_pycurl(TestWithTransport_pycurl, TestCase, TestPost):
397
"""TestPost for pycurl implementation"""
398
399
def test_post_body_is_received_pycurl(self):
400
self._test_post_body_is_received('http+pycurl')
401
402
403
class TestRangeHeader(TestCase):
501
class TestRangeHeader(tests.TestCase):
404
502
"""Test range_header method"""
405
503
406
504
def check_header(self, value, ranges=[], tail=0):
407
505
offsets = [ (start, end - start + 1) for start, end in ranges]
408
coalesce = Transport._coalesce_offsets
506
coalesce = transport.Transport._coalesce_offsets
409
507
coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))
410
range_header = HttpTransportBase._range_header
508
range_header = http.HttpTransportBase._range_header
411
509
self.assertEqual(value, range_header(coalesced, tail))
412
510
413
511
def test_range_header_single(self):
428
526
tail=50)
429
527
430
528
431
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):
432
555
"""Tests exceptions during the connection phase"""
433
556
434
def create_transport_readonly_server(self):
435
return HttpServer(WallRequestHandler)
557
_req_handler_class = WallRequestHandler
436
558
437
559
def test_http_has(self):
438
560
server = self.get_readonly_server()
452
574
t.get, 'foo/bar')
453
575
454
576
455
class TestWallServer_urllib(TestWallServer, TestCaseWithWebserver):
456
"""Tests "wall" server for urllib implementation"""
457
458
_transport = HttpTransport_urllib
459
460
461
class TestWallServer_pycurl(TestWithTransport_pycurl,
462
TestWallServer,
463
TestCaseWithWebserver):
464
"""Tests "wall" server for pycurl implementation"""
465
466
467
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):
468
589
"""Tests bad status from server."""
469
590
470
def create_transport_readonly_server(self):
471
return HttpServer(BadStatusRequestHandler)
591
_req_handler_class = BadStatusRequestHandler
472
592
473
593
def test_http_has(self):
474
594
server = self.get_readonly_server()
481
601
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
482
602
483
603
484
class TestBadStatusServer_urllib(TestBadStatusServer, TestCaseWithWebserver):
485
"""Tests bad status server for urllib implementation"""
486
487
_transport = HttpTransport_urllib
488
489
490
class TestBadStatusServer_pycurl(TestWithTransport_pycurl,
491
TestBadStatusServer,
492
TestCaseWithWebserver):
493
"""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
494
612
495
613
496
614
class TestInvalidStatusServer(TestBadStatusServer):
499
617
Both implementations raises the same error as for a bad status.
500
618
"""
501
619
502
def create_transport_readonly_server(self):
503
return HttpServer(InvalidStatusRequestHandler)
504
505
506
class TestInvalidStatusServer_urllib(TestInvalidStatusServer,
507
TestCaseWithWebserver):
508
"""Tests invalid status server for urllib implementation"""
509
510
_transport = HttpTransport_urllib
511
512
513
class TestInvalidStatusServer_pycurl(TestWithTransport_pycurl,
514
TestInvalidStatusServer,
515
TestCaseWithWebserver):
516
"""Tests invalid status server for pycurl implementation"""
517
518
519
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):
520
650
"""Tests bad protocol from server."""
521
651
522
def create_transport_readonly_server(self):
523
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()
524
659
525
660
def test_http_has(self):
526
661
server = self.get_readonly_server()
533
668
self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')
534
669
535
670
536
class TestBadProtocolServer_urllib(TestBadProtocolServer,
537
TestCaseWithWebserver):
538
"""Tests bad protocol server for urllib implementation"""
539
540
_transport = HttpTransport_urllib
541
542
# curl don't check the protocol version
543
#class TestBadProtocolServer_pycurl(TestWithTransport_pycurl,
544
# TestBadProtocolServer,
545
# TestCaseWithWebserver):
546
# """Tests bad protocol server for pycurl implementation"""
547
548
549
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):
550
682
"""Tests forbidden server"""
551
683
552
def create_transport_readonly_server(self):
553
return HttpServer(ForbiddenRequestHandler)
684
_req_handler_class = ForbiddenRequestHandler
554
685
555
686
def test_http_has(self):
556
687
server = self.get_readonly_server()
563
694
self.assertRaises(errors.TransportError, t.get, 'foo/bar')
564
695
565
696
566
class TestForbiddenServer_urllib(TestForbiddenServer, TestCaseWithWebserver):
567
"""Tests forbidden server for urllib implementation"""
568
569
_transport = HttpTransport_urllib
570
571
572
class TestForbiddenServer_pycurl(TestWithTransport_pycurl,
573
TestForbiddenServer,
574
TestCaseWithWebserver):
575
"""Tests forbidden server for pycurl implementation"""
576
577
578
class TestRecordingServer(TestCase):
697
class TestRecordingServer(tests.TestCase):
579
698
580
699
def test_create(self):
581
700
server = RecordingServer(expect_body_tail=None)
606
725
self.assertEqual('abc', server.received_bytes)
607
726
608
727
609
class TestRangeRequestServer(object):
728
class TestRangeRequestServer(TestSpecificRequestHandler):
610
729
"""Tests readv requests against server.
611
730
612
This MUST be used by daughter classes that also inherit from
613
TestCaseWithWebserver.
614
615
We can't inherit directly from TestCaseWithWebserver or the
616
test framework will try to create an instance which cannot
617
run, its implementation being incomplete.
731
We test against default "normal" server.
618
732
"""
619
733
620
734
def setUp(self):
621
TestCaseWithWebserver.setUp(self)
735
super(TestRangeRequestServer, self).setUp()
622
736
self.build_tree_contents([('a', '0123456789')],)
623
737
624
738
def test_readv(self):
653
767
self.assertListRaises((errors.InvalidRange, errors.ShortReadvError,),
654
768
t.readv, 'a', [(12,2)])
655
769
770
def test_readv_multiple_get_requests(self):
771
server = self.get_readonly_server()
772
t = self._transport(server.get_url())
773
# force transport to issue multiple requests
774
t._max_readv_combine = 1
775
t._max_get_ranges = 1
776
l = list(t.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
777
self.assertEqual(l[0], (0, '0'))
778
self.assertEqual(l[1], (1, '1'))
779
self.assertEqual(l[2], (3, '34'))
780
self.assertEqual(l[3], (9, '9'))
781
# The server should have issued 4 requests
782
self.assertEqual(4, server.GET_request_nb)
783
784
def test_readv_get_max_size(self):
785
server = self.get_readonly_server()
786
t = self._transport(server.get_url())
787
# force transport to issue multiple requests by limiting the number of
788
# bytes by request. Note that this apply to coalesced offsets only, a
789
# single range will keep its size even if bigger than the limit.
790
t._get_max_size = 2
791
l = list(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
792
self.assertEqual(l[0], (0, '0'))
793
self.assertEqual(l[1], (1, '1'))
794
self.assertEqual(l[2], (2, '2345'))
795
self.assertEqual(l[3], (6, '6789'))
796
# The server should have issued 3 requests
797
self.assertEqual(3, server.GET_request_nb)
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
656
838
657
839
class TestSingleRangeRequestServer(TestRangeRequestServer):
658
840
"""Test readv against a server which accept only single range requests"""
659
841
660
def create_transport_readonly_server(self):
661
return HttpServer(SingleRangeRequestHandler)
662
663
664
class TestSingleRangeRequestServer_urllib(TestSingleRangeRequestServer,
665
TestCaseWithWebserver):
666
"""Tests single range requests accepting server for urllib implementation"""
667
668
_transport = HttpTransport_urllib
669
670
671
class TestSingleRangeRequestServer_pycurl(TestWithTransport_pycurl,
672
TestSingleRangeRequestServer,
673
TestCaseWithWebserver):
674
"""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)
675
856
676
857
677
858
class TestSingleOnlyRangeRequestServer(TestRangeRequestServer):
678
859
"""Test readv against a server which only accept single range requests"""
679
860
680
def create_transport_readonly_server(self):
681
return HttpServer(SingleOnlyRangeRequestHandler)
682
683
684
class TestSingleOnlyRangeRequestServer_urllib(TestSingleOnlyRangeRequestServer,
685
TestCaseWithWebserver):
686
"""Tests single range requests accepting server for urllib implementation"""
687
688
_transport = HttpTransport_urllib
689
690
691
class TestSingleOnlyRangeRequestServer_pycurl(TestWithTransport_pycurl,
692
TestSingleOnlyRangeRequestServer,
693
TestCaseWithWebserver):
694
"""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)
695
872
696
873
697
874
class TestNoRangeRequestServer(TestRangeRequestServer):
698
875
"""Test readv against a server which do not accept range requests"""
699
876
700
def create_transport_readonly_server(self):
701
return HttpServer(NoRangeRequestHandler)
702
703
704
class TestNoRangeRequestServer_urllib(TestNoRangeRequestServer,
705
TestCaseWithWebserver):
706
"""Tests range requests refusing server for urllib implementation"""
707
708
_transport = HttpTransport_urllib
709
710
711
class TestNoRangeRequestServer_pycurl(TestWithTransport_pycurl,
712
TestNoRangeRequestServer,
713
TestCaseWithWebserver):
714
"""Tests range requests refusing server for pycurl implementation"""
715
716
717
class TestLimitedRangeRequestServer(object):
718
"""Tests readv requests against server that errors out on too much ranges.
719
720
This MUST be used by daughter classes that also inherit from
721
TestCaseWithWebserver.
722
723
We can't inherit directly from TestCaseWithWebserver or the
724
test framework will try to create an instance which cannot
725
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).
726
915
"""
727
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
728
1019
range_limit = 3
729
1020
730
1021
def create_transport_readonly_server(self):
731
# Requests with more range specifiers will error out
732
return LimitedRangeHTTPServer(range_limit=self.range_limit)
1022
return LimitedRangeHTTPServer(range_limit=self.range_limit,
1023
protocol_version=self._protocol_version)
733
1024
734
1025
def get_transport(self):
735
1026
return self._transport(self.get_readonly_server().get_url())
736
1027
737
1028
def setUp(self):
738
TestCaseWithWebserver.setUp(self)
1029
http_utils.TestCaseWithWebserver.setUp(self)
739
1030
# We need to manipulate ranges that correspond to real chunks in the
740
1031
# response, so we build a content appropriately.
741
filler = ''.join(['abcdefghij' for _ in range(102)])
1032
filler = ''.join(['abcdefghij' for x in range(102)])
742
1033
content = ''.join(['%04d' % v + filler for v in range(16)])
743
1034
self.build_tree_contents([('a', content)],)
744
1035
749
1040
self.assertEqual(l[1], (1024, '0001'))
750
1041
self.assertEqual(1, self.get_readonly_server().GET_request_nb)
751
1042
752
def test_a_lot_of_ranges(self):
1043
def test_more_ranges(self):
753
1044
t = self.get_transport()
754
1045
l = list(t.readv('a', ((0, 4), (1024, 4), (4096, 4), (8192, 4))))
755
1046
self.assertEqual(l[0], (0, '0000'))
757
1048
self.assertEqual(l[2], (4096, '0004'))
758
1049
self.assertEqual(l[3], (8192, '0008'))
759
1050
# The server will refuse to serve the first request (too much ranges),
760
# a second request will succeeds.
1051
# a second request will succeed.
761
1052
self.assertEqual(2, self.get_readonly_server().GET_request_nb)
762
1053
763
1054
764
class TestLimitedRangeRequestServer_urllib(TestLimitedRangeRequestServer,
765
TestCaseWithWebserver):
766
"""Tests limited range requests server for urllib implementation"""
767
768
_transport = HttpTransport_urllib
769
770
771
class TestLimitedRangeRequestServer_pycurl(TestWithTransport_pycurl,
772
TestLimitedRangeRequestServer,
773
TestCaseWithWebserver):
774
"""Tests limited range requests server for pycurl implementation"""
775
776
777
778
class TestHttpProxyWhiteBox(TestCase):
1055
class TestHttpProxyWhiteBox(tests.TestCase):
779
1056
"""Whitebox test proxy http authorization.
780
1057
781
1058
Only the urllib implementation is tested here.
782
1059
"""
783
1060
784
1061
def setUp(self):
785
TestCase.setUp(self)
1062
tests.TestCase.setUp(self)
786
1063
self._old_env = {}
787
1064
788
1065
def tearDown(self):
789
1066
self._restore_env()
1067
tests.TestCase.tearDown(self)
790
1068
791
1069
def _install_env(self, env):
792
1070
for name, value in env.iteritems():
797
1075
osutils.set_or_unset_env(name, value)
798
1076
799
1077
def _proxied_request(self):
800
handler = ProxyHandler()
801
request = Request('GET','http://baz/buzzle')
1078
handler = _urllib2_wrappers.ProxyHandler()
1079
request = _urllib2_wrappers.Request('GET','http://baz/buzzle')
802
1080
handler.set_proxy(request, 'http')
803
1081
return request
804
1082
813
1091
self.assertRaises(errors.InvalidURL, self._proxied_request)
814
1092
815
1093
816
class TestProxyHttpServer(object):
1094
class TestProxyHttpServer(http_utils.TestCaseWithTwoWebservers):
817
1095
"""Tests proxy server.
818
1096
819
This MUST be used by daughter classes that also inherit from
820
TestCaseWithTwoWebservers.
821
822
We can't inherit directly from TestCaseWithTwoWebservers or
823
the test framework will try to create an instance which
824
cannot run, its implementation being incomplete.
825
826
1097
Be aware that we do not setup a real proxy here. Instead, we
827
1098
check that the *connection* goes through the proxy by serving
828
1099
different content (the faked proxy server append '-proxied'
833
1104
# test https connections.
834
1105
835
1106
def setUp(self):
836
TestCaseWithTwoWebservers.setUp(self)
1107
super(TestProxyHttpServer, self).setUp()
837
1108
self.build_tree_contents([('foo', 'contents of foo\n'),
838
1109
('foo-proxied', 'proxied contents of foo\n')])
839
1110
# Let's setup some attributes for tests
840
1111
self.server = self.get_readonly_server()
841
1112
self.proxy_address = '%s:%d' % (self.server.host, self.server.port)
842
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
843
1119
# The secondary server is the proxy
844
1120
self.proxy = self.get_secondary_server()
845
1121
self.proxy_url = self.proxy.get_url()
846
1122
self._old_env = {}
847
1123
1124
def _testing_pycurl(self):
1125
return pycurl_present and self._transport == PyCurlTransport
1126
848
1127
def create_transport_secondary_server(self):
849
1128
"""Creates an http server that will serve files with
850
1129
'-proxied' appended to their names.
851
1130
"""
852
return ProxyServer()
1131
return http_utils.ProxyServer(protocol_version=self._protocol_version)
853
1132
854
1133
def _install_env(self, env):
855
1134
for name, value in env.iteritems():
881
1160
self.proxied_in_env({'http_proxy': self.proxy_url})
882
1161
883
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')
884
1169
self.proxied_in_env({'HTTP_PROXY': self.proxy_url})
885
1170
886
1171
def test_all_proxy(self):
894
1179
'no_proxy': self.no_proxy_host})
895
1180
896
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')
897
1185
self.not_proxied_in_env({'HTTP_PROXY': self.proxy_url,
898
1186
'NO_PROXY': self.no_proxy_host})
899
1187
906
1194
'NO_PROXY': self.no_proxy_host})
907
1195
908
1196
def test_http_proxy_without_scheme(self):
909
self.assertRaises(errors.InvalidURL,
910
self.proxied_in_env,
911
{'http_proxy': self.proxy_address})
912
913
914
class TestProxyHttpServer_urllib(TestProxyHttpServer,
915
TestCaseWithTwoWebservers):
916
"""Tests proxy server for urllib implementation"""
917
918
_transport = HttpTransport_urllib
919
920
921
class TestProxyHttpServer_pycurl(TestWithTransport_pycurl,
922
TestProxyHttpServer,
923
TestCaseWithTwoWebservers):
924
"""Tests proxy server for pycurl implementation"""
925
926
def setUp(self):
927
TestProxyHttpServer.setUp(self)
928
# Oh my ! pycurl does not check for the port as part of
929
# no_proxy :-( So we just test the host part
930
self.no_proxy_host = 'localhost'
931
932
def test_HTTP_PROXY(self):
933
# pycurl does not check HTTP_PROXY for security reasons
934
# (for use in a CGI context that we do not care
935
# about. Should we ?)
936
raise TestSkipped('pycurl does not check HTTP_PROXY '
937
'for security reasons')
938
939
def test_HTTP_PROXY_with_NO_PROXY(self):
940
raise TestSkipped('pycurl does not check HTTP_PROXY '
941
'for security reasons')
942
943
def test_http_proxy_without_scheme(self):
944
# pycurl *ignores* invalid proxy env variables. If that
945
# ever change in the future, this test will fail
946
# indicating that pycurl do not ignore anymore such
947
# variables.
948
self.not_proxied_in_env({'http_proxy': self.proxy_address})
949
950
951
class TestRanges(object):
952
"""Test the Range header in GET methods..
953
954
This MUST be used by daughter classes that also inherit from
955
TestCaseWithWebserver.
956
957
We can't inherit directly from TestCaseWithWebserver or the
958
test framework will try to create an instance which cannot
959
run, its implementation being incomplete.
960
"""
961
962
def setUp(self):
963
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)
964
1213
self.build_tree_contents([('a', '0123456789')],)
965
1214
server = self.get_readonly_server()
966
1215
self.transport = self._transport(server.get_url())
967
1216
1217
def create_transport_readonly_server(self):
1218
return http_server.HttpServer(protocol_version=self._protocol_version)
1219
968
1220
def _file_contents(self, relpath, ranges):
969
1221
offsets = [ (start, end - start + 1) for start, end in ranges]
970
1222
coalesce = self.transport._coalesce_offsets
978
1230
def _file_tail(self, relpath, tail_amount):
979
1231
code, data = self.transport._get(relpath, [], tail_amount)
980
1232
self.assertTrue(code in (200, 206),'_get returns: %d' % code)
981
data.seek(-tail_amount + 1, 2)
1233
data.seek(-tail_amount, 2)
982
1234
return data.read(tail_amount)
983
1235
984
1236
def test_range_header(self):
985
1237
# Valid ranges
986
1238
map(self.assertEqual,['0', '234'],
987
1239
list(self._file_contents('a', [(0,0), (2,4)])),)
988
# Tail
1240
1241
def test_range_header_tail(self):
989
1242
self.assertEqual('789', self._file_tail('a', 3))
990
# Syntactically invalid range
991
self.assertListRaises(errors.InvalidRange,
1243
1244
def test_syntactically_invalid_range_header(self):
1245
self.assertListRaises(errors.InvalidHttpRange,
992
1246
self._file_contents, 'a', [(4, 3)])
993
# Semantically invalid range
994
self.assertListRaises(errors.InvalidRange,
1247
1248
def test_semantically_invalid_range_header(self):
1249
self.assertListRaises(errors.InvalidHttpRange,
995
1250
self._file_contents, 'a', [(42, 128)])
996
1251
997
1252
998
class TestRanges_urllib(TestRanges, TestCaseWithWebserver):
999
"""Test the Range header in GET methods for urllib implementation"""
1000
1001
_transport = HttpTransport_urllib
1002
1003
1004
class TestRanges_pycurl(TestWithTransport_pycurl,
1005
TestRanges,
1006
TestCaseWithWebserver):
1007
"""Test the Range header in GET methods for pycurl implementation"""
1008
1009
1010
class TestHTTPRedirections(object):
1011
"""Test redirection between http servers.
1012
1013
This MUST be used by daughter classes that also inherit from
1014
TestCaseWithRedirectedWebserver.
1015
1016
We can't inherit directly from TestCaseWithTwoWebservers or the
1017
test framework will try to create an instance which cannot
1018
run, its implementation being incomplete.
1019
"""
1253
class TestHTTPRedirections(http_utils.TestCaseWithRedirectedWebserver):
1254
"""Test redirection between http servers."""
1020
1255
1021
1256
def create_transport_secondary_server(self):
1022
1257
"""Create the secondary server redirecting to the primary server"""
1023
1258
new = self.get_readonly_server()
1024
1259
1025
redirecting = HTTPServerRedirecting()
1260
redirecting = http_utils.HTTPServerRedirecting(
1261
protocol_version=self._protocol_version)
1026
1262
redirecting.redirect_to(new.host, new.port)
1027
1263
return redirecting
1028
1264
1048
1284
self.assertEqual([], bundle.revisions)
1049
1285
1050
1286
1051
class TestHTTPRedirections_urllib(TestHTTPRedirections,
1052
TestCaseWithRedirectedWebserver):
1053
"""Tests redirections for urllib implementation"""
1054
1055
_transport = HttpTransport_urllib
1056
1057
1058
1059
class TestHTTPRedirections_pycurl(TestWithTransport_pycurl,
1060
TestHTTPRedirections,
1061
TestCaseWithRedirectedWebserver):
1062
"""Tests redirections for pycurl implementation"""
1063
1064
1065
class RedirectedRequest(Request):
1066
"""Request following redirections"""
1067
1068
init_orig = Request.__init__
1287
class RedirectedRequest(_urllib2_wrappers.Request):
1288
"""Request following redirections. """
1289
1290
init_orig = _urllib2_wrappers.Request.__init__
1069
1291
1070
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.
1071
1299
RedirectedRequest.init_orig(self, method, url, args, kwargs)
1072
1300
self.follow_redirections = True
1073
1301
1074
1302
1075
class TestHTTPSilentRedirections_urllib(TestCaseWithRedirectedWebserver):
1076
"""Test redirections provided by urllib.
1303
class TestHTTPSilentRedirections(http_utils.TestCaseWithRedirectedWebserver):
1304
"""Test redirections.
1077
1305
1078
1306
http implementations do not redirect silently anymore (they
1079
1307
do not redirect at all in fact). The mechanism is still in
1086
1314
-- vila 20070212
1087
1315
"""
1088
1316
1089
_transport = HttpTransport_urllib
1090
1091
1317
def setUp(self):
1092
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()
1093
1322
self.setup_redirected_request()
1094
1323
self.addCleanup(self.cleanup_redirected_request)
1095
1324
self.build_tree_contents([('a','a'),
1116
1345
1117
1346
def create_transport_secondary_server(self):
1118
1347
"""Create the secondary server, redirections are defined in the tests"""
1119
return HTTPServerRedirecting()
1348
return http_utils.HTTPServerRedirecting(
1349
protocol_version=self._protocol_version)
1120
1350
1121
1351
def test_one_redirection(self):
1122
1352
t = self.old_transport
1138
1368
self.old_server.port)
1139
1369
new_prefix = 'http://%s:%s' % (self.new_server.host,
1140
1370
self.new_server.port)
1141
self.old_server.redirections = \
1142
[('/1(.*)', r'%s/2\1' % (old_prefix), 302),
1143
('/2(.*)', r'%s/3\1' % (old_prefix), 303),
1144
('/3(.*)', r'%s/4\1' % (old_prefix), 307),
1145
('/4(.*)', r'%s/5\1' % (new_prefix), 301),
1146
('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),
1147
]
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
]
1148
1378
self.assertEquals('redirected 5 times',t._perform(req).read())
1149
1379
1150
1380
1151
class TestDoCatchRedirections(TestCaseWithRedirectedWebserver):
1152
"""Test transport.do_catching_redirections.
1153
1154
We arbitrarily choose to use urllib transports
1155
"""
1156
1157
_transport = HttpTransport_urllib
1381
class TestDoCatchRedirections(http_utils.TestCaseWithRedirectedWebserver):
1382
"""Test transport.do_catching_redirections."""
1158
1383
1159
1384
def setUp(self):
1160
1385
super(TestDoCatchRedirections, self).setUp()
1170
1395
1171
1396
# We use None for redirected so that we fail if redirected
1172
1397
self.assertEquals('0123456789',
1173
do_catching_redirections(self.get_a, t, None).read())
1398
transport.do_catching_redirections(
1399
self.get_a, t, None).read())
1174
1400
1175
1401
def test_one_redirection(self):
1176
1402
self.redirections = 0
1181
1407
return self._transport(dir)
1182
1408
1183
1409
self.assertEquals('0123456789',
1184
do_catching_redirections(self.get_a,
1185
self.old_transport,
1186
redirected
1187
).read())
1410
transport.do_catching_redirections(
1411
self.get_a, self.old_transport, redirected).read())
1188
1412
self.assertEquals(1, self.redirections)
1189
1413
1190
1414
def test_redirection_loop(self):
1195
1419
# a/a/a
1196
1420
return self.old_transport.clone(exception.target)
1197
1421
1198
self.assertRaises(errors.TooManyRedirections, do_catching_redirections,
1422
self.assertRaises(errors.TooManyRedirections,
1423
transport.do_catching_redirections,
1199
1424
self.get_a, self.old_transport, redirected)
1200
1425
1201
1426
1202
class TestAuth(object):
1203
"""Test some authentication scheme specified by daughter class.
1204
1205
This MUST be used by daughter classes that also inherit from
1206
either TestCaseWithWebserver or TestCaseWithTwoWebservers.
1207
"""
1208
1427
class TestAuth(http_utils.TestCaseWithWebserver):
1428
"""Test authentication scheme"""
1429
1430
_auth_header = 'Authorization'
1209
1431
_password_prompt_prefix = ''
1210
1432
1211
1433
def setUp(self):
1212
"""Set up the test environment
1213
1214
Daughter classes should set up their own environment
1215
(including self.server) and explicitely call this
1216
method. This is needed because we want to reuse the same
1217
tests for proxy and no-proxy accesses which have
1218
different ways of setting self.server.
1219
"""
1434
super(TestAuth, self).setUp()
1435
self.server = self.get_readonly_server()
1220
1436
self.build_tree_contents([('a', 'contents of a\n'),
1221
1437
('b', 'contents of b\n'),])
1222
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
1223
1454
def get_user_url(self, user=None, password=None):
1224
1455
"""Build an url embedding user and password"""
1225
1456
url = '%s://' % self.server._url_protocol
1231
1462
url += '%s:%s/' % (self.server.host, self.server.port)
1232
1463
return url
1233
1464
1465
def get_user_transport(self, user=None, password=None):
1466
return self._transport(self.get_user_url(user, password))
1467
1234
1468
def test_no_user(self):
1235
1469
self.server.add_user('joe', 'foo')
1236
1470
t = self.get_user_transport()
1270
1504
self.assertEqual(2, self.server.auth_required_errors)
1271
1505
1272
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
1273
1512
self.server.add_user('joe', 'foo')
1274
1513
t = self.get_user_transport('joe', None)
1275
stdout = StringIOWrapper()
1276
ui.ui_factory = TestUIFactory(stdin='foo\n', stdout=stdout)
1514
stdout = tests.StringIOWrapper()
1515
ui.ui_factory = tests.TestUIFactory(stdin='foo\n', stdout=stdout)
1277
1516
self.assertEqual('contents of a\n',t.get('a').read())
1278
1517
# stdin should be empty
1279
1518
self.assertEqual('', ui.ui_factory.stdin.readline())
1297
1536
self.assertEquals(expected_prompt, actual_prompt)
1298
1537
1299
1538
def test_no_prompt_for_password_when_using_auth_config(self):
1539
if self._testing_pycurl():
1540
raise tests.TestNotApplicable(
1541
'pycurl does not support authentication.conf'
1542
' since it cannot prompt')
1543
1300
1544
user =' joe'
1301
1545
password = 'foo'
1302
1546
stdin_content = 'bar\n' # Not the right password
1303
1547
self.server.add_user(user, password)
1304
1548
t = self.get_user_transport(user, None)
1305
ui.ui_factory = TestUIFactory(stdin=stdin_content,
1306
stdout=StringIOWrapper())
1549
ui.ui_factory = tests.TestUIFactory(stdin=stdin_content,
1550
stdout=tests.StringIOWrapper())
1307
1551
# Create a minimal config file with the right password
1308
1552
conf = config.AuthenticationConfig()
1309
1553
conf._get_config().update(
1317
1561
# Only one 'Authentication Required' error should occur
1318
1562
self.assertEqual(1, self.server.auth_required_errors)
1319
1563
1320
1321
1322
class TestHTTPAuth(TestAuth):
1323
"""Test HTTP authentication schemes.
1324
1325
Daughter classes MUST inherit from TestCaseWithWebserver too.
1326
"""
1327
1328
_auth_header = 'Authorization'
1329
1330
def setUp(self):
1331
TestCaseWithWebserver.setUp(self)
1332
self.server = self.get_readonly_server()
1333
TestAuth.setUp(self)
1334
1335
def get_user_transport(self, user=None, password=None):
1336
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
1337
1584
1338
1585
1339
1586
class TestProxyAuth(TestAuth):
1340
"""Test proxy authentication schemes.
1587
"""Test proxy authentication schemes."""
1341
1588
1342
Daughter classes MUST also inherit from TestCaseWithWebserver.
1343
"""
1344
1589
_auth_header = 'Proxy-authorization'
1345
_password_prompt_prefix = 'Proxy '
1346
1590
_password_prompt_prefix='Proxy '
1347
1591
1348
1592
def setUp(self):
1349
TestCaseWithWebserver.setUp(self)
1350
self.server = self.get_readonly_server()
1593
super(TestProxyAuth, self).setUp()
1351
1594
self._old_env = {}
1352
1595
self.addCleanup(self._restore_env)
1353
TestAuth.setUp(self)
1354
1596
# Override the contents to avoid false positives
1355
1597
self.build_tree_contents([('a', 'not proxied contents of a\n'),
1356
1598
('b', 'not proxied contents of b\n'),
1358
1600
('b-proxied', 'contents of b\n'),
1359
1601
])
1360
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
1361
1615
def get_user_transport(self, user=None, password=None):
1362
1616
self._install_env({'all_proxy': self.get_user_url(user, password)})
1363
1617
return self._transport(self.server.get_url())
1370
1624
for name, value in self._old_env.iteritems():
1371
1625
osutils.set_or_unset_env(name, value)
1372
1626
1373
1374
class TestHTTPBasicAuth(TestHTTPAuth, TestCaseWithWebserver):
1375
"""Test http basic authentication scheme"""
1376
1377
_transport = HttpTransport_urllib
1378
1379
def create_transport_readonly_server(self):
1380
return HTTPBasicAuthServer()
1381
1382
1383
class TestHTTPProxyBasicAuth(TestProxyAuth, TestCaseWithWebserver):
1384
"""Test proxy basic authentication scheme"""
1385
1386
_transport = HttpTransport_urllib
1387
1388
def create_transport_readonly_server(self):
1389
return ProxyBasicAuthServer()
1390
1391
1392
class TestDigestAuth(object):
1393
"""Digest Authentication specific tests"""
1394
1395
def test_changing_nonce(self):
1396
self.server.add_user('joe', 'foo')
1397
t = self.get_user_transport('joe', 'foo')
1398
self.assertEqual('contents of a\n', t.get('a').read())
1399
self.assertEqual('contents of b\n', t.get('b').read())
1400
# Only one 'Authentication Required' error should have
1401
# occured so far
1402
self.assertEqual(1, self.server.auth_required_errors)
1403
# The server invalidates the current nonce
1404
self.server.auth_nonce = self.server.auth_nonce + '. No, now!'
1405
self.assertEqual('contents of a\n', t.get('a').read())
1406
# Two 'Authentication Required' errors should occur (the
1407
# initial 'who are you' and a second 'who are you' with the new nonce)
1408
self.assertEqual(2, self.server.auth_required_errors)
1409
1410
1411
class TestHTTPDigestAuth(TestHTTPAuth, TestDigestAuth, TestCaseWithWebserver):
1412
"""Test http digest authentication scheme"""
1413
1414
_transport = HttpTransport_urllib
1415
1416
def create_transport_readonly_server(self):
1417
return HTTPDigestAuthServer()
1418
1419
1420
class TestHTTPProxyDigestAuth(TestProxyAuth, TestDigestAuth,
1421
TestCaseWithWebserver):
1422
"""Test proxy digest authentication scheme"""
1423
1424
_transport = HttpTransport_urllib
1425
1426
def create_transport_readonly_server(self):
1427
return ProxyDigestAuthServer()
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
1428
1721
Loggerhead is a web-based interface for Breezy
Version: 2.0.2.dev0