/brz/remove-bazaar

# Copyright (C) 2005-2012, 2015, 2016, 2017 Canonical Ltd

#

# This program is free software; you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA

"""Tests for HTTP implementations.

This module defines a load_tests() method that parametrize tests classes for

transport implementation, http protocol versions and authentication schemes.

"""

# TODO: Should be renamed to breezy.transport.http.tests?

# TODO: What about renaming to breezy.tests.transport.http ?

import httplib

import io

import SimpleHTTPServer

import socket

import sys

import threading

import breezy

from .. import (

    config,

    controldir,

    debug,

    errors,

    osutils,

    remote as _mod_remote,

    tests,

    trace,

    transport,

    ui,

    )

from . import (

    features,

    http_server,

    http_utils,

    test_server,

    )

from .scenarios import (

    load_tests_apply_scenarios,

    multiply_scenarios,

    )

from ..transport import (

    http,

    remote,

    )

from ..transport.http import (

    _urllib,

    _urllib2_wrappers,

    )

if features.pycurl.available():

    from ..transport.http._pycurl import PyCurlTransport

load_tests = load_tests_apply_scenarios

def vary_by_http_client_implementation():

    """Test the two libraries we can use, pycurl and urllib."""

    transport_scenarios = [

        ('urllib', dict(_transport=_urllib.HttpTransport_urllib,

                        _server=http_server.HttpServer_urllib,

                        _url_protocol='http+urllib',)),

        ]

    if features.pycurl.available():

        transport_scenarios.append(

            ('pycurl', dict(_transport=PyCurlTransport,

                            _server=http_server.HttpServer_PyCurl,

                            _url_protocol='http+pycurl',)))

    return transport_scenarios

def vary_by_http_protocol_version():

    """Test on http/1.0 and 1.1"""

    return [

        ('HTTP/1.0',  dict(_protocol_version='HTTP/1.0')),

        ('HTTP/1.1',  dict(_protocol_version='HTTP/1.1')),

        ]

def vary_by_http_auth_scheme():

    scenarios = [

        ('basic', dict(_auth_server=http_utils.HTTPBasicAuthServer)),

        ('digest', dict(_auth_server=http_utils.HTTPDigestAuthServer)),

        ('basicdigest',

            dict(_auth_server=http_utils.HTTPBasicAndDigestAuthServer)),

        ]

    # Add some attributes common to all scenarios

    for scenario_id, scenario_dict in scenarios:

        scenario_dict.update(_auth_header='Authorization',

                             _username_prompt_prefix='',

                             _password_prompt_prefix='')

    return scenarios

def vary_by_http_proxy_auth_scheme():

    scenarios = [

        ('proxy-basic', dict(_auth_server=http_utils.ProxyBasicAuthServer)),

        ('proxy-digest', dict(_auth_server=http_utils.ProxyDigestAuthServer)),

        ('proxy-basicdigest',

            dict(_auth_server=http_utils.ProxyBasicAndDigestAuthServer)),

        ]

    # Add some attributes common to all scenarios

    for scenario_id, scenario_dict in scenarios:

        scenario_dict.update(_auth_header='Proxy-Authorization',

                             _username_prompt_prefix='Proxy ',

                             _password_prompt_prefix='Proxy ')

    return scenarios

def vary_by_http_activity():

    activity_scenarios = [

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

                            _transport=_urllib.HttpTransport_urllib,)),

        ]

    if features.pycurl.available():

        activity_scenarios.append(

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

                                _transport=PyCurlTransport,)),)

    if features.HTTPSServerFeature.available():

        # FIXME: Until we have a better way to handle self-signed certificates

        # (like allowing them in a test specific authentication.conf for

        # example), we need some specialized pycurl/urllib transport for tests.

        # -- vila 2012-01-20

        from . import (

            ssl_certs,

            )

        class HTTPS_urllib_transport(_urllib.HttpTransport_urllib):

            def __init__(self, base, _from_transport=None):

                super(HTTPS_urllib_transport, self).__init__(

                    base, _from_transport=_from_transport,

                    ca_certs=ssl_certs.build_path('ca.crt'))

        activity_scenarios.append(

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

                                  _transport=HTTPS_urllib_transport,)),)

        if features.pycurl.available():

            class HTTPS_pycurl_transport(PyCurlTransport):

                def __init__(self, base, _from_transport=None):

                    super(HTTPS_pycurl_transport, self).__init__(

                        base, _from_transport)

                    self.cabundle = str(ssl_certs.build_path('ca.crt'))

            activity_scenarios.append(

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

                                    _transport=HTTPS_pycurl_transport,)),)

    return activity_scenarios

