/brz/remove-bazaar

# Copyright (C) 2008-2011 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 .. import (

    config,

    errors,

    osutils,

    tests,

    trace,

    )

from ..bzr import (

    btree_index,

    groupcompress,

    knit,

    index as _mod_index,

    versionedfile,

    )

from ..osutils import sha_string

from .test__groupcompress import compiled_groupcompress_feature

from .scenarios import load_tests_apply_scenarios

def group_compress_implementation_scenarios():

    scenarios = [

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

        ]

    if compiled_groupcompress_feature.available():

        scenarios.append(('C',

            {'compressor': groupcompress.PyrexGroupCompressor}))

    return scenarios

load_tests = load_tests_apply_scenarios

class TestGroupCompressor(tests.TestCase):

    def _chunks_to_repr_lines(self, chunks):

        return '\n'.join(map(repr, b''.join(chunks).split(b'\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"""

    scenarios = group_compress_implementation_scenarios()

    compressor = None # Set by scenario

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

            b'strange\ncommon\n', None)

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

        expected_lines = b'f\x0fstrange\ncommon\n'

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

        self.assertEqual(0, start_point)

        self.assertEqual(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',),

                                                                 b'', 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',), b'some\nbytes\n', None)

        self.assertTrue(compressor.endpoint > 0)

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

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

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

        expected_lines = list(compressor.chunks)

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

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

        # get the first out

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

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

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

        # and the second

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

                          b'different\n', sha1_2),

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

    def test_pop_last(self):

        compressor = self.compressor()

        _, _, _, _ = compressor.compress(('key1',),

            b'some text\nfor the first entry\n', None)

        expected_lines = list(compressor.chunks)

        _, _, _, _ = compressor.compress(('key2',),

            b'some text\nfor the second entry\n', None)

        compressor.pop_last()

        self.assertEqual(expected_lines, compressor.chunks)

class TestPyrexGroupCompressor(TestGroupCompressor):

    _test_needs_features = [compiled_groupcompress_feature]

    compressor = groupcompress.PyrexGroupCompressor

    def test_stats(self):

        compressor = self.compressor()

        compressor.compress(('label',),

                            b'strange\n'

                            b'common very very long line\n'

                            b'plus more text\n', None)

        compressor.compress(('newlabel',),

                            b'common very very long line\n'

                            b'plus more text\n'

                            b'different\n'

                            b'moredifferent\n', None)

        compressor.compress(('label3',),

                            b'new\n'

                            b'common very very long line\n'

                            b'plus more text\n'

                            b'different\n'

                            b'moredifferent\n', None)

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

    def test_two_nosha_delta(self):

        compressor = self.compressor()

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

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

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

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

                                    b'that needs a 16 byte match\n'

                                    b'different\n'), sha1_2)

        expected_lines.extend([

            # 'delta', delta length

            b'd\x0f',

            # source and target length

            b'\x36',

            # copy the line common

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

            # add the line different, and the trailing newline

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

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

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

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

        expected_lines = list(compressor.chunks)

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

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

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

            None)

        self.assertEqual(

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

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

            sha1_3)

        expected_lines.extend([

            # 'delta', delta length

            b'd\x0b',

            # source and target length

            b'\x5f'

            # insert new

            b'\x03new',

            # Copy of first parent 'common' range

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

            # Copy of second parent 'different' range

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

                            b'strange\n'

                            b'common very very long line\n'

                            b'plus more text\n', None)

        compressor.compress(('newlabel',),

                            b'common very very long line\n'

                            b'plus more text\n'

                            b'different\n'

                            b'moredifferent\n', None)

        compressor.compress(('label3',),

                            b'new\n'

                            b'common very very long line\n'

                            b'plus more text\n'

                            b'different\n'

                            b'moredifferent\n', None)

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

    def test_two_nosha_delta(self):

        compressor = self.compressor()

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

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

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

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

                                    b'that needs a 16 byte match\n'

                                    b'different\n'), sha1_2)

        expected_lines.extend([

            # 'delta', delta length

            b'd\x0f',

            # target length

            b'\x36',

            # copy the line common

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

            # add the line different, and the trailing newline

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

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

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

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

        expected_lines = list(compressor.chunks)

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

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

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

            None)

        self.assertEqual(

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

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

            sha1_3)

        expected_lines.extend([

            # 'delta', delta length

            b'd\x0c',

            # target length

            b'\x5f'

            # insert new

            b'\x04new\n',

            # Copy of first parent 'common' range

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

            # Copy of second parent 'different' range

            b'\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.items())

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

    def test_from_minimal_bytes(self):

        block = groupcompress.GroupCompressBlock.from_bytes(

            b'gcb1z\n0\n0\n')

        self.assertIsInstance(block, groupcompress.GroupCompressBlock)

        self.assertIs(None, block._content)

        self.assertEqual(b'', block._z_content)

        block._ensure_content()

        self.assertEqual(b'', block._content)

        self.assertEqual(b'', 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,

                          b'this is not a valid header')

    def test_from_bytes(self):

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

        z_content = zlib.compress(content)

        z_bytes = (

            b'gcb1z\n' # group compress block v1 plain

            b'%d\n' # Length of compressed content

            b'%d\n' # Length of uncompressed content

            b'%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_chunks(self):

        content_chunks = [b'this is some content\n',

                          b'this content will be compressed\n']

        content_len = sum(map(len, content_chunks))

        content = b''.join(content_chunks)

        gcb = groupcompress.GroupCompressBlock()

        gcb.set_chunked_content(content_chunks, content_len)

        total_len, block_chunks = gcb.to_chunks()

        block_bytes = b''.join(block_chunks)

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

        self.assertEqual(total_len, len(block_bytes))

        self.assertEqual(gcb._content_length, content_len)

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

                          b'%d\n' # Length of compressed content

                          b'%d\n' # Length of uncompressed content

                         ) % (gcb._z_content_length, gcb._content_length)

        # The first chunk should be the header chunk. It is small, fixed size,

        # and there is no compelling reason to split it up

        self.assertEqual(expected_header, block_chunks[0])

        self.assertStartsWith(block_bytes, expected_header)

        remaining_bytes = block_bytes[len(expected_header):]

        raw_bytes = zlib.decompress(remaining_bytes)

        self.assertEqual(content, raw_bytes)

    def test_to_bytes(self):

        content = (b'this is some content\n'

                   b'this content will be compressed\n')

        gcb = groupcompress.GroupCompressBlock()

        gcb.set_content(content)

        data = gcb.to_bytes()

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

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

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

                          b'%d\n' # Length of compressed content

                          b'%d\n' # Length of uncompressed content

                         ) % (gcb._z_content_length, gcb._content_length)

        self.assertStartsWith(data, expected_header)

        remaining_bytes = data[len(expected_header):]

        raw_bytes = zlib.decompress(remaining_bytes)

        self.assertEqual(content, raw_bytes)

        # we should get the same results if using the chunked version

        gcb = groupcompress.GroupCompressBlock()

        gcb.set_chunked_content([b'this is some content\n'

                                 b'this content will be compressed\n'],

                                 len(content))

        old_data = data

        data = gcb.to_bytes()

        self.assertEqual(old_data, data)

    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 range(2048):

            next_content = b'%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 + b'\n')

        content = b''.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_chunks = (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__ensure_all_content(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 range(2048):

            next_content = b'%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 + b'\n')

        content = b''.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_chunks = (z_content,)

        block._z_content_length = len(z_content)

        block._compressor_name = 'zlib'

        block._content_length = 158634

        self.assertIs(None, block._content)

        # The first _ensure_content got all of the required data

        block._ensure_content(158634)

        self.assertEqualDiff(content, block._content)

        # And we should have released the _z_content_decompressor since it was

        # fully consumed

        self.assertIs(None, block._z_content_decompressor)

    def test__dump(self):

        dup_content = b'some duplicate content\nwhich is sufficiently long\n'

        key_to_text = {(b'1',): dup_content + b'1 unique\n',

                       (b'2',): dup_content + b'2 extra special\n'}

        locs, block = self.make_block(key_to_text)

        self.assertEqual([(b'f', len(key_to_text[(b'1',)])),

                          (b'd', 21, len(key_to_text[(b'2',)]),

                           [(b'c', 2, len(dup_content)),

                            (b'i', len(b'2 extra special\n'), b'')

                           ]),

                         ], block._dump())

class TestCaseWithGroupCompressVersionedFiles(

        tests.TestCaseWithMemoryTransport):

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

                     dir='.', inconsistency_fatal=True):

        t = self.get_transport(dir)

        t.ensure_base()

        vf = groupcompress.make_pack_factory(graph=create_graph,

            delta=False, keylength=keylength,

            inconsistency_fatal=inconsistency_fatal)(t)

        if do_cleanup:

            self.addCleanup(groupcompress.cleanup_pack_group, vf)

        return vf

class TestGroupCompressVersionedFiles(TestCaseWithGroupCompressVersionedFiles):

    def make_g_index(self, name, ref_lists=0, nodes=[]):

        builder = btree_index.BTreeBuilder(ref_lists)

        for node, references, value in nodes:

            builder.add_node(node, references, value)

        stream = builder.finish()

        trans = self.get_transport()

        size = trans.put_file(name, stream)

        return btree_index.BTreeGraphIndex(trans, name, size)

    def make_g_index_missing_parent(self):

        graph_index = self.make_g_index('missing_parent', 1,

            [((b'parent', ), b'2 78 2 10', ([],)),

             ((b'tip', ), b'2 78 2 10',

              ([(b'parent', ), (b'missing-parent', )],)),

              ])

        return graph_index

    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((b'a',), (), [b'lines\n'])

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

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

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

        vf.writer.end()

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

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

                    'as-requested', False)]

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

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

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

                    'as-requested', False)]

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

        # It should work even after being repacked into another VF