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

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.

750

"""

751

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

class LimitedRangeRequestHandler(http_server.TestingHTTPRequestHandler):

908

"""Errors out when range specifiers exceed the limit"""

909

910

def get_multiple_ranges(self, file, file_size, ranges):

911

"""Refuses the multiple ranges request"""

912

tcs = self.server.test_case_server

913

if tcs.range_limit is not None and len(ranges) > tcs.range_limit:

914

file.close()

915

# Emulate apache behavior

916

self.send_error(400, "Bad Request")

917

return

918

return http_server.TestingHTTPRequestHandler.get_multiple_ranges(

919

self, file, file_size, ranges)

920

921

922

class LimitedRangeHTTPServer(http_server.HttpServer):

923

"""An HttpServer erroring out on requests with too much range specifiers"""

924

925

def __init__(self, request_handler=LimitedRangeRequestHandler,

926

protocol_version=None,

927

range_limit=None):

928

http_server.HttpServer.__init__(self, request_handler,

929

protocol_version=protocol_version)

930

self.range_limit = range_limit

931

932

933

class TestLimitedRangeRequestServer(http_utils.TestCaseWithWebserver):

934

"""Tests readv requests against a server erroring out on too much ranges."""

935

936

# Requests with more range specifiers will error out

752

937

range_limit = 3

753

938

754

939

def create_transport_readonly_server(self):

755

# Requests with more range specifiers will error out

756

return LimitedRangeHTTPServer(range_limit=self.range_limit)

940

return LimitedRangeHTTPServer(range_limit=self.range_limit,

941

protocol_version=self._protocol_version)

757

942

758

943

def get_transport(self):

759

944

return self._transport(self.get_readonly_server().get_url())

760

945

761

946

def setUp(self):

762

TestCaseWithWebserver.setUp(self)

947

http_utils.TestCaseWithWebserver.setUp(self)

763

948

# We need to manipulate ranges that correspond to real chunks in the

764

949

# response, so we build a content appropriately.

765

950

filler = ''.join(['abcdefghij' for x in range(102)])

785

970

self.assertEqual(2, self.get_readonly_server().GET_request_nb)

786

971

787

972

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

973

class TestHttpProxyWhiteBox(tests.TestCase):

803

974

"""Whitebox test proxy http authorization.

804

975

821

992

osutils.set_or_unset_env(name, value)

822

993

823

994

def _proxied_request(self):

824

handler = ProxyHandler()

825

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

995

handler = _urllib2_wrappers.ProxyHandler()

996

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

826

997

handler.set_proxy(request, 'http')

827

998

return request

828

999

837

1008

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

838

1009

839

1010

840

class TestProxyHttpServer(object):

1011

class TestProxyHttpServer(http_utils.TestCaseWithTwoWebservers):

841

1012

"""Tests proxy server.

842

1013

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

1014

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

851

1015

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

852

1016

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

857

1021

# test https connections.

858

1022

859

1023

def setUp(self):

860

TestCaseWithTwoWebservers.setUp(self)

1024

super(TestProxyHttpServer, self).setUp()

861

1025

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

862

1026

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

863

1027

# Let's setup some attributes for tests

864

1028

self.server = self.get_readonly_server()

865

1029

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

866

self.no_proxy_host = self.proxy_address

1030

if self._testing_pycurl():

1031

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

1032

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

1033

self.no_proxy_host = 'localhost'

1034

else:

1035

self.no_proxy_host = self.proxy_address

867

1036

# The secondary server is the proxy

868

1037

self.proxy = self.get_secondary_server()

869

1038

self.proxy_url = self.proxy.get_url()

870

1039

self._old_env = {}

871

1040

1041

def _testing_pycurl(self):

1042

return pycurl_present and self._transport == PyCurlTransport

1043

872

1044

def create_transport_secondary_server(self):

873

1045

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

874

1046

'-proxied' appended to their names.

875

1047

"""

876

return ProxyServer()

1048

return http_utils.ProxyServer(protocol_version=self._protocol_version)

877

1049

878

1050

def _install_env(self, env):

879

1051

for name, value in env.iteritems():

905

1077

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

906

1078

907

1079

def test_HTTP_PROXY(self):

1080

if self._testing_pycurl():

1081

# pycurl does not check HTTP_PROXY for security reasons

1082

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

1083

# about. Should we ?)

1084

