78
73
class StoreText(TestBase):
79
74
"""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)
77
idx = k.add([], TEXT_0)
78
self.assertEqual(k.get(idx), TEXT_0)
85
79
self.assertEqual(idx, 0)
88
83
class AnnotateOne(TestBase):
91
k.add_lines('text0', [], TEXT_0)
92
self.assertEqual(k.annotate('text0'),
93
[('text0', TEXT_0[0])])
87
self.assertEqual(k.annotate(0),
96
91
class StoreTwo(TestBase):
100
idx = k.add_lines('text0', [], TEXT_0)
95
idx = k.add([], TEXT_0)
101
96
self.assertEqual(idx, 0)
103
idx = k.add_lines('text1', [], TEXT_1)
98
idx = k.add([], TEXT_1)
104
99
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):
113
k.add_lines('text0', [], 'text0')
114
self.assertEqual('34dc0e430c642a26c3dd1c2beb7a8b4f4445eb79',
116
self.assertRaises(errors.RevisionNotPresent,
118
self.assertRaises(errors.RevisionNotPresent,
101
self.assertEqual(k.get(0), TEXT_0)
102
self.assertEqual(k.get(1), TEXT_1)
104
k.dump(self.TEST_LOG)
122
108
class InvalidAdd(TestBase):
123
109
"""Try to use invalid version number during add."""
124
110
def runTest(self):
127
self.assertRaises(errors.RevisionNotPresent,
113
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
119
class InsertLines(TestBase):
161
120
"""Store a revision that adds one line to the original.
165
124
def runTest(self):
168
k.add_lines('text0', [], ['line 1'])
169
k.add_lines('text1', ['text0'], ['line 1', 'line 2'])
171
self.assertEqual(k.annotate('text0'),
172
[('text0', 'line 1')])
174
self.assertEqual(k.get_lines(1),
127
k.add([], ['line 1'])
128
k.add([0], ['line 1', 'line 2'])
130
self.assertEqual(k.annotate(0),
133
self.assertEqual(k.get(1),
178
self.assertEqual(k.annotate('text1'),
179
[('text0', 'line 1'),
180
('text1', 'line 2')])
182
k.add_lines('text2', ['text0'], ['line 1', 'diverged line'])
184
self.assertEqual(k.annotate('text2'),
185
[('text0', 'line 1'),
186
('text2', 'diverged line')])
137
self.assertEqual(k.annotate(1),
141
k.add([0], ['line 1', 'diverged line'])
143
self.assertEqual(k.annotate(2),
145
(2, 'diverged line')])
188
147
text3 = ['line 1', 'middle line', 'line 2']
193
151
# self.log("changes to text3: " + pformat(list(k._delta(set([0, 1]), text3))))
195
153
self.log("k._weave=" + pformat(k._weave))
197
self.assertEqual(k.annotate('text3'),
198
[('text0', 'line 1'),
199
('text3', 'middle line'),
200
('text1', 'line 2')])
155
self.assertEqual(k.annotate(3),
202
160
# now multiple insertions at different places
204
['text0', 'text1', 'text3'],
205
162
['line 1', 'aaa', 'middle line', 'bbb', 'line 2', 'ccc'])
207
self.assertEqual(k.annotate('text4'),
208
[('text0', 'line 1'),
210
('text3', 'middle line'),
164
self.assertEqual(k.annotate(4),
216
174
class DeleteLines(TestBase):
779
748
self.doMerge(['aaa', 'bbb', 'ccc'],
780
749
['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])))
751
['<<<<', 'aaa', '====', '>>>>', 'ccc'])
757
from unittest import TestSuite, TestLoader
762
suite.addTest(tl.loadTestsFromModule(testweave))
764
return int(not testsweet.run_suite(suite)) # for shell 0=true
767
if __name__ == '__main__':
769
sys.exit(testweave())
added
removed
tools/testweave.py
