/brz/remove-bazaar

# Copyright (C) 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 group compression."""

import zlib

from bzrlib import (

    groupcompress,

    errors,

    osutils,

    tests,

    versionedfile,

    )

from bzrlib.osutils import sha_string

from bzrlib.tests.test__groupcompress import CompiledGroupCompressFeature

def load_tests(standard_tests, module, loader):

    """Parameterize tests for all versions of groupcompress."""

    to_adapt, result = tests.split_suite_by_condition(

        standard_tests, tests.condition_isinstance(TestAllGroupCompressors))

    scenarios = [

        ('python', {'compressor': groupcompress.PythonGroupCompressor}),

        ]

    if CompiledGroupCompressFeature.available():

        scenarios.append(('C',

            {'compressor': groupcompress.PyrexGroupCompressor}))

    return tests.multiply_tests(to_adapt, scenarios, result)

class TestGroupCompressor(tests.TestCase):

    def _chunks_to_repr_lines(self, chunks):

        return '\n'.join(map(repr, ''.join(chunks).split('\n')))

    def assertEqualDiffEncoded(self, expected, actual):

        """Compare the actual content to the expected content.

        :param expected: A group of chunks that we expect to see

        :param actual: The measured 'chunks'

        We will transform the chunks back into lines, and then run 'repr()'

        over them to handle non-ascii characters.

        """

        self.assertEqualDiff(self._chunks_to_repr_lines(expected),

                             self._chunks_to_repr_lines(actual))

class TestAllGroupCompressors(TestGroupCompressor):

    """Tests for GroupCompressor"""

    compressor = None # Set by multiply_tests

    def test_empty_delta(self):

        compressor = self.compressor()

        self.assertEqual([], compressor.chunks)

    def test_one_nosha_delta(self):

        # diff against NUKK

        compressor = self.compressor()

        sha1, start_point, end_point, _ = compressor.compress(('label',),

            'strange\ncommon\n', None)

        self.assertEqual(sha_string('strange\ncommon\n'), sha1)

        expected_lines = 'f' '\x0f' 'strange\ncommon\n'

        self.assertEqual(expected_lines, ''.join(compressor.chunks))

        self.assertEqual(0, start_point)

        self.assertEqual(sum(map(len, expected_lines)), end_point)

    def test_empty_content(self):

        compressor = self.compressor()

        # Adding empty bytes should return the 'null' record

        sha1, start_point, end_point, kind = compressor.compress(('empty',),

                                                                 '', None)

        self.assertEqual(0, start_point)

        self.assertEqual(0, end_point)

        self.assertEqual('fulltext', kind)

        self.assertEqual(groupcompress._null_sha1, sha1)

        self.assertEqual(0, compressor.endpoint)

        self.assertEqual([], compressor.chunks)

        # Even after adding some content

        compressor.compress(('content',), 'some\nbytes\n', None)

        self.assertTrue(compressor.endpoint > 0)

        sha1, start_point, end_point, kind = compressor.compress(('empty2',),

                                                                 '', None)

        self.assertEqual(0, start_point)

        self.assertEqual(0, end_point)

        self.assertEqual('fulltext', kind)

        self.assertEqual(groupcompress._null_sha1, sha1)

    def test_extract_from_compressor(self):

        # Knit fetching will try to reconstruct texts locally which results in

        # reading something that is in the compressor stream already.

        compressor = self.compressor()

        sha1_1, _, _, _ = compressor.compress(('label',),

            'strange\ncommon long line\nthat needs a 16 byte match\n', None)

        expected_lines = list(compressor.chunks)

        sha1_2, _, end_point, _ = compressor.compress(('newlabel',),

            'common long line\nthat needs a 16 byte match\ndifferent\n', None)

        # get the first out

        self.assertEqual(('strange\ncommon long line\n'

                          'that needs a 16 byte match\n', sha1_1),

                         compressor.extract(('label',)))

        # and the second

        self.assertEqual(('common long line\nthat needs a 16 byte match\n'

                          'different\n', sha1_2),

                         compressor.extract(('newlabel',)))

