/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 ?

try:

    from http.client import UnknownProtocol, parse_headers

    from http.server import SimpleHTTPRequestHandler

except ImportError:  # python < 3

    from httplib import UnknownProtocol

    from SimpleHTTPServer import SimpleHTTPRequestHandler

import io

import socket

import sys

import threading

import breezy

from .. import (

    config,

    controldir,

    debug,

    errors,

    osutils,

    tests,

    trace,

    transport,

    ui,

    urlutils,

    )

from ..bzr import (

    remote as _mod_remote,

    )

from ..sixish import PY3

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 (

    HttpTransport,

    _urllib2_wrappers,

    )

load_tests = load_tests_apply_scenarios

def vary_by_http_client_implementation():

    """Test the libraries we can use, currently just urllib."""

    transport_scenarios = [

        ('urllib', dict(_transport=HttpTransport,

                        _server=http_server.HttpServer,

                        _url_protocol='http',)),

        ]

    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=HttpTransport,)),

        ]

    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 urllib transport for tests.

        # -- vila 2012-01-20

        from . import (

            ssl_certs,

            )

        class HTTPS_transport(HttpTransport):

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

                super(HTTPS_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_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 = b''

        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(b'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(u'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(UnknownProtocol,

                          http_server.HttpServer, BogusRequestHandler)

    def test_force_invalid_protocol(self):

        self.assertRaises(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 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(urlutils.InvalidURL,

                          self._transport,

                          'http://example.com:port/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 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) - "-" "Breezy/')

    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 - "-" "Breezy/')

    def test_http_get(self):

        server = self.get_readonly_server()

        t = self.get_readonly_transport()

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

        self.assertEqualDiff(

            fp.read(),

            b'contents of foo/bar\n')

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

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

            '"GET /foo/bar HTTP/1.1" 200 - "-" "Breezy/%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=b'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(b'abc def end-of-body')

        self.assertTrue(

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

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

        self.assertTrue(b'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(b'\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.HttpTransport._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

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.

        self.assertRaises((errors.ConnectionError,

                           errors.InvalidHttpResponse),

                          t.has, 'foo/bar')

    def test_http_get(self):

        t = self.get_readonly_transport()

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

                           errors.InvalidHttpResponse),

                          t.get, 'foo/bar')

class BadStatusRequestHandler(http_server.TestingHTTPRequestHandler):

    """Whatever request comes in, returns a bad status"""

    def parse_request(self):

        """Fakes handling a single HTTP request, returns a bad status"""

        ignored = http_server.TestingHTTPRequestHandler.parse_request(self)

        self.send_response(0, "Bad status")

        self.close_connection = 1

        return False

class TestBadStatusServer(TestSpecificRequestHandler):

    """Tests bad status from server."""

    _req_handler_class = BadStatusRequestHandler

    def setUp(self):

        super(TestBadStatusServer, self).setUp()

        # See https://bugs.launchpad.net/bzr/+bug/1451448 for details.

        # TD;LR: Running both a TCP client and server in the same process and

        # thread uncovers a race in python. The fix is to run the server in a

        # different process. Trying to fix yet another race here is not worth

        # the effort. -- vila 2015-09-06

        if 'HTTP/1.0' in self.id():

            raise tests.TestSkipped(

                'Client/Server in the same process and thread can hang')

    def test_http_has(self):

        t = self.get_readonly_transport()

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

                           errors.InvalidHttpResponse),

                          t.has, 'foo/bar')

    def test_http_get(self):

        t = self.get_readonly_transport()

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

                           errors.InvalidHttpResponse),

                          t.get, 'foo/bar')

class InvalidStatusRequestHandler(http_server.TestingHTTPRequestHandler):