~bzr-pqm/bzr/bzr.dev

# Copyright (C) 2005, 2006 by 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 from HTTP response parsing."""

from cStringIO import StringIO
import mimetools

from bzrlib import errors
from bzrlib.transport import http
from bzrlib.transport.http import response
from bzrlib.tests import TestCase


class TestResponseRange(TestCase):
    """Test the ResponseRange class."""

    def test_cmp(self):
        RR = response.ResponseRange
        r1 = RR(0, 10, 0)
        r2 = RR(15, 20, 10)
        self.assertTrue(r1 < r2)
        self.assertFalse(r1 > r2)
        self.assertTrue(r1 < 5)
        self.assertFalse(r2 < 5)

        self.assertEqual(RR(0, 10, 5), RR(0, 10, 5))
        self.assertNotEqual(RR(0, 10, 5), RR(0, 8, 5))
        self.assertNotEqual(RR(0, 10, 5), RR(0, 10, 6))

    def test_sort_list(self):
        """Ensure longer ranges are sorted after shorter ones"""
        RR = response.ResponseRange
        lst = [RR(3, 8, 0), 5, RR(3, 7, 0), 6]
        lst.sort()
        self.assertEqual([RR(3,7,0), RR(3,8,0), 5, 6], lst)


class TestRangeFile(TestCase):
    """Test RangeFile."""

    def setUp(self):
        content = "abcdefghijklmnopqrstuvwxyz"
        self.fp = response.RangeFile('foo', StringIO(content))
        self.fp._add_range(0,  9,   0)
        self.fp._add_range(20, 29, 10)
        self.fp._add_range(30, 39, 15)

    def test_valid_accesses(self):
        """Test so that valid accesses work to the file."""
        self.fp.seek(0, 0)
        self.assertEquals(self.fp.read(3), 'abc')
        self.assertEquals(self.fp.read(3), 'def')
        self.assertEquals(self.fp.tell(), 6)
        self.fp.seek(20, 0)
        self.assertEquals(self.fp.read(3), 'klm')
        self.assertEquals(self.fp.read(2), 'no')
        self.assertEquals(self.fp.tell(), 25)
        # should wrap over to 30-39 entity
        self.assertEquals(self.fp.read(3), 'pqr')
        self.fp.seek(3)
        self.assertEquals(self.fp.read(3), 'def')
        self.assertEquals(self.fp.tell(), 6)

    def test_invalid_accesses(self):
        """Test so that invalid accesses trigger errors."""
        self.fp.seek(9)
        self.assertRaises(errors.InvalidRange, self.fp.read, 2)
        self.fp.seek(39)
        self.assertRaises(errors.InvalidRange, self.fp.read, 2)
        self.fp.seek(19)
        self.assertRaises(errors.InvalidRange, self.fp.read, 2)

    def test__finish_ranges(self):
        """Test that after RangeFile._finish_ranges the list is sorted."""
        self.fp._add_range(1, 2, 3)
        self.fp._add_range(8, 9, 10)
        self.fp._add_range(3, 4, 5)

        # TODO: jam 20060706 If we switch to inserting
        #       in sorted order, remove this test
        self.assertNotEqual(self.fp._ranges, sorted(self.fp._ranges))

        self.fp._finish_ranges()
        self.assertEqual(self.fp._ranges, sorted(self.fp._ranges))

    def test_seek_and_tell(self):
        # Check for seeking before start
        self.fp.seek(-2, 0)
        self.assertEqual(0, self.fp.tell())

        self.fp.seek(5, 0)
        self.assertEqual(5, self.fp.tell())

        self.fp.seek(-2, 1)
        self.assertEqual(3, self.fp.tell())

        # TODO: jam 20060706 following tests will fail if this 
        #       is not true, and would be difficult to debug
        #       but it is a layering violation
        self.assertEqual(39, self.fp._len)

        self.fp.seek(0, 2)
        self.assertEqual(39, self.fp.tell())

        self.fp.seek(-10, 2)
        self.assertEqual(29, self.fp.tell())

        self.assertRaises(ValueError, self.fp.seek, 0, 4)
        self.assertRaises(ValueError, self.fp.seek, 0, -1)