class FakeManager(object):

    def __init__(self):

        self.credentials = []

    def add_password(self, realm, host, username, password):

        self.credentials.append([realm, host, username, password])

class RecordingServer(object):

    """A fake HTTP server.

    It records the bytes sent to it, and replies with a 200.

    """

    def __init__(self, expect_body_tail=None, scheme=''):

        """Constructor.

        :type expect_body_tail: str

        :param expect_body_tail: a reply won't be sent until this string is

            received.

        """

        self._expect_body_tail = expect_body_tail

        self.host = None

        self.port = None

        self.received_bytes = ''

        self.scheme = scheme

    def get_url(self):

        return '%s://%s:%s/' % (self.scheme, self.host, self.port)

    def start_server(self):

        self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

        self._sock.bind(('127.0.0.1', 0))

        self.host, self.port = self._sock.getsockname()

        self._ready = threading.Event()

        self._thread = test_server.TestThread(

            sync_event=self._ready, target=self._accept_read_and_reply)

        self._thread.start()

        if 'threads' in tests.selftest_debug_flags:

            sys.stderr.write('Thread started: %s\n' % (self._thread.ident,))

        self._ready.wait()

    def _accept_read_and_reply(self):

        self._sock.listen(1)

        self._ready.set()

        conn, address = self._sock.accept()

        if self._expect_body_tail is not None:

            while not self.received_bytes.endswith(self._expect_body_tail):

                self.received_bytes += conn.recv(4096)

            conn.sendall('HTTP/1.1 200 OK\r\n')

        try:

            self._sock.close()

        except socket.error:

            # The client may have already closed the socket.

            pass

    def stop_server(self):

        try:

            # Issue a fake connection to wake up the server and allow it to

            # finish quickly

            fake_conn = osutils.connect_socket((self.host, self.port))

            fake_conn.close()

        except socket.error:

            # We might have already closed it.  We don't care.

            pass

        self.host = None

        self.port = None

        self._thread.join()

        if 'threads' in tests.selftest_debug_flags:

            sys.stderr.write('Thread  joined: %s\n' % (self._thread.ident,))

class TestAuthHeader(tests.TestCase):

    def parse_header(self, header, auth_handler_class=None):

        if auth_handler_class is None:

            auth_handler_class = _urllib2_wrappers.AbstractAuthHandler

        self.auth_handler =  auth_handler_class()

        return self.auth_handler._parse_auth_header(header)

    def test_empty_header(self):

        scheme, remainder = self.parse_header('')

        self.assertEqual('', scheme)

        self.assertIs(None, remainder)

    def test_negotiate_header(self):

        scheme, remainder = self.parse_header('Negotiate')

        self.assertEqual('negotiate', scheme)

        self.assertIs(None, remainder)

    def test_basic_header(self):

        scheme, remainder = self.parse_header(

            'Basic realm="Thou should not pass"')

        self.assertEqual('basic', scheme)

        self.assertEqual('realm="Thou should not pass"', remainder)

    def test_build_basic_header_with_long_creds(self):

        handler = _urllib2_wrappers.BasicAuthHandler()

        user = 'user' * 10  # length 40

        password = 'password' * 5  # length 40

        header = handler.build_auth_header(

            dict(user=user, password=password), None)

        # https://bugs.launchpad.net/bzr/+bug/1606203 was caused by incorrectly

        # creating a header value with an embedded '\n'

        self.assertFalse('\n' in header)

    def test_basic_extract_realm(self):

        scheme, remainder = self.parse_header(

            'Basic realm="Thou should not pass"',

            _urllib2_wrappers.BasicAuthHandler)

        match, realm = self.auth_handler.extract_realm(remainder)

        self.assertTrue(match is not None)

        self.assertEqual('Thou should not pass', realm)

    def test_digest_header(self):

        scheme, remainder = self.parse_header(

            'Digest realm="Thou should not pass"')

        self.assertEqual('digest', scheme)

        self.assertEqual('realm="Thou should not pass"', remainder)

