~bzr-pqm/bzr/bzr.dev : revision 3152.1.3

14

# along with this program; if not, write to the Free Software

15

# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA

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)])

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

821

1074

osutils.set_or_unset_env(name, value)

822

1075

823

1076

def _proxied_request(self):

824

handler = ProxyHandler()

825

request = Request('GET','http://baz/buzzle')

1077

handler = _urllib2_wrappers.ProxyHandler()

1078

request = _urllib2_wrappers.Request('GET','http://baz/buzzle')

826

1079

handler.set_proxy(request, 'http')

827

1080

return request

828

1081

837

1090

self.assertRaises(errors.InvalidURL, self._proxied_request)

838

1091

839

1092

840

class TestProxyHttpServer(object):

1093

class TestProxyHttpServer(http_utils.TestCaseWithTwoWebservers):

841

1094

"""Tests proxy server.

842

1095

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

1096

Be aware that we do not setup a real proxy here. Instead, we

851

1097

check that the *connection* goes through the proxy by serving

852

1098

different content (the faked proxy server append '-proxied'

857

1103

# test https connections.

858

1104

859

1105

def setUp(self):

860

TestCaseWithTwoWebservers.setUp(self)

1106

super(TestProxyHttpServer, self).setUp()

861

1107

self.build_tree_contents([('foo', 'contents of foo\n'),

862

1108

('foo-proxied', 'proxied contents of foo\n')])

863

1109

# Let's setup some attributes for tests

864

1110

self.server = self.get_readonly_server()

865

1111

self.proxy_address = '%s:%d' % (self.server.host, self.server.port)

866

self.no_proxy_host = self.proxy_address

1112

if self._testing_pycurl():

1113

# Oh my ! pycurl does not check for the port as part of

1114

# no_proxy :-( So we just test the host part

1115

self.no_proxy_host = 'localhost'

1116

else:

1117

self.no_proxy_host = self.proxy_address

867

1118

# The secondary server is the proxy

868

1119

self.proxy = self.get_secondary_server()

869

1120

self.proxy_url = self.proxy.get_url()

870

1121

self._old_env = {}

871

1122

1123

def _testing_pycurl(self):

1124

return pycurl_present and self._transport == PyCurlTransport

1125

872

1126

def create_transport_secondary_server(self):

873

1127

"""Creates an http server that will serve files with

874

1128

'-proxied' appended to their names.

875

1129

"""

876

return ProxyServer()

1130

return http_utils.ProxyServer(protocol_version=self._protocol_version)

877

1131

878

1132

def _install_env(self, env):

879

1133

for name, value in env.iteritems():

905

1159

self.proxied_in_env({'http_proxy': self.proxy_url})

906

1160

907

1161

def test_HTTP_PROXY(self):

1162

if self._testing_pycurl():

1163

# pycurl does not check HTTP_PROXY for security reasons

1164

# (for use in a CGI context that we do not care

1165

# about. Should we ?)

1166

raise tests.TestNotApplicable(

1167

'pycurl does not check HTTP_PROXY for security reasons')

908

1168

self.proxied_in_env({'HTTP_PROXY': self.proxy_url})

909

1169

910

1170

def test_all_proxy(self):

918

1178

'no_proxy': self.no_proxy_host})

919

1179

920

1180

def test_HTTP_PROXY_with_NO_PROXY(self):

1181

if self._testing_pycurl():

1182

raise tests.TestNotApplicable(

1183

'pycurl does not check HTTP_PROXY for security reasons')

921

1184

self.not_proxied_in_env({'HTTP_PROXY': self.proxy_url,

922

1185

'NO_PROXY': self.no_proxy_host})

923

1186

930

1193

'NO_PROXY': self.no_proxy_host})

931

1194

932

1195

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)

1196

if self._testing_pycurl():

1197

# pycurl *ignores* invalid proxy env variables. If that ever change

1198

# in the future, this test will fail indicating that pycurl do not

1199

# ignore anymore such variables.

1200

self.not_proxied_in_env({'http_proxy': self.proxy_address})

1201

else:

1202

self.assertRaises(errors.InvalidURL,

1203

self.proxied_in_env,

1204

{'http_proxy': self.proxy_address})

1205

1206

1207

class TestRanges(http_utils.TestCaseWithWebserver):

1208

"""Test the Range header in GET methods."""

1209

1210

def setUp(self):

1211

http_utils.TestCaseWithWebserver.setUp(self)

988

1212

self.build_tree_contents([('a', '0123456789')],)

989

1213

server = self.get_readonly_server()

990

1214

self.transport = self._transport(server.get_url())

991

1215

1216

def create_transport_readonly_server(self):

1217

return http_server.HttpServer(protocol_version=self._protocol_version)

1218

992

1219

def _file_contents(self, relpath, ranges):

993

1220

offsets = [ (start, end - start + 1) for start, end in ranges]

994

1221

coalesce = self.transport._coalesce_offsets

1009

1236

# Valid ranges

1010

1237

map(self.assertEqual,['0', '234'],

1011

1238

list(self._file_contents('a', [(0,0), (2,4)])),)

1012

# Tail

1239

1240

def test_range_header_tail(self):

1013

1241

self.assertEqual('789', self._file_tail('a', 3))

1014

# Syntactically invalid range

1242

1243

def test_syntactically_invalid_range_header(self):

1015

1244

self.assertListRaises(errors.InvalidHttpRange,

1016

1245

self._file_contents, 'a', [(4, 3)])

1017

# Semantically invalid range

1246

1247

def test_semantically_invalid_range_header(self):

1018

1248

self.assertListRaises(errors.InvalidHttpRange,

1019

1249

self._file_contents, 'a', [(42, 128)])

1020

1250

1021

1251

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

"""

