/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_http.py

• browse files at revision 4301.3.2
• compare with another revision
• compare with revision 1663
• download diff
• download tarball
• view history from revision 4301.3.2

Show diffs side-by-side

added

removed

bzrlib/tests/test_http.py

113

113

                                            protocol_scenarios)

114

114

    tests.multiply_tests(tp_tests, tp_scenarios, result)

115

115

 

 

116

    # proxy auth: each auth scheme on all http versions on all implementations.

 

117

    tppa_tests, remaining_tests = tests.split_suite_by_condition(

 

118

        remaining_tests, tests.condition_isinstance((

 

119

                TestProxyAuth,

 

120

                )))

 

121

    proxy_auth_scheme_scenarios = [

 

122

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

 

123

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

 

124

        ('basicdigest',

 

125

         dict(_auth_server=http_utils.ProxyBasicAndDigestAuthServer)),

 

126

        ]

 

127

    tppa_scenarios = tests.multiply_scenarios(tp_scenarios,

 

128

                                              proxy_auth_scheme_scenarios)

 

129

    tests.multiply_tests(tppa_tests, tppa_scenarios, result)

 

130

 

116

131

    # auth: each auth scheme on all http versions on all implementations.

117

132

    tpa_tests, remaining_tests = tests.split_suite_by_condition(

118

133

        remaining_tests, tests.condition_isinstance((

119

134

                TestAuth,

120

135

                )))

121

136

    auth_scheme_scenarios = [

122

 

        ('basic', dict(_auth_scheme='basic')),

123

 

        ('digest', dict(_auth_scheme='digest')),

 

137

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

 

138

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

 

139

        ('basicdigest',

 

140

         dict(_auth_server=http_utils.HTTPBasicAndDigestAuthServer)),

124

141

        ]

125

142

    tpa_scenarios = tests.multiply_scenarios(tp_scenarios,

126

 

        auth_scheme_scenarios)

 

143

                                             auth_scheme_scenarios)

127

144

    tests.multiply_tests(tpa_tests, tpa_scenarios, result)

128

145

 

129

 

    # activity: activity on all http versions on all implementations

 

146

    # activity: on all http[s] versions on all implementations

130

147

    tpact_tests, remaining_tests = tests.split_suite_by_condition(

131

148

        remaining_tests, tests.condition_isinstance((

132

149

                TestActivity,

133

150

                )))

134

151

    activity_scenarios = [

135

 

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

 

152

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

 

153

                             _transport=_urllib.HttpTransport_urllib,)),

136

154

        ]

137

155

    if tests.HTTPSServerFeature.available():

138

156

        activity_scenarios.append(

139

 

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

140

 

    tpact_scenarios = tests.multiply_scenarios(tp_scenarios,

141

 

        activity_scenarios)

 

157

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

 

158

                                  _transport=_urllib.HttpTransport_urllib,)),)

 

159

    if pycurl_present:

 

160

        activity_scenarios.append(

 

161

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

 

162

                                 _transport=PyCurlTransport,)),)

 

163

        if tests.HTTPSServerFeature.available():

 

164

            from bzrlib.tests import (

 

165

                ssl_certs,

 

166

                )

 

167

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

 

168

            # certificates (like allowing them in a test specific

 

169

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

 

170

            # transport for tests.

 

171

            class HTTPS_pycurl_transport(PyCurlTransport):

 

172

 

 

173

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

 

174

                    super(HTTPS_pycurl_transport, self).__init__(

 

175

                        base, _from_transport)

 

176

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

 

177

 

 

178

            activity_scenarios.append(

 

179

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

 

180

                                      _transport=HTTPS_pycurl_transport,)),)

 

181

 

 

182

    tpact_scenarios = tests.multiply_scenarios(activity_scenarios,

 

183

                                               protocol_scenarios)

142

184

    tests.multiply_tests(tpact_tests, tpact_scenarios, result)

143

185

 

144

186

    # No parametrization for the remaining tests

1451

1493

    _auth_header = 'Authorization'

1452

1494

    _password_prompt_prefix = ''

1453

1495

    _username_prompt_prefix = ''

 

1496

    # Set by load_tests

 

1497

    _auth_server = None

1454

1498

 

1455

1499

    def setUp(self):

1456

1500

        super(TestAuth, self).setUp()