class TestRegexes(TestCase):

    def assertRegexMatches(self, groups, text):
        """Check that the regex matches and returns the right values"""
        m = self.regex.match(text)
        self.assertNotEqual(None, m, "text %s did not match regex" % (text,))

        self.assertEqual(groups, m.groups())

    def test_range_re(self):
        """Test that we match valid ranges."""
        self.regex = response.HttpRangeResponse._CONTENT_RANGE_RE
        self.assertRegexMatches(('bytes', '1', '10', '11'),
                           'bytes 1-10/11')
        self.assertRegexMatches(('bytes', '1', '10', '11'),
                           '\tbytes  1-10/11   ')
        self.assertRegexMatches(('bytes', '2123', '4242', '1231'),
                           '\tbytes  2123-4242/1231   ')
        self.assertRegexMatches(('chars', '1', '2', '3'),
                           ' chars 1-2/3')

    def test_content_type_re(self):
        self.regex = response.HttpMultipartRangeResponse._CONTENT_TYPE_RE
        self.assertRegexMatches(('xxyyzz',),
                                'multipart/byteranges; boundary = xxyyzz')
        self.assertRegexMatches(('xxyyzz',),
                                'multipart/byteranges;boundary=xxyyzz')
        self.assertRegexMatches(('xx yy zz',),
                                ' multipart/byteranges ; boundary= xx yy zz ')
        self.assertEqual(None,
                self.regex.match('multipart byteranges;boundary=xx'))


simple_data = """
--xxyyzz\r
foo\r
Content-range: bytes 1-10/20\r
\r
1234567890
--xxyyzz\r
Content-Range: bytes 21-30/20\r
bar\r
\r
abcdefghij
--xxyyzz\r
content-range: bytes 41-50/20\r
\r
zyxwvutsrq
--xxyyzz\r
content-range: bytes 51-60/20\r
\r
xxyyzz fbd
"""


class TestHelpers(TestCase):
    """Test the helper functions"""

    def test__parse_range(self):
        """Test that _parse_range acts reasonably."""
        content = StringIO('')
        parse_range = response.HttpRangeResponse._parse_range
        self.assertEqual((1,2), parse_range('bytes 1-2/3'))
        self.assertEqual((10,20), parse_range('bytes 10-20/2'))

        self.assertRaises(errors.InvalidHttpRange, parse_range, 'char 1-3/2')
        self.assertRaises(errors.InvalidHttpRange, parse_range, 'bytes a-3/2')

        try:
            parse_range('bytes x-10/3', path='http://foo/bar')
        except errors.InvalidHttpRange, e:
            self.assertContainsRe(str(e), 'http://foo/bar')
            self.assertContainsRe(str(e), 'bytes x-10/3')
        else:
            self.fail('Did not raise InvalidHttpRange')

    def test__parse_boundary_simple(self):
        """Test that _parse_boundary handles Content-type properly"""
        parse_boundary = response.HttpMultipartRangeResponse._parse_boundary
        m = parse_boundary(' multipart/byteranges; boundary=xxyyzz')
        self.assertNotEqual(None, m)
        # Check that the returned regex is capable of splitting simple_data
        matches = list(m.finditer(simple_data))
        self.assertEqual(4, len(matches))

        # match.group() should be the content-range entry
        # and match.end() should be the start of the content
        self.assertEqual(' bytes 1-10/20', matches[0].group(1))
        self.assertEqual(simple_data.find('1234567890'), matches[0].end())
        self.assertEqual(' bytes 21-30/20', matches[1].group(1))
        self.assertEqual(simple_data.find('abcdefghij'), matches[1].end())
        self.assertEqual(' bytes 41-50/20', matches[2].group(1))
        self.assertEqual(simple_data.find('zyxwvutsrq'), matches[2].end())
        self.assertEqual(' bytes 51-60/20', matches[3].group(1))
        self.assertEqual(simple_data.find('xxyyzz fbd'), matches[3].end())

    def test__parse_boundary_invalid(self):
        parse_boundary = response.HttpMultipartRangeResponse._parse_boundary
        try:
            parse_boundary(' multipart/bytes;boundary=xxyyzz',
                           path='http://foo/bar')
        except errors.InvalidHttpContentType, e:
            self.assertContainsRe(str(e), 'http://foo/bar')
            self.assertContainsRe(str(e), 'multipart/bytes;boundary=xxyyzz')
        else:
            self.fail('Did not raise InvalidHttpContentType')


class TestHttpRangeResponse(TestCase):

    def test_smoketest(self):
        """A basic test that HttpRangeResponse is reasonable."""
        content = StringIO('0123456789')
        f = response.HttpRangeResponse('http://foo', 'bytes 1-10/9', content)
        self.assertEqual([response.ResponseRange(1,10,0)], f._ranges)

        f.seek(0)
        self.assertRaises(errors.InvalidRange, f.read, 2)
        f.seek(1)
        self.assertEqual('012345', f.read(6))

    def test_invalid(self):
        try:
            f = response.HttpRangeResponse('http://foo', 'bytes x-10/9',
                                           StringIO('0123456789'))
        except errors.InvalidHttpRange, e:
            self.assertContainsRe(str(e), 'http://foo')
            self.assertContainsRe(str(e), 'bytes x-10/9')
        else:
            self.fail('Failed to raise InvalidHttpRange')


class TestHttpMultipartRangeResponse(TestCase):
    """Test the handling of multipart range responses"""

    def test_simple(self):
        content = StringIO(simple_data)
        multi = response.HttpMultipartRangeResponse('http://foo',
                    'multipart/byteranges; boundary = xxyyzz', content)

        self.assertEqual(4, len(multi._ranges))

        multi.seek(1)
        self.assertEqual('1234567890', multi.read(10))
        multi.seek(21)
        self.assertEqual('abcdefghij', multi.read(10))
        multi.seek(41)
        self.assertEqual('zyxwvutsrq', multi.read(10))
        multi.seek(51)
        self.assertEqual('xxyyzz fbd', multi.read(10))
        # TODO: jam 20060706 Currently RangeFile does not support
        #       reading across ranges. Consider adding it.
        multi.seek(41)
        # self.assertEqual('zyxwvutsrqxxyyzz fbd', multi.read(20))
        self.assertRaises(errors.InvalidRange, multi.read, 20)

        multi.seek(21)
        self.assertRaises(errors.InvalidRange, multi.read, 11)
        multi.seek(31)
        self.assertRaises(errors.InvalidRange, multi.read, 10)

    def test_invalid(self):
        content = StringIO('')
        try:
            response.HttpMultipartRangeResponse('http://foo',
                        'multipart/byte;boundary=invalid', content)
        except errors.InvalidHttpContentType, e:
            self.assertContainsRe(str(e), 'http://foo')
            self.assertContainsRe(str(e), 'multipart/byte;')


# Taken from real request responses
_full_text_response = (200, """HTTP/1.1 200 OK\r
Date: Tue, 11 Jul 2006 04:32:56 GMT\r
Server: Apache/2.0.54 (Fedora)\r
Last-Modified: Sun, 23 Apr 2006 19:35:20 GMT\r
ETag: "56691-23-38e9ae00"\r
Accept-Ranges: bytes\r
Content-Length: 35\r
Connection: close\r
Content-Type: text/plain; charset=UTF-8\r
\r
""", """Bazaar-NG meta directory, format 1
""")


_missing_response = (404, """HTTP/1.1 404 Not Found\r
Date: Tue, 11 Jul 2006 04:32:56 GMT\r
Server: Apache/2.0.54 (Fedora)\r
Content-Length: 336\r
Connection: close\r
Content-Type: text/html; charset=iso-8859-1\r
\r
""", """<!DOCTYPE HTML PUBLIC "-//IETF//DTD HTML 2.0//EN">
<html><head>
<title>404 Not Found</title>
</head><body>
<h1>Not Found</h1>
<p>The requested URL /branches/bzr/jam-integration/.bzr/repository/format was not found on this server.</p>
<hr>
<address>Apache/2.0.54 (Fedora) Server at bzr.arbash-meinel.com Port 80</address>
</body></html>
""")