raise tests.TestNotApplicable(

1085

'pycurl does not check HTTP_PROXY for security reasons')

908

1086

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

909

1087

910

1088

def test_all_proxy(self):

918

1096

'no_proxy': self.no_proxy_host})

919

1097

920

1098

def test_HTTP_PROXY_with_NO_PROXY(self):

1099

if self._testing_pycurl():

1100

raise tests.TestNotApplicable(

1101

'pycurl does not check HTTP_PROXY for security reasons')

921

1102

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

922

1103

'NO_PROXY': self.no_proxy_host})

923

1104

930

1111

'NO_PROXY': self.no_proxy_host})

931

1112

932

1113

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)

1114

if self._testing_pycurl():

1115

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

1116

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

1117

# ignore anymore such variables.

1118

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

1119

else:

1120

self.assertRaises(errors.InvalidURL,

1121

self.proxied_in_env,

1122

{'http_proxy': self.proxy_address})

1123

1124

1125

class TestRanges(http_utils.TestCaseWithWebserver):

1126

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

1127

1128

def setUp(self):

1129

http_utils.TestCaseWithWebserver.setUp(self)

988

1130

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

989

1131

server = self.get_readonly_server()

990

1132

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

991

1133

1134

def create_transport_readonly_server(self):

1135

return http_server.HttpServer(protocol_version=self._protocol_version)

1136

992

1137

def _file_contents(self, relpath, ranges):

993

1138

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

994

1139

coalesce = self.transport._coalesce_offsets

1009

1154

# Valid ranges

1010

1155

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

1011

1156

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

1012

# Tail

1157

1158

def test_range_header_tail(self):

1013

1159

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

1014

# Syntactically invalid range

1160

1161

def test_syntactically_invalid_range_header(self):

1015

1162

self.assertListRaises(errors.InvalidHttpRange,

1016

1163

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

1017

# Semantically invalid range

1164

1165

def test_semantically_invalid_range_header(self):

1018

1166

self.assertListRaises(errors.InvalidHttpRange,

1019

1167

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

1020

1168

1021

1169

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

"""

1170

class TestHTTPRedirections(http_utils.TestCaseWithRedirectedWebserver):

1171

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

1044

1172

1045

1173

def create_transport_secondary_server(self):

1046

1174

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

1047

1175

new = self.get_readonly_server()

1048

1176

1049

redirecting = HTTPServerRedirecting()

1177

redirecting = http_utils.HTTPServerRedirecting(

1178

protocol_version=self._protocol_version)

1050

1179

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

1051

1180

return redirecting

1052

1181

1072

1201

self.assertEqual([], bundle.revisions)

1073

1202

1074

1203

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__

1204

class RedirectedRequest(_urllib2_wrappers.Request):

1205

"""Request following redirections. """

1206

1207

init_orig = _urllib2_wrappers.Request.__init__

1093

1208

1094

1209

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

1210

"""Constructor.

1211

1212

"""

1213

# Since the tests using this class will replace

1214

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

1215

# or we'll loop.

1095

1216

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

1096

1217

self.follow_redirections = True

1097

1218

1098

1219

1099

class TestHTTPSilentRedirections_urllib(TestCaseWithRedirectedWebserver):

1100

"""Test redirections provided by urllib.

1220

class TestHTTPSilentRedirections(http_utils.TestCaseWithRedirectedWebserver):

1221

"""Test redirections.

1101

1222

1102

1223

http implementations do not redirect silently anymore (they

1103

1224

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

1110

1231

-- vila 20070212

1111

1232

"""

1112

1233

1113

_transport = HttpTransport_urllib

1114

1115

1234

def setUp(self):

1116

super(TestHTTPSilentRedirections_urllib, self).setUp()

1235

if pycurl_present and self._transport == PyCurlTransport:

1236

raise tests.TestNotApplicable(

1237

"pycurl doesn't redirect silently annymore")

1238

super(TestHTTPSilentRedirections, self).setUp()

1117

1239

self.setup_redirected_request()

1118

1240

self.addCleanup(self.cleanup_redirected_request)

1119

1241

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

1140

1262

1141

1263

def create_transport_secondary_server(self):

1142

1264

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

1143

return HTTPServerRedirecting()

1265

return http_utils.HTTPServerRedirecting(

1266

protocol_version=self._protocol_version)

1144

1267

1145

1268

def test_one_redirection(self):

1146

1269

t = self.old_transport

1162

1285

self.old_server.port)

1163

1286

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

1164

1287

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

]

