← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_http.py

  • Committer: Patch Queue Manager
  • Date: 2015-12-17 18:39:00 UTC
  • mfrom: (6606.1.2 fix-float)
  • Revision ID: pqm@pqm.ubuntu.com-20151217183900-0719du2uv1kwu3lc
(vila) Inline testtools private method to fix an issue in xenial (the
 private implementation has changed in an backward incompatible way).
 (Jelmer Vernooij)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_http.py
1
 
# Copyright (C) 2005-2011 Canonical Ltd
 
1
# Copyright (C) 2005-2012, 2015 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):
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)])
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):
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
 
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