_single_range_response = (206, """HTTP/1.1 206 Partial Content\r
Date: Tue, 11 Jul 2006 04:45:22 GMT\r
Server: Apache/2.0.54 (Fedora)\r
Last-Modified: Thu, 06 Jul 2006 20:22:05 GMT\r
ETag: "238a3c-16ec2-805c5540"\r
Accept-Ranges: bytes\r
Content-Length: 100\r
Content-Range: bytes 100-199/93890\r
Connection: close\r
Content-Type: text/plain; charset=UTF-8\r
\r
""", """mbp@sourcefrog.net-20050309040815-13242001617e4a06
mbp@sourcefrog.net-20050309040929-eee0eb3e6d1e762""")


_multipart_range_response = (206, """HTTP/1.1 206 Partial Content\r
Date: Tue, 11 Jul 2006 04:49:48 GMT\r
Server: Apache/2.0.54 (Fedora)\r
Last-Modified: Thu, 06 Jul 2006 20:22:05 GMT\r
ETag: "238a3c-16ec2-805c5540"\r
Accept-Ranges: bytes\r
Content-Length: 1534\r
Connection: close\r
Content-Type: multipart/byteranges; boundary=418470f848b63279b\r
\r
\r""", """--418470f848b63279b\r
Content-type: text/plain; charset=UTF-8\r
Content-range: bytes 0-254/93890\r
\r
mbp@sourcefrog.net-20050309040815-13242001617e4a06
mbp@sourcefrog.net-20050309040929-eee0eb3e6d1e7627
mbp@sourcefrog.net-20050309040957-6cad07f466bb0bb8
mbp@sourcefrog.net-20050309041501-c840e09071de3b67
mbp@sourcefrog.net-20050309044615-c24a3250be83220a
\r
--418470f848b63279b\r
Content-type: text/plain; charset=UTF-8\r
Content-range: bytes 1000-2049/93890\r
\r
40-fd4ec249b6b139ab
mbp@sourcefrog.net-20050311063625-07858525021f270b
mbp@sourcefrog.net-20050311231934-aa3776aff5200bb9
mbp@sourcefrog.net-20050311231953-73aeb3a131c3699a
mbp@sourcefrog.net-20050311232353-f5e33da490872c6a
mbp@sourcefrog.net-20050312071639-0a8f59a34a024ff0
mbp@sourcefrog.net-20050312073432-b2c16a55e0d6e9fb
mbp@sourcefrog.net-20050312073831-a47c3335ece1920f
mbp@sourcefrog.net-20050312085412-13373aa129ccbad3
mbp@sourcefrog.net-20050313052251-2bf004cb96b39933
mbp@sourcefrog.net-20050313052856-3edd84094687cb11
mbp@sourcefrog.net-20050313053233-e30a4f28aef48f9d
mbp@sourcefrog.net-20050313053853-7c64085594ff3072
mbp@sourcefrog.net-20050313054757-a86c3f5871069e22
mbp@sourcefrog.net-20050313061422-418f1f73b94879b9
mbp@sourcefrog.net-20050313120651-497bd231b19df600
mbp@sourcefrog.net-20050314024931-eae0170ef25a5d1a
mbp@sourcefrog.net-20050314025438-d52099f915fe65fc
mbp@sourcefrog.net-20050314025539-637a636692c055cf
mbp@sourcefrog.net-20050314025737-55eb441f430ab4ba
mbp@sourcefrog.net-20050314025901-d74aa93bb7ee8f62
mbp@source\r
--418470f848b63279b--\r\n'
""")