1252

class TestHTTPRedirections(http_utils.TestCaseWithRedirectedWebserver):

1253

"""Test redirection between http servers."""

1044

1254

1045

1255

def create_transport_secondary_server(self):

1046

1256

"""Create the secondary server redirecting to the primary server"""

1047

1257

new = self.get_readonly_server()

1048

1258

1049

redirecting = HTTPServerRedirecting()

1259

redirecting = http_utils.HTTPServerRedirecting(

1260

protocol_version=self._protocol_version)

1050

1261

redirecting.redirect_to(new.host, new.port)

1051

1262

return redirecting

1052

1263

1072

1283

self.assertEqual([], bundle.revisions)

1073

1284

1074

1285

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__

1286

class RedirectedRequest(_urllib2_wrappers.Request):

1287

"""Request following redirections. """

1288

1289

init_orig = _urllib2_wrappers.Request.__init__

1093

1290

1094

1291

def __init__(self, method, url, *args, **kwargs):

1292

"""Constructor.

1293

1294

"""

1295

# Since the tests using this class will replace

1296

# _urllib2_wrappers.Request, we can't just call the base class __init__

1297

# or we'll loop.

1095

1298

RedirectedRequest.init_orig(self, method, url, args, kwargs)

1096

1299

self.follow_redirections = True

1097

1300

1098

1301

1099

class TestHTTPSilentRedirections_urllib(TestCaseWithRedirectedWebserver):

1100

"""Test redirections provided by urllib.

1302

class TestHTTPSilentRedirections(http_utils.TestCaseWithRedirectedWebserver):

1303

"""Test redirections.

1101

1304

1102

1305

http implementations do not redirect silently anymore (they

1103

1306

do not redirect at all in fact). The mechanism is still in

1110

1313

-- vila 20070212

1111

1314

