~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_http.py

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

• reverse the comparison (6616.1.1 to 6015.59.1)
• stop comparing with revision 6015.59.1

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

        ]

131

 

    if tests.HTTPSServerFeature.available():

132

 

        activity_scenarios.append(

133

 

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

134

 

                                _transport=_urllib.HttpTransport_urllib,)),)

135

130

    if features.pycurl.available():

136

131

        activity_scenarios.append(

137

132

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

138

133

                                _transport=PyCurlTransport,)),)

139

 

        if tests.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.

 

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

 

 

149

        activity_scenarios.append(

 

150

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

 

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

282

288

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

283

289

 

284

290

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

285

 

        self.assertEquals(ranges,

 

291

        self.assertEqual(ranges,

286

292

                          self.req_handler._parse_ranges(header, file_size))

287

293

 

288

294

    def test_simple_range(self):

378

384

    _transport = property(_get_pycurl_maybe)

379

385

 

380

386

 

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

387

class TestHttpTransportUrls(tests.TestCase):

400

388

    """Test the http urls."""

401

389

 

481

469

        )

482

470

 

483

471

    def setUp(self):

484

 

        http_utils.TestCaseWithWebserver.setUp(self)

 

472

        super(TestHTTPConnections, self).setUp()

485

473

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

486

474

                        transport=self.get_transport())

487

475

 

533

521

    scenarios = vary_by_http_client_implementation()

534

522

 

535

523

    def test_http_registered(self):

536

 

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

 

524

        t = transport.get_transport_from_url(

 

525

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

537

526

        self.assertIsInstance(t, transport.Transport)

538

527

        self.assertIsInstance(t, self._transport)

539

528

 

551

540

        self.start_server(server)

552

541

        url = server.get_url()

553

542

        # FIXME: needs a cleanup -- vila 20100611

554

 

        http_transport = transport.get_transport(url)

 

543

        http_transport = transport.get_transport_from_url(url)

555

544

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

556

545

        self.assertTrue(

557

546

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

667

656

 

668

657

    _req_handler_class = BadStatusRequestHandler

669

658

 

 

659

    def setUp(self):

 

660

        super(TestBadStatusServer, self).setUp()

 

661

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

 

662

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

 

663

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

 

664

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

 

665

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

 

666

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

 

667

            raise tests.TestSkipped(

 

668

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

 

669

 

670

670

    def test_http_has(self):

671

671

        t = self.get_readonly_transport()

672

 

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

 

672

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

 

673

                           errors.InvalidHttpResponse),

 

674

                          t.has, 'foo/bar')

673

675

 

674

676

    def test_http_get(self):

675

677

        t = self.get_readonly_transport()

676

 

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

 

678

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

 

679

                           errors.InvalidHttpResponse),

 

680

                          t.get, 'foo/bar')

677

681

 

678

682

 

679

683

class InvalidStatusRequestHandler(http_server.TestingHTTPRequestHandler):

1156

1160

                                      protocol_version=self._protocol_version)

1157

1161

 

1158

1162

    def setUp(self):

1159

 

        http_utils.TestCaseWithWebserver.setUp(self)

 

1163

        super(TestLimitedRangeRequestServer, self).setUp()

1160

1164

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

1161

1165

        # response, so we build a content appropriately.

1162

1166

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

1196

1200

 

1197

1201

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

1198

1202

        handler = _urllib2_wrappers.ProxyHandler()

1199

 

        self.assertEquals(expected,

 

1203

        self.assertEqual(expected,

1200

1204

                          handler.evaluate_proxy_bypass(host, no_proxy))

1201

1205

 

1202

1206

    def test_empty_user(self):

1348

1352

        )

1349

1353

 

1350

1354

    def setUp(self):

1351

 

        http_utils.TestCaseWithWebserver.setUp(self)

 

1355

        super(TestRanges, self).setUp()

1352

1356

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

1353

1357

 

1354

1358

    def create_transport_readonly_server(self):

1595

1599

            path='/',

1596

1600

            realm='Realm',

1597

1601

            ))

1598

 

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

 

1602

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

1599

1603

 

1600

1604

 

1601

1605

class TestAuth(http_utils.TestCaseWithWebserver):

1635

1639

        return url

1636

1640

 

1637

1641

    def get_user_transport(self, user, password):

1638

 

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

 

1642

        t = transport.get_transport_from_url(

 

1643

            self.get_user_url(user, password))

1639

1644

        return t

1640

1645

 

1641

1646

    def test_no_user(self):

1767

1772

                                     http_utils.ProxyDigestAuthServer):

1768

1773

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

1769

1774

        if self._testing_pycurl():