class TestPyrexGroupCompressor(TestGroupCompressor):

    _test_needs_features = [CompiledGroupCompressFeature]

    compressor = groupcompress.PyrexGroupCompressor

    def test_stats(self):

        compressor = self.compressor()

        compressor.compress(('label',),

                            'strange\n'

                            'common very very long line\n'

                            'plus more text\n', None)

        compressor.compress(('newlabel',),

                            'common very very long line\n'

                            'plus more text\n'

                            'different\n'

                            'moredifferent\n', None)

        compressor.compress(('label3',),

                            'new\n'

                            'common very very long line\n'

                            'plus more text\n'

                            'different\n'

                            'moredifferent\n', None)

        self.assertAlmostEqual(1.9, compressor.ratio(), 1)

    def test_two_nosha_delta(self):

        compressor = self.compressor()

        sha1_1, _, _, _ = compressor.compress(('label',),

            'strange\ncommon long line\nthat needs a 16 byte match\n', None)

        expected_lines = list(compressor.chunks)

        sha1_2, start_point, end_point, _ = compressor.compress(('newlabel',),

            'common long line\nthat needs a 16 byte match\ndifferent\n', None)

        self.assertEqual(sha_string('common long line\n'

                                    'that needs a 16 byte match\n'

                                    'different\n'), sha1_2)

        expected_lines.extend([

            # 'delta', delta length

            'd\x0f',

            # source and target length

            '\x36',

            # copy the line common

            '\x91\x0a\x2c', #copy, offset 0x0a, len 0x2c

            # add the line different, and the trailing newline

            '\x0adifferent\n', # insert 10 bytes

            ])

        self.assertEqualDiffEncoded(expected_lines, compressor.chunks)

        self.assertEqual(sum(map(len, expected_lines)), end_point)

    def test_three_nosha_delta(self):

        # The first interesting test: make a change that should use lines from

        # both parents.

        compressor = self.compressor()

        sha1_1, _, _, _ = compressor.compress(('label',),

            'strange\ncommon very very long line\nwith some extra text\n', None)

        sha1_2, _, _, _ = compressor.compress(('newlabel',),

            'different\nmoredifferent\nand then some more\n', None)

        expected_lines = list(compressor.chunks)

        sha1_3, start_point, end_point, _ = compressor.compress(('label3',),

            'new\ncommon very very long line\nwith some extra text\n'

            'different\nmoredifferent\nand then some more\n',

            None)

        self.assertEqual(

            sha_string('new\ncommon very very long line\nwith some extra text\n'

                       'different\nmoredifferent\nand then some more\n'),

            sha1_3)

        expected_lines.extend([

            # 'delta', delta length

            'd\x0b',

            # source and target length

            '\x5f'

            # insert new

            '\x03new',

            # Copy of first parent 'common' range

            '\x91\x09\x31' # copy, offset 0x09, 0x31 bytes

            # Copy of second parent 'different' range

            '\x91\x3c\x2b' # copy, offset 0x3c, 0x2b bytes

            ])

        self.assertEqualDiffEncoded(expected_lines, compressor.chunks)

        self.assertEqual(sum(map(len, expected_lines)), end_point)

