~bzr-pqm/bzr/bzr.dev

# Copyright (C) 2005 by 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., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

from StringIO import StringIO

import bzrlib

import bzrlib.errors as errors

from bzrlib.errors import (

                           RevisionNotPresent, 

                           RevisionAlreadyPresent,

                           WeaveParentMismatch

                           )

from bzrlib.knit import KnitVersionedFile, \

     KnitAnnotateFactory

from bzrlib.tests import TestCaseWithTransport

from bzrlib.tests.HTTPTestUtil import TestCaseWithWebserver

from bzrlib.trace import mutter

from bzrlib.transport import get_transport

from bzrlib.transport.memory import MemoryTransport

from bzrlib.tsort import topo_sort

import bzrlib.versionedfile as versionedfile

from bzrlib.weave import WeaveFile

from bzrlib.weavefile import read_weave, write_weave

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

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

            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,

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

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

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

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

            origins = f.annotate('r1')

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

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

            origins = f.annotate('r2')

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

            self.assertEquals(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_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_get_delta(self):

        f = self.get_file()

        sha1s = self._setup_for_deltas(f)

        expected_delta = (None, '6bfa09d82ce3e898ad4641ae13dd4fdb9cf0d76b', False, 

                          [(0, 0, 1, [('base', 'line\n')])])

        self.assertEqual(expected_delta, f.get_delta('base'))

        next_parent = 'base'

        text_name = 'chain1-'

        for depth in range(26):

            new_version = text_name + '%s' % depth

            expected_delta = (next_parent, sha1s[depth], 

                              False,

                              [(depth + 1, depth + 1, 1, [(new_version, 'line\n')])])

            self.assertEqual(expected_delta, f.get_delta(new_version))

            next_parent = new_version

        next_parent = 'base'

        text_name = 'chain2-'

        for depth in range(26):

            new_version = text_name + '%s' % depth

            expected_delta = (next_parent, sha1s[depth], False,

                              [(depth + 1, depth + 1, 1, [(new_version, 'line\n')])])

            self.assertEqual(expected_delta, f.get_delta(new_version))

            next_parent = new_version

        # smoke test for eol support

        expected_delta = ('base', '264f39cab871e4cfd65b3a002f7255888bb5ed97', True, [])

        self.assertEqual(['line'], f.get_lines('noeol'))

        self.assertEqual(expected_delta, f.get_delta('noeol'))

    def test_get_deltas(self):

        f = self.get_file()

        sha1s = self._setup_for_deltas(f)

        deltas = f.get_deltas(f.versions())

        expected_delta = (None, '6bfa09d82ce3e898ad4641ae13dd4fdb9cf0d76b', False, 

                          [(0, 0, 1, [('base', 'line\n')])])

        self.assertEqual(expected_delta, deltas['base'])

        next_parent = 'base'

        text_name = 'chain1-'

        for depth in range(26):

            new_version = text_name + '%s' % depth

            expected_delta = (next_parent, sha1s[depth], 

                              False,

                              [(depth + 1, depth + 1, 1, [(new_version, 'line\n')])])

            self.assertEqual(expected_delta, deltas[new_version])

            next_parent = new_version

        next_parent = 'base'

        text_name = 'chain2-'

        for depth in range(26):

            new_version = text_name + '%s' % depth

            expected_delta = (next_parent, sha1s[depth], False,

                              [(depth + 1, depth + 1, 1, [(new_version, 'line\n')])])

            self.assertEqual(expected_delta, deltas[new_version])

            next_parent = new_version

        # smoke tests for eol support

        expected_delta = ('base', '264f39cab871e4cfd65b3a002f7255888bb5ed97', True, [])

        self.assertEqual(['line'], f.get_lines('noeol'))

        self.assertEqual(expected_delta, deltas['noeol'])

        # smoke tests for eol support - two noeol in a row same content

        expected_deltas = (('noeol', '3ad7ee82dbd8f29ecba073f96e43e414b3f70a4d', True, 

                          [(0, 1, 2, [(u'noeolsecond', 'line\n'), (u'noeolsecond', 'line\n')])]),

                          ('noeol', '3ad7ee82dbd8f29ecba073f96e43e414b3f70a4d', True, 

                           [(0, 0, 1, [('noeolsecond', 'line\n')]), (1, 1, 0, [])]))

        self.assertEqual(['line\n', 'line'], f.get_lines('noeolsecond'))

        self.assertTrue(deltas['noeolsecond'] in expected_deltas)

        # two no-eol in a row, different content

        expected_delta = ('noeolsecond', '8bb553a84e019ef1149db082d65f3133b195223b', True, 

                          [(1, 2, 1, [(u'noeolnotshared', 'phone\n')])])

        self.assertEqual(['line\n', 'phone'], f.get_lines('noeolnotshared'))

        self.assertEqual(expected_delta, deltas['noeolnotshared'])

        # eol folling a no-eol with content change

        expected_delta = ('noeol', 'a61f6fb6cfc4596e8d88c34a308d1e724caf8977', False, 

                          [(0, 1, 1, [(u'eol', 'phone\n')])])

        self.assertEqual(['phone\n'], f.get_lines('eol'))

        self.assertEqual(expected_delta, deltas['eol'])

        # eol folling a no-eol with content change

        expected_delta = ('noeol', '6bfa09d82ce3e898ad4641ae13dd4fdb9cf0d76b', False, 

                          [(0, 1, 1, [(u'eolline', 'line\n')])])

        self.assertEqual(['line\n'], f.get_lines('eolline'))

        self.assertEqual(expected_delta, deltas['eolline'])

        # eol with no parents

        expected_delta = (None, '264f39cab871e4cfd65b3a002f7255888bb5ed97', True, 

                          [(0, 0, 1, [(u'noeolbase', 'line\n')])])

        self.assertEqual(['line'], f.get_lines('noeolbase'))

        self.assertEqual(expected_delta, deltas['noeolbase'])

        # eol with two parents, in inverse insertion order

        expected_deltas = (('noeolbase', '264f39cab871e4cfd65b3a002f7255888bb5ed97', True,

                            [(0, 1, 1, [(u'eolbeforefirstparent', 'line\n')])]),

                           ('noeolbase', '264f39cab871e4cfd65b3a002f7255888bb5ed97', True,

                            [(0, 1, 1, [(u'eolbeforefirstparent', 'line\n')])]))

        self.assertEqual(['line'], f.get_lines('eolbeforefirstparent'))

        #self.assertTrue(deltas['eolbeforefirstparent'] in expected_deltas)

    def _setup_for_deltas(self, f):

        self.assertRaises(errors.RevisionNotPresent, f.get_delta, '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_add_delta(self):

        # tests for the add-delta facility.

        # at this point, optimising for speed, we assume no checks when deltas are inserted.

        # this may need to be revisited.

        source = self.get_file('source')

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

        next_parent = 'base'

        text_name = 'chain1-'

        text = ['line\n']

        for depth in range(26):

            new_version = text_name + '%s' % depth

            text = text + ['line\n']

            source.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']

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

            next_parent = new_version

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

        target = self.get_file('target')

        for version in source.versions():

            parent, sha1, noeol, delta = source.get_delta(version)

            target.add_delta(version,

                             source.get_parents(version),

                             parent,

                             sha1,

                             noeol,

                             delta)

        self.assertRaises(RevisionAlreadyPresent,

                          target.add_delta, 'base', [], None, '', False, [])

        for version in source.versions():

            self.assertEqual(source.get_lines(version),

                             target.get_lines(version))

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

    def test_mutate_after_finish(self):

        f = self.get_file()

        f.transaction_finished()

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

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

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

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

        self.assertRaises(errors.OutSideTransaction, f.join, '')

        self.assertRaises(errors.OutSideTransaction, f.clone_text, 'base', 'bar', ['foo'])

    def test_clear_cache(self):

        f = self.get_file()

        # on a new file it should not error

        f.clear_cache()

        # and after adding content, doing a clear_cache and a get should work.

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

        f.clear_cache()

        self.assertEqual(['a'], f.get_lines('0'))

    def test_clone_text(self):

        f = self.get_file()

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

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

        def verify_file(f):

            self.assertEquals(f.get_lines('r1'), f.get_lines('r0'))

            self.assertEquals(f.get_lines('r1'), ['a\n', 'b\n'])

            self.assertEquals(f.get_parents('r1'), ['r0'])

            self.assertRaises(RevisionNotPresent,

                f.clone_text, 'r2', 'rX', [])

            self.assertRaises(RevisionAlreadyPresent,

                f.clone_text, 'r1', 'r0', [])

        verify_file(f)

        verify_file(self.reopen_file())

    def test_create_empty(self):

        f = self.get_file()

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

        new_f = f.create_empty('t', MemoryTransport())

        # smoke test, specific types should check it is honoured correctly for

        # non type attributes

        self.assertEqual([], new_f.versions())

        self.assertTrue(isinstance(new_f, f.__class__))

    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 f.__class__.get_suffixes():

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

    def test_get_suffixes(self):

        f = self.get_file()

        # should be the same

        self.assertEqual(f.__class__.get_suffixes(), f.__class__.get_suffixes())

        # and should be a list

        self.assertTrue(isinstance(f.__class__.get_suffixes(), list))

    def build_graph(self, file, graph):

        for node in topo_sort(graph.items()):

            file.add_lines(node, graph[node], [])

    def test_get_graph(self):

        f = self.get_file()

        graph = {

            'v1': [],

            'v2': ['v1'],

            'v3': ['v2']}

        self.build_graph(f, graph)

        self.assertEqual(graph, f.get_graph())

    def test_get_graph_partial(self):

        f = self.get_file()

        complex_graph = {}

        simple_a = {

            'c': [],

            'b': ['c'],

            'a': ['b'],

            }

        complex_graph.update(simple_a)

        simple_b = {

            'c': [],

            'b': ['c'],

            }

        complex_graph.update(simple_b)

        simple_gam = {

            'c': [],

            'oo': [],

            'bar': ['oo', 'c'],

            'gam': ['bar'],

            }

        complex_graph.update(simple_gam)

        simple_b_gam = {}

        simple_b_gam.update(simple_gam)

        simple_b_gam.update(simple_b)

        self.build_graph(f, complex_graph)

        self.assertEqual(simple_a, f.get_graph(['a']))

        self.assertEqual(simple_b, f.get_graph(['b']))

        self.assertEqual(simple_gam, f.get_graph(['gam']))

        self.assertEqual(simple_b_gam, f.get_graph(['b', 'gam']))

    def test_get_parents(self):

        f = self.get_file()

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

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

        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.assertEquals(f.get_parents('m'), ['r0', 'r1', 'r2', 'r3'])

        self.assertRaises(RevisionNotPresent,

            f.get_parents, 'y')

    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.assertEquals(origins[0][0], 'r1')

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

        self.assertRaises(RevisionNotPresent,

            f.annotate, 'foo')

    def test_walk(self):

        # tests that walk returns all the inclusions for the requested

        # revisions as well as the revisions changes themselves.

        f = self.get_file('1')

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

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

        f.add_lines('rX', ['r1'], ['d\n', 'b\n'])

        f.add_lines('rY', ['r1'], ['c\n', 'e\n'])

        lines = {}

        for lineno, insert, dset, text in f.walk(['rX', 'rY']):

            lines[text] = (insert, dset)

        self.assertTrue(lines['a\n'], ('r0', set(['r1'])))

        self.assertTrue(lines['b\n'], ('r0', set(['rY'])))

        self.assertTrue(lines['c\n'], ('r1', set(['rX'])))

        self.assertTrue(lines['d\n'], ('rX', set([])))

        self.assertTrue(lines['e\n'], ('rY', set([])))

    def test_detection(self):

        # Test weaves detect corruption.

        #

        # Weaves contain a checksum of their texts.

        # When a text is extracted, this checksum should be

        # verified.

        w = self.get_file_corrupted_text()

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

        self.assertRaises(errors.WeaveInvalidChecksum, w.get_text, 'v2')

        self.assertRaises(errors.WeaveInvalidChecksum, w.get_lines, 'v2')

        self.assertRaises(errors.WeaveInvalidChecksum, w.check)

        w = self.get_file_corrupted_checksum()

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

        self.assertRaises(errors.WeaveInvalidChecksum, w.get_text, 'v2')

        self.assertRaises(errors.WeaveInvalidChecksum, w.get_lines, 'v2')

        self.assertRaises(errors.WeaveInvalidChecksum, w.check)

    def get_file_corrupted_text(self):

        """Return a versioned file with corrupt text but valid metadata."""

        raise NotImplementedError(self.get_file_corrupted_text)

    def reopen_file(self, name='foo'):

        """Open the versioned file from disk again."""

        raise NotImplementedError(self.reopen_file)

    def test_iter_lines_added_or_present_in_versions(self):

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

        # versions in the weave 

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

        # more changes to muck up.

        vf = self.get_file()

        # add a base to get included

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

        # add a ancestor to be included on one side

        vf.add_lines('lancestor', [], ['lancestor\n'])

        # add a ancestor to be included on the other side

        vf.add_lines('rancestor', ['base'], ['rancestor\n'])

        # add a child of rancestor with no eofile-nl

        vf.add_lines('child', ['rancestor'], ['base\n', 'child\n'])

        # add a child of lancestor and base to join the two roots

        vf.add_lines('otherchild',

                     ['lancestor', 'base'],

                     ['base\n', 'lancestor\n', 'otherchild\n'])

        def iter_with_versions(versions):

            # now we need to see what lines are returned, and how often.

            lines = {'base\n':0,

                     'lancestor\n':0,

                     'rancestor\n':0,

                     'child\n':0,

                     'otherchild\n':0,

                     }

            # iterate over the lines

            for line in vf.iter_lines_added_or_present_in_versions(versions):

                lines[line] += 1

            return lines

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

        # we must see child and otherchild

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

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

        # we dont care if we got more than that.

        # test all lines

        lines = iter_with_versions(None)

        # all lines must be seen at least once

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

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

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

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

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

    def test_fix_parents(self):

        # some versioned files allow incorrect parents to be corrected after

        # insertion - this may not fix ancestry..

        # if they do not supported, they just do not implement it.

        # we test this as an interface test to ensure that those that *do*

        # implementent it get it right.

        vf = self.get_file()

        vf.add_lines('notbase', [], [])

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

        try:

            vf.fix_parents('notbase', ['base'])

        except NotImplementedError:

            return

        self.assertEqual(['base'], vf.get_parents('notbase'))

        # open again, check it stuck.

        vf = self.get_file()

        self.assertEqual(['base'], vf.get_parents('notbase'))

    def test_fix_parents_with_ghosts(self):

        # when fixing parents, ghosts that are listed should not be ghosts

        # anymore.

        vf = self.get_file()

        try:

            vf.add_lines_with_ghosts('notbase', ['base', 'stillghost'], [])

        except NotImplementedError: