/brz/remove-bazaar

# Copyright (C) 2006-2012, 2016 Canonical Ltd

#

# Authors:

#   Johan Rydberg <jrydberg@gnu.org>

#

# 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

# TODO: might be nice to create a versionedfile with some type of corruption

# considered typical and check that it can be detected/corrected.

from gzip import GzipFile

import itertools

from .. import (

    errors,

    graph as _mod_graph,

    osutils,

    progress,

    transport,

    ui,

    )

from ..bzr import (

    groupcompress,

    knit as _mod_knit,

    )

from ..errors import (

    RevisionNotPresent,

    RevisionAlreadyPresent,

    )

from ..bzr.knit import (

    cleanup_pack_knit,

    make_file_factory,

    make_pack_factory,

    )

from ..sixish import (

    BytesIO,

    zip,

    )

from . import (

    TestCase,

    TestCaseWithMemoryTransport,

    TestNotApplicable,

    TestSkipped,

    )

from .http_utils import TestCaseWithWebserver

from ..transport.memory import MemoryTransport

from ..bzr import versionedfile as versionedfile

from ..bzr.versionedfile import (

    ConstantMapper,

    HashEscapedPrefixMapper,

    PrefixMapper,

    VirtualVersionedFiles,

    make_versioned_files_factory,

    )

from ..bzr.weave import (

    WeaveFile,

    WeaveInvalidChecksum,

    )

from ..bzr.weavefile import write_weave

from .scenarios import load_tests_apply_scenarios

load_tests = load_tests_apply_scenarios

def get_diamond_vf(f, trailing_eol=True, left_only=False):

    """Get a diamond graph to exercise deltas and merges.

    :param trailing_eol: If True end the last line with \n.

    """

    parents = {

        b'origin': (),

        b'base': ((b'origin',),),

        b'left': ((b'base',),),

        b'right': ((b'base',),),

        b'merged': ((b'left',), (b'right',)),

        }

    # insert a diamond graph to exercise deltas and merges.

    if trailing_eol:

        last_char = b'\n'

    else:

        last_char = b''

    f.add_lines(b'origin', [], [b'origin' + last_char])

    f.add_lines(b'base', [b'origin'], [b'base' + last_char])

    f.add_lines(b'left', [b'base'], [b'base\n', b'left' + last_char])

    if not left_only:

        f.add_lines(b'right', [b'base'],

            [b'base\n', b'right' + last_char])

        f.add_lines(b'merged', [b'left', b'right'],

            [b'base\n', b'left\n', b'right\n', b'merged' + last_char])

    return f, parents

def get_diamond_files(files, key_length, trailing_eol=True, left_only=False,

    nograph=False, nokeys=False):

    """Get a diamond graph to exercise deltas and merges.

    This creates a 5-node graph in files. If files supports 2-length keys two

    graphs are made to exercise the support for multiple ids.

    :param trailing_eol: If True end the last line with \n.

    :param key_length: The length of keys in files. Currently supports length 1

        and 2 keys.

    :param left_only: If True do not add the right and merged nodes.

    :param nograph: If True, do not provide parents to the add_lines calls;

        this is useful for tests that need inserted data but have graphless

        stores.

    :param nokeys: If True, pass None is as the key for all insertions.

        Currently implies nograph.

    :return: The results of the add_lines calls.

    """

    if nokeys:

        nograph = True

    if key_length == 1:

        prefixes = [()]

    else:

        prefixes = [(b'FileA',), (b'FileB',)]

    # insert a diamond graph to exercise deltas and merges.

    if trailing_eol:

        last_char = b'\n'

    else:

        last_char = b''

    result = []

    def get_parents(suffix_list):

        if nograph:

            return ()

        else:

            result = [prefix + suffix for suffix in suffix_list]

            return result

    def get_key(suffix):

        if nokeys:

            return (None, )

        else:

            return (suffix,)

    # we loop over each key because that spreads the inserts across prefixes,

    # which is how commit operates.

    for prefix in prefixes:

        result.append(files.add_lines(prefix + get_key(b'origin'), (),

            [b'origin' + last_char]))

    for prefix in prefixes:

        result.append(files.add_lines(prefix + get_key(b'base'),

            get_parents([(b'origin',)]), [b'base' + last_char]))

    for prefix in prefixes:

        result.append(files.add_lines(prefix + get_key(b'left'),

            get_parents([(b'base',)]),

            [b'base\n', b'left' + last_char]))

    if not left_only:

        for prefix in prefixes:

            result.append(files.add_lines(prefix + get_key(b'right'),

                get_parents([(b'base',)]),

                [b'base\n', b'right' + last_char]))

        for prefix in prefixes:

            result.append(files.add_lines(prefix + get_key(b'merged'),

                get_parents([(b'left',), (b'right',)]),

                [b'base\n', b'left\n', b'right\n', b'merged' + last_char]))

    return result

