78
74
class StoreText(TestBase):
79
75
"""Store and retrieve a simple text."""
81
def test_storing_text(self):
83
idx = k.add_lines('text0', [], TEXT_0)
84
self.assertEqual(k.get_lines(idx), TEXT_0)
78
idx = k.add([], TEXT_0)
79
self.assertEqual(k.get(idx), TEXT_0)
85
80
self.assertEqual(idx, 0)
88
84
class AnnotateOne(TestBase):
91
k.add_lines('text0', [], TEXT_0)
92
self.assertEqual(k.annotate('text0'),
93
[('text0', TEXT_0[0])])
88
self.assertEqual(k.annotate(0),
96
92
class StoreTwo(TestBase):
100
idx = k.add_lines('text0', [], TEXT_0)
96
idx = k.add([], TEXT_0)
101
97
self.assertEqual(idx, 0)
103
idx = k.add_lines('text1', [], TEXT_1)
99
idx = k.add([], TEXT_1)
104
100
self.assertEqual(idx, 1)
106
self.assertEqual(k.get_lines(0), TEXT_0)
107
self.assertEqual(k.get_lines(1), TEXT_1)
110
class GetSha1(TestBase):
111
def test_get_sha1(self):
102
self.assertEqual(k.get(0), TEXT_0)
103
self.assertEqual(k.get(1), TEXT_1)
105
k.dump(self.TEST_LOG)
109
class DeltaAdd(TestBase):
110
"""Detection of changes prior to inserting new revision."""
113
k.add_lines('text0', [], 'text0')
114
self.assertEqual('34dc0e430c642a26c3dd1c2beb7a8b4f4445eb79',
116
self.assertRaises(errors.RevisionNotPresent,
118
self.assertRaises(errors.RevisionNotPresent,
113
k.add([], ['line 1'])
115
self.assertEqual(k._l,
121
changes = list(k._delta(set([0]),
125
self.log('raw changes: ' + pformat(changes))
127
# currently there are 3 lines in the weave, and we insert after them
128
self.assertEquals(changes,
129
[(3, 3, ['new line'])])
131
changes = k._delta(set([0]),
135
self.assertEquals(list(changes),
136
[(1, 1, ['top line'])])
138
self.check_read_write(k)
122
141
class InvalidAdd(TestBase):
123
142
"""Try to use invalid version number during add."""
124
143
def runTest(self):
127
self.assertRaises(errors.RevisionNotPresent,
146
self.assertRaises(IndexError,
134
class RepeatedAdd(TestBase):
135
"""Add the same version twice; harmless."""
138
idx = k.add_lines('text0', [], TEXT_0)
139
idx2 = k.add_lines('text0', [], TEXT_0)
140
self.assertEqual(idx, idx2)
143
class InvalidRepeatedAdd(TestBase):
146
k.add_lines('basis', [], TEXT_0)
147
idx = k.add_lines('text0', [], TEXT_0)
148
self.assertRaises(errors.RevisionAlreadyPresent,
152
['not the same text'])
153
self.assertRaises(errors.RevisionAlreadyPresent,
156
['basis'], # not the right parents
160
152
class InsertLines(TestBase):
161
153
"""Store a revision that adds one line to the original.
672
ver = k.add_lines('text%d' % i,
674
parents.add('text%d' % i)
666
ver = k.add(list(parents), t)
677
self.log("k._weave=" + pformat(k._weave))
669
self.log("k._l=" + pformat(k._l))
679
671
for i, t in enumerate(texts):
680
self.assertEqual(k.get_lines(i), t)
672
self.assertEqual(k.get(i), t)
682
674
self.check_read_write(k)
685
class MergeCases(TestBase):
686
def doMerge(self, base, a, b, mp):
687
from cStringIO import StringIO
688
from textwrap import dedent
694
w.add_lines('text0', [], map(addcrlf, base))
695
w.add_lines('text1', ['text0'], map(addcrlf, a))
696
w.add_lines('text2', ['text0'], map(addcrlf, b))
698
self.log('weave is:')
701
self.log(tmpf.getvalue())
703
self.log('merge plan:')
704
p = list(w.plan_merge('text1', 'text2'))
705
for state, line in p:
707
self.log('%12s | %s' % (state, line[:-1]))
711
mt.writelines(w.weave_merge(p))
713
self.log(mt.getvalue())
715
mp = map(addcrlf, mp)
716
self.assertEqual(mt.readlines(), mp)
719
def testOneInsert(self):
725
def testSeparateInserts(self):
726
self.doMerge(['aaa', 'bbb', 'ccc'],
727
['aaa', 'xxx', 'bbb', 'ccc'],
728
['aaa', 'bbb', 'yyy', 'ccc'],
729
['aaa', 'xxx', 'bbb', 'yyy', 'ccc'])
731
def testSameInsert(self):
732
self.doMerge(['aaa', 'bbb', 'ccc'],
733
['aaa', 'xxx', 'bbb', 'ccc'],
734
['aaa', 'xxx', 'bbb', 'yyy', 'ccc'],
735
['aaa', 'xxx', 'bbb', 'yyy', 'ccc'])
737
def testOverlappedInsert(self):
738
self.doMerge(['aaa', 'bbb'],
739
['aaa', 'xxx', 'yyy', 'bbb'],
740
['aaa', 'xxx', 'bbb'],
741
['aaa', '<<<<<<< ', 'xxx', 'yyy', '=======', 'xxx',
744
# really it ought to reduce this to
745
# ['aaa', 'xxx', 'yyy', 'bbb']
748
def testClashReplace(self):
749
self.doMerge(['aaa'],
752
['<<<<<<< ', 'xxx', '=======', 'yyy', 'zzz',
755
def testNonClashInsert(self):
756
self.doMerge(['aaa'],
759
['<<<<<<< ', 'xxx', 'aaa', '=======', 'yyy', 'zzz',
762
self.doMerge(['aaa'],
768
def testDeleteAndModify(self):
769
"""Clashing delete and modification.
771
If one side modifies a region and the other deletes it then
772
there should be a conflict with one side blank.
775
#######################################
776
# skippd, not working yet
779
self.doMerge(['aaa', 'bbb', 'ccc'],
780
['aaa', 'ddd', 'ccc'],
782
['<<<<<<<< ', 'aaa', '=======', '>>>>>>> ', 'ccc'])
784
def _test_merge_from_strings(self, base, a, b, expected):
786
w.add_lines('text0', [], base.splitlines(True))
787
w.add_lines('text1', ['text0'], a.splitlines(True))
788
w.add_lines('text2', ['text0'], b.splitlines(True))
789
self.log('merge plan:')
790
p = list(w.plan_merge('text1', 'text2'))
791
for state, line in p:
793
self.log('%12s | %s' % (state, line[:-1]))
794
self.log('merge result:')
795
result_text = ''.join(w.weave_merge(p))
796
self.log(result_text)
797
self.assertEqualDiff(result_text, expected)
799
def test_deletion_extended(self):
800
"""One side deletes, the other deletes more.
817
self._test_merge_from_strings(base, a, b, result)
819
def test_deletion_overlap(self):
820
"""Delete overlapping regions with no other conflict.
822
Arguably it'd be better to treat these as agreement, rather than
823
conflict, but for now conflict is safer.
851
self._test_merge_from_strings(base, a, b, result)
853
def test_agreement_deletion(self):
854
"""Agree to delete some lines, without conflicts."""
876
self._test_merge_from_strings(base, a, b, result)
878
def test_sync_on_deletion(self):
879
"""Specific case of merge where we can synchronize incorrectly.
881
A previous version of the weave merge concluded that the two versions
882
agreed on deleting line 2, and this could be a synchronization point.
883
Line 1 was then considered in isolation, and thought to be deleted on
886
It's better to consider the whole thing as a disagreement region.
897
a's replacement line 2
910
a's replacement line 2
917
self._test_merge_from_strings(base, a, b, result)
920
class JoinWeavesTests(TestBase):
922
super(JoinWeavesTests, self).setUp()
923
self.weave1 = Weave()
924
self.lines1 = ['hello\n']
925
self.lines3 = ['hello\n', 'cruel\n', 'world\n']
926
self.weave1.add_lines('v1', [], self.lines1)
927
self.weave1.add_lines('v2', ['v1'], ['hello\n', 'world\n'])
928
self.weave1.add_lines('v3', ['v2'], self.lines3)
930
def test_join_empty(self):
931
"""Join two empty weaves."""
932
eq = self.assertEqual
938
def test_join_empty_to_nonempty(self):
939
"""Join empty weave onto nonempty."""
940
self.weave1.join(Weave())
941
self.assertEqual(len(self.weave1), 3)
943
def test_join_unrelated(self):
944
"""Join two weaves with no history in common."""
946
wb.add_lines('b1', [], ['line from b\n'])
949
eq = self.assertEqual
951
eq(sorted(w1.versions()),
952
['b1', 'v1', 'v2', 'v3'])
954
def test_join_related(self):
955
wa = self.weave1.copy()
956
wb = self.weave1.copy()
957
wa.add_lines('a1', ['v3'], ['hello\n', 'sweet\n', 'world\n'])
958
wb.add_lines('b1', ['v3'], ['hello\n', 'pale blue\n', 'world\n'])
959
eq = self.assertEquals
964
eq(wa.get_lines('b1'),
965
['hello\n', 'pale blue\n', 'world\n'])
967
def test_join_parent_disagreement(self):
968
#join reconciles differening parents into a union.
971
wa.add_lines('v1', [], ['hello\n'])
972
wb.add_lines('v0', [], [])
973
wb.add_lines('v1', ['v0'], ['hello\n'])
975
self.assertEqual(['v0'], wa.get_parents('v1'))
977
def test_join_text_disagreement(self):
978
"""Cannot join weaves with different texts for a version."""
981
wa.add_lines('v1', [], ['hello\n'])
982
wb.add_lines('v1', [], ['not\n', 'hello\n'])
983
self.assertRaises(WeaveError,
986
def test_join_unordered(self):
987
"""Join weaves where indexes differ.
989
The source weave contains a different version at index 0."""
990
wa = self.weave1.copy()
992
wb.add_lines('x1', [], ['line from x1\n'])
993
wb.add_lines('v1', [], ['hello\n'])
994
wb.add_lines('v2', ['v1'], ['hello\n', 'world\n'])
996
eq = self.assertEquals
997
eq(sorted(wa.versions()), ['v1', 'v2', 'v3', 'x1',])
998
eq(wa.get_text('x1'), 'line from x1\n')
1000
def test_written_detection(self):
1001
# Test detection of weave file corruption.
1003
# Make sure that we can detect if a weave file has
1004
# been corrupted. This doesn't test all forms of corruption,
1005
# but it at least helps verify the data you get, is what you want.
1006
from cStringIO import StringIO
1009
w.add_lines('v1', [], ['hello\n'])
1010
w.add_lines('v2', ['v1'], ['hello\n', 'there\n'])
1013
write_weave(w, tmpf)
1015
# Because we are corrupting, we need to make sure we have the exact text
1016
self.assertEquals('# bzr weave file v5\n'
1017
'i\n1 f572d396fae9206628714fb2ce00f72e94f2258f\nn v1\n\n'
1018
'i 0\n1 90f265c6e75f1c8f9ab76dcf85528352c5f215ef\nn v2\n\n'
1019
'w\n{ 0\n. hello\n}\n{ 1\n. there\n}\nW\n',
1022
# Change a single letter
1023
tmpf = StringIO('# bzr weave file v5\n'
1024
'i\n1 f572d396fae9206628714fb2ce00f72e94f2258f\nn v1\n\n'
1025
'i 0\n1 90f265c6e75f1c8f9ab76dcf85528352c5f215ef\nn v2\n\n'
1026
'w\n{ 0\n. hello\n}\n{ 1\n. There\n}\nW\n')
1028
w = read_weave(tmpf)
1030
self.assertEqual('hello\n', w.get_text('v1'))
1031
self.assertRaises(errors.WeaveInvalidChecksum, w.get_text, 'v2')
1032
self.assertRaises(errors.WeaveInvalidChecksum, w.get_lines, 'v2')
1033
self.assertRaises(errors.WeaveInvalidChecksum, w.check)
1035
# Change the sha checksum
1036
tmpf = StringIO('# bzr weave file v5\n'
1037
'i\n1 f572d396fae9206628714fb2ce00f72e94f2258f\nn v1\n\n'
1038
'i 0\n1 f0f265c6e75f1c8f9ab76dcf85528352c5f215ef\nn v2\n\n'
1039
'w\n{ 0\n. hello\n}\n{ 1\n. there\n}\nW\n')
1041
w = read_weave(tmpf)
1043
self.assertEqual('hello\n', w.get_text('v1'))
1044
self.assertRaises(errors.WeaveInvalidChecksum, w.get_text, 'v2')
1045
self.assertRaises(errors.WeaveInvalidChecksum, w.get_lines, 'v2')
1046
self.assertRaises(errors.WeaveInvalidChecksum, w.check)
1049
class InstrumentedWeave(Weave):
1050
"""Keep track of how many times functions are called."""
1052
def __init__(self, weave_name=None):
1053
self._extract_count = 0
1054
Weave.__init__(self, weave_name=weave_name)
1056
def _extract(self, versions):
1057
self._extract_count += 1
1058
return Weave._extract(self, versions)
1061
class JoinOptimization(TestCase):
1062
"""Test that Weave.join() doesn't extract all texts, only what must be done."""
1064
def test_join(self):
1065
w1 = InstrumentedWeave()
1066
w2 = InstrumentedWeave()
1069
txt1 = ['a\n', 'b\n']
1070
txt2 = ['a\n', 'c\n']
1071
txt3 = ['a\n', 'b\n', 'c\n']
1073
w1.add_lines('txt0', [], txt0) # extract 1a
1074
w2.add_lines('txt0', [], txt0) # extract 1b
1075
w1.add_lines('txt1', ['txt0'], txt1)# extract 2a
1076
w2.add_lines('txt2', ['txt0'], txt2)# extract 2b
1077
w1.join(w2) # extract 3a to add txt2
1078
w2.join(w1) # extract 3b to add txt1
1080
w1.add_lines('txt3', ['txt1', 'txt2'], txt3) # extract 4a
1081
w2.add_lines('txt3', ['txt2', 'txt1'], txt3) # extract 4b
1082
# These secretly have inverted parents
1084
# This should not have to do any extractions
1085
w1.join(w2) # NO extract, texts already present with same parents
1086
w2.join(w1) # NO extract, texts already present with same parents
1088
self.assertEqual(4, w1._extract_count)
1089
self.assertEqual(4, w2._extract_count)
1091
def test_double_parent(self):
1092
# It should not be considered illegal to add
1093
# a revision with the same parent twice
1094
w1 = InstrumentedWeave()
1095
w2 = InstrumentedWeave()
1098
txt1 = ['a\n', 'b\n']
1099
txt2 = ['a\n', 'c\n']
1100
txt3 = ['a\n', 'b\n', 'c\n']
1102
w1.add_lines('txt0', [], txt0)
1103
w2.add_lines('txt0', [], txt0)
1104
w1.add_lines('txt1', ['txt0'], txt1)
1105
w2.add_lines('txt1', ['txt0', 'txt0'], txt1)
1106
# Same text, effectively the same, because the
1107
# parent is only repeated
1108
w1.join(w2) # extract 3a to add txt2
1109
w2.join(w1) # extract 3b to add txt1
1112
class TestNeedsReweave(TestCase):
1113
"""Internal corner cases for when reweave is needed."""
1115
def test_compatible_parents(self):
1117
my_parents = set([1, 2, 3])
1119
self.assertTrue(w1._compatible_parents(my_parents, set([3])))
1121
self.assertTrue(w1._compatible_parents(my_parents, set(my_parents)))
1122
# same empty corner case
1123
self.assertTrue(w1._compatible_parents(set(), set()))
1124
# other cannot contain stuff my_parents does not
1125
self.assertFalse(w1._compatible_parents(set(), set([1])))
1126
self.assertFalse(w1._compatible_parents(my_parents, set([1, 2, 3, 4])))
1127
self.assertFalse(w1._compatible_parents(my_parents, set([4])))
679
from unittest import TestSuite, TestLoader
684
suite.addTest(tl.loadTestsFromModule(testweave))
686
return int(not testsweet.run_suite(suite)) # for shell 0=true
689
if __name__ == '__main__':
691
sys.exit(testweave())
added
removed
tools/testweave.py