_redirect_response = (206, """HTTP/1.1 301 Moved Permanently\r
Date: Tue, 18 Jul 2006 20:29:22 GMT\r
Server: Apache/2.0.54 (Ubuntu) PHP/4.4.0-3ubuntu1 mod_ssl/2.0.54 OpenSSL/0.9.7g\r
Location: http://bazaar-vcs.org/bzr/bzr.dev/.bzr/repository/inventory.knit\r
Content-Length: 272\r
Keep-Alive: timeout=15, max=100\r
Connection: Keep-Alive\r
Content-Type: text/html; charset=iso-8859-1\r
\r
HTTP/1.1 206 Partial Content\r
Date: Tue, 18 Jul 2006 20:29:23 GMT\r
Server: Apache/2.0.54 (Ubuntu) PHP/4.4.0-3ubuntu1 mod_ssl/2.0.54 OpenSSL/0.9.7g\r
Last-Modified: Tue, 18 Jul 2006 20:24:59 GMT\r
ETag: "be8213-83958c-f0d3dcc0"\r
Accept-Ranges: bytes\r
Content-Length: 425\r
Content-Range: bytes 8623075-8623499/8623500\r
Keep-Alive: timeout=15, max=100\r
Connection: Keep-Alive\r
Content-Type: text/plain; charset=UTF-8\r
\r
""", """this data intentionally removed, 
this is not meant to be tested by
handle_response, just _extract_headers
""")


# This is made up
_invalid_response = (444, """HTTP/1.1 444 Bad Response\r
Date: Tue, 11 Jul 2006 04:32:56 GMT\r
Connection: close\r
Content-Type: text/html; charset=iso-8859-1\r
\r
""", """<!DOCTYPE HTML PUBLIC "-//IETF//DTD HTML 2.0//EN">
<html><head>
<title>404 Not Found</title>
</head><body>
<h1>Not Found</h1>
<p>I don't know what I'm doing</p>
<hr>
</body></html>
""")


# This should be in test_http.py, but the headers we
# want to parse are here
class TestExtractHeader(TestCase):
    
    def use_response(self, response):
        self.headers = http._extract_headers(response[1], 'http://foo')

    def check_header(self, header, value):
        self.assertEqual(value, self.headers[header])
        
    def test_full_text(self):
        self.use_response(_full_text_response)

        self.check_header('Date', 'Tue, 11 Jul 2006 04:32:56 GMT')
        self.check_header('date', 'Tue, 11 Jul 2006 04:32:56 GMT')
        self.check_header('Content-Length', '35')
        self.check_header('Content-Type', 'text/plain; charset=UTF-8')
        self.check_header('content-type', 'text/plain; charset=UTF-8')

    def test_missing_response(self):
        self.use_response(_missing_response)

        self.check_header('Content-Length', '336')
        self.check_header('Content-Type', 'text/html; charset=iso-8859-1')

    def test_single_range(self):
        self.use_response(_single_range_response)

        self.check_header('Content-Length', '100')
        self.check_header('Content-Range', 'bytes 100-199/93890')
        self.check_header('Content-Type', 'text/plain; charset=UTF-8')

    def test_multi_range(self):
        self.use_response(_multipart_range_response)

        self.check_header('Content-Length', '1534')
        self.check_header('Content-Type',
                          'multipart/byteranges; boundary=418470f848b63279b')

    def test_redirect(self):
        """We default to returning the last group of headers in the file."""
        self.use_response(_redirect_response)
        self.check_header('Content-Range', 'bytes 8623075-8623499/8623500')
        self.check_header('Content-Type', 'text/plain; charset=UTF-8')

    def test_empty(self):
        self.assertRaises(errors.InvalidHttpResponse,
            http._extract_headers, '', 'bad url')

    def test_no_opening_http(self):
        # Remove the HTTP line from the header
        first, txt = _full_text_response[1].split('\r\n', 1)
        self.assertRaises(errors.InvalidHttpResponse,
            http._extract_headers, txt, 'missing HTTTP')

    def test_trailing_whitespace(self):
        # Test that we ignore bogus whitespace on the end
        code, txt, body = _full_text_response
        txt += '\r\n\n\n\n\n'
        self.use_response((code, txt, body))

        self.check_header('Date', 'Tue, 11 Jul 2006 04:32:56 GMT')
        self.check_header('Content-Length', '35')
        self.check_header('Content-Type', 'text/plain; charset=UTF-8')

    def test_trailing_non_http(self):
        # Test that we ignore bogus stuff on the end
        code, txt, body = _full_text_response
        txt = txt + 'Foo: Bar\r\nBaz: Bling\r\n\r\n'
        self.use_response((code, txt, body))

        self.check_header('Date', 'Tue, 11 Jul 2006 04:32:56 GMT')
        self.check_header('Content-Length', '35')
        self.check_header('Content-Type', 'text/plain; charset=UTF-8')
        self.assertRaises(KeyError, self.headers.__getitem__, 'Foo')

    def test_extra_whitespace(self):
        # Test that we read an HTTP response, even with extra whitespace
        code, txt, body = _redirect_response
        # Find the second HTTP location
        loc = txt.find('HTTP', 5)
        txt = txt[:loc] + '\r\n\n' + txt[loc:]
        self.use_response((code, txt, body))
        self.check_header('Content-Range', 'bytes 8623075-8623499/8623500')
        self.check_header('Content-Type', 'text/plain; charset=UTF-8')