class TestHTTPRangeParsing(tests.TestCase):

    def setUp(self):

        super(TestHTTPRangeParsing, self).setUp()

        # We focus on range  parsing here and ignore everything else

        class RequestHandler(http_server.TestingHTTPRequestHandler):

            def setup(self): pass

            def handle(self): pass

            def finish(self): pass

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

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

        self.assertEqual(ranges,

                          self.req_handler._parse_ranges(header, file_size))

    def test_simple_range(self):

        self.assertRanges([(0,2)], 'bytes=0-2', 12)

    def test_tail(self):

        self.assertRanges([(8, 11)], 'bytes=-4', 12)

    def test_tail_bigger_than_file(self):

        self.assertRanges([(0, 11)], 'bytes=-99', 12)

    def test_range_without_end(self):

        self.assertRanges([(4, 11)], 'bytes=4-', 12)

    def test_invalid_ranges(self):

        self.assertRanges(None, 'bytes=12-22', 12)

        self.assertRanges(None, 'bytes=1-3,12-22', 12)

        self.assertRanges(None, 'bytes=-', 12)

class TestHTTPServer(tests.TestCase):

    """Test the HTTP servers implementations."""

    def test_invalid_protocol(self):

        class BogusRequestHandler(http_server.TestingHTTPRequestHandler):

            protocol_version = 'HTTP/0.1'

        self.assertRaises(httplib.UnknownProtocol,

                          http_server.HttpServer, BogusRequestHandler)

    def test_force_invalid_protocol(self):

        self.assertRaises(httplib.UnknownProtocol,

                          http_server.HttpServer, protocol_version='HTTP/0.1')

    def test_server_start_and_stop(self):

        server = http_server.HttpServer()

        self.addCleanup(server.stop_server)

        server.start_server()

        self.assertTrue(server.server is not None)

        self.assertTrue(server.server.serving is not None)

        self.assertTrue(server.server.serving)

    def test_create_http_server_one_zero(self):

        class RequestHandlerOneZero(http_server.TestingHTTPRequestHandler):

            protocol_version = 'HTTP/1.0'

        server = http_server.HttpServer(RequestHandlerOneZero)

        self.start_server(server)

        self.assertIsInstance(server.server, http_server.TestingHTTPServer)

    def test_create_http_server_one_one(self):

        class RequestHandlerOneOne(http_server.TestingHTTPRequestHandler):

            protocol_version = 'HTTP/1.1'

        server = http_server.HttpServer(RequestHandlerOneOne)

        self.start_server(server)

        self.assertIsInstance(server.server,

                              http_server.TestingThreadingHTTPServer)

    def test_create_http_server_force_one_one(self):

        class RequestHandlerOneZero(http_server.TestingHTTPRequestHandler):

            protocol_version = 'HTTP/1.0'

        server = http_server.HttpServer(RequestHandlerOneZero,

                                        protocol_version='HTTP/1.1')

        self.start_server(server)

        self.assertIsInstance(server.server,

                              http_server.TestingThreadingHTTPServer)

    def test_create_http_server_force_one_zero(self):

        class RequestHandlerOneOne(http_server.TestingHTTPRequestHandler):

            protocol_version = 'HTTP/1.1'

        server = http_server.HttpServer(RequestHandlerOneOne,

                                        protocol_version='HTTP/1.0')

        self.start_server(server)

        self.assertIsInstance(server.server,

                              http_server.TestingHTTPServer)

class TestWithTransport_pycurl(object):

    """Test case to inherit from if pycurl is present"""

    def _get_pycurl_maybe(self):

        self.requireFeature(features.pycurl)

        return PyCurlTransport

    _transport = property(_get_pycurl_maybe)

