/brz/remove-bazaar

# Copyright (C) 2005, 2006 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., 59 Temple Place, Suite 330, Boston, MA  02111-1307  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.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

class TransportAdapter(tests.TestScenarioApplier):

    """Generate the same test for each transport implementation."""

    def __init__(self):

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

        self.scenarios = transport_scenarios

class TransportProtocolAdapter(TransportAdapter):

    """Generate the same test for each protocol implementation.

    In addition to the transport adaptatation that we inherit from.

    """

    def __init__(self):

        super(TransportProtocolAdapter, self).__init__()

        protocol_scenarios = [

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

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

            ]

        self.scenarios = tests.multiply_scenarios(self.scenarios,

                                                  protocol_scenarios)

class TransportProtocolAuthenticationAdapter(TransportProtocolAdapter):

    """Generate the same test for each authentication scheme implementation.

    In addition to the protocol adaptatation that we inherit from.

    """

    def __init__(self):

        super(TransportProtocolAuthenticationAdapter, self).__init__()

        auth_scheme_scenarios = [

            ('basic', dict(_auth_scheme='basic')),

            ('digest', dict(_auth_scheme='digest')),

            ]

        self.scenarios = tests.multiply_scenarios(self.scenarios,

                                                  auth_scheme_scenarios)

def load_tests(standard_tests, module, loader):

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

    # one for each transport

    t_adapter = TransportAdapter()

    t_classes= (TestHttpTransportRegistration,

                TestHttpTransportUrls,

                )

    is_testing_for_transports = tests.condition_isinstance(t_classes)

    # multiplied by one for each protocol version

    tp_adapter = TransportProtocolAdapter()

    tp_classes= (SmartHTTPTunnellingTest,

                 TestDoCatchRedirections,

                 TestHTTPConnections,

                 TestHTTPRedirections,

                 TestHTTPSilentRedirections,

                 TestLimitedRangeRequestServer,

                 TestPost,

                 TestProxyHttpServer,

                 TestRanges,

                 TestSpecificRequestHandler,

                 )

    is_also_testing_for_protocols = tests.condition_isinstance(tp_classes)

    # multiplied by one for each authentication scheme

    tpa_adapter = TransportProtocolAuthenticationAdapter()

    tpa_classes = (TestAuth,

                   )

    is_also_testing_for_authentication = tests.condition_isinstance(

        tpa_classes)

    result = loader.suiteClass()

    for test_class in tests.iter_suite_tests(standard_tests):

        # Each test class is either standalone or testing for some combination

        # of transport, protocol version, authentication scheme. Use the right

        # adpater (or none) depending on the class.

        if is_testing_for_transports(test_class):

            result.addTests(t_adapter.adapt(test_class))

        elif is_also_testing_for_protocols(test_class):

            result.addTests(tp_adapter.adapt(test_class))

        elif is_also_testing_for_authentication(test_class):

            result.addTests(tpa_adapter.adapt(test_class))

        else:

            result.addTest(test_class)

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

        self.assertTrue(server._http_running)

        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)

        server.setUp()

        self.addCleanup(server.tearDown)

        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)

        server.setUp()

        self.addCleanup(server.tearDown)

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

        server.setUp()

        self.addCleanup(server.tearDown)

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

        server.setUp()

        self.addCleanup(server.tearDown)

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

        try:

            server.setUp()

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

        # For HTTP, get_smart_medium should return the transport object.

        server = self.get_readonly_server()

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

        medium = http_transport.get_smart_medium()

        self.assertIs(medium, http_transport)

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

        server.setUp()

        self.addCleanup(server.tearDown)

        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.

        self.assertRaises((errors.ConnectionError, 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.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):

        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_get()

class BadProtocolRequestHandler(http_server.TestingHTTPRequestHandler):

    """Whatever request comes in, returns a bad protocol version"""

    def parse_request(self):

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

        ignored = http_server.TestingHTTPRequestHandler.parse_request(self)

        # Returns an invalid protocol version, but curl just

        # ignores it and those cannot be tested.

        self.wfile.write("%s %d %s\r\n" % ('HTTP/0.0',

                                           404,

                                           'Look at my protocol version'))

        return False

class TestBadProtocolServer(TestSpecificRequestHandler):

    """Tests bad protocol from server."""

    _req_handler_class = BadProtocolRequestHandler

    def setUp(self):

        if pycurl_present and self._transport == PyCurlTransport:

            raise tests.TestNotApplicable(

                "pycurl doesn't check the protocol version")

        super(TestBadProtocolServer, self).setUp()

    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 ForbiddenRequestHandler(http_server.TestingHTTPRequestHandler):

    """Whatever request comes in, returns a 403 code"""

    def parse_request(self):

        """Handle a single HTTP request, by replying we cannot handle it"""

        ignored = http_server.TestingHTTPRequestHandler.parse_request(self)

        self.send_error(403)

        return False

class TestForbiddenServer(TestSpecificRequestHandler):

    """Tests forbidden server"""

    _req_handler_class = ForbiddenRequestHandler

    def test_http_has(self):

        server = self.get_readonly_server()

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

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

    def test_http_get(self):

        server = self.get_readonly_server()

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

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

class TestRecordingServer(tests.TestCase):

    def test_create(self):

        server = RecordingServer(expect_body_tail=None)

        self.assertEqual('', server.received_bytes)

        self.assertEqual(None, server.host)

        self.assertEqual(None, server.port)

    def test_setUp_and_tearDown(self):

        server = RecordingServer(expect_body_tail=None)

        server.setUp()

        try:

            self.assertNotEqual(None, server.host)

            self.assertNotEqual(None, server.port)

        finally:

            server.tearDown()

        self.assertEqual(None, server.host)

        self.assertEqual(None, server.port)