/brz/remove-bazaar

# Copyright (C) 2005, 2006, 2007, 2008, 2009 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 bzrlib.transport.http.tests?

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

from cStringIO import StringIO

import httplib

import os

import select

import SimpleHTTPServer

import socket

import sys

import threading

import bzrlib

from bzrlib import (

    bzrdir,

    config,

    errors,

    osutils,

    remote as _mod_remote,

    tests,

    transport,

    ui,

    urlutils,

    )

from bzrlib.symbol_versioning import (

    deprecated_in,

    )

from bzrlib.tests import (

    http_server,

    http_utils,

    )

from bzrlib.transport import (

    http,

    remote,

    )

from bzrlib.transport.http import (

    _urllib,

    _urllib2_wrappers,

    )

try:

    from bzrlib.transport.http._pycurl import PyCurlTransport

    pycurl_present = True

except errors.DependencyNotPresent:

    pycurl_present = False

def load_tests(standard_tests, module, loader):

    """Multiply tests for http clients and protocol versions."""

    result = loader.suiteClass()

    # one for each transport implementation

    t_tests, remaining_tests = tests.split_suite_by_condition(

        standard_tests, tests.condition_isinstance((

                TestHttpTransportRegistration,

                TestHttpTransportUrls,

                Test_redirected_to,

                )))

    transport_scenarios = [

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

                        _server=http_server.HttpServer_urllib,

                        _qualified_prefix='http+urllib',)),

        ]

    if pycurl_present:

        transport_scenarios.append(

            ('pycurl', dict(_transport=PyCurlTransport,

                            _server=http_server.HttpServer_PyCurl,

                            _qualified_prefix='http+pycurl',)))

    tests.multiply_tests(t_tests, transport_scenarios, result)

    # each implementation tested with each HTTP version

    tp_tests, remaining_tests = tests.split_suite_by_condition(

        remaining_tests, tests.condition_isinstance((

                SmartHTTPTunnellingTest,

                TestDoCatchRedirections,

                TestHTTPConnections,

                TestHTTPRedirections,

                TestHTTPSilentRedirections,

                TestLimitedRangeRequestServer,

                TestPost,

                TestProxyHttpServer,

                TestRanges,

                TestSpecificRequestHandler,

                )))

    protocol_scenarios = [

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

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

            ]

    tp_scenarios = tests.multiply_scenarios(transport_scenarios,

                                            protocol_scenarios)

    tests.multiply_tests(tp_tests, tp_scenarios, result)

    # proxy auth: each auth scheme on all http versions on all implementations.

    tppa_tests, remaining_tests = tests.split_suite_by_condition(

        remaining_tests, tests.condition_isinstance((

                TestProxyAuth,

                )))

    proxy_auth_scheme_scenarios = [

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

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

        ('basicdigest',

         dict(_auth_server=http_utils.ProxyBasicAndDigestAuthServer)),

        ]

    tppa_scenarios = tests.multiply_scenarios(tp_scenarios,

                                              proxy_auth_scheme_scenarios)

    tests.multiply_tests(tppa_tests, tppa_scenarios, result)

    # auth: each auth scheme on all http versions on all implementations.

    tpa_tests, remaining_tests = tests.split_suite_by_condition(

        remaining_tests, tests.condition_isinstance((

                TestAuth,

                )))

    auth_scheme_scenarios = [

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

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

        ('basicdigest',

         dict(_auth_server=http_utils.HTTPBasicAndDigestAuthServer)),

        ]

    tpa_scenarios = tests.multiply_scenarios(tp_scenarios,

                                             auth_scheme_scenarios)

    tests.multiply_tests(tpa_tests, tpa_scenarios, result)

    # activity: on all http[s] versions on all implementations

    tpact_tests, remaining_tests = tests.split_suite_by_condition(

        remaining_tests, tests.condition_isinstance((

                TestActivity,

                )))

    activity_scenarios = [

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

                             _transport=_urllib.HttpTransport_urllib,)),

        ]

    if tests.HTTPSServerFeature.available():

        activity_scenarios.append(

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

                                  _transport=_urllib.HttpTransport_urllib,)),)

    if pycurl_present:

        activity_scenarios.append(

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

                                 _transport=PyCurlTransport,)),)

        if tests.HTTPSServerFeature.available():

            from bzrlib.tests import (

                ssl_certs,

                )

            # 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

            # transport for tests.

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

    tpact_scenarios = tests.multiply_scenarios(activity_scenarios,

                                               protocol_scenarios)

    tests.multiply_tests(tpact_tests, tpact_scenarios, result)

    # No parametrization for the remaining tests

    result.addTests(remaining_tests)

    return result

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):

        """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 = ''

    def setUp(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 = threading.Thread(target=self._accept_read_and_reply)

        self._thread.setDaemon(True)

        self._thread.start()

        self._ready.wait(5)

    def _accept_read_and_reply(self):

        self._sock.listen(1)

        self._ready.set()

        self._sock.settimeout(5)

        try:

            conn, address = self._sock.accept()

            # On win32, the accepted connection will be non-blocking to start

            # with because we're using settimeout.

            conn.setblocking(True)

            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')

        except socket.timeout:

            # Make sure the client isn't stuck waiting for us to e.g. accept.

            self._sock.close()

        except socket.error:

            # The client may have already closed the socket.

            pass

    def tearDown(self):

        try:

            self._sock.close()

        except socket.error:

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

            pass

        self.host = None

        self.port = None

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.assertEquals('', scheme)

        self.assertIs(None, remainder)

    def test_negotiate_header(self):

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

        self.assertEquals('negotiate', scheme)

        self.assertIs(None, remainder)

    def test_basic_header(self):

        scheme, remainder = self.parse_header(

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

        self.assertEquals('basic', scheme)

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

    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.assertEquals('Thou should not pass', realm)

    def test_digest_header(self):

        scheme, remainder = self.parse_header(

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

        self.assertEquals('digest', scheme)

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

class TestHTTPServer(tests.TestCase):

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

    def test_invalid_protocol(self):

        class BogusRequestHandler(http_server.TestingHTTPRequestHandler):

            protocol_version = 'HTTP/0.1'

        server = http_server.HttpServer(BogusRequestHandler)

        try:

            self.assertRaises(httplib.UnknownProtocol, server.setUp)

        except:

            server.tearDown()

            self.fail('HTTP Server creation did not raise UnknownProtocol')

    def test_force_invalid_protocol(self):

        server = http_server.HttpServer(protocol_version='HTTP/0.1')

        try:

            self.assertRaises(httplib.UnknownProtocol, server.setUp)

        except:

            server.tearDown()

            self.fail('HTTP Server creation did not raise UnknownProtocol')

    def test_server_start_and_stop(self):

        server = http_server.HttpServer()

        server.setUp()

        try:

            self.assertTrue(server._http_running)

        finally:

            server.tearDown()

        self.assertFalse(server._http_running)

    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._httpd, 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._httpd,

                              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._httpd,

                              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._httpd,

                              http_server.TestingHTTPServer)

class TestWithTransport_pycurl(object):

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

    def _get_pycurl_maybe(self):

        try:

            from bzrlib.transport.http._pycurl import PyCurlTransport

            return PyCurlTransport

        except errors.DependencyNotPresent:

            raise tests.TestSkipped('pycurl not present')

    _transport = property(_get_pycurl_maybe)

class TestHttpUrls(tests.TestCase):

    # TODO: This should be moved to authorization tests once they

    # are written.

    def test_url_parsing(self):

        f = FakeManager()

        url = http.extract_auth('http://example.com', f)

        self.assertEquals('http://example.com', url)

        self.assertEquals(0, len(f.credentials))

        url = http.extract_auth(

            'http://user:pass@www.bazaar-vcs.org/bzr/bzr.dev', f)

        self.assertEquals('http://www.bazaar-vcs.org/bzr/bzr.dev', url)

        self.assertEquals(1, len(f.credentials))

        self.assertEquals([None, 'www.bazaar-vcs.org', 'user', 'pass'],

                          f.credentials[0])

class TestHttpTransportUrls(tests.TestCase):

    """Test the http urls."""

    def test_abs_url(self):

        """Construction of absolute http URLs"""

        t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')

        eq = self.assertEqualDiff

        eq(t.abspath('.'), 'http://bazaar-vcs.org/bzr/bzr.dev')

        eq(t.abspath('foo/bar'), 'http://bazaar-vcs.org/bzr/bzr.dev/foo/bar')

        eq(t.abspath('.bzr'), 'http://bazaar-vcs.org/bzr/bzr.dev/.bzr')

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

           'http://bazaar-vcs.org/bzr/bzr.dev/.bzr/1/2/3')

    def test_invalid_http_urls(self):

        """Trap invalid construction of urls"""

        t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')

        self.assertRaises(errors.InvalidURL,

                          self._transport,

                          'http://http://bazaar-vcs.org/bzr/bzr.dev/')

    def test_http_root_urls(self):

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

        t = self._transport('http://bzr.ozlabs.org/')

        eq = self.assertEqualDiff

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

           'http://bzr.ozlabs.org/.bzr/tree-version')

    def test_http_impl_urls(self):

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

        server = self._server()

        server.setUp()

        try:

            url = server.get_url()

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

        finally:

            server.tearDown()

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.

        """

        try:

            import pycurl

        except ImportError:

            raise tests.TestSkipped('pycurl not present')

        version_info_orig = pycurl.version_info

        try:

            # Now that we have pycurl imported, we can fake its version_info

            # This was taken from a windows pycurl without SSL

            # (thanks to bialix)

            pycurl.version_info = 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')

        finally:

            # Restore the right function

            pycurl.version_info = version_info_orig

