/brz/remove-bazaar

#

# Authors:

#   Johan Rydberg <jrydberg@gnu.org>

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

 

 

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

from itertools import chain, izip

from StringIO import StringIO

 

from bzrlib import (

    errors,

    osutils,

    progress,

    )

from bzrlib.errors import (

                           RevisionNotPresent,

                           RevisionAlreadyPresent,

                           WeaveParentMismatch

                           )

from bzrlib.knit import (

    cleanup_pack_knit,

    make_file_factory,

    KnitAnnotateFactory,

    KnitPlainFactory,

    )

from bzrlib.tests import (

    TestCase,

    TestCaseWithMemoryTransport,

    TestNotApplicable,

    TestSkipped,

    condition_isinstance,

    split_suite_by_condition,

    )

from bzrlib.tests.http_utils import TestCaseWithWebserver

from bzrlib.trace import mutter

    """Parameterize VersionedFiles tests for different implementations."""

    to_adapt, result = split_suite_by_condition(

        standard_tests, condition_isinstance(TestVersionedFiles))

    # We want to be sure of behaviour for:

    # weaves prefix layout (weave texts)

    # individually named weaves (weave inventories)

    # individual graph knits in packs (inventories)

    # individual graph nocompression knits in packs (revisions)

    # plain text knits in packs (texts)

        ('weave-named', {

            'cleanup':None,

            'factory':make_versioned_files_factory(WeaveFile,

            'key_length':1,

            'support_partial_insertion': False,

            }),

        ]

        ('weave-prefix', {

            'cleanup':None,

            'factory':make_versioned_files_factory(WeaveFile,

            'key_length':2,

            'support_partial_insertion': True,

            }),

        ]

 

 

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

 

 

