/breezy/unstable

To get this branch, use:
bzr branch https://code.breezy-vcs.org/breezy/unstable

« back to all changes in this revision

Viewing changes to breezy/tests/test_http.py

  • Committer: Jelmer Vernooij
  • Date: 2017-05-24 01:39:33 UTC
  • mfrom: (3815.3776.6)
  • Revision ID: jelmer@jelmer.uk-20170524013933-ir4y4tqtrsiz2ka2
New upstream snapshot.

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2005-2012, 2015, 2016 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
20
20
transport implementation, http protocol versions and authentication schemes.
21
21
"""
22
22
 
23
 
# TODO: Should be renamed to bzrlib.transport.http.tests?
24
 
# TODO: What about renaming to bzrlib.tests.transport.http ?
 
23
# TODO: Should be renamed to breezy.transport.http.tests?
 
24
# TODO: What about renaming to breezy.tests.transport.http ?
25
25
 
26
26
import httplib
 
27
import io
27
28
import SimpleHTTPServer
28
29
import socket
29
30
import sys
30
31
import threading
31
32
 
32
 
import bzrlib
33
 
from bzrlib import (
 
33
import breezy
 
34
from .. import (
34
35
    config,
35
36
    controldir,
36
37
    debug,
42
43
    transport,
43
44
    ui,
44
45
    )
45
 
from bzrlib.tests import (
 
46
from . import (
46
47
    features,
47
48
    http_server,
48
49
    http_utils,
49
50
    test_server,
50
51
    )
51
 
from bzrlib.tests.scenarios import (
 
52
from .scenarios import (
52
53
    load_tests_apply_scenarios,
53
54
    multiply_scenarios,
54
55
    )
55
 
from bzrlib.transport import (
 
56
from ..transport import (
56
57
    http,
57
58
    remote,
58
59
    )
59
 
from bzrlib.transport.http import (
 
60
from ..transport.http import (
60
61
    _urllib,
61
62
    _urllib2_wrappers,
62
63
    )
63
64
 
64
65
 
65
66
if features.pycurl.available():
66
 
    from bzrlib.transport.http._pycurl import PyCurlTransport
 
67
    from ..transport.http._pycurl import PyCurlTransport
67
68
 
68
69
 
69
70
load_tests = load_tests_apply_scenarios
136
137
        # (like allowing them in a test specific authentication.conf for
137
138
        # example), we need some specialized pycurl/urllib transport for tests.
138
139
        # -- vila 2012-01-20
139
 
        from bzrlib.tests import (
 
140
        from . import (
140
141
            ssl_certs,
141
142
            )
142
143
        class HTTPS_urllib_transport(_urllib.HttpTransport_urllib):
260
261
        self.assertEqual('basic', scheme)
261
262
        self.assertEqual('realm="Thou should not pass"', remainder)
262
263
 
 
264
    def test_build_basic_header_with_long_creds(self):
 
265
        handler = _urllib2_wrappers.BasicAuthHandler()
 
266
        user = 'user' * 10  # length 40
 
267
        password = 'password' * 5  # length 40
 
268
        header = handler.build_auth_header(
 
269
            dict(user=user, password=password), None)
 
270
        # https://bugs.launchpad.net/bzr/+bug/1606203 was caused by incorrectly
 
271
        # creating a header value with an embedded '\n'
 
272
        self.assertFalse('\n' in header)
 
273
 
263
274
    def test_basic_extract_realm(self):
264
275
        scheme, remainder = self.parse_header(
265
276
            'Basic realm="Thou should not pass"',
498
509
        self.assertEqual(len(server.logs), 1)
499
510
        self.assertTrue(server.logs[0].find(
500
511
            '"GET /foo/bar HTTP/1.1" 200 - "-" "bzr/%s'
501
 
            % bzrlib.__version__) > -1)
 
512
            % breezy.__version__) > -1)
502
513
 
503
514
    def test_has_on_bogus_host(self):
504
515
        # Get a free address and don't 'accept' on it, so that we
1206
1217
    def test_empty_user(self):
1207
1218
        self.overrideEnv('http_proxy', 'http://bar.com')
1208
1219
        request = self._proxied_request()
1209
 
        self.assertFalse(request.headers.has_key('Proxy-authorization'))
 
1220
        self.assertFalse('Proxy-authorization' in request.headers)
1210
1221
 
1211
1222
    def test_user_with_at(self):
1212
1223
        self.overrideEnv('http_proxy',
1213
1224
                         'http://username@domain:password@proxy_host:1234')
1214
1225
        request = self._proxied_request()
1215
 
        self.assertFalse(request.headers.has_key('Proxy-authorization'))
 
1226
        self.assertFalse('Proxy-authorization' in request.headers)
1216
1227
 
1217
1228
    def test_invalid_proxy(self):
1218
1229
        """A proxy env variable without scheme"""
1273
1284
            self.no_proxy_host = self.server_host_port
1274
1285
        # The secondary server is the proxy
1275
1286
        self.proxy_url = self.get_secondary_url()
 
1287
        if self._testing_pycurl():
 
1288
            self.proxy_url = self.proxy_url.replace('+pycurl', '')
1276
1289
 
1277
1290
    def _testing_pycurl(self):
1278
1291
        # TODO: This is duplicated for lots of the classes in this file
1689
1702
 
1690
1703
        self.server.add_user('joe', 'foo')
1691
1704
        t = self.get_user_transport(None, None)
1692
 
        stdout = tests.StringIOWrapper()
1693
 
        stderr = tests.StringIOWrapper()
1694
 
        ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',
1695
 
                                            stdout=stdout, stderr=stderr)
 
1705
        ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n')
 
1706
        stdout, stderr = ui.ui_factory.stdout, ui.ui_factory.stderr
1696
1707
        self.assertEqual('contents of a\n',t.get('a').read())
1697
1708
        # stdin should be empty
1698
1709
        self.assertEqual('', ui.ui_factory.stdin.readline())
1711
1722
 
1712
1723
        self.server.add_user('joe', 'foo')
1713
1724
        t = self.get_user_transport('joe', None)
1714
 
        stdout = tests.StringIOWrapper()
1715
 
        stderr = tests.StringIOWrapper()
1716
 
        ui.ui_factory = tests.TestUIFactory(stdin='foo\n',
1717
 
                                            stdout=stdout, stderr=stderr)
 
1725
        ui.ui_factory = tests.TestUIFactory(stdin='foo\n')
 
1726
        stdout, stderr = ui.ui_factory.stdout, ui.ui_factory.stderr
1718
1727
        self.assertEqual('contents of a\n', t.get('a').read())
1719
1728
        # stdin should be empty
1720
1729
        self.assertEqual('', ui.ui_factory.stdin.readline())
1755
1764
        stdin_content = 'bar\n'  # Not the right password
1756
1765
        self.server.add_user(user, password)
1757
1766
        t = self.get_user_transport(user, None)
1758
 
        ui.ui_factory = tests.TestUIFactory(stdin=stdin_content,
1759
 
                                            stderr=tests.StringIOWrapper())
 
1767
        ui.ui_factory = tests.TestUIFactory(stdin=stdin_content)
1760
1768
        # Create a minimal config file with the right password
1761
1769
        _setup_authentication_config(scheme='http', port=self.server.port,
1762
1770
                                     user=user, password=password)
1804
1812
        self.assertEqual(1, self.server.auth_required_errors)
1805
1813
 
1806
1814
    def test_no_credential_leaks_in_log(self):
1807
 
        self.overrideAttr(debug, 'debug_flags', set(['http']))
 
1815
        self.overrideAttr(debug, 'debug_flags', {'http'})
1808
1816
        user = 'joe'
1809
1817
        password = 'very-sensitive-password'
1810
1818
        self.server.add_user(user, password)
1853
1861
                                  ])
1854
1862
 
1855
1863
    def get_user_transport(self, user, password):
1856
 
        self.overrideEnv('all_proxy', self.get_user_url(user, password))
 
1864
        proxy_url = self.get_user_url(user, password)
 
1865
        if self._testing_pycurl():
 
1866
            proxy_url = proxy_url.replace('+pycurl', '')
 
1867
        self.overrideEnv('all_proxy', proxy_url)
1857
1868
        return TestAuth.get_user_transport(self, user, password)
1858
1869
 
1859
1870
    def test_empty_pass(self):
1865
1876
        super(TestProxyAuth, self).test_empty_pass()
1866
1877
 
1867
1878
 
 
1879
class NonClosingBytesIO(io.BytesIO):
 
1880
 
 
1881
    def close(self):
 
1882
        """Ignore and leave file open."""
 
1883
 
 
1884
 
1868
1885
class SampleSocket(object):
1869
1886
    """A socket-like object for use in testing the HTTP request handler."""
1870
1887
 
1873
1890
 
1874
1891
        :param socket_read_content: a byte sequence
1875
1892
        """