1288

self.old_server.redirections = [

1289

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

1290

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

1291

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

1292

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

1293

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

1294

]

1172

1295

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

1173

1296

1174

1297

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

1298

class TestDoCatchRedirections(http_utils.TestCaseWithRedirectedWebserver):

1299

"""Test transport.do_catching_redirections."""

1182

1300

1183

1301

def setUp(self):

1184

1302

super(TestDoCatchRedirections, self).setUp()

1194

1312

1195

1313

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

1196

1314

self.assertEquals('0123456789',

1197

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

1315

transport.do_catching_redirections(

1316

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

1198

1317

1199

1318

def test_one_redirection(self):

1200

1319

self.redirections = 0

1205

1324

return self._transport(dir)

1206

1325

1207

1326

self.assertEquals('0123456789',

1208

do_catching_redirections(self.get_a,

1209

self.old_transport,

1210

redirected

1211

).read())

1327

transport.do_catching_redirections(

1328

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

1212

1329

self.assertEquals(1, self.redirections)

1213

1330

1214

1331

def test_redirection_loop(self):

1219

1336

# a/a/a

1220

1337

return self.old_transport.clone(exception.target)

1221

1338

1222

self.assertRaises(errors.TooManyRedirections, do_catching_redirections,

1339

self.assertRaises(errors.TooManyRedirections,

1340

transport.do_catching_redirections,

1223

1341

self.get_a, self.old_transport, redirected)

1224

1342

1225

1343

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

1344

class TestAuth(http_utils.TestCaseWithWebserver):

1345

"""Test authentication scheme"""

1346

1347

_auth_header = 'Authorization'

1233

1348

_password_prompt_prefix = ''

1234

1349

1235

1350

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

"""

1351

super(TestAuth, self).setUp()

1352

self.server = self.get_readonly_server()

1244

1353

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

1245

1354

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

1246

1355

1356

def create_transport_readonly_server(self):

1357

if self._auth_scheme == 'basic':

1358

server = http_utils.HTTPBasicAuthServer(

1359

protocol_version=self._protocol_version)

1360

else:

1361

if self._auth_scheme != 'digest':

1362

raise AssertionError('Unknown auth scheme: %r'

1363

% self._auth_scheme)

1364

server = http_utils.HTTPDigestAuthServer(

1365

protocol_version=self._protocol_version)

1366

return server

1367

1368

def _testing_pycurl(self):

1369

return pycurl_present and self._transport == PyCurlTransport

1370

1247

1371

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

1248

1372

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

1249

1373

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

1255

1379

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

1256

1380

return url

1257

1381

1382

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

1383

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

1384

1258

1385

def test_no_user(self):

1259

1386

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

1260

1387

t = self.get_user_transport()

1294

1421

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

1295

1422

1296

1423

def test_prompt_for_password(self):

1424

if self._testing_pycurl():

1425

raise tests.TestNotApplicable(

1426

'pycurl cannot prompt, it handles auth by embedding'

1427

' user:pass in urls only')

1428

1297

1429

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

1298

1430

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

1299

1431

stdout = tests.StringIOWrapper()

1321

1453

self.assertEquals(expected_prompt, actual_prompt)

1322

1454

1323

1455

def test_no_prompt_for_password_when_using_auth_config(self):

1324

user ='joe'

1456

if self._testing_pycurl():

1457

raise tests.TestNotApplicable(

1458

'pycurl does not support authentication.conf'

1459

' since it cannot prompt')

1460

1461

user =' joe'

1325

1462

password = 'foo'

1326

1463

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

1327

1464

self.server.add_user(user, password)

1341

1478

# Only one 'Authentication Required' error should occur

1342

1479

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

1343

1480

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

1481

def test_changing_nonce(self):

1482

if self._auth_scheme != 'digest':

1483

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

1484

if self._testing_pycurl():

1485

raise tests.KnownFailure(

1486

'pycurl does not handle a nonce change')

1487

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

1488

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

1489

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

1490

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

1491

# Only one 'Authentication Required' error should have

1492

# occured so far

1493

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

1494

# The server invalidates the current nonce

1495

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

1496

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

1497

# Two 'Authentication Required' errors should occur (the

1498

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

1499

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

1500

1361

1501

1362

1502

1363

1503

class TestProxyAuth(TestAuth):

1364

"""Test proxy authentication schemes.

1504

"""Test proxy authentication schemes."""

1365

1505

1366

Daughter classes MUST also inherit from TestCaseWithWebserver.

1367

"""

1368

1506

_auth_header = 'Proxy-authorization'

1369

_password_prompt_prefix = 'Proxy '

1370

1507

_password_prompt_prefix='Proxy '

1371

1508

1372

1509

def setUp(self):

1373

TestCaseWithWebserver.setUp(self)

1374

self.server = self.get_readonly_server()

1510

super(TestProxyAuth, self).setUp()

1375

1511

self._old_env = {}

1376

1512

self.addCleanup(self._restore_env)

1377

TestAuth.setUp(self)

1378

1513

# Override the contents to avoid false positives

1379

1514

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

1380

1515

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

1382

1517

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

1383

1518

])

1384

1519

1520

def create_transport_readonly_server(self):

1521

if self._auth_scheme == 'basic':

1522

server = http_utils.ProxyBasicAuthServer(

1523

protocol_version=self._protocol_version)

1524

else:

1525

if self._auth_scheme != 'digest':

1526

raise AssertionError('Unknown auth scheme: %r'

1527

% self._auth_scheme)

1528

server = http_utils.ProxyDigestAuthServer(

1529

protocol_version=self._protocol_version)

1530

return server

1531

1385

1532

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

1386

1533

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

1387

1534

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

1394

1541

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

1395

1542

osutils.set_or_unset_env(name, value)

1396

1543

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

1544

def test_empty_pass(self):

1545

if self._testing_pycurl():

1546

import pycurl

1547

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

1548

raise tests.KnownFailure(

1549

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

1550

super(TestProxyAuth, self).test_empty_pass()

1551

1552

1553

class SampleSocket(object):

1554

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

1555

1556

def __init__(self, socket_read_content):

1557

"""Constructs a sample socket.

1558

1559

:param socket_read_content: a byte sequence

1560

"""

1561

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

1562

# not discard the contents.

1563

from StringIO import StringIO

1564

self.readfile = StringIO(socket_read_content)

1565

self.writefile = StringIO()

1566

self.writefile.close = lambda: None

1567

1568

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

1569

if 'r' in mode:

1570

return self.readfile

1571

else:

1572

return self.writefile

1573

1574

1575

class SmartHTTPTunnellingTest(tests.TestCaseWithTransport):

1576

1577

def setUp(self):

1578

super(SmartHTTPTunnellingTest, self).setUp()

1579

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

1580

self._captureVar('BZR_NO_SMART_VFS', None)

1581

self.transport_readonly_server = http_utils.HTTPServerWithSmarts

1582

1583

def create_transport_readonly_server(self):

1584

return http_utils.HTTPServerWithSmarts(

1585

protocol_version=self._protocol_version)

1586

1587

def test_bulk_data(self):

1588

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

1589

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

1590

# bulk data, so we use that.

1591

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

1592

http_server = self.get_readonly_server()

1593

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

1594

medium = http_transport.get_smart_medium()

1595

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

1596

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

1597

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

1598

medium=medium)

1599

self.assertEqual(

1600

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

1601

1602

def test_http_send_smart_request(self):

1603

1604

post_body = 'hello\n'

1605

expected_reply_body = 'ok\x012\n'

1606

1607

http_server = self.get_readonly_server()

1608

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

1609

medium = http_transport.get_smart_medium()

1610

response = medium.send_http_smart_request(post_body)

1611

reply_body = response.read()

1612

self.assertEqual(expected_reply_body, reply_body)

1613

1614

def test_smart_http_server_post_request_handler(self):

1615

httpd = self.get_readonly_server()._get_httpd()

1616

1617

socket = SampleSocket(

1618

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

1619

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

1620

# for 1.0)

1621

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

1622

'\r\n'

1623

'hello\n')

1624

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

1625

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

1626

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

1627

# address. The test framework never uses this client

1628

# address, so far...

1629

request_handler = http_utils.SmartRequestHandler(socket,

1630

('localhost', 80),

1631

httpd)

1632

response = socket.writefile.getvalue()

1633

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

1634

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

1635

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

1636

self.assertEndsWith(response, expected_end_of_response)

1637

1502

1638