class TestPythonGroupCompressor(TestGroupCompressor):

    compressor = groupcompress.PythonGroupCompressor

    def test_stats(self):

        compressor = self.compressor()

        compressor.compress(('label',),

                            'strange\n'

                            'common very very long line\n'

                            'plus more text\n', None)

        compressor.compress(('newlabel',),

                            'common very very long line\n'

                            'plus more text\n'

                            'different\n'

                            'moredifferent\n', None)

        compressor.compress(('label3',),

                            'new\n'

                            'common very very long line\n'

                            'plus more text\n'

                            'different\n'

                            'moredifferent\n', None)

        self.assertAlmostEqual(1.9, compressor.ratio(), 1)

    def test_two_nosha_delta(self):

        compressor = self.compressor()

        sha1_1, _, _, _ = compressor.compress(('label',),

            'strange\ncommon long line\nthat needs a 16 byte match\n', None)

        expected_lines = list(compressor.chunks)

        sha1_2, start_point, end_point, _ = compressor.compress(('newlabel',),

            'common long line\nthat needs a 16 byte match\ndifferent\n', None)

        self.assertEqual(sha_string('common long line\n'

                                    'that needs a 16 byte match\n'

                                    'different\n'), sha1_2)

        expected_lines.extend([

            # 'delta', delta length

            'd\x0f',

            # target length

            '\x36',

            # copy the line common

            '\x91\x0a\x2c', #copy, offset 0x0a, len 0x2c

            # add the line different, and the trailing newline

            '\x0adifferent\n', # insert 10 bytes

            ])

        self.assertEqualDiffEncoded(expected_lines, compressor.chunks)

        self.assertEqual(sum(map(len, expected_lines)), end_point)

    def test_three_nosha_delta(self):

        # The first interesting test: make a change that should use lines from

        # both parents.

        compressor = self.compressor()

        sha1_1, _, _, _ = compressor.compress(('label',),

            'strange\ncommon very very long line\nwith some extra text\n', None)

        sha1_2, _, _, _ = compressor.compress(('newlabel',),

            'different\nmoredifferent\nand then some more\n', None)

        expected_lines = list(compressor.chunks)

        sha1_3, start_point, end_point, _ = compressor.compress(('label3',),

            'new\ncommon very very long line\nwith some extra text\n'

            'different\nmoredifferent\nand then some more\n',

            None)

        self.assertEqual(

            sha_string('new\ncommon very very long line\nwith some extra text\n'

                       'different\nmoredifferent\nand then some more\n'),

            sha1_3)

        expected_lines.extend([

            # 'delta', delta length

            'd\x0c',

            # target length

            '\x5f'

            # insert new

            '\x04new\n',

            # Copy of first parent 'common' range

            '\x91\x0a\x30' # copy, offset 0x0a, 0x30 bytes

            # Copy of second parent 'different' range

            '\x91\x3c\x2b' # copy, offset 0x3c, 0x2b bytes

            ])

        self.assertEqualDiffEncoded(expected_lines, compressor.chunks)

        self.assertEqual(sum(map(len, expected_lines)), end_point)

