~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_http.py

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

• reverse the comparison (6621 to 6170)
• stop comparing with revision 6170

Show diffs side-by-side

added

removed

bzrlib/tests/test_http.py

1

 

# Copyright (C) 2005-2011 Canonical Ltd

 

1

# Copyright (C) 2005-2012, 2015, 2016, 2017 Canonical Ltd

2

2

#

3

3

# This program is free software; you can redistribute it and/or modify

4

4

# it under the terms of the GNU General Public License as published by

31

31

 

32

32

import bzrlib

33

33

from bzrlib import (

34

 

    bzrdir,

35

 

    cethread,

36

34

    config,

 

35

    controldir,

37

36

    debug,

38

37

    errors,

39

38

    osutils,

128

127

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

129

128

                            _transport=_urllib.HttpTransport_urllib,)),

130

129

        ]

 

130

    if features.pycurl.available():

 

131

        activity_scenarios.append(

 

132

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

 

133

                                _transport=PyCurlTransport,)),)

131

134

    if features.HTTPSServerFeature.available():

 

135

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

 

136

        # (like allowing them in a test specific authentication.conf for

 

137

        # example), we need some specialized pycurl/urllib transport for tests.

 

138

        # -- vila 2012-01-20

 

139

        from bzrlib.tests import (

 

140

            ssl_certs,

 

141

            )

 

142

        class HTTPS_urllib_transport(_urllib.HttpTransport_urllib):

 

143

 

 

144

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

 

145

                super(HTTPS_urllib_transport, self).__init__(

 

146

                    base, _from_transport=_from_transport,

 

147

                    ca_certs=ssl_certs.build_path('ca.crt'))

 

148

 

132

149

        activity_scenarios.append(

133

150

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

134

 

                                _transport=_urllib.HttpTransport_urllib,)),)

135

 

    if features.pycurl.available():

136

 

        activity_scenarios.append(

137

 

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

138

 

                                _transport=PyCurlTransport,)),)

139

 

        if features.HTTPSServerFeature.available():

140

 

            from bzrlib.tests import (

141

 

                ssl_certs,

142

 

                )

143

 

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

144

 

            # certificates (like allowing them in a test specific

145

 

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

146

 

            # transport for tests.

 

151

                                  _transport=HTTPS_urllib_transport,)),)

 

152

        if features.pycurl.available():

147

153

            class HTTPS_pycurl_transport(PyCurlTransport):

148

154

 

149

155

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

254

260

        self.assertEqual('basic', scheme)

255

261

        self.assertEqual('realm="Thou should not pass"', remainder)

256

262

 

 

263

    def test_build_basic_header_with_long_creds(self):

 

264

        handler = _urllib2_wrappers.BasicAuthHandler()

 

265

        user = 'user' * 10  # length 40

 

266

        password = 'password' * 5  # length 40

 

267

        header = handler.build_auth_header(

 

268

            dict(user=user, password=password), None)

 

269

        # https://bugs.launchpad.net/bzr/+bug/1606203 was caused by incorrectly

 

270

        # creating a header value with an embedded '\n'

 

271

        self.assertFalse('\n' in header)

 

272

 

257

273

    def test_basic_extract_realm(self):

258

