~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_http.py

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

• reverse the comparison (6615 to 6309)
• stop comparing with revision 6309

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

        ]

 

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

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

 

668

656

 

669

657

    _req_handler_class = BadStatusRequestHandler

670

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

 

671

670

    def test_http_has(self):

672

671

        t = self.get_readonly_transport()

673

 

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

 

672

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

 

673

                           errors.InvalidHttpResponse),

 

674

                          t.has, 'foo/bar')

674

675

 

675

676

    def test_http_get(self):

676

677

        t = self.get_readonly_transport()

677

 

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

 

678

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

 

679

                           errors.InvalidHttpResponse),

 

680

                          t.get, 'foo/bar')

678

681

 

679

682

 

680

683

class InvalidStatusRequestHandler(http_server.TestingHTTPRequestHandler):

1157

1160

                                      protocol_version=self._protocol_version)

1158

1161

 

1159

1162

    def setUp(self):

1160

 

        http_utils.TestCaseWithWebserver.setUp(self)

 

1163

        super(TestLimitedRangeRequestServer, self).setUp()

1161

1164

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

1162

1165

        # response, so we build a content appropriately.

1163

1166

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

1197

1200

 

1198

1201

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

1199

1202

        handler = _urllib2_wrappers.ProxyHandler()

1200

 

        self.assertEquals(expected,

 

1203

        self.assertEqual(expected,

1201

1204

                          handler.evaluate_proxy_bypass(host, no_proxy))

1202

1205

 

1203

1206

    def test_empty_user(self):

1349

1352

        )

1350

1353

 

1351

1354

    def setUp(self):

1352

 

        http_utils.TestCaseWithWebserver.setUp(self)

 

1355

        super(TestRanges, self).setUp()

1353

1356

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

1354

1357

 

1355

1358

    def create_transport_readonly_server(self):

1596

1599

            path='/',

1597

1600

            realm='Realm',

1598

1601

            ))

1599

 

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

 

1602

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

1600

1603

 

1601

1604

 

1602

1605

class TestAuth(http_utils.TestCaseWithWebserver):

1905

1908

        server._url_protocol = self._url_protocol

1906

1909

        return server

1907

1910

 

1908

 

    def test_open_bzrdir(self):

 

1911

    def test_open_controldir(self):

1909

1912

        branch = self.make_branch('relpath')

1910

1913

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

1911

 

        bd = bzrdir.BzrDir.open(url)

 

1914

        bd = controldir.ControlDir.open(url)

1912

1915

        self.addCleanup(bd.transport.disconnect)

1913

1916

        self.assertIsInstance(bd, _mod_remote.RemoteBzrDir)

1914

1917

 

1997

2000

        self.assertIsInstance(r, type(t))

1998

2001

        # Both transports share the some connection

1999

2002

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

2000

 

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

 

2003

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

2001

2004

 

2002

2005

    def test_redirected_to_self_with_slash(self):

2003

2006

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

2014

2017

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

2015

2018

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

2016

2019

        self.assertIsInstance(r, type(t))

2017

 

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

 

2020

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

2018

2021

            r.external_url())

2019

2022

 

2020

2023

    def test_redirected_to_same_host_sibling_protocol(self):

2022

2025

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

2023

2026

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

2024

2027

        self.assertIsInstance(r, type(t))

2025

 

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

 

2028

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

2026

2029

            r.external_url())

2027

2030

 

2028

2031

    def test_redirected_to_same_host_different_protocol(self):

2029

2032

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

2030

2033

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

2031

2034

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

2032

 

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

2033

 

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

 

2035

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

 

2036

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

2034

2037

 

2035

2038

    def test_redirected_to_same_host_specific_implementation(self):

2036

2039

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

2037

2040

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

2038

2041

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

2039

 

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

 

2042

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

2040

2043

 

2041

2044

    def test_redirected_to_different_host_same_user(self):

2042

2045

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

2044

2047

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

2045

2048

        self.assertIsInstance(r, type(t))

2046

2049

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

2047

 

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

 

2050

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

2048

2051

 

2049

2052

 

2050

2053

class PredefinedRequestHandler(http_server.TestingHTTPRequestHandler):

2117

2120

    """

2118

2121

 

2119

2122

    def setUp(self):

2120

 

        tests.TestCase.setUp(self)

2121

2123

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

2122

2124

        self.server.start_server()

 

2125

        self.addCleanup(self.server.stop_server)

2123

2126

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

2124

2127

        def report_activity(t, bytes, direction):

2125

2128

            count = _activities.get(direction, 0)

2126

2129

            count += bytes

2127

2130

            _activities[direction] = count

2128

2131

        self.activities = _activities

2129

 

 

2130

2132

        # We override at class level because constructors may propagate the

2131

2133

        # bound method and render instance overriding ineffective (an

2132

2134

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

2133

2135

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

2134

 

        self.addCleanup(self.server.stop_server)

2135

2136

 

2136

2137

    def get_transport(self):

2137

2138

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

2261

2262

        )

2262

2263

 

2263

2264

    def setUp(self):

 

2265

        super(TestActivity, self).setUp()

2264

2266

        TestActivityMixin.setUp(self)

2265

2267

 

2266

2268

 

2275

2277

    _protocol_version = 'HTTP/1.1'

2276

2278

 

2277

2279

    def setUp(self):

 

2280

        super(TestNoReportActivity, self).setUp()

2278

2281

        self._transport =_urllib.HttpTransport_urllib

2279

2282

        TestActivityMixin.setUp(self)

2280

2283

 

• Older »

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