class TestGroupCompressBlock(tests.TestCase):

    def make_block(self, key_to_text):

        """Create a GroupCompressBlock, filling it with the given texts."""

        compressor = groupcompress.GroupCompressor()

        start = 0

        for key in sorted(key_to_text):

            compressor.compress(key, key_to_text[key], None)

        locs = dict((key, (start, end)) for key, (start, _, end, _)

                    in compressor.labels_deltas.iteritems())

        block = compressor.flush()

        raw_bytes = block.to_bytes()

        # Go through from_bytes(to_bytes()) so that we start with a compressed

        # content object

        return locs, groupcompress.GroupCompressBlock.from_bytes(raw_bytes)

    def test_from_empty_bytes(self):

        self.assertRaises(ValueError,

                          groupcompress.GroupCompressBlock.from_bytes, '')

    def test_from_minimal_bytes(self):

        block = groupcompress.GroupCompressBlock.from_bytes(

            'gcb1z\n0\n0\n')

        self.assertIsInstance(block, groupcompress.GroupCompressBlock)

        self.assertIs(None, block._content)

        self.assertEqual('', block._z_content)

        block._ensure_content()

        self.assertEqual('', block._content)

        self.assertEqual('', block._z_content)

        block._ensure_content() # Ensure content is safe to call 2x

    def test_from_invalid(self):

        self.assertRaises(ValueError,

                          groupcompress.GroupCompressBlock.from_bytes,

                          'this is not a valid header')

    def test_from_bytes(self):

        content = ('a tiny bit of content\n')

        z_content = zlib.compress(content)

        z_bytes = (

            'gcb1z\n' # group compress block v1 plain

            '%d\n' # Length of compressed content

            '%d\n' # Length of uncompressed content

            '%s'   # Compressed content

            ) % (len(z_content), len(content), z_content)

        block = groupcompress.GroupCompressBlock.from_bytes(

            z_bytes)

        self.assertEqual(z_content, block._z_content)

        self.assertIs(None, block._content)

        self.assertEqual(len(z_content), block._z_content_length)

        self.assertEqual(len(content), block._content_length)

        block._ensure_content()

        self.assertEqual(z_content, block._z_content)

        self.assertEqual(content, block._content)

    def test_to_bytes(self):

        content = ('this is some content\n'

                   'this content will be compressed\n')

        gcb = groupcompress.GroupCompressBlock()

        gcb.set_content(content)

        bytes = gcb.to_bytes()

        self.assertEqual(gcb._z_content_length, len(gcb._z_content))

        self.assertEqual(gcb._content_length, len(content))

        expected_header =('gcb1z\n' # group compress block v1 zlib

                          '%d\n' # Length of compressed content

                          '%d\n' # Length of uncompressed content

                         ) % (gcb._z_content_length, gcb._content_length)

        self.assertStartsWith(bytes, expected_header)

        remaining_bytes = bytes[len(expected_header):]

        raw_bytes = zlib.decompress(remaining_bytes)

        self.assertEqual(content, raw_bytes)

    def test_partial_decomp(self):

        content_chunks = []

        # We need a sufficient amount of data so that zlib.decompress has

        # partial decompression to work with. Most auto-generated data

        # compresses a bit too well, we want a combination, so we combine a sha

        # hash with compressible data.

        for i in xrange(2048):

            next_content = '%d\nThis is a bit of duplicate text\n' % (i,)

            content_chunks.append(next_content)

            next_sha1 = osutils.sha_string(next_content)

            content_chunks.append(next_sha1 + '\n')

        content = ''.join(content_chunks)

        self.assertEqual(158634, len(content))

        z_content = zlib.compress(content)

        self.assertEqual(57182, len(z_content))

        block = groupcompress.GroupCompressBlock()

        block._z_content = z_content

        block._z_content_length = len(z_content)

        block._compressor_name = 'zlib'

        block._content_length = 158634

        self.assertIs(None, block._content)

        block._ensure_content(100)

        self.assertIsNot(None, block._content)

        # We have decompressed at least 100 bytes

        self.assertTrue(len(block._content) >= 100)

        # We have not decompressed the whole content

        self.assertTrue(len(block._content) < 158634)

        self.assertEqualDiff(content[:len(block._content)], block._content)

        # ensuring content that we already have shouldn't cause any more data

        # to be extracted

        cur_len = len(block._content)

        block._ensure_content(cur_len - 10)

        self.assertEqual(cur_len, len(block._content))

        # Now we want a bit more content

        cur_len += 10

        block._ensure_content(cur_len)

        self.assertTrue(len(block._content) >= cur_len)

        self.assertTrue(len(block._content) < 158634)

        self.assertEqualDiff(content[:len(block._content)], block._content)

        # And now lets finish

        block._ensure_content(158634)

        self.assertEqualDiff(content, block._content)

        # And the decompressor is finalized

        self.assertIs(None, block._z_content_decompressor)

    def test_partial_decomp_no_known_length(self):

        content_chunks = []

        for i in xrange(2048):

            next_content = '%d\nThis is a bit of duplicate text\n' % (i,)

            content_chunks.append(next_content)

            next_sha1 = osutils.sha_string(next_content)

            content_chunks.append(next_sha1 + '\n')

        content = ''.join(content_chunks)

        self.assertEqual(158634, len(content))

        z_content = zlib.compress(content)

        self.assertEqual(57182, len(z_content))

        block = groupcompress.GroupCompressBlock()

        block._z_content = z_content

        block._z_content_length = len(z_content)

        block._compressor_name = 'zlib'

        block._content_length = None # Don't tell the decompressed length

        self.assertIs(None, block._content)

        block._ensure_content(100)

        self.assertIsNot(None, block._content)

        # We have decompressed at least 100 bytes

        self.assertTrue(len(block._content) >= 100)

        # We have not decompressed the whole content

        self.assertTrue(len(block._content) < 158634)

        self.assertEqualDiff(content[:len(block._content)], block._content)

        # ensuring content that we already have shouldn't cause any more data

        # to be extracted

        cur_len = len(block._content)

        block._ensure_content(cur_len - 10)

        self.assertEqual(cur_len, len(block._content))

        # Now we want a bit more content

        cur_len += 10

        block._ensure_content(cur_len)

        self.assertTrue(len(block._content) >= cur_len)

        self.assertTrue(len(block._content) < 158634)

        self.assertEqualDiff(content[:len(block._content)], block._content)

        # And now lets finish

        block._ensure_content()

        self.assertEqualDiff(content, block._content)

        # And the decompressor is finalized

        self.assertIs(None, block._z_content_decompressor)