class VersionedFileTestMixIn(object):

    """A mixin test class for testing VersionedFiles.

    This is not an adaptor-style test at this point because

    theres no dynamic substitution of versioned file implementations,

    they are strictly controlled by their owning repositories.

    """

    def get_transaction(self):

        if not hasattr(self, '_transaction'):

            self._transaction = None

        return self._transaction

    def test_add(self):

        f = self.get_file()

        f.add_lines(b'r0', [], [b'a\n', b'b\n'])

        f.add_lines(b'r1', [b'r0'], [b'b\n', b'c\n'])

        def verify_file(f):

            versions = f.versions()

            self.assertTrue(b'r0' in versions)

            self.assertTrue(b'r1' in versions)

            self.assertEqual(f.get_lines(b'r0'), [b'a\n', b'b\n'])

            self.assertEqual(f.get_text(b'r0'), b'a\nb\n')

            self.assertEqual(f.get_lines(b'r1'), [b'b\n', b'c\n'])

            self.assertEqual(2, len(f))

            self.assertEqual(2, f.num_versions())

            self.assertRaises(RevisionNotPresent,

                f.add_lines, b'r2', [b'foo'], [])

            self.assertRaises(RevisionAlreadyPresent,

                f.add_lines, b'r1', [], [])

        verify_file(f)

        # this checks that reopen with create=True does not break anything.

        f = self.reopen_file(create=True)

        verify_file(f)

    def test_adds_with_parent_texts(self):

        f = self.get_file()

        parent_texts = {}

        _, _, parent_texts[b'r0'] = f.add_lines(b'r0', [], [b'a\n', b'b\n'])

        try:

            _, _, parent_texts[b'r1'] = f.add_lines_with_ghosts(b'r1',

                [b'r0', b'ghost'], [b'b\n', b'c\n'], parent_texts=parent_texts)

        except NotImplementedError:

            # if the format doesn't support ghosts, just add normally.

            _, _, parent_texts[b'r1'] = f.add_lines(b'r1',

                [b'r0'], [b'b\n', b'c\n'], parent_texts=parent_texts)

        f.add_lines(b'r2', [b'r1'], [b'c\n', b'd\n'], parent_texts=parent_texts)

        self.assertNotEqual(None, parent_texts[b'r0'])

        self.assertNotEqual(None, parent_texts[b'r1'])

        def verify_file(f):

            versions = f.versions()

            self.assertTrue(b'r0' in versions)

            self.assertTrue(b'r1' in versions)

            self.assertTrue(b'r2' in versions)

            self.assertEqual(f.get_lines(b'r0'), [b'a\n', b'b\n'])

            self.assertEqual(f.get_lines(b'r1'), [b'b\n', b'c\n'])

            self.assertEqual(f.get_lines(b'r2'), [b'c\n', b'd\n'])

            self.assertEqual(3, f.num_versions())

            origins = f.annotate(b'r1')

            self.assertEqual(origins[0][0], b'r0')

            self.assertEqual(origins[1][0], b'r1')

            origins = f.annotate(b'r2')

            self.assertEqual(origins[0][0], b'r1')

            self.assertEqual(origins[1][0], b'r2')

        verify_file(f)

        f = self.reopen_file()

        verify_file(f)

    def test_add_unicode_content(self):

        # unicode content is not permitted in versioned files.

        # versioned files version sequences of bytes only.

        vf = self.get_file()

        self.assertRaises(errors.BzrBadParameterUnicode,

            vf.add_lines, b'a', [], [b'a\n', u'b\n', b'c\n'])

        self.assertRaises(

            (errors.BzrBadParameterUnicode, NotImplementedError),

            vf.add_lines_with_ghosts, b'a', [], [b'a\n', u'b\n', b'c\n'])

    def test_add_follows_left_matching_blocks(self):

        """If we change left_matching_blocks, delta changes

        Note: There are multiple correct deltas in this case, because

        we start with 1 "a" and we get 3.

        """

        vf = self.get_file()

        if isinstance(vf, WeaveFile):

            raise TestSkipped("WeaveFile ignores left_matching_blocks")

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

        vf.add_lines(b'2', [b'1'], [b'a\n', b'a\n', b'a\n'],

                     left_matching_blocks=[(0, 0, 1), (1, 3, 0)])

        self.assertEqual([b'a\n', b'a\n', b'a\n'], vf.get_lines(b'2'))

        vf.add_lines(b'3', [b'1'], [b'a\n', b'a\n', b'a\n'],

                     left_matching_blocks=[(0, 2, 1), (1, 3, 0)])

        self.assertEqual([b'a\n', b'a\n', b'a\n'], vf.get_lines(b'3'))

    def test_inline_newline_throws(self):

        # \r characters are not permitted in lines being added

        vf = self.get_file()

        self.assertRaises(errors.BzrBadParameterContainsNewline,

            vf.add_lines, b'a', [], [b'a\n\n'])

        self.assertRaises(

            (errors.BzrBadParameterContainsNewline, NotImplementedError),

            vf.add_lines_with_ghosts, b'a', [], [b'a\n\n'])

        # but inline CR's are allowed

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

        try:

            vf.add_lines_with_ghosts(b'b', [], [b'a\r\n'])

        except NotImplementedError:

            pass

    def test_add_reserved(self):

        vf = self.get_file()

        self.assertRaises(errors.ReservedId,

            vf.add_lines, b'a:', [], [b'a\n', b'b\n', b'c\n'])

    def test_add_lines_nostoresha(self):

        """When nostore_sha is supplied using old content raises."""

        vf = self.get_file()

        empty_text = (b'a', [])

        sample_text_nl = (b'b', [b"foo\n", b"bar\n"])

        sample_text_no_nl = (b'c', [b"foo\n", b"bar"])

        shas = []

        for version, lines in (empty_text, sample_text_nl, sample_text_no_nl):

            sha, _, _ = vf.add_lines(version, [], lines)

            shas.append(sha)

        # we now have a copy of all the lines in the vf.

        for sha, (version, lines) in zip(

            shas, (empty_text, sample_text_nl, sample_text_no_nl)):

            self.assertRaises(errors.ExistingContent,

                vf.add_lines, version + b"2", [], lines,

                nostore_sha=sha)

            # and no new version should have been added.

            self.assertRaises(errors.RevisionNotPresent, vf.get_lines,

                version + b"2")

    def test_add_lines_with_ghosts_nostoresha(self):

        """When nostore_sha is supplied using old content raises."""

        vf = self.get_file()

        empty_text = (b'a', [])

        sample_text_nl = (b'b', [b"foo\n", b"bar\n"])

        sample_text_no_nl = (b'c', [b"foo\n", b"bar"])

        shas = []

        for version, lines in (empty_text, sample_text_nl, sample_text_no_nl):

            sha, _, _ = vf.add_lines(version, [], lines)

            shas.append(sha)

        # we now have a copy of all the lines in the vf.

        # is the test applicable to this vf implementation?

        try:

            vf.add_lines_with_ghosts(b'd', [], [])

        except NotImplementedError:

            raise TestSkipped("add_lines_with_ghosts is optional")

        for sha, (version, lines) in zip(

            shas, (empty_text, sample_text_nl, sample_text_no_nl)):

            self.assertRaises(errors.ExistingContent,

                vf.add_lines_with_ghosts, version + b"2", [], lines,

                nostore_sha=sha)

            # and no new version should have been added.

            self.assertRaises(errors.RevisionNotPresent, vf.get_lines,

                version + b"2")

    def test_add_lines_return_value(self):

        # add_lines should return the sha1 and the text size.

        vf = self.get_file()

        empty_text = (b'a', [])

        sample_text_nl = (b'b', [b"foo\n", b"bar\n"])

        sample_text_no_nl = (b'c', [b"foo\n", b"bar"])

        # check results for the three cases:

        for version, lines in (empty_text, sample_text_nl, sample_text_no_nl):

            # the first two elements are the same for all versioned files:

            # - the digest and the size of the text. For some versioned files

            #   additional data is returned in additional tuple elements.

            result = vf.add_lines(version, [], lines)

            self.assertEqual(3, len(result))

            self.assertEqual((osutils.sha_strings(lines), sum(map(len, lines))),

                result[0:2])

        # parents should not affect the result:

        lines = sample_text_nl[1]

        self.assertEqual((osutils.sha_strings(lines), sum(map(len, lines))),

            vf.add_lines(b'd', [b'b', b'c'], lines)[0:2])

    def test_get_reserved(self):

        vf = self.get_file()

        self.assertRaises(errors.ReservedId, vf.get_texts, [b'b:'])

        self.assertRaises(errors.ReservedId, vf.get_lines, b'b:')

        self.assertRaises(errors.ReservedId, vf.get_text, b'b:')

    def test_add_unchanged_last_line_noeol_snapshot(self):

        """Add a text with an unchanged last line with no eol should work."""

        # Test adding this in a number of chain lengths; because the interface

        # for VersionedFile does not allow forcing a specific chain length, we

        # just use a small base to get the first snapshot, then a much longer

        # first line for the next add (which will make the third add snapshot)

        # and so on. 20 has been chosen as an aribtrary figure - knits use 200

        # as a capped delta length, but ideally we would have some way of

        # tuning the test to the store (e.g. keep going until a snapshot

        # happens).

        for length in range(20):

            version_lines = {}

            vf = self.get_file('case-%d' % length)

            prefix = b'step-%d'

            parents = []

            for step in range(length):

                version = prefix % step

                lines = ([b'prelude \n'] * step) + [b'line']

                vf.add_lines(version, parents, lines)

                version_lines[version] = lines

                parents = [version]

            vf.add_lines(b'no-eol', parents, [b'line'])

            vf.get_texts(version_lines.keys())

            self.assertEqualDiff(b'line', vf.get_text(b'no-eol'))

    def test_get_texts_eol_variation(self):

        # similar to the failure in <http://bugs.launchpad.net/234748>

        vf = self.get_file()

        sample_text_nl = [b"line\n"]

        sample_text_no_nl = [b"line"]

        versions = []

        version_lines = {}

        parents = []

        for i in range(4):

            version = b'v%d' % i

            if i % 2:

                lines = sample_text_nl

            else:

                lines = sample_text_no_nl

            # left_matching blocks is an internal api; it operates on the

            # *internal* representation for a knit, which is with *all* lines

            # being normalised to end with \n - even the final line in a no_nl

            # file. Using it here ensures that a broken internal implementation

            # (which is what this test tests) will generate a correct line

            # delta (which is to say, an empty delta).

            vf.add_lines(version, parents, lines,

                left_matching_blocks=[(0, 0, 1)])

            parents = [version]

            versions.append(version)

            version_lines[version] = lines

        vf.check()

        vf.get_texts(versions)

        vf.get_texts(reversed(versions))

    def test_add_lines_with_matching_blocks_noeol_last_line(self):

        """Add a text with an unchanged last line with no eol should work."""

        from breezy import multiparent

        # Hand verified sha1 of the text we're adding.

        sha1 = '6a1d115ec7b60afb664dc14890b5af5ce3c827a4'

        # Create a mpdiff which adds a new line before the trailing line, and

        # reuse the last line unaltered (which can cause annotation reuse).

        # Test adding this in two situations:

        # On top of a new insertion

        vf = self.get_file('fulltext')

        vf.add_lines(b'noeol', [], [b'line'])

        vf.add_lines(b'noeol2', [b'noeol'], [b'newline\n', b'line'],

            left_matching_blocks=[(0, 1, 1)])

        self.assertEqualDiff(b'newline\nline', vf.get_text(b'noeol2'))

        # On top of a delta

        vf = self.get_file('delta')

        vf.add_lines(b'base', [], [b'line'])

        vf.add_lines(b'noeol', [b'base'], [b'prelude\n', b'line'])

        vf.add_lines(b'noeol2', [b'noeol'], [b'newline\n', b'line'],

            left_matching_blocks=[(1, 1, 1)])

        self.assertEqualDiff(b'newline\nline', vf.get_text(b'noeol2'))

    def test_make_mpdiffs(self):

        from breezy import multiparent

        vf = self.get_file('foo')

        sha1s = self._setup_for_deltas(vf)

        new_vf = self.get_file('bar')

        for version in multiparent.topo_iter(vf):

            mpdiff = vf.make_mpdiffs([version])[0]

            new_vf.add_mpdiffs([(version, vf.get_parent_map([version])[version],

                                 vf.get_sha1s([version])[version], mpdiff)])

            self.assertEqualDiff(vf.get_text(version),

                                 new_vf.get_text(version))

    def test_make_mpdiffs_with_ghosts(self):

        vf = self.get_file('foo')

        try:

            vf.add_lines_with_ghosts(b'text', [b'ghost'], [b'line\n'])

        except NotImplementedError:

            # old Weave formats do not allow ghosts

            return

        self.assertRaises(errors.RevisionNotPresent, vf.make_mpdiffs, [b'ghost'])

    def _setup_for_deltas(self, f):

        self.assertFalse(f.has_version('base'))

        # add texts that should trip the knit maximum delta chain threshold

        # as well as doing parallel chains of data in knits.

        # this is done by two chains of 25 insertions

        f.add_lines(b'base', [], [b'line\n'])

        f.add_lines(b'noeol', [b'base'], [b'line'])

        # detailed eol tests:

        # shared last line with parent no-eol

        f.add_lines(b'noeolsecond', [b'noeol'], [b'line\n', b'line'])

        # differing last line with parent, both no-eol

        f.add_lines(b'noeolnotshared', [b'noeolsecond'], [b'line\n', b'phone'])

        # add eol following a noneol parent, change content

        f.add_lines(b'eol', [b'noeol'], [b'phone\n'])

        # add eol following a noneol parent, no change content

        f.add_lines(b'eolline', [b'noeol'], [b'line\n'])

        # noeol with no parents:

        f.add_lines(b'noeolbase', [], [b'line'])

        # noeol preceeding its leftmost parent in the output:

        # this is done by making it a merge of two parents with no common

        # anestry: noeolbase and noeol with the

        # later-inserted parent the leftmost.

        f.add_lines(b'eolbeforefirstparent', [b'noeolbase', b'noeol'], [b'line'])

        # two identical eol texts

        f.add_lines(b'noeoldup', [b'noeol'], [b'line'])

        next_parent = b'base'

        text_name = b'chain1-'

        text = [b'line\n']

        sha1s = {0: b'da6d3141cb4a5e6f464bf6e0518042ddc7bfd079',

                 1: b'45e21ea146a81ea44a821737acdb4f9791c8abe7',

                 2: b'e1f11570edf3e2a070052366c582837a4fe4e9fa',

                 3: b'26b4b8626da827088c514b8f9bbe4ebf181edda1',

                 4: b'e28a5510be25ba84d31121cff00956f9970ae6f6',

                 5: b'd63ec0ce22e11dcf65a931b69255d3ac747a318d',

                 6: b'2c2888d288cb5e1d98009d822fedfe6019c6a4ea',

                 7: b'95c14da9cafbf828e3e74a6f016d87926ba234ab',

                 8: b'779e9a0b28f9f832528d4b21e17e168c67697272',

                 9: b'1f8ff4e5c6ff78ac106fcfe6b1e8cb8740ff9a8f',

                 10: b'131a2ae712cf51ed62f143e3fbac3d4206c25a05',

                 11: b'c5a9d6f520d2515e1ec401a8f8a67e6c3c89f199',

                 12: b'31a2286267f24d8bedaa43355f8ad7129509ea85',

                 13: b'dc2a7fe80e8ec5cae920973973a8ee28b2da5e0a',

                 14: b'2c4b1736566b8ca6051e668de68650686a3922f2',

                 15: b'5912e4ecd9b0c07be4d013e7e2bdcf9323276cde',

                 16: b'b0d2e18d3559a00580f6b49804c23fea500feab3',

                 17: b'8e1d43ad72f7562d7cb8f57ee584e20eb1a69fc7',

                 18: b'5cf64a3459ae28efa60239e44b20312d25b253f3',

                 19: b'1ebed371807ba5935958ad0884595126e8c4e823',

                 20: b'2aa62a8b06fb3b3b892a3292a068ade69d5ee0d3',

                 21: b'01edc447978004f6e4e962b417a4ae1955b6fe5d',

                 22: b'd8d8dc49c4bf0bab401e0298bb5ad827768618bb',

                 23: b'c21f62b1c482862983a8ffb2b0c64b3451876e3f',

                 24: b'c0593fe795e00dff6b3c0fe857a074364d5f04fc',

                 25: b'dd1a1cf2ba9cc225c3aff729953e6364bf1d1855',

                 }

        for depth in range(26):

            new_version = text_name + b'%d' % depth

            text = text + [b'line\n']

            f.add_lines(new_version, [next_parent], text)

            next_parent = new_version

        next_parent = b'base'

        text_name = b'chain2-'

        text = [b'line\n']

        for depth in range(26):

            new_version = text_name + b'%d' % depth

            text = text + [b'line\n']

            f.add_lines(new_version, [next_parent], text)

            next_parent = new_version

        return sha1s

    def test_ancestry(self):

        f = self.get_file()

        self.assertEqual([], f.get_ancestry([]))

        f.add_lines(b'r0', [], [b'a\n', b'b\n'])

        f.add_lines(b'r1', [b'r0'], [b'b\n', b'c\n'])

        f.add_lines(b'r2', [b'r0'], [b'b\n', b'c\n'])

        f.add_lines(b'r3', [b'r2'], [b'b\n', b'c\n'])

        f.add_lines(b'rM', [b'r1', b'r2'], [b'b\n', b'c\n'])

        self.assertEqual([], f.get_ancestry([]))

        versions = f.get_ancestry([b'rM'])

        # there are some possibilities:

        # r0 r1 r2 rM r3

        # r0 r1 r2 r3 rM

        # etc

        # so we check indexes

        r0 = versions.index(b'r0')

        r1 = versions.index(b'r1')

        r2 = versions.index(b'r2')

        self.assertFalse(b'r3' in versions)

        rM = versions.index(b'rM')

        self.assertTrue(r0 < r1)

        self.assertTrue(r0 < r2)

        self.assertTrue(r1 < rM)

        self.assertTrue(r2 < rM)

        self.assertRaises(RevisionNotPresent,

            f.get_ancestry, [b'rM', b'rX'])