def get_diamond_files(files, key_length, trailing_eol=True, left_only=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 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.

    :return: The results of the add_lines calls.

    """

    if key_length == 1:

        prefixes = [()]

    else:

        else:

            result = [prefix + suffix for suffix in suffix_list]

            return result

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

    # which is how commit operates.

    for prefix in prefixes:

            ['origin' + last_char]))

    for prefix in prefixes:

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

    for prefix in prefixes:

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

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

    if not left_only:

        for prefix in prefixes:

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

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

        for prefix in prefixes:

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

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

    return result

            self.assertEquals(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,

        verify_file(f)

 

    def test_add_unicode_content(self):

        # versioned files version sequences of bytes only.

        vf = self.get_file()

        self.assertRaises(errors.BzrBadParameterUnicode,

    def test_inline_newline_throws(self):

        # \r characters are not permitted in lines being added

        vf = self.get_file()

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

        self.assertRaises(

            (errors.BzrBadParameterContainsNewline, NotImplementedError),

        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

        # later-inserted parent the leftmost.

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

        # two identical eol texts

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

 

    def test_iter_lines_added_or_present_in_versions(self):

        # test that we get at least an equalset of the lines added by

        # the ordering here is to make a tree so that dumb searches have

        # more changes to muck up.

 

 

            def __init__(self):

                self.updates = []

 

            def update(self, msg=None, current=None, total=None):

                self.assertEqual(expected, progress.updates)

            return lines

        lines = iter_with_versions(['child', 'otherchild'],

        # we must see child and otherchild

        self.assertTrue(lines[('child\n', 'child')] > 0)

        self.assertTrue(lines[('otherchild\n', 'otherchild')] > 0)

        # we dont care if we got more than that.

        # test all lines

        # all lines must be seen at least once

        self.assertTrue(lines[('base\n', 'base')] > 0)

        self.assertTrue(lines[('lancestor\n', 'lancestor')] > 0)

                          'base',

                          [],

                          [])

    def test_get_sha1s(self):

        # check the sha1 data is available

        vf = self.get_file()

            'b': '3f786850e387550fdab836ed7e6dc881de23001b',

            },

            vf.get_sha1s(['a', 'c', 'b']))

 

class TestWeave(TestCaseWithMemoryTransport, VersionedFileTestMixIn):

 

            get_scope=self.get_transaction)

        w.add_lines('v1', [], ['hello\n'])

        w.add_lines('v2', ['v1'], ['hello\n', 'there\n'])

        # We are going to invasively corrupt the text

        # Make sure the internals of weave are the same

        self.assertEqual([('{', 0)

                        , 'there\n'

                        , ('}', None)

                        ], w._weave)

        self.assertEqual(['f572d396fae9206628714fb2ce00f72e94f2258f'

                        , '90f265c6e75f1c8f9ab76dcf85528352c5f215ef'

                        ], w._sha1s)

        w.check()

        # Corrupted

        w._weave[4] = 'There\n'

        return w

        # Corrected

        w._weave[4] = 'there\n'

        self.assertEqual('hello\nthere\n', w.get_text('v2'))

        #Invalid checksum, first digit changed

        w._sha1s[1] =  'f0f265c6e75f1c8f9ab76dcf85528352c5f215ef'

        return w

 

        def addcrlf(x):

            return x + '\n'

        w = self.get_file()

        w.add_lines('text0', [], map(addcrlf, base))

        w.add_lines('text1', ['text0'], map(addcrlf, a))

 

        mp = map(addcrlf, mp)

        self.assertEqual(mt.readlines(), mp)

    def testOneInsert(self):

        self.doMerge([],

                     ['aa'],

                     ['aaa', 'xxx', 'yyy', 'bbb'],

                     ['aaa', 'xxx', 'bbb'], self.overlappedInsertExpected)

 

        # ['aaa', 'xxx', 'yyy', 'bbb']

 

 

        self.doMerge(['aaa'],

                     ['xxx'],

                     ['yyy', 'zzz'],

                      '>>>>>>> '])

 

    def testNonClashInsert1(self):

        self.doMerge(['aaa'],

                     ['xxx', 'aaa'],

                     ['yyy', 'zzz'],

                      '>>>>>>> '])

 

    def testNonClashInsert2(self):

        #######################################

        # skippd, not working yet

        return

        self.doMerge(['aaa', 'bbb', 'ccc'],

                     ['aaa', 'ddd', 'ccc'],

                     ['aaa', 'ccc'],

            """

        result = """\

            line 1

            """

        self._test_merge_from_strings(base, a, b, result)

 

    def test_deletion_overlap(self):

        """Delete overlapping regions with no other conflict.

 

        conflict, but for now conflict is safer.

        """

        base = """\

            """

        result = """\

            start context

            int a() {}

=======

            int c() {}

            end context

            """

        self._test_merge_from_strings(base, a, b, result)

 

    def test_sync_on_deletion(self):

        """Specific case of merge where we can synchronize incorrectly.

        A previous version of the weave merge concluded that the two versions

        agreed on deleting line 2, and this could be a synchronization point.

        both sides.

 

        It's better to consider the whole thing as a disagreement region.

            """

        result = """\

            start context

            base line 1

            a's replacement line 2

=======

            b replaces

            both lines

            end context

            """

        self._test_merge_from_strings(base, a, b, result)

        write_weave(w, tmpf)

        self.log(tmpf.getvalue())

 

                                'xxx', '>>>>>>> ', 'bbb']

 

 

 

    def test_unannotated_to_fulltext(self):

        """Test adapting unannotated knits to full texts.

        This is used for -> weaves, and for -> annotated knits.

        """

        # we need a full text, and a delta

 

    def test_unannotated_to_fulltext_no_eol(self):

        """Test adapting unannotated knits to full texts.

        This is used for -> weaves, and for -> annotated knits.

        """

        # we need a full text, and a delta

            transport.mkdir('.')

        files = self.factory(transport)

        if self.cleanup is not None:

        return files

 

    def test_annotate(self):

        files = self.get_versionedfiles()

        self.get_diamond_files(files)

        self.assertRaises(RevisionNotPresent,

            files.annotate, prefix + ('missing-key',))

 

    def test_construct(self):

        """Each parameterised test can be constructed on a transport."""

        files = self.get_versionedfiles()

 

        return get_diamond_files(files, self.key_length,

            trailing_eol=trailing_eol, nograph=not self.graph,

 

    def test_add_lines_return(self):

        files = self.get_versionedfiles()

                ('ed8bce375198ea62444dc71952b22cfc2b09226d', 23)],

                results)

 

    def test_empty_lines(self):

        """Empty files can be stored."""

        f = self.get_versionedfiles()

            f.get_record_stream([key_b], 'unordered', True

                ).next().get_bytes_as('fulltext'))

 

    def test_get_record_stream_empty(self):

        """An empty stream can be requested without error."""

        f = self.get_versionedfiles()

             'knit-ft', 'knit-delta', 'chunked', 'fulltext',

             'knit-annotated-ft-gz', 'knit-annotated-delta-gz', 'knit-ft-gz',

             'knit-delta-gz',

 

        """Capture a stream for testing."""

        for factory in entries:

            on_seen(factory.key)

            self.assertValidStorageKind(factory.storage_kind)

            self.assertEqual(parents[factory.key], factory.parents)

            self.assertIsInstance(factory.get_bytes_as(factory.storage_kind),

                str)

 

    def test_get_record_stream_interface(self):

        """each item in a stream has to provide a regular interface."""

        self.capture_stream(files, entries, seen.add, parent_map)

        self.assertEqual(set(keys), seen)

 

    def get_keys_and_sort_order(self):

        """Get diamond test keys list, and their sort ordering."""

        if self.key_length == 1:

                }

        return keys, sort_order

 

    def test_get_record_stream_interface_ordered(self):

        """each item in a stream has to provide a regular interface."""

        files = self.get_versionedfiles()

 

        self.assertStreamOrder(sort_order, seen, keys)

 

    def assertStreamOrder(self, sort_order, seen, keys):

        self.assertEqual(len(set(seen)), len(keys))

        if self.key_length == 1:

        for factory in entries:

            seen.add(factory.key)

            self.assertValidStorageKind(factory.storage_kind)

            self.assertEqual(parent_map[factory.key], factory.parents)

            # currently no stream emits mpdiff

            self.assertRaises(errors.UnavailableRepresentation,

 

    def assertStreamMetaEqual(self, records, expected, stream):

        """Assert that streams expected and stream have the same records.

        :param records: A list to collect the seen records.

        :return: A generator of the records in stream.

        """

 

        :param skipped_records: A list with one element to increment when a

            record is skipped.

            checking chunked or fulltext stored records.

        :param stream: A record_stream.

        :return: An iterator over the bytes of each record.

                self.assertEqual(None, factory.parents)

            else:

                self.assertValidStorageKind(factory.storage_kind)

                self.assertEqual(parents[factory.key], factory.parents)

                self.assertIsInstance(factory.get_bytes_as(factory.storage_kind),

                    str)

        else:

            return None