class TestCaseWithGroupCompressVersionedFiles(tests.TestCaseWithTransport):

    def make_test_vf(self, create_graph, keylength=1, do_cleanup=True,

                     dir='.'):

        t = self.get_transport(dir)

        t.ensure_base()

        vf = groupcompress.make_pack_factory(graph=create_graph,

            delta=False, keylength=keylength)(t)

        if do_cleanup:

            self.addCleanup(groupcompress.cleanup_pack_group, vf)

        return vf

class TestGroupCompressVersionedFiles(TestCaseWithGroupCompressVersionedFiles):

    def test_get_record_stream_as_requested(self):

        # Consider promoting 'as-requested' to general availability, and

        # make this a VF interface test

        vf = self.make_test_vf(False, dir='source')

        vf.add_lines(('a',), (), ['lines\n'])

        vf.add_lines(('b',), (), ['lines\n'])

        vf.add_lines(('c',), (), ['lines\n'])

        vf.add_lines(('d',), (), ['lines\n'])

        vf.writer.end()

        keys = [record.key for record in vf.get_record_stream(

                    [('a',), ('b',), ('c',), ('d',)],

                    'as-requested', False)]

        self.assertEqual([('a',), ('b',), ('c',), ('d',)], keys)

        keys = [record.key for record in vf.get_record_stream(

                    [('b',), ('a',), ('d',), ('c',)],

                    'as-requested', False)]

        self.assertEqual([('b',), ('a',), ('d',), ('c',)], keys)

        # It should work even after being repacked into another VF

        vf2 = self.make_test_vf(False, dir='target')

        vf2.insert_record_stream(vf.get_record_stream(

                    [('b',), ('a',), ('d',), ('c',)], 'as-requested', False))

        vf2.writer.end()

        keys = [record.key for record in vf2.get_record_stream(

                    [('a',), ('b',), ('c',), ('d',)],

                    'as-requested', False)]

        self.assertEqual([('a',), ('b',), ('c',), ('d',)], keys)

        keys = [record.key for record in vf2.get_record_stream(

                    [('b',), ('a',), ('d',), ('c',)],

                    'as-requested', False)]

        self.assertEqual([('b',), ('a',), ('d',), ('c',)], keys)

    def test_insert_record_stream_re_uses_blocks(self):

        vf = self.make_test_vf(True, dir='source')

        def grouped_stream(revision_ids, first_parents=()):

            parents = first_parents

            for revision_id in revision_ids:

                key = (revision_id,)

                record = versionedfile.FulltextContentFactory(

                    key, parents, None,

                    'some content that is\n'

                    'identical except for\n'

                    'revision_id:%s\n' % (revision_id,))

                yield record

                parents = (key,)

        # One group, a-d

        vf.insert_record_stream(grouped_stream(['a', 'b', 'c', 'd']))

        # Second group, e-h

        vf.insert_record_stream(grouped_stream(['e', 'f', 'g', 'h'],

                                               first_parents=(('d',),)))

        block_bytes = {}

        stream = vf.get_record_stream([(r,) for r in 'abcdefgh'],

                                      'unordered', False)

        num_records = 0

        for record in stream:

            if record.key in [('a',), ('e',)]:

                self.assertEqual('groupcompress-block', record.storage_kind)

            else:

                self.assertEqual('groupcompress-block-ref',

                                 record.storage_kind)

            block_bytes[record.key] = record._manager._block._z_content

            num_records += 1

        self.assertEqual(8, num_records)

        for r in 'abcd':

            key = (r,)

            self.assertIs(block_bytes[key], block_bytes[('a',)])

            self.assertNotEqual(block_bytes[key], block_bytes[('e',)])

        for r in 'efgh':

            key = (r,)

            self.assertIs(block_bytes[key], block_bytes[('e',)])

            self.assertNotEqual(block_bytes[key], block_bytes[('a',)])

        # Now copy the blocks into another vf, and ensure that the blocks are

        # preserved without creating new entries

        vf2 = self.make_test_vf(True, dir='target')

        # ordering in 'groupcompress' order, should actually swap the groups in

        # the target vf, but the groups themselves should not be disturbed.

        vf2.insert_record_stream(vf.get_record_stream(

            [(r,) for r in 'abcdefgh'], 'groupcompress', False))

        stream = vf2.get_record_stream([(r,) for r in 'abcdefgh'],

                                       'groupcompress', False)

        vf2.writer.end()

        num_records = 0

        for record in stream:

            num_records += 1

            self.assertEqual(block_bytes[record.key],

                             record._manager._block._z_content)

        self.assertEqual(8, num_records)

    def test__insert_record_stream_no_reuse_block(self):

        vf = self.make_test_vf(True, dir='source')

        def grouped_stream(revision_ids, first_parents=()):

            parents = first_parents

            for revision_id in revision_ids:

                key = (revision_id,)

                record = versionedfile.FulltextContentFactory(

                    key, parents, None,

                    'some content that is\n'

                    'identical except for\n'

                    'revision_id:%s\n' % (revision_id,))

                yield record

                parents = (key,)

        # One group, a-d

        vf.insert_record_stream(grouped_stream(['a', 'b', 'c', 'd']))

        # Second group, e-h

        vf.insert_record_stream(grouped_stream(['e', 'f', 'g', 'h'],

                                               first_parents=(('d',),)))

        vf.writer.end()

        self.assertEqual(8, len(list(vf.get_record_stream(

                                        [(r,) for r in 'abcdefgh'],

                                        'unordered', False))))

        # Now copy the blocks into another vf, and ensure that the blocks are

        # preserved without creating new entries

        vf2 = self.make_test_vf(True, dir='target')

        # ordering in 'groupcompress' order, should actually swap the groups in

        # the target vf, but the groups themselves should not be disturbed.

        list(vf2._insert_record_stream(vf.get_record_stream(

            [(r,) for r in 'abcdefgh'], 'groupcompress', False),

            reuse_blocks=False))

        vf2.writer.end()

        # After inserting with reuse_blocks=False, we should have everything in

        # a single new block.

        stream = vf2.get_record_stream([(r,) for r in 'abcdefgh'],

                                       'groupcompress', False)

        block = None

        for record in stream:

            if block is None:

                block = record._manager._block

            else:

                self.assertIs(block, record._manager._block)

class TestLazyGroupCompress(tests.TestCaseWithTransport):

    _texts = {