class TestHTTPConnections(http_utils.TestCaseWithWebserver):

    """Test the http connections."""

    def setUp(self):

        http_utils.TestCaseWithWebserver.setUp(self)

        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._transport(server.get_url())

        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._transport(server.get_url())

        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._transport(server.get_url())

        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'

            % bzrlib.__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"""

    def test_http_registered(self):

        t = transport.get_transport('%s://foo.com/' % self._qualified_prefix)

        self.assertIsInstance(t, transport.Transport)

        self.assertIsInstance(t, self._transport)

class TestPost(tests.TestCase):

    def test_post_body_is_received(self):

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

        self.start_server(server)

        scheme = self._qualified_prefix

        url = '%s://%s:%s/' % (scheme, server.host, server.port)

        http_transport = self._transport(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())

        # 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

    """

    # Provide a useful default

    _req_handler_class = http_server.TestingHTTPRequestHandler

    def create_transport_readonly_server(self):

        return http_server.HttpServer(self._req_handler_class,

                                      protocol_version=self._protocol_version)

    def _testing_pycurl(self):

        return pycurl_present 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):

        server = self.get_readonly_server()

        t = self._transport(server.get_url())

        # 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.

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

                            errors.InvalidHttpResponse),

                          t.has, 'foo/bar')

    def test_http_get(self):

        server = self.get_readonly_server()

        t = self._transport(server.get_url())

        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 test_http_has(self):

        server = self.get_readonly_server()

        t = self._transport(server.get_url())

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

    def test_http_get(self):

        server = self.get_readonly_server()

        t = self._transport(server.get_url())

        self.assertRaises(errors.InvalidHttpResponse, t.get, 'foo/bar')

class InvalidStatusRequestHandler(http_server.TestingHTTPRequestHandler):

    """Whatever request comes in, returns an invalid status"""

    def parse_request(self):

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

        ignored = http_server.TestingHTTPRequestHandler.parse_request(self)

        self.wfile.write("Invalid status line\r\n")

        return False

class TestInvalidStatusServer(TestBadStatusServer):

    """Tests invalid status from server.

    Both implementations raises the same error as for a bad status.

    """

    _req_handler_class = InvalidStatusRequestHandler

    def test_http_has(self):

        if self._testing_pycurl() and self._protocol_version == 'HTTP/1.1':

            raise tests.KnownFailure(

                'pycurl hangs if the server send back garbage')

        super(TestInvalidStatusServer, self).test_http_has()

    def test_http_get(self):