1459

1503

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

1460

1504

 

1461

1505

    def create_transport_readonly_server(self):

1462

 

        if self._auth_scheme == 'basic':

1463

 

            server = http_utils.HTTPBasicAuthServer(

1464

 

                protocol_version=self._protocol_version)

1465

 

        else:

1466

 

            if self._auth_scheme != 'digest':

1467

 

                raise AssertionError('Unknown auth scheme: %r'

1468

 

                                     % self._auth_scheme)

1469

 

            server = http_utils.HTTPDigestAuthServer(

1470

 

                protocol_version=self._protocol_version)

1471

 

        return server

 

1506

        return self._auth_server(protocol_version=self._protocol_version)

1472

1507

 

1473

1508

    def _testing_pycurl(self):

1474

1509

        return pycurl_present and self._transport == PyCurlTransport

1534

1569

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

1535

1570

        t = self.get_user_transport(None, None)

1536

1571

        stdout = tests.StringIOWrapper()

1537

 

        ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n', stdout=stdout)

 

1572

        stderr = tests.StringIOWrapper()

 

1573

        ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',

 

1574

                                            stdout=stdout, stderr=stderr)

1538

1575

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

1539

1576

        # stdin should be empty

1540

1577

        self.assertEqual('', ui.ui_factory.stdin.readline())

1541

 

        stdout.seek(0)

 

1578

        stderr.seek(0)

1542

1579

        expected_prompt = self._expected_username_prompt(t._unqualified_scheme)

1543

 

        self.assertEquals(expected_prompt, stdout.read(len(expected_prompt)))

 

1580

        self.assertEquals(expected_prompt, stderr.read(len(expected_prompt)))

 

1581

        self.assertEquals('', stdout.getvalue())

1544

1582

        self._check_password_prompt(t._unqualified_scheme, 'joe',

1545

 

                                    stdout.readline())

 

1583

                                    stderr.readline())

1546

1584

 

1547

1585

    def test_prompt_for_password(self):

1548

1586

        if self._testing_pycurl():

1553

1591

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

1554

1592

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

1555

1593

        stdout = tests.StringIOWrapper()

1556

 

        ui.ui_factory = tests.TestUIFactory(stdin='foo\n', stdout=stdout)

1557

 

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

 

1594

        stderr = tests.StringIOWrapper()

 

1595

        ui.ui_factory = tests.TestUIFactory(stdin='foo\n',

 

1596

                                            stdout=stdout, stderr=stderr)

 

1597

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

1558

1598

        # stdin should be empty

1559

1599

        self.assertEqual('', ui.ui_factory.stdin.readline())

1560

1600

        self._check_password_prompt(t._unqualified_scheme, 'joe',

1561

 

                                    stdout.getvalue())

 

1601

                                    stderr.getvalue())

 

1602

        self.assertEquals('', stdout.getvalue())

1562

1603

        # And we shouldn't prompt again for a different request

1563

1604

        # against the same transport.

1564

1605

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

1628

1669

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

1629

1670

 

1630

1671

    def test_changing_nonce(self):

1631

 

        if self._auth_scheme != 'digest':

1632

 

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

 

1672

        if self._auth_server not in (http_utils.HTTPDigestAuthServer,

 

1673

                                     http_utils.ProxyDigestAuthServer):

 

1674

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

1633

1675

        if self._testing_pycurl():

1634

1676

            raise tests.KnownFailure(

1635

1677

                'pycurl does not handle a nonce change')

1667

1709

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

1668

1710

                                  ])

1669

1711

 

1670

 

    def create_transport_readonly_server(self):

1671

 

        if self._auth_scheme == 'basic':

1672

 

            server = http_utils.ProxyBasicAuthServer(

1673

 

                protocol_version=self._protocol_version)

1674

 

        else:

1675

 

            if self._auth_scheme != 'digest':

1676

 

                raise AssertionError('Unknown auth scheme: %r'

1677

 

                                     % self._auth_scheme)

1678

 

            server = http_utils.ProxyDigestAuthServer(

1679

 

                protocol_version=self._protocol_version)

1680

 

        return server

1681

 

 

1682

1712

    def get_user_transport(self, user, password):

1683

1713

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

1684

1714

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

• Older »

Loggerhead is a web-based interface for Breezy
Version: 2.0.2.dev0