~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_http.py

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

• reverse the comparison (6622 to 6079)
• stop comparing with revision 6079

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

 

533

531

    scenarios = vary_by_http_client_implementation()

534

532

 

535

533

    def test_http_registered(self):

536

 

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

 

534

        t = transport.get_transport_from_url(

 

535

            '%s://foo.com/' % self._url_protocol)

537

536

        self.assertIsInstance(t, transport.Transport)

538

537

        self.assertIsInstance(t, self._transport)

539

538

 

551

550

        self.start_server(server)

552

551

        url = server.get_url()

553

552

        # FIXME: needs a cleanup -- vila 20100611

554

 

        http_transport = transport.get_transport(url)

 

553

        http_transport = transport.get_transport_from_url(url)

555

554

        code, response = http_transport._post('abc def end-of-body')

556

555

        self.assertTrue(

557

556

            server.received_bytes.startswith('POST /.bzr/smart HTTP/1.'))

667

666

 

668

667

    _req_handler_class = BadStatusRequestHandler

669

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

 

670

680

    def test_http_has(self):

671

681

        t = self.get_readonly_transport()

672

 

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

 

682

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

 

683

                           errors.InvalidHttpResponse),

 

684

                          t.has, 'foo/bar')

673

685

 

674

686

    def test_http_get(self):

675

687

        t = self.get_readonly_transport()

676

 

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

 

688

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

 

689

                           errors.InvalidHttpResponse),

 

690

                          t.get, 'foo/bar')

677

691

 

678

692

 

679

693

class InvalidStatusRequestHandler(http_server.TestingHTTPRequestHandler):

1156

1170

                                      protocol_version=self._protocol_version)

1157

1171

 

1158

1172

    def setUp(self):

1159

 

        http_utils.TestCaseWithWebserver.setUp(self)

 

1173

        super(TestLimitedRangeRequestServer, self).setUp()

1160

1174

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

1161

1175

        # response, so we build a content appropriately.

1162

1176

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

1196

1210

 

1197

1211

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

1198

1212

        handler = _urllib2_wrappers.ProxyHandler()

1199

 

        self.assertEquals(expected,

 

1213

        self.assertEqual(expected,

1200

1214

                          handler.evaluate_proxy_bypass(host, no_proxy))

1201

1215

 

1202

1216

    def test_empty_user(self):

1269

1283

            self.no_proxy_host = self.server_host_port

1270

1284

        # The secondary server is the proxy

1271

1285

        self.proxy_url = self.get_secondary_url()

 

1286

        if self._testing_pycurl():

 

1287

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

1272

1288

 

1273

1289

    def _testing_pycurl(self):

1274

1290

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

1348

1364

        )

1349

1365

 

1350

1366

    def setUp(self):

1351

 

        http_utils.TestCaseWithWebserver.setUp(self)

 

1367

        super(TestRanges, self).setUp()

1352

1368

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

1353

1369

 

1354

1370

    def create_transport_readonly_server(self):

1595

1611

            path='/',

1596

1612

            realm='Realm',

1597

1613

            ))

1598

 

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

 

1614

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

1599

1615

 

1600

1616

 

1601

1617

class TestAuth(http_utils.TestCaseWithWebserver):

1635

1651

        return url

1636

1652

 

1637

1653

    def get_user_transport(self, user, password):

1638

 

        t = transport.get_transport(self.get_user_url(user, password))

 

1654

        t = transport.get_transport_from_url(

 

1655

            self.get_user_url(user, password))

1639

1656

        return t

1640

1657

 

1641

1658

    def test_no_user(self):

1848

1865

                                  ])

1849

1866

 

1850

1867

    def get_user_transport(self, user, password):

1851

 

        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)

1852

1872

        return TestAuth.get_user_transport(self, user, password)

1853

1873

 

1854

1874

    def test_empty_pass(self):

1903

1923

        server._url_protocol = self._url_protocol

1904

1924

        return server

1905

1925

 

1906

 

    def test_open_bzrdir(self):

 

1926

    def test_open_controldir(self):

