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

50

http_utils,

51

test_server,

52

)

53

from bzrlib.tests.variations import (

54

TestVariation,

55

load_tests_from_their_variations,

56

multiply_tests_by_variations,

57

multiply_tests_by_their_variations,

53

from bzrlib.tests.scenarios import (

54

load_tests_from_scenarios,

55

multiply_scenarios,

58

56

)

59

57

from bzrlib.transport import (

60

58

http,

70

68

from bzrlib.transport.http._pycurl import PyCurlTransport

71

69

72

70

73

load_tests = load_tests_from_their_variations

74

75

76

class VaryByHttpClientImplementation(TestVariation):

71

load_tests = load_tests_from_scenarios

72

73

74

def vary_by_http_client_implementation():

77

75

"""Test the two libraries we can use, pycurl and urllib."""

78

79

def scenarios(self):

80

transport_scenarios = [

81

('urllib', dict(_transport=_urllib.HttpTransport_urllib,

82

_server=http_server.HttpServer_urllib,

83

_url_protocol='http+urllib',)),

84

]

85

if features.pycurl.available():

86

transport_scenarios.append(

87

('pycurl', dict(_transport=PyCurlTransport,

88

_server=http_server.HttpServer_PyCurl,

89

_url_protocol='http+pycurl',)))

90

return transport_scenarios

91

92

93

class VaryByHttpProtocolVersion(TestVariation):

76

transport_scenarios = [

77

('urllib', dict(_transport=_urllib.HttpTransport_urllib,

78

_server=http_server.HttpServer_urllib,

79

_url_protocol='http+urllib',)),

80

]

81

if features.pycurl.available():

82

transport_scenarios.append(

83

('pycurl', dict(_transport=PyCurlTransport,

84

_server=http_server.HttpServer_PyCurl,

85

_url_protocol='http+pycurl',)))

86

return transport_scenarios

87

88

89

def vary_by_http_protocol_version():

94

90

"""Test on http/1.0 and 1.1"""

95

96

def scenarios(self):

97

return [

98

('HTTP/1.0', dict(_protocol_version='HTTP/1.0')),

99

('HTTP/1.1', dict(_protocol_version='HTTP/1.1')),

100

]

101

102

103

class VaryByHttpProxyAuthScheme(TestVariation):

104

105

def scenarios(self):

106

return [

107

('basic', dict(_auth_server=http_utils.ProxyBasicAuthServer)),

108

('digest', dict(_auth_server=http_utils.ProxyDigestAuthServer)),

109

('basicdigest',

110

dict(_auth_server=http_utils.ProxyBasicAndDigestAuthServer)),

111

]

112

113

114

class VaryByHttpAuthScheme(TestVariation):

115

116

def scenarios(self):

117

return [

118

('basic', dict(_auth_server=http_utils.HTTPBasicAuthServer)),

119

('digest', dict(_auth_server=http_utils.HTTPDigestAuthServer)),

120

('basicdigest',

121

dict(_auth_server=http_utils.HTTPBasicAndDigestAuthServer)),

122

]

123

124

125

class VaryByHttpActivity(TestVariation):

126

127

def scenarios(self):

128

activity_scenarios = [

129

('urllib,http', dict(_activity_server=ActivityHTTPServer,

130

_transport=_urllib.HttpTransport_urllib,)),

131

]

91

return [

92

('HTTP/1.0', dict(_protocol_version='HTTP/1.0')),

93

('HTTP/1.1', dict(_protocol_version='HTTP/1.1')),

94

]

95

96

97

def vary_by_http_proxy_auth_scheme():

98

return [

99

('basic', dict(_auth_server=http_utils.ProxyBasicAuthServer)),

100

('digest', dict(_auth_server=http_utils.ProxyDigestAuthServer)),

101

('basicdigest',

102

dict(_auth_server=http_utils.ProxyBasicAndDigestAuthServer)),

103

]

104

105

106

def vary_by_http_auth_scheme():

107

return [

108

('basic', dict(_auth_server=http_utils.HTTPBasicAuthServer)),

109

('digest', dict(_auth_server=http_utils.HTTPDigestAuthServer)),

110

('basicdigest',

111

dict(_auth_server=http_utils.HTTPBasicAndDigestAuthServer)),

112

]

113

114

115

def vary_by_http_activity():

116

activity_scenarios = [

117

('urllib,http', dict(_activity_server=ActivityHTTPServer,

118

_transport=_urllib.HttpTransport_urllib,)),

119

]

120

if tests.HTTPSServerFeature.available():

121

activity_scenarios.append(

122

('urllib,https', dict(_activity_server=ActivityHTTPSServer,

123

_transport=_urllib.HttpTransport_urllib,)),)

124

if features.pycurl.available():

125

activity_scenarios.append(

126

('pycurl,http', dict(_activity_server=ActivityHTTPServer,

127

_transport=PyCurlTransport,)),)

132

128

if tests.HTTPSServerFeature.available():

133

activity_scenarios.append(

134

('urllib,https', dict(_activity_server=ActivityHTTPSServer,

135

_transport=_urllib.HttpTransport_urllib,)),)

136

if features.pycurl.available():

137

activity_scenarios.append(

138

('pycurl,http', dict(_activity_server=ActivityHTTPServer,

139

_transport=PyCurlTransport,)),)

140

if tests.HTTPSServerFeature.available():

141

from bzrlib.tests import (

142

ssl_certs,

143

)

144

# FIXME: Until we have a better way to handle self-signed

145

# certificates (like allowing them in a test specific

146

# authentication.conf for example), we need some specialized pycurl

147

# transport for tests.

148

class HTTPS_pycurl_transport(PyCurlTransport):

149

150

def __init__(self, base, _from_transport=None):

151

super(HTTPS_pycurl_transport, self).__init__(

152

base, _from_transport)

153

self.cabundle = str(ssl_certs.build_path('ca.crt'))

154

155

activity_scenarios.append(

156

('pycurl,https', dict(_activity_server=ActivityHTTPSServer,

157

_transport=HTTPS_pycurl_transport,)),)

158

return activity_scenarios

129

from bzrlib.tests import (

130

ssl_certs,

131

)

132

# FIXME: Until we have a better way to handle self-signed

133

# certificates (like allowing them in a test specific

134

# authentication.conf for example), we need some specialized pycurl

135

# transport for tests.

136

class HTTPS_pycurl_transport(PyCurlTransport):

137

138

def __init__(self, base, _from_transport=None):

139

super(HTTPS_pycurl_transport, self).__init__(

140

base, _from_transport)

141

self.cabundle = str(ssl_certs.build_path('ca.crt'))

142

143

activity_scenarios.append(

144

('pycurl,https', dict(_activity_server=ActivityHTTPSServer,

145

_transport=HTTPS_pycurl_transport,)),)

146

return activity_scenarios

159

147

160

148

161

149

class FakeManager(object):

366

354

class TestHttpTransportUrls(tests.TestCase):

367

355

"""Test the http urls."""

368

356

369

variations = [VaryByHttpClientImplementation()]

357

scenarios = vary_by_http_client_implementation()

370

358

371

359

def test_abs_url(self):

372

360

"""Construction of absolute http URLs"""

442

430

class TestHTTPConnections(http_utils.TestCaseWithWebserver):

443

431

"""Test the http connections."""

444

432

445

variations = [

446

VaryByHttpClientImplementation(),

447

VaryByHttpProtocolVersion(),

448

]

433

scenarios = multiply_scenarios(

434

vary_by_http_client_implementation(),

435

vary_by_http_protocol_version(),

436

)

449

437

450

438

def setUp(self):

451

439

http_utils.TestCaseWithWebserver.setUp(self)

497

485

class TestHttpTransportRegistration(tests.TestCase):

498

486

"""Test registrations of various http implementations"""

499

487

500

variations = [VaryByHttpClientImplementation()]

488

scenarios = vary_by_http_client_implementation()

501

489

502

490

def test_http_registered(self):

503

491

t = transport.get_transport('%s://foo.com/' % self._url_protocol)

507

495

508

496

class TestPost(tests.TestCase):

509

497

510

variations = [

511

VaryByHttpClientImplementation(),

512

VaryByHttpProtocolVersion(),

513

]

498

scenarios = multiply_scenarios(

499

vary_by_http_client_implementation(),

500

vary_by_http_protocol_version(),

501

)

514

502

515

503

def test_post_body_is_received(self):

516

504

server = RecordingServer(expect_body_tail='end-of-body',

564

552

Daughter classes are expected to override _req_handler_class

565

553

"""

566

554

567

variations = [

568

VaryByHttpClientImplementation(),

569

VaryByHttpProtocolVersion(),

570

]

555

scenarios = multiply_scenarios(

556

vary_by_http_client_implementation(),

557

vary_by_http_protocol_version(),

558

)

571

559

572

560

# Provide a useful default

573

561

_req_handler_class = http_server.TestingHTTPRequestHandler

1039

1027

class TestLimitedRangeRequestServer(http_utils.TestCaseWithWebserver):

1040

1028

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

1041

1029

1042

variations = [

1043

VaryByHttpClientImplementation(),

1044

VaryByHttpProtocolVersion(),

1045

]

1030

scenarios = multiply_scenarios(

1031

vary_by_http_client_implementation(),

1032

vary_by_http_protocol_version(),

1033

)

1046

1034

1047

1035

# Requests with more range specifiers will error out

1048

1036

range_limit = 3

1123

1111

to the file names).

1124

1112

"""

1125

1113

1126

variations = [

1127

VaryByHttpClientImplementation(),

1128

VaryByHttpProtocolVersion(),

1129

]

1114

scenarios = multiply_scenarios(

1115

vary_by_http_client_implementation(),

1116

vary_by_http_protocol_version(),

1117

)

1130

1118

1131

1119

# FIXME: We don't have an https server available, so we don't

1132

1120

# test https connections. --vila toolongago

1230

1218

class TestRanges(http_utils.TestCaseWithWebserver):

1231

1219

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

1232

1220

1233

variations = [

1234

VaryByHttpClientImplementation(),

1235

VaryByHttpProtocolVersion(),

1236

]

1221

scenarios = multiply_scenarios(

1222

vary_by_http_client_implementation(),

1223

vary_by_http_protocol_version(),

1224

)

1237

1225

1238

1226

def setUp(self):

1239

1227

http_utils.TestCaseWithWebserver.setUp(self)

1280

1268

class TestHTTPRedirections(http_utils.TestCaseWithRedirectedWebserver):

1281

1269

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

1282

1270

1283

variations = [

1284

VaryByHttpClientImplementation(),

1285

VaryByHttpProtocolVersion(),

1286

]

1271

scenarios = multiply_scenarios(

1272

vary_by_http_client_implementation(),

1273

vary_by_http_protocol_version(),

1274

)

1287

1275

1288

1276

def setUp(self):

1289

1277

super(TestHTTPRedirections, self).setUp()

1353

1341

-- vila 20070212

1354

1342

"""

1355

1343

1356

variations = [

1357

VaryByHttpClientImplementation(),

1358

VaryByHttpProtocolVersion(),

1359

]

1344

scenarios = multiply_scenarios(

1345

vary_by_http_client_implementation(),

1346

vary_by_http_protocol_version(),

1347

)

1360

1348

1361

1349

def setUp(self):

1362

1350

if (features.pycurl.available()

1408

1396

class TestDoCatchRedirections(http_utils.TestCaseWithRedirectedWebserver):

1409

1397

"""Test transport.do_catching_redirections."""

1410

1398

1411

variations = [

1412

VaryByHttpClientImplementation(),

1413

VaryByHttpProtocolVersion(),

1414

]

1399

scenarios = multiply_scenarios(

1400

vary_by_http_client_implementation(),

1401

vary_by_http_protocol_version(),

1402

)

1415

1403

1416

1404

def setUp(self):

1417

1405

super(TestDoCatchRedirections, self).setUp()

1460

1448

class TestAuth(http_utils.TestCaseWithWebserver):

1461

1449

"""Test authentication scheme"""

1462

1450

1463

variations = [

1464

VaryByHttpClientImplementation(),

1465

VaryByHttpProtocolVersion(),

1466

VaryByHttpAuthScheme(),

1467

]

1451

scenarios = multiply_scenarios(

1452

vary_by_http_client_implementation(),

1453

vary_by_http_protocol_version(),

1454

vary_by_http_auth_scheme(),

1455

)

1468

1456

1469

1457

_auth_header = 'Authorization'

1470

1458

_password_prompt_prefix = ''

1675

1663

class TestProxyAuth(TestAuth):

1676

1664

"""Test proxy authentication schemes."""

1677

1665

1678

variations = [

1679

VaryByHttpClientImplementation(),

1680

VaryByHttpProtocolVersion(),

1681

VaryByHttpProxyAuthScheme(),

1682

]

1666

scenarios = multiply_scenarios(

1667

vary_by_http_client_implementation(),

1668

vary_by_http_protocol_version(),

1669

vary_by_http_proxy_auth_scheme(),

1670

)

1683

1671

1684

1672

_auth_header = 'Proxy-authorization'

1685

1673

_password_prompt_prefix = 'Proxy '

1742

1730

1743

1731

class SmartHTTPTunnellingTest(tests.TestCaseWithTransport):

1744

1732

1745

variations = [

1746

VaryByHttpClientImplementation(),

1747

VaryByHttpProtocolVersion(),

1748

]

1733

scenarios = multiply_scenarios(

1734

vary_by_http_client_implementation(),

1735

vary_by_http_protocol_version(),

1736

)

1749

1737

1750

1738

def setUp(self):

1751

1739

super(SmartHTTPTunnellingTest, self).setUp()

1841

1829

1842

1830

class Test_redirected_to(tests.TestCase):

1843

1831

1844

variations = [VaryByHttpClientImplementation()]

1832

scenarios = vary_by_http_client_implementation()

1845

1833

1846

1834

def test_redirected_to_subdir(self):

1847

1835

t = self._transport('http://www.example.com/foo')

2094

2082

2095

2083

class TestActivity(tests.TestCase, TestActivityMixin):

2096

2084

2097

variations = [

2098

VaryByHttpActivity(),

2099

VaryByHttpProtocolVersion(),

2100

]

2085

scenarios = multiply_scenarios(

2086

vary_by_http_activity(),

2087

vary_by_http_protocol_version(),

2088

)

2101

2089

2102

2090

def setUp(self):

2103

2091

TestActivityMixin.setUp(self)

2125

2113

class TestAuthOnRedirected(http_utils.TestCaseWithRedirectedWebserver):

2126

2114

"""Test authentication on the redirected http server."""

2127

2115

2128

variations = [VaryByHttpProtocolVersion()]

2116

scenarios = vary_by_http_protocol_version()

2129

2117

2130

2118

_auth_header = 'Authorization'

2131

2119

_password_prompt_prefix = ''