274

        scheme, remainder = self.parse_header(

259

275

            'Basic realm="Thou should not pass"',

282

298

        self.req_handler = RequestHandler(None, None, None)

283

299

 

284

300

    def assertRanges(self, ranges, header, file_size):

285

 

        self.assertEquals(ranges,

 

301

        self.assertEqual(ranges,

286

302

                          self.req_handler._parse_ranges(header, file_size))

287

303

 

288

304

    def test_simple_range(self):

378

394

    _transport = property(_get_pycurl_maybe)

379

395

 

380

396

 

381

 

class TestHttpUrls(tests.TestCase):

382

 

 

383

 

    # TODO: This should be moved to authorization tests once they

384

 

    # are written.

385

 

 

386

 

    def test_url_parsing(self):

387

 

        f = FakeManager()

388

 

        url = http.extract_auth('http://example.com', f)

389

 

        self.assertEqual('http://example.com', url)

390

 

        self.assertEqual(0, len(f.credentials))

391

 

        url = http.extract_auth(

392

 

            'http://user:pass@example.com/bzr/bzr.dev', f)

393

 

        self.assertEqual('http://example.com/bzr/bzr.dev', url)

394

 

        self.assertEqual(1, len(f.credentials))

395

 

        self.assertEqual([None, 'example.com', 'user', 'pass'],

396

 

                         f.credentials[0])

397

 

 

398

 

 

399

397

class TestHttpTransportUrls(tests.TestCase):

400

398

    """Test the http urls."""

401

399

 

481

479

        )

482

480

 

483

481

    def setUp(self):

484

 

        http_utils.TestCaseWithWebserver.setUp(self)

 

482

        super(TestHTTPConnections, self).setUp()

485

483

        self.build_tree(['foo/', 'foo/bar'], line_endings='binary',

486

484

                        transport=self.get_transport())

487

485

 

668

666

 

669

667

    _req_handler_class = BadStatusRequestHandler

670

668

 

 

669

    def setUp(self):

 

670

        super(TestBadStatusServer, self).setUp()

 

671

        # See https://bugs.launchpad.net/bzr/+bug/1451448 for details.

 

672

        # TD;LR: Running both a TCP client and server in the same process and

 

673

        # thread uncovers a race in python. The fix is to run the server in a

 

674

        # different process. Trying to fix yet another race here is not worth

 

675

        # the effort. -- vila 2015-09-06

 

676

        if 'HTTP/1.0' in self.id():

 

677

            raise tests.TestSkipped(

 

678

                'Client/Server in the same process and thread can hang')

 

679

 

671

680

    def test_http_has(self):

672

681

        t = self.get_readonly_transport()

673

 

        self.assertRaises(errors.InvalidHttpResponse, t.has, 'foo/bar')

 

682

        self.assertRaises((errors.ConnectionError, errors.ConnectionReset,

 

683

                           errors.InvalidHttpResponse),

 

684

                          t.has, 'foo/bar')

674

685

 

675

686

    def test_http_get(self):

676

687

        t = self.get_readonly_transport()

677

 

        self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')

 

688

        self.assertRaises((errors.ConnectionError, errors.ConnectionReset,

 

689

                           errors.InvalidHttpResponse),

 

690

                          t.get, 'foo/bar')

678

691

 

679

692

 

680

693

class InvalidStatusRequestHandler(http_server.TestingHTTPRequestHandler):

1157

1170

                                      protocol_version=self._protocol_version)

1158

1171

 

1159

1172

    def setUp(self):

1160

 

        http_utils.TestCaseWithWebserver.setUp(self)

 

1173

        super(TestLimitedRangeRequestServer, self).setUp()

1161

1174

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

1162

1175

        # response, so we build a content appropriately.

1163

1176

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

1197

1210

 

1198

1211

    def assertEvaluateProxyBypass(self, expected, host, no_proxy):

1199

1212

        handler = _urllib2_wrappers.ProxyHandler()

1200

 

        self.assertEquals(expected,

 

1213

        self.assertEqual(expected,

1201

1214

                          handler.evaluate_proxy_bypass(host, no_proxy))

1202

1215

 

1203

1216

    def test_empty_user(self):

1270

1283

            self.no_proxy_host = self.server_host_port

1271

1284

        # The secondary server is the proxy

1272

1285

        self.proxy_url = self.get_secondary_url()

 

1286

        if self._testing_pycurl():

 

1287

            self.proxy_url = self.proxy_url.replace('+pycurl', '')

1273

1288

 

1274

1289

    def _testing_pycurl(self):

1275

1290

        # TODO: This is duplicated for lots of the classes in this file

1349

1364

        )

1350

1365

 

1351

1366

    def setUp(self):

1352

 

        http_utils.TestCaseWithWebserver.setUp(self)

 

1367

        super(TestRanges, self).setUp()

1353

1368

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

1354

1369

 

1355

1370

    def create_transport_readonly_server(self):

1596

1611

            path='/',

1597

1612

            realm='Realm',

1598

1613

            ))

1599

 

        self.assertEquals((user, password), got_pass)

 

1614

        self.assertEqual((user, password), got_pass)

1600

1615

 

1601

1616

 

1602

1617

class TestAuth(http_utils.TestCaseWithWebserver):

1850

1865

                                  ])

1851

1866

 

1852

1867

    def get_user_transport(self, user, password):

1853

 

        self.overrideEnv('all_proxy', self.get_user_url(user, password))

 

1868

        proxy_url = self.get_user_url(user, password)

 

1869

        if self._testing_pycurl():

 

1870

            proxy_url = proxy_url.replace('+pycurl', '')

 

1871

        self.overrideEnv('all_proxy', proxy_url)

1854

1872

        return TestAuth.get_user_transport(self, user, password)

1855

1873

 

1856

1874

    def test_empty_pass(self):

1905

1923

        server._url_protocol = self._url_protocol

1906

1924

        return server

1907

1925

 

1908

 

    def test_open_bzrdir(self):

 

1926

    def test_open_controldir(self):

1909

1927

        branch = self.make_branch('relpath')

1910

1928

        url = self.http_server.get_url() + 'relpath'

1911

 

        bd = bzrdir.BzrDir.open(url)

 

1929

        bd = controldir.ControlDir.open(url)

1912

