/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,

    groupcompress,

    knit as _mod_knit,

    osutils,

    progress,

    transport,

    ui,

    )

from ..errors import (

                           RevisionNotPresent,

                           RevisionAlreadyPresent,

                           )

from ..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 .. import versionedfile as versionedfile

from ..versionedfile import (

    ConstantMapper,

    HashEscapedPrefixMapper,

    PrefixMapper,

    VirtualVersionedFiles,

    make_versioned_files_factory,

    )

from ..weave import WeaveFile

from ..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 = {

        'origin': (),

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

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

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

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

        }

    # insert a diamond graph to exercise deltas and merges.

    if trailing_eol:

        last_char = '\n'

    else:

        last_char = ''

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

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

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

    if not left_only:

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

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

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

            ['base\n', 'left\n', 'right\n', '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 = [('FileA',), ('FileB',)]

    # insert a diamond graph to exercise deltas and merges.

    if trailing_eol:

        last_char = '\n'

    else:

        last_char = ''

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

            ['origin' + last_char]))

    for prefix in prefixes:

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

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

    for prefix in prefixes:

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

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

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

    if not left_only:

        for prefix in prefixes:

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

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

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

        for prefix in prefixes:

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

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

                ['base\n', 'left\n', 'right\n', '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('r0', [], ['a\n', 'b\n'])

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

        def verify_file(f):

            versions = f.versions()

            self.assertTrue('r0' in versions)

            self.assertTrue('r1' in versions)

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

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

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

            self.assertEqual(2, len(f))

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

            self.assertRaises(RevisionNotPresent,

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

            self.assertRaises(RevisionAlreadyPresent,

                f.add_lines, '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['r0'] = f.add_lines('r0', [], ['a\n', 'b\n'])

        try:

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

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

        except NotImplementedError:

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

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

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

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

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

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

        def verify_file(f):

            versions = f.versions()

            self.assertTrue('r0' in versions)

            self.assertTrue('r1' in versions)

            self.assertTrue('r2' in versions)

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

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

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

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

            origins = f.annotate('r1')

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

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

            origins = f.annotate('r2')

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

            self.assertEqual(origins[1][0], '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, 'a', [], ['a\n', u'b\n', 'c\n'])

        self.assertRaises(

            (errors.BzrBadParameterUnicode, NotImplementedError),

            vf.add_lines_with_ghosts, 'a', [], ['a\n', u'b\n', '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('1', [], ['a\n'])

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

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

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

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

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

        self.assertEqual(['a\n', 'a\n', 'a\n'], vf.get_lines('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, 'a', [], ['a\n\n'])

        self.assertRaises(

            (errors.BzrBadParameterContainsNewline, NotImplementedError),

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

        # but inline CR's are allowed

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

        try:

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

        except NotImplementedError:

            pass

    def test_add_reserved(self):

        vf = self.get_file()

        self.assertRaises(errors.ReservedId,

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

    def test_add_lines_nostoresha(self):

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

        vf = self.get_file()

        empty_text = ('a', [])

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

        sample_text_no_nl = ('c', ["foo\n", "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 + "2", [], lines,

                nostore_sha=sha)

            # and no new version should have been added.

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

                version + "2")

    def test_add_lines_with_ghosts_nostoresha(self):

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

        vf = self.get_file()

        empty_text = ('a', [])

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

        sample_text_no_nl = ('c', ["foo\n", "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('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 + "2", [], lines,

                nostore_sha=sha)

            # and no new version should have been added.

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

                version + "2")

    def test_add_lines_return_value(self):

        # add_lines should return the sha1 and the text size.

        vf = self.get_file()

        empty_text = ('a', [])

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

        sample_text_no_nl = ('c', ["foo\n", "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('d', ['b', 'c'], lines)[0:2])

    def test_get_reserved(self):

        vf = self.get_file()

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

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

        self.assertRaises(errors.ReservedId, vf.get_text, '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 = 'step-%d'

            parents = []

            for step in range(length):

                version = prefix % step

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

                vf.add_lines(version, parents, lines)

                version_lines[version] = lines

                parents = [version]

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

            vf.get_texts(version_lines.keys())

            self.assertEqualDiff('line', vf.get_text('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 = ["line\n"]

        sample_text_no_nl = ["line"]

        versions = []

        version_lines = {}

        parents = []

        for i in range(4):

            version = '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('noeol', [], ['line'])

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

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

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

        # On top of a delta

        vf = self.get_file('delta')

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

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

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

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

        self.assertEqualDiff('newline\nline', vf.get_text('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('text', ['ghost'], ['line\n'])

        except NotImplementedError:

            # old Weave formats do not allow ghosts

            return

        self.assertRaises(errors.RevisionNotPresent, vf.make_mpdiffs, ['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('base', [], ['line\n'])

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

        # detailed eol tests:

        # shared last line with parent no-eol

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

        # differing last line with parent, both no-eol

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

        # add eol following a noneol parent, change content

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

        # add eol following a noneol parent, no change content

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

        # noeol with no parents:

        f.add_lines('noeolbase', [], ['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('eolbeforefirstparent', ['noeolbase', 'noeol'], ['line'])

        # two identical eol texts

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

        next_parent = 'base'

        text_name = 'chain1-'

        text = ['line\n']

        sha1s = {0 :'da6d3141cb4a5e6f464bf6e0518042ddc7bfd079',

                 1 :'45e21ea146a81ea44a821737acdb4f9791c8abe7',

                 2 :'e1f11570edf3e2a070052366c582837a4fe4e9fa',

                 3 :'26b4b8626da827088c514b8f9bbe4ebf181edda1',

                 4 :'e28a5510be25ba84d31121cff00956f9970ae6f6',

                 5 :'d63ec0ce22e11dcf65a931b69255d3ac747a318d',

                 6 :'2c2888d288cb5e1d98009d822fedfe6019c6a4ea',

                 7 :'95c14da9cafbf828e3e74a6f016d87926ba234ab',

                 8 :'779e9a0b28f9f832528d4b21e17e168c67697272',

                 9 :'1f8ff4e5c6ff78ac106fcfe6b1e8cb8740ff9a8f',

                 10:'131a2ae712cf51ed62f143e3fbac3d4206c25a05',

                 11:'c5a9d6f520d2515e1ec401a8f8a67e6c3c89f199',

                 12:'31a2286267f24d8bedaa43355f8ad7129509ea85',

                 13:'dc2a7fe80e8ec5cae920973973a8ee28b2da5e0a',

                 14:'2c4b1736566b8ca6051e668de68650686a3922f2',

                 15:'5912e4ecd9b0c07be4d013e7e2bdcf9323276cde',

                 16:'b0d2e18d3559a00580f6b49804c23fea500feab3',

                 17:'8e1d43ad72f7562d7cb8f57ee584e20eb1a69fc7',

                 18:'5cf64a3459ae28efa60239e44b20312d25b253f3',

                 19:'1ebed371807ba5935958ad0884595126e8c4e823',

                 20:'2aa62a8b06fb3b3b892a3292a068ade69d5ee0d3',

                 21:'01edc447978004f6e4e962b417a4ae1955b6fe5d',

                 22:'d8d8dc49c4bf0bab401e0298bb5ad827768618bb',

                 23:'c21f62b1c482862983a8ffb2b0c64b3451876e3f',

                 24:'c0593fe795e00dff6b3c0fe857a074364d5f04fc',

                 25:'dd1a1cf2ba9cc225c3aff729953e6364bf1d1855',

                 }

        for depth in range(26):

            new_version = text_name + '%s' % depth

            text = text + ['line\n']

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

            next_parent = new_version

        next_parent = 'base'

        text_name = 'chain2-'

        text = ['line\n']

        for depth in range(26):

            new_version = text_name + '%s' % depth

            text = text + ['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('r0', [], ['a\n', 'b\n'])

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

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

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

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

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

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

        # there are some possibilities:

        # r0 r1 r2 rM r3

        # r0 r1 r2 r3 rM

        # etc

        # so we check indexes

        r0 = versions.index('r0')

        r1 = versions.index('r1')

        r2 = versions.index('r2')

        self.assertFalse('r3' in versions)

        rM = versions.index('rM')

        self.assertTrue(r0 < r1)

        self.assertTrue(r0 < r2)

        self.assertTrue(r1 < rM)

        self.assertTrue(r2 < rM)

        self.assertRaises(RevisionNotPresent,

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

        self.assertEqual(set(f.get_ancestry('rM')),

            set(f.get_ancestry('rM', topo_sorted=False)))

    def test_mutate_after_finish(self):

        self._transaction = 'before'

        f = self.get_file()

        self._transaction = 'after'

        self.assertRaises(errors.OutSideTransaction, f.add_lines, '', [], [])

        self.assertRaises(errors.OutSideTransaction, f.add_lines_with_ghosts, '', [], [])

    def test_copy_to(self):

        f = self.get_file()

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

        t = MemoryTransport()

        f.copy_to('foo', t)

        for suffix in self.get_factory().get_suffixes():

            self.assertTrue(t.has('foo' + suffix))

    def test_get_suffixes(self):

        f = self.get_file()

        # and should be a list

        self.assertTrue(isinstance(self.get_factory().get_suffixes(), list))

    def test_get_parent_map(self):

        f = self.get_file()

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

        self.assertEqual(

            {'r0':()}, f.get_parent_map(['r0']))

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

        self.assertEqual(

            {'r1':('r0',)}, f.get_parent_map(['r1']))

        self.assertEqual(

            {'r0':(),

             'r1':('r0',)},

            f.get_parent_map(['r0', 'r1']))

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

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

        f.add_lines('m', ['r0', 'r1', 'r2', 'r3'], ['a\n', 'b\n'])

        self.assertEqual(

            {'m':('r0', 'r1', 'r2', 'r3')}, f.get_parent_map(['m']))

        self.assertEqual({}, f.get_parent_map('y'))

        self.assertEqual(

            {'r0':(),

             'r1':('r0',)},

            f.get_parent_map(['r0', 'y', 'r1']))

    def test_annotate(self):

        f = self.get_file()

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

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

        origins = f.annotate('r1')

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

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