class TestHttpTransportUrls(tests.TestCase):

    """Test the http urls."""

    scenarios = vary_by_http_client_implementation()

    def test_abs_url(self):

        """Construction of absolute http URLs"""

        t = self._transport('http://example.com/bzr/bzr.dev/')

        eq = self.assertEqualDiff

        eq(t.abspath('.'), 'http://example.com/bzr/bzr.dev')

        eq(t.abspath('foo/bar'), 'http://example.com/bzr/bzr.dev/foo/bar')

        eq(t.abspath('.bzr'), 'http://example.com/bzr/bzr.dev/.bzr')

        eq(t.abspath('.bzr/1//2/./3'),

           'http://example.com/bzr/bzr.dev/.bzr/1/2/3')

    def test_invalid_http_urls(self):

        """Trap invalid construction of urls"""

        self._transport('http://example.com/bzr/bzr.dev/')

        self.assertRaises(errors.InvalidURL,

                          self._transport,

                          'http://http://example.com/bzr/bzr.dev/')

    def test_http_root_urls(self):

        """Construction of URLs from server root"""

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

        eq = self.assertEqualDiff

        eq(t.abspath('.bzr/tree-version'),

           'http://example.com/.bzr/tree-version')

    def test_http_impl_urls(self):

        """There are servers which ask for particular clients to connect"""

        server = self._server()

        server.start_server()

        try:

            url = server.get_url()

            self.assertTrue(url.startswith('%s://' % self._url_protocol))

        finally:

            server.stop_server()

class TestHttps_pycurl(TestWithTransport_pycurl, tests.TestCase):

    # TODO: This should really be moved into another pycurl

    # specific test. When https tests will be implemented, take

    # this one into account.

    def test_pycurl_without_https_support(self):

        """Test that pycurl without SSL do not fail with a traceback.

        For the purpose of the test, we force pycurl to ignore

        https by supplying a fake version_info that do not

        support it.

        """

        self.requireFeature(features.pycurl)

        # Import the module locally now that we now it's available.

        pycurl = features.pycurl.module

        self.overrideAttr(pycurl, 'version_info',

                          # Fake the pycurl version_info This was taken from

                          # a windows pycurl without SSL (thanks to bialix)

                          lambda : (2,

                                    '7.13.2',

                                    462082,

                                    'i386-pc-win32',

                                    2576,

                                    None,

                                    0,

                                    None,

                                    ('ftp', 'gopher', 'telnet',

                                     'dict', 'ldap', 'http', 'file'),

                                    None,

                                    0,

                                    None))

        self.assertRaises(errors.DependencyNotPresent, self._transport,

                          'https://launchpad.net')

class TestHTTPConnections(http_utils.TestCaseWithWebserver):

    """Test the http connections."""

    scenarios = multiply_scenarios(

        vary_by_http_client_implementation(),

        vary_by_http_protocol_version(),

        )

    def setUp(self):

        super(TestHTTPConnections, self).setUp()

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

                        transport=self.get_transport())

    def test_http_has(self):

        server = self.get_readonly_server()

        t = self.get_readonly_transport()

        self.assertEqual(t.has('foo/bar'), True)

        self.assertEqual(len(server.logs), 1)

        self.assertContainsRe(server.logs[0],

            r'"HEAD /foo/bar HTTP/1.." (200|302) - "-" "bzr/')

    def test_http_has_not_found(self):

        server = self.get_readonly_server()

        t = self.get_readonly_transport()

        self.assertEqual(t.has('not-found'), False)

        self.assertContainsRe(server.logs[1],

            r'"HEAD /not-found HTTP/1.." 404 - "-" "bzr/')

    def test_http_get(self):

        server = self.get_readonly_server()

        t = self.get_readonly_transport()

        fp = t.get('foo/bar')

        self.assertEqualDiff(

            fp.read(),

            'contents of foo/bar\n')

        self.assertEqual(len(server.logs), 1)

        self.assertTrue(server.logs[0].find(

            '"GET /foo/bar HTTP/1.1" 200 - "-" "bzr/%s'

            % breezy.__version__) > -1)

    def test_has_on_bogus_host(self):

        # Get a free address and don't 'accept' on it, so that we

        # can be sure there is no http handler there, but set a

        # reasonable timeout to not slow down tests too much.

        default_timeout = socket.getdefaulttimeout()

        try:

            socket.setdefaulttimeout(2)

            s = socket.socket()

            s.bind(('localhost', 0))

            t = self._transport('http://%s:%s/' % s.getsockname())

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

        finally:

            socket.setdefaulttimeout(default_timeout)

