/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to breezy/tests/test_http.py

  • Committer: Jelmer Vernooij
  • Date: 2017-05-22 00:56:52 UTC
  • mfrom: (6621.2.26 py3_pokes)
  • Revision ID: jelmer@jelmer.uk-20170522005652-yjahcr9hwmjkno7n
Merge Python3 porting work ('py3 pokes')

Show diffs side-by-side

added added

removed removed

Lines of Context:
breezy/tests/test_http.py
24
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
33
import breezy
33
 
from breezy import (
 
34
from .. import (
34
35
    config,
35
36
    controldir,
36
37
    debug,
42
43
    transport,
43
44
    ui,
44
45
    )
45
 
from breezy.tests import (
 
46
from . import (
46
47
    features,
47
48
    http_server,
48
49
    http_utils,
49
50
    test_server,
50
51
    )
51
 
from breezy.tests.scenarios import (
 
52
from .scenarios import (
52
53
    load_tests_apply_scenarios,
53
54
    multiply_scenarios,
54
55
    )
55
 
from breezy.transport import (
 
56
from ..transport import (
56
57
    http,
57
58
    remote,
58
59
    )
59
 
from breezy.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 breezy.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 breezy.tests import (
 
140
        from . import (
140
141
            ssl_certs,
141
142
            )
142
143
        class HTTPS_urllib_transport(_urllib.HttpTransport_urllib):
1216
1217
    def test_empty_user(self):
1217
1218
        self.overrideEnv('http_proxy', 'http://bar.com')
1218
1219
        request = self._proxied_request()
1219
 
        self.assertFalse(request.headers.has_key('Proxy-authorization'))
 
1220
        self.assertFalse('Proxy-authorization' in request.headers)
1220
1221
 
1221
1222
    def test_user_with_at(self):
1222
1223
        self.overrideEnv('http_proxy',
1223
1224
                         'http://username@domain:password@proxy_host:1234')
1224
1225
        request = self._proxied_request()
1225
 
        self.assertFalse(request.headers.has_key('Proxy-authorization'))
 
1226
        self.assertFalse('Proxy-authorization' in request.headers)
1226
1227
 
1227
1228
    def test_invalid_proxy(self):
1228
1229
        """A proxy env variable without scheme"""
1701
1702
 
1702
1703
        self.server.add_user('joe', 'foo')
1703
1704
        t = self.get_user_transport(None, None)
1704
 
        stdout = tests.StringIOWrapper()
1705
 
        stderr = tests.StringIOWrapper()
1706
 
        ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',
1707
 
                                            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
1708
1707
        self.assertEqual('contents of a\n',t.get('a').read())
1709
1708
        # stdin should be empty
1710
1709
        self.assertEqual('', ui.ui_factory.stdin.readline())
1723
1722
 
1724
1723
        self.server.add_user('joe', 'foo')
1725
1724
        t = self.get_user_transport('joe', None)
1726
 
        stdout = tests.StringIOWrapper()
1727
 
        stderr = tests.StringIOWrapper()
1728
 
        ui.ui_factory = tests.TestUIFactory(stdin='foo\n',
1729
 
                                            stdout=stdout, stderr=stderr)
 
1725
        ui.ui_factory = tests.TestUIFactory(stdin='foo\n')
 
1726
        stdout, stderr = ui.ui_factory.stdout, ui.ui_factory.stderr
1730
1727
        self.assertEqual('contents of a\n', t.get('a').read())
1731
1728
        # stdin should be empty
1732
1729
        self.assertEqual('', ui.ui_factory.stdin.readline())
1767
1764
        stdin_content = 'bar\n'  # Not the right password
1768
1765
        self.server.add_user(user, password)
1769
1766
        t = self.get_user_transport(user, None)
1770
 
        ui.ui_factory = tests.TestUIFactory(stdin=stdin_content,
1771
 
                                            stderr=tests.StringIOWrapper())
 
1767
        ui.ui_factory = tests.TestUIFactory(stdin=stdin_content)
1772
1768
        # Create a minimal config file with the right password
1773
1769
        _setup_authentication_config(scheme='http', port=self.server.port,
1774
1770
                                     user=user, password=password)
1816
1812
        self.assertEqual(1, self.server.auth_required_errors)
1817
1813
 
1818
1814
    def test_no_credential_leaks_in_log(self):
1819
 
        self.overrideAttr(debug, 'debug_flags', set(['http']))
 
1815
        self.overrideAttr(debug, 'debug_flags', {'http'})
1820
1816
        user = 'joe'
1821
1817
        password = 'very-sensitive-password'
1822
1818
        self.server.add_user(user, password)
1880
1876
        super(TestProxyAuth, self).test_empty_pass()
1881
1877
 
1882
1878
 
 
1879
class NonClosingBytesIO(io.BytesIO):
 
1880
 
 
1881
    def close(self):
 
1882
        """Ignore and leave file open."""
 
1883
 
 
1884
 
1883
1885
class SampleSocket(object):
1884
1886
    """A socket-like object for use in testing the HTTP request handler."""
1885
1887
 
1888
1890
 
1889
1891
        :param socket_read_content: a byte sequence
1890
1892
        """
1891
 
        # Use plain python StringIO so we can monkey-patch the close method to
1892
 
        # not discard the contents.
1893
 
        from StringIO import StringIO
1894
 
        self.readfile = StringIO(socket_read_content)
1895
 
        self.writefile = StringIO()
1896
 
        self.writefile.close = lambda: None
1897
 
        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."""
1898
1898
 
1899
1899
    def makefile(self, mode='r', bufsize=None):
1900
1900
        if 'r' in mode:
2122
2122
 
2123
2123
 
2124
2124
if features.HTTPSServerFeature.available():
2125
 
    from breezy.tests import https_server
 
2125
    from . import https_server
2126
2126
    class ActivityHTTPSServer(ActivityServerMixin, https_server.HTTPSServer):
2127
2127
        pass
2128
2128
 
2343
2343
            self.addCleanup(redirected_t.disconnect)
2344
2344
            return redirected_t
2345
2345
 
2346
 
        stdout = tests.StringIOWrapper()
2347
 
        stderr = tests.StringIOWrapper()
2348
 
        ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',
2349
 
                                            stdout=stdout, stderr=stderr)
 
2346
        ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n')
2350
2347
        self.assertEqual('redirected once',
2351
2348
                         transport.do_catching_redirections(
2352
2349
                self.get_a, self.old_transport, redirected).read())
2354
2351
        # stdin should be empty
2355
2352
        self.assertEqual('', ui.ui_factory.stdin.readline())
2356
2353
        # stdout should be empty, stderr will contains the prompts
2357
 
        self.assertEqual('', stdout.getvalue())
 
2354
        self.assertEqual('', ui.ui_factory.stdout.getvalue())
2358
2355
 
2359
2356
    def test_auth_on_redirected_via_following_redirections(self):
2360
2357
        self.new_server.add_user('joe', 'foo')
2361
 
        stdout = tests.StringIOWrapper()
2362
 
        stderr = tests.StringIOWrapper()
2363
 
        ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n',
2364
 
                                            stdout=stdout, stderr=stderr)
 
2358
        ui.ui_factory = tests.TestUIFactory(stdin='joe\nfoo\n')
2365
2359
        t = self.old_transport
2366
2360
        req = RedirectedRequest('GET', t.abspath('a'))
2367
2361
        new_prefix = 'http://%s:%s' % (self.new_server.host,
2372
2366
        # stdin should be empty
2373
2367
        self.assertEqual('', ui.ui_factory.stdin.readline())
2374
2368
        # stdout should be empty, stderr will contains the prompts
2375
 
        self.assertEqual('', stdout.getvalue())
 
2369
        self.assertEqual('', ui.ui_factory.stdout.getvalue())
2376
2370