"""

1112

1315

1113

_transport = HttpTransport_urllib

1114

1115

1316

def setUp(self):

1116

super(TestHTTPSilentRedirections_urllib, self).setUp()

1317

if pycurl_present and self._transport == PyCurlTransport:

1318

raise tests.TestNotApplicable(

1319

"pycurl doesn't redirect silently annymore")

1320

super(TestHTTPSilentRedirections, self).setUp()

1117

1321

self.setup_redirected_request()

1118

1322

self.addCleanup(self.cleanup_redirected_request)

1119

1323

self.build_tree_contents([('a','a'),

1140

1344

1141

1345

def create_transport_secondary_server(self):

1142

1346

"""Create the secondary server, redirections are defined in the tests"""

1143

return HTTPServerRedirecting()

1347

return http_utils.HTTPServerRedirecting(

1348

protocol_version=self._protocol_version)

1144

1349

1145

1350

def test_one_redirection(self):

1146

1351

t = self.old_transport

1162

1367

self.old_server.port)

1163

1368

new_prefix = 'http://%s:%s' % (self.new_server.host,

1164

1369

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

]

1370

self.old_server.redirections = [

1371

('/1(.*)', r'%s/2\1' % (old_prefix), 302),

1372

('/2(.*)', r'%s/3\1' % (old_prefix), 303),

1373

('/3(.*)', r'%s/4\1' % (old_prefix), 307),

1374

('/4(.*)', r'%s/5\1' % (new_prefix), 301),

1375

('(/[^/]+)', r'%s/1\1' % (old_prefix), 301),

1376

]

1172

1377

self.assertEquals('redirected 5 times',t._perform(req).read())

1173

1378

1174

1379

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

1380

class TestDoCatchRedirections(http_utils.TestCaseWithRedirectedWebserver):

1381

"""Test transport.do_catching_redirections."""

1182

1382

1183

1383

def setUp(self):

1184

1384

super(TestDoCatchRedirections, self).setUp()

1194

1394

1195

1395

# We use None for redirected so that we fail if redirected

1196

1396

self.assertEquals('0123456789',

1197

do_catching_redirections(self.get_a, t, None).read())

1397

transport.do_catching_redirections(

1398

self.get_a, t, None).read())

1198

1399

1199

1400

def test_one_redirection(self):

1200

1401

self.redirections = 0

1205

1406

return self._transport(dir)

1206

1407

1207

1408

self.assertEquals('0123456789',

1208

do_catching_redirections(self.get_a,

1209

self.old_transport,

1210

redirected

1211

).read())

1409

transport.do_catching_redirections(

1410

self.get_a, self.old_transport, redirected).read())

1212

1411

self.assertEquals(1, self.redirections)

1213

1412

1214

1413

def test_redirection_loop(self):

1219

1418

# a/a/a

1220

1419

return self.old_transport.clone(exception.target)

1221

1420

1222

self.assertRaises(errors.TooManyRedirections, do_catching_redirections,

1421

self.assertRaises(errors.TooManyRedirections,

1422

transport.do_catching_redirections,

1223

1423

self.get_a, self.old_transport, redirected)

1224

1424

1225

1425

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

1426

class TestAuth(http_utils.TestCaseWithWebserver):

1427

"""Test authentication scheme"""

1428

1429

_auth_header = 'Authorization'

1233

1430

_password_prompt_prefix = ''

1234

1431

1235

1432

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

"""

1433

super(TestAuth, self).setUp()

1434

self.server = self.get_readonly_server()

1244

1435

self.build_tree_contents([('a', 'contents of a\n'),

1245

1436

('b', 'contents of b\n'),])

1246

1437

1438

def create_transport_readonly_server(self):

1439

if self._auth_scheme == 'basic':

1440

server = http_utils.HTTPBasicAuthServer(

1441

protocol_version=self._protocol_version)

1442

else:

1443

if self._auth_scheme != 'digest':

1444

raise AssertionError('Unknown auth scheme: %r'

1445

% self._auth_scheme)

1446

server = http_utils.HTTPDigestAuthServer(

1447

protocol_version=self._protocol_version)

1448

return server

1449

1450

def _testing_pycurl(self):

1451

return pycurl_present and self._transport == PyCurlTransport

1452

1247

1453

def get_user_url(self, user=None, password=None):

1248

1454

"""Build an url embedding user and password"""

1249

1455

url = '%s://' % self.server._url_protocol

1255

1461

url += '%s:%s/' % (self.server.host, self.server.port)

1256

1462

return url

1257

1463

1464

def get_user_transport(self, user=None, password=None):

1465

return self._transport(self.get_user_url(user, password))

1466

1258

1467

def test_no_user(self):

1259

1468

self.server.add_user('joe', 'foo')