1907

1927

        branch = self.make_branch('relpath')

1908

1928

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

1909

 

        bd = bzrdir.BzrDir.open(url)

 

1929

        bd = controldir.ControlDir.open(url)

1910

1930

        self.addCleanup(bd.transport.disconnect)

1911

1931

        self.assertIsInstance(bd, _mod_remote.RemoteBzrDir)

1912

1932

 

1915

1935

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

1916

1936

        # bulk data, so we use that.

1917

1937

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

1918

 

        http_transport = transport.get_transport(self.http_server.get_url())

 

1938

        http_transport = transport.get_transport_from_url(

 

1939

            self.http_server.get_url())

1919

1940

        medium = http_transport.get_smart_medium()

1920

1941

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

1921

1942

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

1929

1950

        post_body = 'hello\n'

1930

1951

        expected_reply_body = 'ok\x012\n'

1931

1952

 

1932

 

        http_transport = transport.get_transport(self.http_server.get_url())

 

1953

        http_transport = transport.get_transport_from_url(

 

1954

            self.http_server.get_url())

1933

1955

        medium = http_transport.get_smart_medium()

1934

1956

        response = medium.send_http_smart_request(post_body)

1935

1957

        reply_body = response.read()

1993

2015

        self.assertIsInstance(r, type(t))

1994

2016

        # Both transports share the some connection

1995

2017

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

 

2018

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

1996

2019

 

1997

2020

    def test_redirected_to_self_with_slash(self):

1998

2021

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

2009

2032

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

2010

2033

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

2011

2034

        self.assertIsInstance(r, type(t))

 

2035

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

 

2036

            r.external_url())

2012

2037

 

2013

2038

    def test_redirected_to_same_host_sibling_protocol(self):

2014

2039

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

2015

2040

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

2016

2041

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

2017

2042

        self.assertIsInstance(r, type(t))

 

2043

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

 

2044

            r.external_url())

2018

2045

 

2019

2046

    def test_redirected_to_same_host_different_protocol(self):

2020

2047

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

2021

2048

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

2022

2049

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

2023

 

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

 

2050

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

 

2051

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

 

2052

 

 

2053

    def test_redirected_to_same_host_specific_implementation(self):

 

2054

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

 

2055

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

 

2056

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

 

2057

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

2024

2058

 

2025

2059

    def test_redirected_to_different_host_same_user(self):

2026

2060

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

2028

2062

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

2029

2063

        self.assertIsInstance(r, type(t))

2030

2064

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

 

2065

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

2031

2066

 

2032

2067

 

2033

2068

class PredefinedRequestHandler(http_server.TestingHTTPRequestHandler):

2100

2135

    """

2101

2136

 

2102

2137

    def setUp(self):

2103

 

        tests.TestCase.setUp(self)

2104

2138

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

2105

2139

        self.server.start_server()

 

2140

        self.addCleanup(self.server.stop_server)

2106

2141

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

2107

2142

        def report_activity(t, bytes, direction):

2108

2143

            count = _activities.get(direction, 0)

2109

2144

            count += bytes

2110

2145

            _activities[direction] = count

2111

2146

        self.activities = _activities

2112

 

 

2113

2147

        # We override at class level because constructors may propagate the

2114

2148

        # bound method and render instance overriding ineffective (an

2115

2149

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

2116

2150

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

2117

 

        self.addCleanup(self.server.stop_server)

2118

2151

 

2119

2152

    def get_transport(self):

2120

2153

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

2244

2277

        )

2245

2278

 

2246

2279

    def setUp(self):

 

2280

        super(TestActivity, self).setUp()

2247

2281

        TestActivityMixin.setUp(self)

2248

2282

 

2249

2283

 

2258

2292

    _protocol_version = 'HTTP/1.1'

2259

2293

 

2260

2294

    def setUp(self):

 

2295

        super(TestNoReportActivity, self).setUp()

2261

2296

        self._transport =_urllib.HttpTransport_urllib

2262

2297

        TestActivityMixin.setUp(self)

2263

2298

 

• Older »

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