class TestHandleResponse(TestCase):
    
    def get_response(self, a_response):
        """Process a supplied response, and return the result."""
        headers = http._extract_headers(a_response[1], 'http://foo')
        return response.handle_response('http://foo', a_response[0], headers,
                                        StringIO(a_response[2]))

    def test_full_text(self):
        out = self.get_response(_full_text_response)
        # It is a StringIO from the original data
        self.assertEqual(_full_text_response[2], out.read())

    def test_missing_response(self):
        self.assertRaises(errors.NoSuchFile,
            self.get_response, _missing_response)

    def test_single_range(self):
        out = self.get_response(_single_range_response)
        self.assertIsInstance(out, response.HttpRangeResponse)

        self.assertRaises(errors.InvalidRange, out.read, 20)

        out.seek(100)
        self.assertEqual(_single_range_response[2], out.read(100))

    def test_multi_range(self):
        out = self.get_response(_multipart_range_response)
        self.assertIsInstance(out, response.HttpMultipartRangeResponse)

        # Just make sure we can read the right contents
        out.seek(0)
        out.read(255)

        out.seek(1000)
        out.read(1050)

    def test_invalid_response(self):
        self.assertRaises(errors.InvalidHttpResponse,
            self.get_response, _invalid_response)

    def test_full_text_no_content_type(self):
        # We should not require Content-Type for a full response
        a_response = _full_text_response
        headers = http._extract_headers(a_response[1], 'http://foo')
        del headers['Content-Type']
        out = response.handle_response('http://foo', a_response[0], headers,
                                        StringIO(a_response[2]))
        self.assertEqual(_full_text_response[2], out.read())

    def test_missing_no_content_type(self):
        # Without Content-Type we should still raise NoSuchFile on a 404
        a_response = _missing_response
        headers = http._extract_headers(a_response[1], 'http://missing')
        del headers['Content-Type']
        self.assertRaises(errors.NoSuchFile,
            response.handle_response, 'http://missing', a_response[0], headers,
                                      StringIO(a_response[2]))

    def test_missing_content_type(self):
        a_response = _single_range_response
        headers = http._extract_headers(a_response[1], 'http://nocontent')
        del headers['Content-Type']
        self.assertRaises(errors.InvalidHttpContentType,
            response.handle_response, 'http://nocontent', a_response[0],
                                      headers, StringIO(a_response[2]))

    def test_missing_content_range(self):
        a_response = _single_range_response
        headers = http._extract_headers(a_response[1], 'http://nocontent')
        del headers['Content-Range']
        self.assertRaises(errors.InvalidHttpResponse,
            response.handle_response, 'http://nocontent', a_response[0],
                                      headers, StringIO(a_response[2]))