1930

        self.addCleanup(bd.transport.disconnect)

1913

1931

        self.assertIsInstance(bd, _mod_remote.RemoteBzrDir)

1914

1932

 

1997

2015

        self.assertIsInstance(r, type(t))

1998

2016

        # Both transports share the some connection

1999

2017

        self.assertEqual(t._get_connection(), r._get_connection())

2000

 

        self.assertEquals('http://www.example.com/foo/subdir/', r.base)

 

2018

        self.assertEqual('http://www.example.com/foo/subdir/', r.base)

2001

2019

 

2002

2020

    def test_redirected_to_self_with_slash(self):

2003

2021

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

2014

2032

        r = t._redirected_to('http://www.example.com/foo',

2015

2033

                             'http://foo.example.com/foo/subdir')

2016

2034

        self.assertIsInstance(r, type(t))

2017

 

        self.assertEquals('http://foo.example.com/foo/subdir/',

 

2035

        self.assertEqual('http://foo.example.com/foo/subdir/',

2018

2036

            r.external_url())

2019

2037

 

2020

2038

    def test_redirected_to_same_host_sibling_protocol(self):

2022

2040

        r = t._redirected_to('http://www.example.com/foo',

2023

2041

                             'https://www.example.com/foo')

2024

2042

        self.assertIsInstance(r, type(t))

2025

 

        self.assertEquals('https://www.example.com/foo/',

 

2043

        self.assertEqual('https://www.example.com/foo/',

2026

2044

            r.external_url())

2027

2045

 

2028

2046

    def test_redirected_to_same_host_different_protocol(self):

2029

2047

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

2030

2048

        r = t._redirected_to('http://www.example.com/foo',

2031

2049

                             'ftp://www.example.com/foo')

2032

 

        self.assertNotEquals(type(r), type(t))

2033

 

        self.assertEquals('ftp://www.example.com/foo/', r.external_url())

 

2050

        self.assertNotEqual(type(r), type(t))

 

2051

        self.assertEqual('ftp://www.example.com/foo/', r.external_url())

2034

2052

 

2035

2053

    def test_redirected_to_same_host_specific_implementation(self):

2036

2054

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

2037

2055

        r = t._redirected_to('http://www.example.com/foo',

2038

2056

                             'https+urllib://www.example.com/foo')

2039

 

        self.assertEquals('https://www.example.com/foo/', r.external_url())

 

2057

        self.assertEqual('https://www.example.com/foo/', r.external_url())

2040

2058

 

2041

2059

    def test_redirected_to_different_host_same_user(self):

2042

2060

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

2044

2062

                             'https://foo.example.com/foo')

2045

2063

        self.assertIsInstance(r, type(t))

2046

2064

        self.assertEqual(t._parsed_url.user, r._parsed_url.user)

2047

 

        self.assertEquals('https://joe@foo.example.com/foo/', r.external_url())

 

2065

        self.assertEqual('https://joe@foo.example.com/foo/', r.external_url())

2048

2066

 

2049

2067

 

2050

2068

class PredefinedRequestHandler(http_server.TestingHTTPRequestHandler):

2117

2135

    """

2118

2136

 

2119

2137

    def setUp(self):

2120

 

        tests.TestCase.setUp(self)

2121

2138

        self.server = self._activity_server(self._protocol_version)

2122

2139

        self.server.start_server()

 

2140

        self.addCleanup(self.server.stop_server)

2123

2141

        _activities = {} # Don't close over self and create a cycle

2124

2142

        def report_activity(t, bytes, direction):

2125

2143

            count = _activities.get(direction, 0)

2126

2144

            count += bytes

2127

2145

            _activities[direction] = count

2128

2146

        self.activities = _activities

2129

 

 

2130

2147

        # We override at class level because constructors may propagate the

2131

2148

        # bound method and render instance overriding ineffective (an

2132

2149

        # alternative would be to define a specific ui factory instead...)

2133

2150

        self.overrideAttr(self._transport, '_report_activity', report_activity)

2134

 

        self.addCleanup(self.server.stop_server)

2135

2151

 

2136

2152

    def get_transport(self):

2137

2153

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

2261

2277

        )

2262

2278

 

2263

2279

    def setUp(self):

 

2280

        super(TestActivity, self).setUp()

2264

2281

        TestActivityMixin.setUp(self)

2265

2282

 

2266

2283

 

2275

2292

    _protocol_version = 'HTTP/1.1'

2276

2293

 

2277

2294

    def setUp(self):

 

2295

        super(TestNoReportActivity, self).setUp()

2278

2296

        self._transport =_urllib.HttpTransport_urllib

2279

2297

        TestActivityMixin.setUp(self)

2280

2298

 

• Older »

Loggerhead is a web-based interface for Breezy
Version: 2.0.1