1876
 
        # Use plain python StringIO so we can monkey-patch the close method to
1877
 
        # not discard the contents.
1878
 
        from StringIO import StringIO
1879
 
        self.readfile = StringIO(socket_read_content)
1880
 
        self.writefile = StringIO()
1881
 
        self.writefile.close = lambda: None
1882
 
        self.close = lambda: None
 
1893
        self.readfile = io.BytesIO(socket_read_content)
 
1894
        self.writefile = NonClosingBytesIO()
 
1895
 
 
1896
    def close(self):
 
1897
        """Ignore and leave files alone."""
1883
1898
 
1884
1899
    def makefile(self, mode='r', bufsize=None):
1885
1900
        if 'r' in mode:
1898
1913
    def setUp(self):
1899
1914
        super(SmartHTTPTunnellingTest, self).setUp()
1900
1915
        # We use the VFS layer as part of HTTP tunnelling tests.
1901
 
        self.overrideEnv('BZR_NO_SMART_VFS', None)
 
1916
        self.overrideEnv('BRZ_NO_SMART_VFS', None)
1902
1917
        self.transport_readonly_server = http_utils.HTTPServerWithSmarts
1903
1918
        self.http_server = self.get_readonly_server()
1904
1919
 
2107
2122
 
2108
2123
 
