638
ver = k.add_lines('text%d' % i,
686
ver = k.add('text%d' % i,
639
687
list(parents), t)
640
parents.add('text%d' % i)
643
691
self.log("k._weave=" + pformat(k._weave))
645
693
for i, t in enumerate(texts):
646
self.assertEqual(k.get_lines(i), t)
694
self.assertEqual(k.get(i), t)
648
696
self.check_read_write(k)
700
class MergeCases(TestBase):
701
def doMerge(self, base, a, b, mp):
702
from cStringIO import StringIO
703
from textwrap import dedent
709
w.add('text0', [], map(addcrlf, base))
710
w.add('text1', [0], map(addcrlf, a))
711
w.add('text2', [0], map(addcrlf, b))
713
self.log('weave is:')
716
self.log(tmpf.getvalue())
718
self.log('merge plan:')
719
p = list(w.plan_merge(1, 2))
720
for state, line in p:
722
self.log('%12s | %s' % (state, line[:-1]))
726
mt.writelines(w.weave_merge(p))
728
self.log(mt.getvalue())
730
mp = map(addcrlf, mp)
731
self.assertEqual(mt.readlines(), mp)
734
def testOneInsert(self):
740
def testSeparateInserts(self):
741
self.doMerge(['aaa', 'bbb', 'ccc'],
742
['aaa', 'xxx', 'bbb', 'ccc'],
743
['aaa', 'bbb', 'yyy', 'ccc'],
744
['aaa', 'xxx', 'bbb', 'yyy', 'ccc'])
746
def testSameInsert(self):
747
self.doMerge(['aaa', 'bbb', 'ccc'],
748
['aaa', 'xxx', 'bbb', 'ccc'],
749
['aaa', 'xxx', 'bbb', 'yyy', 'ccc'],
750
['aaa', 'xxx', 'bbb', 'yyy', 'ccc'])
752
def testOverlappedInsert(self):
753
self.doMerge(['aaa', 'bbb'],
754
['aaa', 'xxx', 'yyy', 'bbb'],
755
['aaa', 'xxx', 'bbb'],
756
['aaa', '<<<<', 'xxx', 'yyy', '====', 'xxx', '>>>>', 'bbb'])
758
# really it ought to reduce this to
759
# ['aaa', 'xxx', 'yyy', 'bbb']
762
def testClashReplace(self):
763
self.doMerge(['aaa'],
766
['<<<<', 'xxx', '====', 'yyy', 'zzz', '>>>>'])
768
def testNonClashInsert(self):
769
self.doMerge(['aaa'],
772
['<<<<', 'xxx', 'aaa', '====', 'yyy', 'zzz', '>>>>'])
774
self.doMerge(['aaa'],
780
def testDeleteAndModify(self):
781
"""Clashing delete and modification.
783
If one side modifies a region and the other deletes it then
784
there should be a conflict with one side blank.
787
#######################################
788
# skippd, not working yet
791
self.doMerge(['aaa', 'bbb', 'ccc'],
792
['aaa', 'ddd', 'ccc'],
794
['<<<<', 'aaa', '====', '>>>>', 'ccc'])
651
797
class JoinWeavesTests(TestBase):
653
799
super(JoinWeavesTests, self).setUp()
654
800
self.weave1 = Weave()
655
801
self.lines1 = ['hello\n']
656
802
self.lines3 = ['hello\n', 'cruel\n', 'world\n']
657
self.weave1.add_lines('v1', [], self.lines1)
658
self.weave1.add_lines('v2', ['v1'], ['hello\n', 'world\n'])
659
self.weave1.add_lines('v3', ['v2'], self.lines3)
661
def test_written_detection(self):
662
# Test detection of weave file corruption.
664
# Make sure that we can detect if a weave file has
665
# been corrupted. This doesn't test all forms of corruption,
666
# but it at least helps verify the data you get, is what you want.
667
from cStringIO import StringIO
670
w.add_lines('v1', [], ['hello\n'])
671
w.add_lines('v2', ['v1'], ['hello\n', 'there\n'])
676
# Because we are corrupting, we need to make sure we have the exact text
677
self.assertEquals('# bzr weave file v5\n'
678
'i\n1 f572d396fae9206628714fb2ce00f72e94f2258f\nn v1\n\n'
679
'i 0\n1 90f265c6e75f1c8f9ab76dcf85528352c5f215ef\nn v2\n\n'
680
'w\n{ 0\n. hello\n}\n{ 1\n. there\n}\nW\n',
683
# Change a single letter
684
tmpf = StringIO('# bzr weave file v5\n'
685
'i\n1 f572d396fae9206628714fb2ce00f72e94f2258f\nn v1\n\n'
686
'i 0\n1 90f265c6e75f1c8f9ab76dcf85528352c5f215ef\nn v2\n\n'
687
'w\n{ 0\n. hello\n}\n{ 1\n. There\n}\nW\n')
691
self.assertEqual('hello\n', w.get_text('v1'))
692
self.assertRaises(errors.WeaveInvalidChecksum, w.get_text, 'v2')
693
self.assertRaises(errors.WeaveInvalidChecksum, w.get_lines, 'v2')
694
self.assertRaises(errors.WeaveInvalidChecksum, w.check)
696
# Change the sha checksum
697
tmpf = StringIO('# bzr weave file v5\n'
698
'i\n1 f572d396fae9206628714fb2ce00f72e94f2258f\nn v1\n\n'
699
'i 0\n1 f0f265c6e75f1c8f9ab76dcf85528352c5f215ef\nn v2\n\n'
700
'w\n{ 0\n. hello\n}\n{ 1\n. there\n}\nW\n')
704
self.assertEqual('hello\n', w.get_text('v1'))
705
self.assertRaises(errors.WeaveInvalidChecksum, w.get_text, 'v2')
706
self.assertRaises(errors.WeaveInvalidChecksum, w.get_lines, 'v2')
707
self.assertRaises(errors.WeaveInvalidChecksum, w.check)
710
class TestWeave(TestCase):
712
def test_allow_reserved_false(self):
713
w = Weave('name', allow_reserved=False)
714
# Add lines is checked at the WeaveFile level, not at the Weave level
715
w.add_lines('name:', [], TEXT_1)
716
# But get_lines is checked at this level
717
self.assertRaises(errors.ReservedId, w.get_lines, 'name:')
719
def test_allow_reserved_true(self):
720
w = Weave('name', allow_reserved=True)
721
w.add_lines('name:', [], TEXT_1)
722
self.assertEqual(TEXT_1, w.get_lines('name:'))
725
class InstrumentedWeave(Weave):
726
"""Keep track of how many times functions are called."""
728
def __init__(self, weave_name=None):
729
self._extract_count = 0
730
Weave.__init__(self, weave_name=weave_name)
732
def _extract(self, versions):
733
self._extract_count += 1
734
return Weave._extract(self, versions)
737
class TestNeedsReweave(TestCase):
738
"""Internal corner cases for when reweave is needed."""
740
def test_compatible_parents(self):
742
my_parents = set([1, 2, 3])
744
self.assertTrue(w1._compatible_parents(my_parents, set([3])))
746
self.assertTrue(w1._compatible_parents(my_parents, set(my_parents)))
747
# same empty corner case
748
self.assertTrue(w1._compatible_parents(set(), set()))
749
# other cannot contain stuff my_parents does not
750
self.assertFalse(w1._compatible_parents(set(), set([1])))
751
self.assertFalse(w1._compatible_parents(my_parents, set([1, 2, 3, 4])))
752
self.assertFalse(w1._compatible_parents(my_parents, set([4])))
755
class TestWeaveFile(TestCaseInTempDir):
757
def test_empty_file(self):
758
f = open('empty.weave', 'wb+')
760
self.assertRaises(errors.WeaveFormatError,
803
self.weave1.add('v1', [], self.lines1)
804
self.weave1.add('v2', [0], ['hello\n', 'world\n'])
805
self.weave1.add('v3', [1], self.lines3)
807
def test_join_empty(self):
808
"""Join two empty weaves."""
809
eq = self.assertEqual
813
eq(w1.numversions(), 0)
815
def test_join_empty_to_nonempty(self):
816
"""Join empty weave onto nonempty."""
817
self.weave1.join(Weave())
818
self.assertEqual(len(self.weave1), 3)
820
def test_join_unrelated(self):
821
"""Join two weaves with no history in common."""
823
wb.add('b1', [], ['line from b\n'])
826
eq = self.assertEqual
828
eq(sorted(list(w1.iter_names())),
829
['b1', 'v1', 'v2', 'v3'])
831
def test_join_related(self):
832
wa = self.weave1.copy()
833
wb = self.weave1.copy()
834
wa.add('a1', ['v3'], ['hello\n', 'sweet\n', 'world\n'])
835
wb.add('b1', ['v3'], ['hello\n', 'pale blue\n', 'world\n'])
836
eq = self.assertEquals
841
eq(wa.get_lines('b1'),
842
['hello\n', 'pale blue\n', 'world\n'])
844
def test_join_parent_disagreement(self):
845
"""Cannot join weaves with different parents for a version."""
848
wa.add('v1', [], ['hello\n'])
850
wb.add('v1', ['v0'], ['hello\n'])
851
self.assertRaises(WeaveError,
854
def test_join_text_disagreement(self):
855
"""Cannot join weaves with different texts for a version."""
858
wa.add('v1', [], ['hello\n'])
859
wb.add('v1', [], ['not\n', 'hello\n'])
860
self.assertRaises(WeaveError,
863
def test_join_unordered(self):
864
"""Join weaves where indexes differ.
866
The source weave contains a different version at index 0."""
867
wa = self.weave1.copy()
869
wb.add('x1', [], ['line from x1\n'])
870
wb.add('v1', [], ['hello\n'])
871
wb.add('v2', ['v1'], ['hello\n', 'world\n'])
873
eq = self.assertEquals
874
eq(sorted(wa.iter_names()), ['v1', 'v2', 'v3', 'x1',])
875
eq(wa.get_text('x1'), 'line from x1\n')
877
def test_reweave_with_empty(self):
879
wr = reweave(self.weave1, wb)
880
eq = self.assertEquals
881
eq(sorted(wr.iter_names()), ['v1', 'v2', 'v3'])
882
eq(wr.get_lines('v3'), ['hello\n', 'cruel\n', 'world\n'])
883
self.weave1.reweave(wb)
884
self.assertEquals(wr, self.weave1)
886
def test_join_with_ghosts_raises_parent_mismatch(self):
887
wa = self.weave1.copy()
889
wb.add('x1', [], ['line from x1\n'])
890
wb.add('v1', [], ['hello\n'])
891
wb.add('v2', ['v1', 'x1'], ['hello\n', 'world\n'])
892
self.assertRaises(errors.WeaveParentMismatch, wa.join, wb)
894
def test_reweave_with_ghosts(self):
895
"""Join that inserts parents of an existing revision.
897
This can happen when merging from another branch who
898
knows about revisions the destination does not. In
899
this test the second weave knows of an additional parent of
900
v2. Any revisions which are in common still have to have the
902
wa = self.weave1.copy()
904
wb.add('x1', [], ['line from x1\n'])
905
wb.add('v1', [], ['hello\n'])
906
wb.add('v2', ['v1', 'x1'], ['hello\n', 'world\n'])
908
eq = self.assertEquals
909
eq(sorted(wc.iter_names()), ['v1', 'v2', 'v3', 'x1',])
910
eq(wc.get_text('x1'), 'line from x1\n')
911
eq(wc.get_lines('v2'), ['hello\n', 'world\n'])
912
eq(wc.parent_names('v2'), ['v1', 'x1'])
913
self.weave1.reweave(wb)
914
self.assertEquals(wc, self.weave1)
917
if __name__ == '__main__':
920
sys.exit(unittest.main())
added
removed
bzrlib/selftest/test_weave.py