1260

1469

t = self.get_user_transport()

1294

1503

self.assertEqual(2, self.server.auth_required_errors)

1295

1504

1296

1505

def test_prompt_for_password(self):

1506

if self._testing_pycurl():

1507

raise tests.TestNotApplicable(

1508

'pycurl cannot prompt, it handles auth by embedding'

1509

' user:pass in urls only')

1510

1297

1511

self.server.add_user('joe', 'foo')

1298

1512

t = self.get_user_transport('joe', None)

1299

1513

stdout = tests.StringIOWrapper()

1321

1535

self.assertEquals(expected_prompt, actual_prompt)

1322

1536

1323

1537

def test_no_prompt_for_password_when_using_auth_config(self):

1324

user ='joe'

1538

if self._testing_pycurl():

1539

raise tests.TestNotApplicable(

1540

'pycurl does not support authentication.conf'

1541

' since it cannot prompt')

1542

1543

user =' joe'

1325

1544

password = 'foo'

1326

1545

stdin_content = 'bar\n' # Not the right password

1327

1546

self.server.add_user(user, password)

1341

1560

# Only one 'Authentication Required' error should occur

1342

1561

self.assertEqual(1, self.server.auth_required_errors)

1343

1562

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))

1563

def test_changing_nonce(self):

1564

if self._auth_scheme != 'digest':

1565

raise tests.TestNotApplicable('HTTP auth digest only test')

1566

if self._testing_pycurl():

1567

raise tests.KnownFailure(

1568

'pycurl does not handle a nonce change')

1569

self.server.add_user('joe', 'foo')

1570

t = self.get_user_transport('joe', 'foo')

1571

self.assertEqual('contents of a\n', t.get('a').read())

1572

self.assertEqual('contents of b\n', t.get('b').read())

1573

# Only one 'Authentication Required' error should have

1574

# occured so far

1575

self.assertEqual(1, self.server.auth_required_errors)

1576

# The server invalidates the current nonce

1577

self.server.auth_nonce = self.server.auth_nonce + '. No, now!'

1578

self.assertEqual('contents of a\n', t.get('a').read())

1579

# Two 'Authentication Required' errors should occur (the

1580

# initial 'who are you' and a second 'who are you' with the new nonce)

1581

self.assertEqual(2, self.server.auth_required_errors)

1582

1361

1583

1362

1584

1363

1585

class TestProxyAuth(TestAuth):

1364

"""Test proxy authentication schemes.

1586

"""Test proxy authentication schemes."""

1365

1587

1366

Daughter classes MUST also inherit from TestCaseWithWebserver.

1367

"""

1368

1588

_auth_header = 'Proxy-authorization'

1369

_password_prompt_prefix = 'Proxy '

1370

1589

_password_prompt_prefix='Proxy '

1371

1590

1372

1591

def setUp(self):

1373

TestCaseWithWebserver.setUp(self)

1374

self.server = self.get_readonly_server()

1592

super(TestProxyAuth, self).setUp()

1375

1593

self._old_env = {}

1376

1594

self.addCleanup(self._restore_env)

1377

TestAuth.setUp(self)

1378

1595

# Override the contents to avoid false positives

1379

1596

self.build_tree_contents([('a', 'not proxied contents of a\n'),

1380

1597

('b', 'not proxied contents of b\n'),

1382

1599

('b-proxied', 'contents of b\n'),

1383

1600

])

1384

1601

1602

def create_transport_readonly_server(self):

1603

if self._auth_scheme == 'basic':

1604

server = http_utils.ProxyBasicAuthServer(

1605

protocol_version=self._protocol_version)

1606

else:

1607

if self._auth_scheme != 'digest':

1608

raise AssertionError('Unknown auth scheme: %r'

1609

% self._auth_scheme)

1610

server = http_utils.ProxyDigestAuthServer(

1611

protocol_version=self._protocol_version)

1612

return server

1613

1385

1614

def get_user_transport(self, user=None, password=None):

1386

1615

self._install_env({'all_proxy': self.get_user_url(user, password)})

1387

1616

return self._transport(self.server.get_url())

1394

1623

