30
30
from bzrlib.tests import TestCaseInTempDir
31
31
from bzrlib.trace import mutter
32
32
from bzrlib.transport.local import LocalTransport
33
from bzrlib.weave import Weave
33
from bzrlib.weave import WeaveFile
34
from bzrlib.weavefile import read_weave
36
37
class VersionedFileTestMixIn(object):
45
46
f = self.get_file()
46
47
f.add_lines('r0', [], ['a\n', 'b\n'])
47
48
f.add_lines('r1', ['r0'], ['b\n', 'c\n'])
48
versions = f.versions()
49
self.assertTrue('r0' in versions)
50
self.assertTrue('r1' in versions)
51
self.assertEquals(f.get_lines('r0'), ['a\n', 'b\n'])
52
self.assertEquals(f.get_text('r0'), 'a\nb\n')
53
self.assertEquals(f.get_lines('r1'), ['b\n', 'c\n'])
55
self.assertRaises(RevisionNotPresent,
56
f.add_lines, 'r2', ['foo'], [])
57
self.assertRaises(RevisionAlreadyPresent,
58
f.add_lines, 'r1', [], [])
50
versions = f.versions()
51
self.assertTrue('r0' in versions)
52
self.assertTrue('r1' in versions)
53
self.assertEquals(f.get_lines('r0'), ['a\n', 'b\n'])
54
self.assertEquals(f.get_text('r0'), 'a\nb\n')
55
self.assertEquals(f.get_lines('r1'), ['b\n', 'c\n'])
57
self.assertRaises(RevisionNotPresent,
58
f.add_lines, 'r2', ['foo'], [])
59
self.assertRaises(RevisionAlreadyPresent,
60
f.add_lines, 'r1', [], [])
62
f = self.reopen_file()
60
65
def test_ancestry(self):
61
66
f = self.get_file()
83
88
f = self.get_file()
84
89
f.add_lines('r0', [], ['a\n', 'b\n'])
85
90
f.clone_text('r1', 'r0', ['r0'])
86
self.assertEquals(f.get_lines('r1'), f.get_lines('r0'))
87
self.assertEquals(f.get_lines('r1'), ['a\n', 'b\n'])
88
self.assertEquals(f.get_parents('r1'), ['r0'])
90
self.assertRaises(RevisionNotPresent,
91
f.clone_text, 'r2', 'rX', [])
92
self.assertRaises(RevisionAlreadyPresent,
93
f.clone_text, 'r1', 'r0', [])
92
self.assertEquals(f.get_lines('r1'), f.get_lines('r0'))
93
self.assertEquals(f.get_lines('r1'), ['a\n', 'b\n'])
94
self.assertEquals(f.get_parents('r1'), ['r0'])
96
self.assertRaises(RevisionNotPresent,
97
f.clone_text, 'r2', 'rX', [])
98
self.assertRaises(RevisionAlreadyPresent,
99
f.clone_text, 'r1', 'r0', [])
101
verify_file(self.reopen_file())
95
103
def test_get_parents(self):
96
104
f = self.get_file()
121
129
f1.add_lines('r1', ['r0'], ['c\n', 'b\n'])
122
130
f2 = self.get_file('2')
123
131
f2.join(f1, None)
124
self.assertTrue(f2.has_version('r0'))
125
self.assertTrue(f2.has_version('r1'))
133
self.assertTrue(f.has_version('r0'))
134
self.assertTrue(f.has_version('r1'))
136
verify_file(self.reopen_file('2'))
127
138
self.assertRaises(RevisionNotPresent,
128
139
f2.join, f1, version_ids=['r3'])
131
141
#f3 = self.get_file('1')
132
142
#f3.add_lines('r0', ['a\n', 'b\n'], [])
133
143
#f3.add_lines('r1', ['c\n', 'b\n'], ['r0'])
176
186
self.assertRaises(errors.WeaveInvalidChecksum, list, w.get_iter('v2'))
177
187
self.assertRaises(errors.WeaveInvalidChecksum, w.check)
180
189
def get_file_corrupted_text(self):
181
190
"""Return a versioned file with corrupt text but valid metadata."""
182
191
raise NotImplementedError(self.get_file_corrupted_text)
193
def reopen_file(self, name='foo'):
194
"""Open the versioned file from disk again."""
195
raise NotImplementedError(self.reopen_file)
185
198
class TestWeave(TestCaseInTempDir, VersionedFileTestMixIn):
187
200
def get_file(self, name='foo'):
201
return WeaveFile(name, LocalTransport('.'))
190
203
def get_file_corrupted_text(self):
204
w = WeaveFile('foo', LocalTransport('.'))
192
205
w.add('v1', [], ['hello\n'])
193
206
w.add('v2', ['v1'], ['hello\n', 'there\n'])
221
234
w._sha1s[1] = 'f0f265c6e75f1c8f9ab76dcf85528352c5f215ef'
237
def reopen_file(self, name='foo'):
238
return WeaveFile(name, LocalTransport('.'))
225
241
class TestKnit(TestCaseInTempDir, VersionedFileTestMixIn):
233
249
knit.add_lines('v1', [], ['hello\n'])
234
250
knit.add_lines('v2', ['v1'], ['hello\n', 'there\n'])
253
def reopen_file(self, name='foo'):
254
return KnitVersionedFile(LocalTransport('.'),
255
name, 'w', KnitAnnotateFactory(), delta=True)