1770

 

            raise tests.KnownFailure(

 

1775

            self.knownFailure(

1771

1776

                'pycurl does not handle a nonce change')

1772

1777

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

1773

1778

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

1855

1860

        if self._testing_pycurl():

1856

1861

            import pycurl

1857

1862

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

1858

 

                raise tests.KnownFailure(

 

1863

                self.knownFailure(

1859

1864

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

1860

1865

        super(TestProxyAuth, self).test_empty_pass()

1861

1866

 

1903

1908

        server._url_protocol = self._url_protocol

1904

1909

        return server

1905

1910

 

1906

 

    def test_open_bzrdir(self):

 

1911

    def test_open_controldir(self):

1907

1912

        branch = self.make_branch('relpath')

1908

1913

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

1909

 

        bd = bzrdir.BzrDir.open(url)

 

1914

        bd = controldir.ControlDir.open(url)

1910

1915

        self.addCleanup(bd.transport.disconnect)

1911

1916

        self.assertIsInstance(bd, _mod_remote.RemoteBzrDir)

1912

1917

 

1915

1920

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

1916

1921

        # bulk data, so we use that.

1917

1922

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

1918

 

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

 

1923

        http_transport = transport.get_transport_from_url(

 

1924

            self.http_server.get_url())

1919

1925

        medium = http_transport.get_smart_medium()

1920

1926

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

1921

1927

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

1929

1935

        post_body = 'hello\n'

1930

1936

        expected_reply_body = 'ok\x012\n'

1931

1937

 

1932

 

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

 

1938

        http_transport = transport.get_transport_from_url(

 

1939

            self.http_server.get_url())

1933

1940

        medium = http_transport.get_smart_medium()

1934

1941

        response = medium.send_http_smart_request(post_body)

1935

1942

        reply_body = response.read()

1993

2000

        self.assertIsInstance(r, type(t))

1994

2001

        # Both transports share the some connection

1995

2002

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

 

2003

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

1996

2004

 

1997

2005

    def test_redirected_to_self_with_slash(self):

1998

2006

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

2009

2017

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

2010

2018

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

2011

2019

        self.assertIsInstance(r, type(t))

 

2020

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

 

2021

            r.external_url())

2012

2022

 

2013

2023

    def test_redirected_to_same_host_sibling_protocol(self):

2014

2024

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

2015

2025

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

2016

2026

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

2017

2027

        self.assertIsInstance(r, type(t))

 

2028

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

 

2029

            r.external_url())

2018

2030

 

2019

2031

    def test_redirected_to_same_host_different_protocol(self):

2020

2032

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

2021

2033

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

2022

2034

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

2023

 

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

 

2035

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

 

2036

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

 

2037

 

 

2038

    def test_redirected_to_same_host_specific_implementation(self):

 

2039

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

 

2040

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

 

2041

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

 

2042

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

2024

2043

 

2025

2044

    def test_redirected_to_different_host_same_user(self):

2026

2045

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

2027

2046

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

2028

2047

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

2029

2048

        self.assertIsInstance(r, type(t))

2030

 

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

 

2049

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

 

2050

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

2031

2051

 

2032

2052

 

2033

2053

class PredefinedRequestHandler(http_server.TestingHTTPRequestHandler):

2086

2106

    pass

2087

2107

 

2088

2108

 

2089

 

if tests.HTTPSServerFeature.available():

 

2109

if features.HTTPSServerFeature.available():

2090

2110

    from bzrlib.tests import https_server

2091

2111

    class ActivityHTTPSServer(ActivityServerMixin, https_server.HTTPSServer):

2092

2112

        pass

2100

2120

    """

2101

2121

 

2102

2122

    def setUp(self):

2103

 

        tests.TestCase.setUp(self)

2104

2123

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

2105

2124

        self.server.start_server()

 

2125

        self.addCleanup(self.server.stop_server)

2106

2126

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

2107

2127

        def report_activity(t, bytes, direction):

2108

2128

            count = _activities.get(direction, 0)

2109

2129

            count += bytes

2110

2130

            _activities[direction] = count

2111

2131

        self.activities = _activities

2112

 

 

2113

2132

        # We override at class level because constructors may propagate the

2114

2133

        # bound method and render instance overriding ineffective (an

2115

2134

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

2116

2135

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

2117

 

        self.addCleanup(self.server.stop_server)

2118

2136

 

2119

2137

    def get_transport(self):

2120

2138

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

2244

2262

        )

2245

2263

 

2246

2264

    def setUp(self):

 

2265

        super(TestActivity, self).setUp()

2247

2266

        TestActivityMixin.setUp(self)

2248

2267

 

2249

2268

 

2258

2277

    _protocol_version = 'HTTP/1.1'

2259

2278

 

2260

2279

    def setUp(self):

 

2280

        super(TestNoReportActivity, self).setUp()

2261

2281

        self._transport =_urllib.HttpTransport_urllib

2262

2282

        TestActivityMixin.setUp(self)

2263

2283

 

• Older »

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