class TestHttpTransportRegistration(tests.TestCase):

    """Test registrations of various http implementations"""

    scenarios = vary_by_http_client_implementation()

    def test_http_registered(self):

        t = transport.get_transport_from_url(

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

        self.assertIsInstance(t, transport.Transport)

        self.assertIsInstance(t, self._transport)

class TestPost(tests.TestCase):

    scenarios = multiply_scenarios(

        vary_by_http_client_implementation(),

        vary_by_http_protocol_version(),

        )

    def test_post_body_is_received(self):

        server = RecordingServer(expect_body_tail='end-of-body',

                                 scheme=self._url_protocol)

        self.start_server(server)

        url = server.get_url()

        # FIXME: needs a cleanup -- vila 20100611

        http_transport = transport.get_transport_from_url(url)

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

        self.assertTrue(

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

        self.assertTrue('content-length: 19\r' in server.received_bytes.lower())

        self.assertTrue('content-type: application/octet-stream\r'

                        in server.received_bytes.lower())

        # The transport should not be assuming that the server can accept

        # chunked encoding the first time it connects, because HTTP/1.1, so we

        # check for the literal string.

        self.assertTrue(

            server.received_bytes.endswith('\r\n\r\nabc def end-of-body'))

class TestRangeHeader(tests.TestCase):

    """Test range_header method"""

    def check_header(self, value, ranges=[], tail=0):

        offsets = [ (start, end - start + 1) for start, end in ranges]

        coalesce = transport.Transport._coalesce_offsets

        coalesced = list(coalesce(offsets, limit=0, fudge_factor=0))

        range_header = http.HttpTransportBase._range_header

        self.assertEqual(value, range_header(coalesced, tail))

    def test_range_header_single(self):

        self.check_header('0-9', ranges=[(0,9)])

        self.check_header('100-109', ranges=[(100,109)])

    def test_range_header_tail(self):

        self.check_header('-10', tail=10)

        self.check_header('-50', tail=50)

    def test_range_header_multi(self):

        self.check_header('0-9,100-200,300-5000',

                          ranges=[(0,9), (100, 200), (300,5000)])

    def test_range_header_mixed(self):

        self.check_header('0-9,300-5000,-50',

                          ranges=[(0,9), (300,5000)],

                          tail=50)

class TestSpecificRequestHandler(http_utils.TestCaseWithWebserver):

    """Tests a specific request handler.

    Daughter classes are expected to override _req_handler_class

    """

    scenarios = multiply_scenarios(

        vary_by_http_client_implementation(),

        vary_by_http_protocol_version(),

        )

    # Provide a useful default

    _req_handler_class = http_server.TestingHTTPRequestHandler

    def create_transport_readonly_server(self):

        server = http_server.HttpServer(self._req_handler_class,

                                        protocol_version=self._protocol_version)

        server._url_protocol = self._url_protocol

        return server

    def _testing_pycurl(self):

        # TODO: This is duplicated for lots of the classes in this file

        return (features.pycurl.available()

                and self._transport == PyCurlTransport)

class WallRequestHandler(http_server.TestingHTTPRequestHandler):

    """Whatever request comes in, close the connection"""

    def _handle_one_request(self):

        """Handle a single HTTP request, by abruptly closing the connection"""

        self.close_connection = 1

class TestWallServer(TestSpecificRequestHandler):

    """Tests exceptions during the connection phase"""

    _req_handler_class = WallRequestHandler

    def test_http_has(self):

        t = self.get_readonly_transport()

        # Unfortunately httplib (see HTTPResponse._read_status

        # for details) make no distinction between a closed

        # socket and badly formatted status line, so we can't

        # just test for ConnectionError, we have to test

        # InvalidHttpResponse too. And pycurl may raise ConnectionReset

        # instead of ConnectionError too.