for name, value in self._old_env.iteritems():

1395

1624

osutils.set_or_unset_env(name, value)

1396

1625

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')

1626

def test_empty_pass(self):

1627

if self._testing_pycurl():

1628

import pycurl

1629

if pycurl.version_info()[1] < '7.16.0':

1630

raise tests.KnownFailure(

1631

'pycurl < 7.16.0 does not handle empty proxy passwords')

1632

super(TestProxyAuth, self).test_empty_pass()

1633

1634

1635

class SampleSocket(object):

1636

"""A socket-like object for use in testing the HTTP request handler."""

1637

1638

def __init__(self, socket_read_content):

1639

"""Constructs a sample socket.

1640

1641

:param socket_read_content: a byte sequence

1642

"""

1643

# Use plain python StringIO so we can monkey-patch the close method to

1644

# not discard the contents.

1645

from StringIO import StringIO

1646

self.readfile = StringIO(socket_read_content)

1647

self.writefile = StringIO()

1648

self.writefile.close = lambda: None

1649

1650

def makefile(self, mode='r', bufsize=None):

1651

if 'r' in mode:

1652

return self.readfile

1653

else:

1654

return self.writefile

1655

1656

1657

class SmartHTTPTunnellingTest(tests.TestCaseWithTransport):

1658

1659

def setUp(self):

1660

super(SmartHTTPTunnellingTest, self).setUp()

1661

# We use the VFS layer as part of HTTP tunnelling tests.

1662

self._captureVar('BZR_NO_SMART_VFS', None)

1663

self.transport_readonly_server = http_utils.HTTPServerWithSmarts

1664

1665

def create_transport_readonly_server(self):

1666

return http_utils.HTTPServerWithSmarts(

1667

protocol_version=self._protocol_version)

1668

1669

def test_bulk_data(self):

1670

# We should be able to send and receive bulk data in a single message.

1671

# The 'readv' command in the smart protocol both sends and receives

1672

# bulk data, so we use that.

1673

self.build_tree(['data-file'])

1674

http_server = self.get_readonly_server()

1675

http_transport = self._transport(http_server.get_url())

1676

medium = http_transport.get_smart_medium()

1677

# Since we provide the medium, the url below will be mostly ignored

1678

# during the test, as long as the path is '/'.

1679

remote_transport = remote.RemoteTransport('bzr://fake_host/',

1680

medium=medium)

1681

self.assertEqual(

1682

[(0, "c")], list(remote_transport.readv("data-file", [(0,1)])))

1683

1684

def test_http_send_smart_request(self):

1685

1686

post_body = 'hello\n'

1687

expected_reply_body = 'ok\x012\n'

1688

1689

http_server = self.get_readonly_server()

1690

http_transport = self._transport(http_server.get_url())

1691

medium = http_transport.get_smart_medium()

1692

response = medium.send_http_smart_request(post_body)

1693

reply_body = response.read()

1694

self.assertEqual(expected_reply_body, reply_body)

1695

1696

def test_smart_http_server_post_request_handler(self):

1697

httpd = self.get_readonly_server()._get_httpd()

1698

1699

socket = SampleSocket(

1700

'POST /.bzr/smart %s \r\n' % self._protocol_version

1701

# HTTP/1.1 posts must have a Content-Length (but it doesn't hurt

1702

# for 1.0)

1703

+ 'Content-Length: 6\r\n'

1704

'\r\n'

1705

'hello\n')

1706

# Beware: the ('localhost', 80) below is the

1707

# client_address parameter, but we don't have one because

1708

# we have defined a socket which is not bound to an

1709

# address. The test framework never uses this client

1710

# address, so far...

1711

request_handler = http_utils.SmartRequestHandler(socket,

1712

('localhost', 80),

1713

httpd)

1714

response = socket.writefile.getvalue()

1715

self.assertStartsWith(response, '%s 200 ' % self._protocol_version)

1716

# This includes the end of the HTTP headers, and all the body.

1717

expected_end_of_response = '\r\n\r\nok\x012\n'

1718

self.assertEndsWith(response, expected_end_of_response)

1719

1502

1720