2109
2124
if features.HTTPSServerFeature.available():
2110
 
    from bzrlib.tests import https_server
 
2125
    from . import https_server
2111
2126
    class ActivityHTTPSServer(ActivityServerMixin, https_server.HTTPSServer):
2112
2127
        pass
2113
2128
 
2328
2343
            self.addCleanup(redirected_t.disconnect)
2329
2344
            return redirected_t
2330
2345
 
2331
 
        stdout = tests.StringIOWrapper()
2332
 
        stderr = tests.StringIOWrapper()
2333
 
        ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',
2334
 
                                            stdout=stdout, stderr=stderr)
 
2346
        ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n')
2335
2347
        self.assertEqual('redirected once',
2336
2348
                         transport.do_catching_redirections(
2337
2349
                self.get_a, self.old_transport, redirected).read())
2339
2351
        # stdin should be empty
2340
2352
        self.assertEqual('', ui.ui_factory.stdin.readline())
2341
2353
        # stdout should be empty, stderr will contains the prompts
2342
 
        self.assertEqual('', stdout.getvalue())
 
2354
        self.assertEqual('', ui.ui_factory.stdout.getvalue())
2343
2355
 
2344
2356
    def test_auth_on_redirected_via_following_redirections(self):
2345
2357
        self.new_server.add_user('joe', 'foo')
2346
 
        stdout = tests.StringIOWrapper()
2347
 
        stderr = tests.StringIOWrapper()
2348
 
        ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',
2349
 
                                            stdout=stdout, stderr=stderr)
 
2358
        ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n')
2350
2359
        t = self.old_transport
2351
2360
        req = RedirectedRequest('GET', t.abspath('a'))
2352
2361
        new_prefix = 'http://%s:%s' % (self.new_server.host,
2357
2366
        # stdin should be empty
2358
2367
        self.assertEqual('', ui.ui_factory.stdin.readline())
2359
2368
        # stdout should be empty, stderr will contains the prompts
2360
 
        self.assertEqual('', stdout.getvalue())
 
2369
        self.assertEqual('', ui.ui_factory.stdout.getvalue())
2361
2370