1
# Copyright (C) 2005 by Canonical Ltd
4
# Johan Rydberg <jrydberg@gnu.org>
6
# This program is free software; you can redistribute it and/or modify
7
# it under the terms of the GNU General Public License as published by
8
# the Free Software Foundation; either version 2 of the License, or
9
# (at your option) any later version.
11
# This program is distributed in the hope that it will be useful,
12
# but WITHOUT ANY WARRANTY; without even the implied warranty of
13
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14
# GNU General Public License for more details.
16
# You should have received a copy of the GNU General Public License
17
# along with this program; if not, write to the Free Software
18
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
21
# TODO: might be nice to create a versionedfile with some type of corruption
22
# considered typical and check that it can be detected/corrected.
24
from StringIO import StringIO
31
from bzrlib.errors import (
33
RevisionAlreadyPresent,
36
from bzrlib.knit import KnitVersionedFile, \
38
from bzrlib.tests import TestCaseWithTransport
39
from bzrlib.tests.HTTPTestUtil import TestCaseWithWebserver
40
from bzrlib.trace import mutter
41
from bzrlib.transport import get_transport
42
from bzrlib.transport.memory import MemoryTransport
43
from bzrlib.tsort import topo_sort
44
import bzrlib.versionedfile as versionedfile
45
from bzrlib.weave import WeaveFile
46
from bzrlib.weavefile import read_weave, write_weave
49
class VersionedFileTestMixIn(object):
50
"""A mixin test class for testing VersionedFiles.
52
This is not an adaptor-style test at this point because
53
theres no dynamic substitution of versioned file implementations,
54
they are strictly controlled by their owning repositories.
59
f.add_lines('r0', [], ['a\n', 'b\n'])
60
f.add_lines('r1', ['r0'], ['b\n', 'c\n'])
62
versions = f.versions()
63
self.assertTrue('r0' in versions)
64
self.assertTrue('r1' in versions)
65
self.assertEquals(f.get_lines('r0'), ['a\n', 'b\n'])
66
self.assertEquals(f.get_text('r0'), 'a\nb\n')
67
self.assertEquals(f.get_lines('r1'), ['b\n', 'c\n'])
68
self.assertEqual(2, len(f))
69
self.assertEqual(2, f.num_versions())
71
self.assertRaises(RevisionNotPresent,
72
f.add_lines, 'r2', ['foo'], [])
73
self.assertRaises(RevisionAlreadyPresent,
74
f.add_lines, 'r1', [], [])
76
# this checks that reopen with create=True does not break anything.
77
f = self.reopen_file(create=True)
80
def test_adds_with_parent_texts(self):
83
parent_texts['r0'] = f.add_lines('r0', [], ['a\n', 'b\n'])
85
parent_texts['r1'] = f.add_lines_with_ghosts('r1',
88
parent_texts=parent_texts)
89
except NotImplementedError:
90
# if the format doesn't support ghosts, just add normally.
91
parent_texts['r1'] = f.add_lines('r1',
94
parent_texts=parent_texts)
95
f.add_lines('r2', ['r1'], ['c\n', 'd\n'], parent_texts=parent_texts)
96
self.assertNotEqual(None, parent_texts['r0'])
97
self.assertNotEqual(None, parent_texts['r1'])
99
versions = f.versions()
100
self.assertTrue('r0' in versions)
101
self.assertTrue('r1' in versions)
102
self.assertTrue('r2' in versions)
103
self.assertEquals(f.get_lines('r0'), ['a\n', 'b\n'])
104
self.assertEquals(f.get_lines('r1'), ['b\n', 'c\n'])
105
self.assertEquals(f.get_lines('r2'), ['c\n', 'd\n'])
106
self.assertEqual(3, f.num_versions())
107
origins = f.annotate('r1')
108
self.assertEquals(origins[0][0], 'r0')
109
self.assertEquals(origins[1][0], 'r1')
110
origins = f.annotate('r2')
111
self.assertEquals(origins[0][0], 'r1')
112
self.assertEquals(origins[1][0], 'r2')
115
f = self.reopen_file()
118
def test_add_unicode_content(self):
119
# unicode content is not permitted in versioned files.
120
# versioned files version sequences of bytes only.
122
self.assertRaises(errors.BzrBadParameterUnicode,
123
vf.add_lines, 'a', [], ['a\n', u'b\n', 'c\n'])
125
(errors.BzrBadParameterUnicode, NotImplementedError),
126
vf.add_lines_with_ghosts, 'a', [], ['a\n', u'b\n', 'c\n'])
128
def test_inline_newline_throws(self):
129
# \r characters are not permitted in lines being added
131
self.assertRaises(errors.BzrBadParameterContainsNewline,
132
vf.add_lines, 'a', [], ['a\n\n'])
134
(errors.BzrBadParameterContainsNewline, NotImplementedError),
135
vf.add_lines_with_ghosts, 'a', [], ['a\n\n'])
136
# but inline CR's are allowed
137
vf.add_lines('a', [], ['a\r\n'])
139
vf.add_lines_with_ghosts('b', [], ['a\r\n'])
140
except NotImplementedError:
143
def test_get_delta(self):
145
sha1s = self._setup_for_deltas(f)
146
expected_delta = (None, '6bfa09d82ce3e898ad4641ae13dd4fdb9cf0d76b', False,
147
[(0, 0, 1, [('base', 'line\n')])])
148
self.assertEqual(expected_delta, f.get_delta('base'))
150
text_name = 'chain1-'
151
for depth in range(26):
152
new_version = text_name + '%s' % depth
153
expected_delta = (next_parent, sha1s[depth],
155
[(depth + 1, depth + 1, 1, [(new_version, 'line\n')])])
156
self.assertEqual(expected_delta, f.get_delta(new_version))
157
next_parent = new_version
159
text_name = 'chain2-'
160
for depth in range(26):
161
new_version = text_name + '%s' % depth
162
expected_delta = (next_parent, sha1s[depth], False,
163
[(depth + 1, depth + 1, 1, [(new_version, 'line\n')])])
164
self.assertEqual(expected_delta, f.get_delta(new_version))
165
next_parent = new_version
166
# smoke test for eol support
167
expected_delta = ('base', '264f39cab871e4cfd65b3a002f7255888bb5ed97', True, [])
168
self.assertEqual(['line'], f.get_lines('noeol'))
169
self.assertEqual(expected_delta, f.get_delta('noeol'))
171
def test_get_deltas(self):
173
sha1s = self._setup_for_deltas(f)
174
deltas = f.get_deltas(f.versions())
175
expected_delta = (None, '6bfa09d82ce3e898ad4641ae13dd4fdb9cf0d76b', False,
176
[(0, 0, 1, [('base', 'line\n')])])
177
self.assertEqual(expected_delta, deltas['base'])
179
text_name = 'chain1-'
180
for depth in range(26):
181
new_version = text_name + '%s' % depth
182
expected_delta = (next_parent, sha1s[depth],
184
[(depth + 1, depth + 1, 1, [(new_version, 'line\n')])])
185
self.assertEqual(expected_delta, deltas[new_version])
186
next_parent = new_version
188
text_name = 'chain2-'
189
for depth in range(26):
190
new_version = text_name + '%s' % depth
191
expected_delta = (next_parent, sha1s[depth], False,
192
[(depth + 1, depth + 1, 1, [(new_version, 'line\n')])])
193
self.assertEqual(expected_delta, deltas[new_version])
194
next_parent = new_version
195
# smoke tests for eol support
196
expected_delta = ('base', '264f39cab871e4cfd65b3a002f7255888bb5ed97', True, [])
197
self.assertEqual(['line'], f.get_lines('noeol'))
198
self.assertEqual(expected_delta, deltas['noeol'])
199
# smoke tests for eol support - two noeol in a row same content
200
expected_deltas = (('noeol', '3ad7ee82dbd8f29ecba073f96e43e414b3f70a4d', True,
201
[(0, 1, 2, [(u'noeolsecond', 'line\n'), (u'noeolsecond', 'line\n')])]),
202
('noeol', '3ad7ee82dbd8f29ecba073f96e43e414b3f70a4d', True,
203
[(0, 0, 1, [('noeolsecond', 'line\n')]), (1, 1, 0, [])]))
204
self.assertEqual(['line\n', 'line'], f.get_lines('noeolsecond'))
205
self.assertTrue(deltas['noeolsecond'] in expected_deltas)
206
# two no-eol in a row, different content
207
expected_delta = ('noeolsecond', '8bb553a84e019ef1149db082d65f3133b195223b', True,
208
[(1, 2, 1, [(u'noeolnotshared', 'phone\n')])])
209
self.assertEqual(['line\n', 'phone'], f.get_lines('noeolnotshared'))
210
self.assertEqual(expected_delta, deltas['noeolnotshared'])
211
# eol folling a no-eol with content change
212
expected_delta = ('noeol', 'a61f6fb6cfc4596e8d88c34a308d1e724caf8977', False,
213
[(0, 1, 1, [(u'eol', 'phone\n')])])
214
self.assertEqual(['phone\n'], f.get_lines('eol'))
215
self.assertEqual(expected_delta, deltas['eol'])
216
# eol folling a no-eol with content change
217
expected_delta = ('noeol', '6bfa09d82ce3e898ad4641ae13dd4fdb9cf0d76b', False,
218
[(0, 1, 1, [(u'eolline', 'line\n')])])
219
self.assertEqual(['line\n'], f.get_lines('eolline'))
220
self.assertEqual(expected_delta, deltas['eolline'])
221
# eol with no parents
222
expected_delta = (None, '264f39cab871e4cfd65b3a002f7255888bb5ed97', True,
223
[(0, 0, 1, [(u'noeolbase', 'line\n')])])
224
self.assertEqual(['line'], f.get_lines('noeolbase'))
225
self.assertEqual(expected_delta, deltas['noeolbase'])
226
# eol with two parents, in inverse insertion order
227
expected_deltas = (('noeolbase', '264f39cab871e4cfd65b3a002f7255888bb5ed97', True,
228
[(0, 1, 1, [(u'eolbeforefirstparent', 'line\n')])]),
229
('noeolbase', '264f39cab871e4cfd65b3a002f7255888bb5ed97', True,
230
[(0, 1, 1, [(u'eolbeforefirstparent', 'line\n')])]))
231
self.assertEqual(['line'], f.get_lines('eolbeforefirstparent'))
232
#self.assertTrue(deltas['eolbeforefirstparent'] in expected_deltas)
234
def _setup_for_deltas(self, f):
235
self.assertRaises(errors.RevisionNotPresent, f.get_delta, 'base')
236
# add texts that should trip the knit maximum delta chain threshold
237
# as well as doing parallel chains of data in knits.
238
# this is done by two chains of 25 insertions
239
f.add_lines('base', [], ['line\n'])
240
f.add_lines('noeol', ['base'], ['line'])
241
# detailed eol tests:
242
# shared last line with parent no-eol
243
f.add_lines('noeolsecond', ['noeol'], ['line\n', 'line'])
244
# differing last line with parent, both no-eol
245
f.add_lines('noeolnotshared', ['noeolsecond'], ['line\n', 'phone'])
246
# add eol following a noneol parent, change content
247
f.add_lines('eol', ['noeol'], ['phone\n'])
248
# add eol following a noneol parent, no change content
249
f.add_lines('eolline', ['noeol'], ['line\n'])
250
# noeol with no parents:
251
f.add_lines('noeolbase', [], ['line'])
252
# noeol preceeding its leftmost parent in the output:
253
# this is done by making it a merge of two parents with no common
254
# anestry: noeolbase and noeol with the
255
# later-inserted parent the leftmost.
256
f.add_lines('eolbeforefirstparent', ['noeolbase', 'noeol'], ['line'])
257
# two identical eol texts
258
f.add_lines('noeoldup', ['noeol'], ['line'])
260
text_name = 'chain1-'
262
sha1s = {0 :'da6d3141cb4a5e6f464bf6e0518042ddc7bfd079',
263
1 :'45e21ea146a81ea44a821737acdb4f9791c8abe7',
264
2 :'e1f11570edf3e2a070052366c582837a4fe4e9fa',
265
3 :'26b4b8626da827088c514b8f9bbe4ebf181edda1',
266
4 :'e28a5510be25ba84d31121cff00956f9970ae6f6',
267
5 :'d63ec0ce22e11dcf65a931b69255d3ac747a318d',
268
6 :'2c2888d288cb5e1d98009d822fedfe6019c6a4ea',
269
7 :'95c14da9cafbf828e3e74a6f016d87926ba234ab',
270
8 :'779e9a0b28f9f832528d4b21e17e168c67697272',
271
9 :'1f8ff4e5c6ff78ac106fcfe6b1e8cb8740ff9a8f',
272
10:'131a2ae712cf51ed62f143e3fbac3d4206c25a05',
273
11:'c5a9d6f520d2515e1ec401a8f8a67e6c3c89f199',
274
12:'31a2286267f24d8bedaa43355f8ad7129509ea85',
275
13:'dc2a7fe80e8ec5cae920973973a8ee28b2da5e0a',
276
14:'2c4b1736566b8ca6051e668de68650686a3922f2',
277
15:'5912e4ecd9b0c07be4d013e7e2bdcf9323276cde',
278
16:'b0d2e18d3559a00580f6b49804c23fea500feab3',
279
17:'8e1d43ad72f7562d7cb8f57ee584e20eb1a69fc7',
280
18:'5cf64a3459ae28efa60239e44b20312d25b253f3',
281
19:'1ebed371807ba5935958ad0884595126e8c4e823',
282
20:'2aa62a8b06fb3b3b892a3292a068ade69d5ee0d3',
283
21:'01edc447978004f6e4e962b417a4ae1955b6fe5d',
284
22:'d8d8dc49c4bf0bab401e0298bb5ad827768618bb',
285
23:'c21f62b1c482862983a8ffb2b0c64b3451876e3f',
286
24:'c0593fe795e00dff6b3c0fe857a074364d5f04fc',
287
25:'dd1a1cf2ba9cc225c3aff729953e6364bf1d1855',
289
for depth in range(26):
290
new_version = text_name + '%s' % depth
291
text = text + ['line\n']
292
f.add_lines(new_version, [next_parent], text)
293
next_parent = new_version
295
text_name = 'chain2-'
297
for depth in range(26):
298
new_version = text_name + '%s' % depth
299
text = text + ['line\n']
300
f.add_lines(new_version, [next_parent], text)
301
next_parent = new_version
304
def test_add_delta(self):
305
# tests for the add-delta facility.
306
# at this point, optimising for speed, we assume no checks when deltas are inserted.
307
# this may need to be revisited.
308
source = self.get_file('source')
309
source.add_lines('base', [], ['line\n'])
311
text_name = 'chain1-'
313
for depth in range(26):
314
new_version = text_name + '%s' % depth
315
text = text + ['line\n']
316
source.add_lines(new_version, [next_parent], text)
317
next_parent = new_version
319
text_name = 'chain2-'
321
for depth in range(26):
322
new_version = text_name + '%s' % depth
323
text = text + ['line\n']
324
source.add_lines(new_version, [next_parent], text)
325
next_parent = new_version
326
source.add_lines('noeol', ['base'], ['line'])
328
target = self.get_file('target')
329
for version in source.versions():
330
parent, sha1, noeol, delta = source.get_delta(version)
331
target.add_delta(version,
332
source.get_parents(version),
337
self.assertRaises(RevisionAlreadyPresent,
338
target.add_delta, 'base', [], None, '', False, [])
339
for version in source.versions():
340
self.assertEqual(source.get_lines(version),
341
target.get_lines(version))
343
def test_ancestry(self):
345
self.assertEqual([], f.get_ancestry([]))
346
f.add_lines('r0', [], ['a\n', 'b\n'])
347
f.add_lines('r1', ['r0'], ['b\n', 'c\n'])
348
f.add_lines('r2', ['r0'], ['b\n', 'c\n'])
349
f.add_lines('r3', ['r2'], ['b\n', 'c\n'])
350
f.add_lines('rM', ['r1', 'r2'], ['b\n', 'c\n'])
351
self.assertEqual([], f.get_ancestry([]))
352
versions = f.get_ancestry(['rM'])
353
# there are some possibilities:
357
# so we check indexes
358
r0 = versions.index('r0')
359
r1 = versions.index('r1')
360
r2 = versions.index('r2')
361
self.assertFalse('r3' in versions)
362
rM = versions.index('rM')
363
self.assertTrue(r0 < r1)
364
self.assertTrue(r0 < r2)
365
self.assertTrue(r1 < rM)
366
self.assertTrue(r2 < rM)
368
self.assertRaises(RevisionNotPresent,
369
f.get_ancestry, ['rM', 'rX'])
371
def test_mutate_after_finish(self):
373
f.transaction_finished()
374
self.assertRaises(errors.OutSideTransaction, f.add_delta, '', [], '', '', False, [])
375
self.assertRaises(errors.OutSideTransaction, f.add_lines, '', [], [])
376
self.assertRaises(errors.OutSideTransaction, f.add_lines_with_ghosts, '', [], [])
377
self.assertRaises(errors.OutSideTransaction, f.fix_parents, '', [])
378
self.assertRaises(errors.OutSideTransaction, f.join, '')
379
self.assertRaises(errors.OutSideTransaction, f.clone_text, 'base', 'bar', ['foo'])
381
def test_clear_cache(self):
383
# on a new file it should not error
385
# and after adding content, doing a clear_cache and a get should work.
386
f.add_lines('0', [], ['a'])
388
self.assertEqual(['a'], f.get_lines('0'))
390
def test_clone_text(self):
392
f.add_lines('r0', [], ['a\n', 'b\n'])
393
f.clone_text('r1', 'r0', ['r0'])
395
self.assertEquals(f.get_lines('r1'), f.get_lines('r0'))
396
self.assertEquals(f.get_lines('r1'), ['a\n', 'b\n'])
397
self.assertEquals(f.get_parents('r1'), ['r0'])
399
self.assertRaises(RevisionNotPresent,
400
f.clone_text, 'r2', 'rX', [])
401
self.assertRaises(RevisionAlreadyPresent,
402
f.clone_text, 'r1', 'r0', [])
404
verify_file(self.reopen_file())
406
def test_create_empty(self):
408
f.add_lines('0', [], ['a\n'])
409
new_f = f.create_empty('t', MemoryTransport())
410
# smoke test, specific types should check it is honoured correctly for
411
# non type attributes
412
self.assertEqual([], new_f.versions())
413
self.assertTrue(isinstance(new_f, f.__class__))
415
def test_copy_to(self):
417
f.add_lines('0', [], ['a\n'])
418
t = MemoryTransport()
420
for suffix in f.__class__.get_suffixes():
421
self.assertTrue(t.has('foo' + suffix))
423
def test_get_suffixes(self):
426
self.assertEqual(f.__class__.get_suffixes(), f.__class__.get_suffixes())
427
# and should be a list
428
self.assertTrue(isinstance(f.__class__.get_suffixes(), list))
430
def build_graph(self, file, graph):
431
for node in topo_sort(graph.items()):
432
file.add_lines(node, graph[node], [])
434
def test_get_graph(self):
440
self.build_graph(f, graph)
441
self.assertEqual(graph, f.get_graph())
443
def test_get_graph_partial(self):
451
complex_graph.update(simple_a)
456
complex_graph.update(simple_b)
463
complex_graph.update(simple_gam)
465
simple_b_gam.update(simple_gam)
466
simple_b_gam.update(simple_b)
467
self.build_graph(f, complex_graph)
468
self.assertEqual(simple_a, f.get_graph(['a']))
469
self.assertEqual(simple_b, f.get_graph(['b']))
470
self.assertEqual(simple_gam, f.get_graph(['gam']))
471
self.assertEqual(simple_b_gam, f.get_graph(['b', 'gam']))
473
def test_get_parents(self):
475
f.add_lines('r0', [], ['a\n', 'b\n'])
476
f.add_lines('r1', [], ['a\n', 'b\n'])
477
f.add_lines('r2', [], ['a\n', 'b\n'])
478
f.add_lines('r3', [], ['a\n', 'b\n'])
479
f.add_lines('m', ['r0', 'r1', 'r2', 'r3'], ['a\n', 'b\n'])
480
self.assertEquals(f.get_parents('m'), ['r0', 'r1', 'r2', 'r3'])
482
self.assertRaises(RevisionNotPresent,
485
def test_annotate(self):
487
f.add_lines('r0', [], ['a\n', 'b\n'])
488
f.add_lines('r1', ['r0'], ['c\n', 'b\n'])
489
origins = f.annotate('r1')
490
self.assertEquals(origins[0][0], 'r1')
491
self.assertEquals(origins[1][0], 'r0')
493
self.assertRaises(RevisionNotPresent,
497
# tests that walk returns all the inclusions for the requested
498
# revisions as well as the revisions changes themselves.
499
f = self.get_file('1')
500
f.add_lines('r0', [], ['a\n', 'b\n'])
501
f.add_lines('r1', ['r0'], ['c\n', 'b\n'])
502
f.add_lines('rX', ['r1'], ['d\n', 'b\n'])
503
f.add_lines('rY', ['r1'], ['c\n', 'e\n'])
506
for lineno, insert, dset, text in f.walk(['rX', 'rY']):
507
lines[text] = (insert, dset)
509
self.assertTrue(lines['a\n'], ('r0', set(['r1'])))
510
self.assertTrue(lines['b\n'], ('r0', set(['rY'])))
511
self.assertTrue(lines['c\n'], ('r1', set(['rX'])))
512
self.assertTrue(lines['d\n'], ('rX', set([])))
513
self.assertTrue(lines['e\n'], ('rY', set([])))
515
def test_detection(self):
516
# Test weaves detect corruption.
518
# Weaves contain a checksum of their texts.
519
# When a text is extracted, this checksum should be
522
w = self.get_file_corrupted_text()
524
self.assertEqual('hello\n', w.get_text('v1'))
525
self.assertRaises(errors.WeaveInvalidChecksum, w.get_text, 'v2')
526
self.assertRaises(errors.WeaveInvalidChecksum, w.get_lines, 'v2')
527
self.assertRaises(errors.WeaveInvalidChecksum, w.check)
529
w = self.get_file_corrupted_checksum()
531
self.assertEqual('hello\n', w.get_text('v1'))
532
self.assertRaises(errors.WeaveInvalidChecksum, w.get_text, 'v2')
533
self.assertRaises(errors.WeaveInvalidChecksum, w.get_lines, 'v2')
534
self.assertRaises(errors.WeaveInvalidChecksum, w.check)
536
def get_file_corrupted_text(self):
537
"""Return a versioned file with corrupt text but valid metadata."""
538
raise NotImplementedError(self.get_file_corrupted_text)
540
def reopen_file(self, name='foo'):
541
"""Open the versioned file from disk again."""
542
raise NotImplementedError(self.reopen_file)
544
def test_iter_lines_added_or_present_in_versions(self):
545
# test that we get at least an equalset of the lines added by
546
# versions in the weave
547
# the ordering here is to make a tree so that dumb searches have
548
# more changes to muck up.
550
class InstrumentedProgress(progress.DummyProgress):
554
progress.DummyProgress.__init__(self)
557
def update(self, msg=None, current=None, total=None):
558
self.updates.append((msg, current, total))
561
# add a base to get included
562
vf.add_lines('base', [], ['base\n'])
563
# add a ancestor to be included on one side
564
vf.add_lines('lancestor', [], ['lancestor\n'])
565
# add a ancestor to be included on the other side
566
vf.add_lines('rancestor', ['base'], ['rancestor\n'])
567
# add a child of rancestor with no eofile-nl
568
vf.add_lines('child', ['rancestor'], ['base\n', 'child\n'])
569
# add a child of lancestor and base to join the two roots
570
vf.add_lines('otherchild',
571
['lancestor', 'base'],
572
['base\n', 'lancestor\n', 'otherchild\n'])
573
def iter_with_versions(versions, expected):
574
# now we need to see what lines are returned, and how often.
581
progress = InstrumentedProgress()
582
# iterate over the lines
583
for line in vf.iter_lines_added_or_present_in_versions(versions,
586
if []!= progress.updates:
587
self.assertEqual(expected, progress.updates)
589
lines = iter_with_versions(['child', 'otherchild'],
590
[('Walking content.', 0, 2),
591
('Walking content.', 0, 2),
592
('Walking content.', 3, 2),
593
('Walking content.', 2, 2)])
594
# we must see child and otherchild
595
self.assertTrue(lines['child\n'] > 0)
596
self.assertTrue(lines['otherchild\n'] > 0)
597
# we dont care if we got more than that.
600
lines = iter_with_versions(None, [('Walking content.', 0, 5),
601
('Walking content.', 0, 5),
602
('Walking content.', 1, 5),
603
('Walking content.', 2, 5),
604
('Walking content.', 2, 5),
605
('Walking content.', 3, 5),
606
('Walking content.', 5, 5)])
607
# all lines must be seen at least once
608
self.assertTrue(lines['base\n'] > 0)
609
self.assertTrue(lines['lancestor\n'] > 0)
610
self.assertTrue(lines['rancestor\n'] > 0)
611
self.assertTrue(lines['child\n'] > 0)
612
self.assertTrue(lines['otherchild\n'] > 0)
614
def test_fix_parents(self):
615
# some versioned files allow incorrect parents to be corrected after
616
# insertion - this may not fix ancestry..
617
# if they do not supported, they just do not implement it.
618
# we test this as an interface test to ensure that those that *do*
619
# implementent it get it right.
621
vf.add_lines('notbase', [], [])
622
vf.add_lines('base', [], [])
624
vf.fix_parents('notbase', ['base'])
625
except NotImplementedError:
627
self.assertEqual(['base'], vf.get_parents('notbase'))
628
# open again, check it stuck.
630
self.assertEqual(['base'], vf.get_parents('notbase'))
632
def test_fix_parents_with_ghosts(self):
633
# when fixing parents, ghosts that are listed should not be ghosts
638
vf.add_lines_with_ghosts('notbase', ['base', 'stillghost'], [])
639
except NotImplementedError:
641
vf.add_lines('base', [], [])
642
vf.fix_parents('notbase', ['base', 'stillghost'])
643
self.assertEqual(['base'], vf.get_parents('notbase'))
644
# open again, check it stuck.
646
self.assertEqual(['base'], vf.get_parents('notbase'))
647
# and check the ghosts
648
self.assertEqual(['base', 'stillghost'],
649
vf.get_parents_with_ghosts('notbase'))
651
def test_add_lines_with_ghosts(self):
652
# some versioned file formats allow lines to be added with parent
653
# information that is > than that in the format. Formats that do
654
# not support this need to raise NotImplementedError on the
655
# add_lines_with_ghosts api.
657
# add a revision with ghost parents
659
vf.add_lines_with_ghosts(u'notbxbfse', [u'b\xbfse'], [])
660
except NotImplementedError:
661
# check the other ghost apis are also not implemented
662
self.assertRaises(NotImplementedError, vf.has_ghost, 'foo')
663
self.assertRaises(NotImplementedError, vf.get_ancestry_with_ghosts, ['foo'])
664
self.assertRaises(NotImplementedError, vf.get_parents_with_ghosts, 'foo')
665
self.assertRaises(NotImplementedError, vf.get_graph_with_ghosts)
667
# test key graph related apis: getncestry, _graph, get_parents
669
# - these are ghost unaware and must not be reflect ghosts
670
self.assertEqual([u'notbxbfse'], vf.get_ancestry(u'notbxbfse'))
671
self.assertEqual([], vf.get_parents(u'notbxbfse'))
672
self.assertEqual({u'notbxbfse':[]}, vf.get_graph())
673
self.assertFalse(vf.has_version(u'b\xbfse'))
674
# we have _with_ghost apis to give us ghost information.
675
self.assertEqual([u'b\xbfse', u'notbxbfse'], vf.get_ancestry_with_ghosts([u'notbxbfse']))
676
self.assertEqual([u'b\xbfse'], vf.get_parents_with_ghosts(u'notbxbfse'))
677
self.assertEqual({u'notbxbfse':[u'b\xbfse']}, vf.get_graph_with_ghosts())
678
self.assertTrue(vf.has_ghost(u'b\xbfse'))
679
# if we add something that is a ghost of another, it should correct the
680
# results of the prior apis
681
vf.add_lines(u'b\xbfse', [], [])
682
self.assertEqual([u'b\xbfse', u'notbxbfse'], vf.get_ancestry([u'notbxbfse']))
683
self.assertEqual([u'b\xbfse'], vf.get_parents(u'notbxbfse'))
684
self.assertEqual({u'b\xbfse':[],
685
u'notbxbfse':[u'b\xbfse'],
688
self.assertTrue(vf.has_version(u'b\xbfse'))
689
# we have _with_ghost apis to give us ghost information.
690
self.assertEqual([u'b\xbfse', u'notbxbfse'], vf.get_ancestry_with_ghosts([u'notbxbfse']))
691
self.assertEqual([u'b\xbfse'], vf.get_parents_with_ghosts(u'notbxbfse'))
692
self.assertEqual({u'b\xbfse':[],
693
u'notbxbfse':[u'b\xbfse'],
695
vf.get_graph_with_ghosts())
696
self.assertFalse(vf.has_ghost(u'b\xbfse'))
698
def test_add_lines_with_ghosts_after_normal_revs(self):
699
# some versioned file formats allow lines to be added with parent
700
# information that is > than that in the format. Formats that do
701
# not support this need to raise NotImplementedError on the
702
# add_lines_with_ghosts api.
704
# probe for ghost support
707
except NotImplementedError:
709
vf.add_lines_with_ghosts('base', [], ['line\n', 'line_b\n'])
710
vf.add_lines_with_ghosts('references_ghost',
712
['line\n', 'line_b\n', 'line_c\n'])
713
origins = vf.annotate('references_ghost')
714
self.assertEquals(('base', 'line\n'), origins[0])
715
self.assertEquals(('base', 'line_b\n'), origins[1])
716
self.assertEquals(('references_ghost', 'line_c\n'), origins[2])
718
def test_readonly_mode(self):
719
transport = get_transport(self.get_url('.'))
720
factory = self.get_factory()
721
vf = factory('id', transport, 0777, create=True, access_mode='w')
722
vf = factory('id', transport, access_mode='r')
723
self.assertRaises(errors.ReadOnlyError, vf.add_delta, '', [], '', '', False, [])
724
self.assertRaises(errors.ReadOnlyError, vf.add_lines, 'base', [], [])
725
self.assertRaises(errors.ReadOnlyError,
726
vf.add_lines_with_ghosts,
730
self.assertRaises(errors.ReadOnlyError, vf.fix_parents, 'base', [])
731
self.assertRaises(errors.ReadOnlyError, vf.join, 'base')
732
self.assertRaises(errors.ReadOnlyError, vf.clone_text, 'base', 'bar', ['foo'])
734
def test_get_sha1(self):
735
# check the sha1 data is available
738
vf.add_lines('a', [], ['a\n'])
739
# the same file, different metadata
740
vf.add_lines('b', ['a'], ['a\n'])
741
# a file differing only in last newline.
742
vf.add_lines('c', [], ['a'])
744
'3f786850e387550fdab836ed7e6dc881de23001b', vf.get_sha1('a'))
746
'3f786850e387550fdab836ed7e6dc881de23001b', vf.get_sha1('b'))
748
'86f7e437faa5a7fce15d1ddcb9eaeaea377667b8', vf.get_sha1('c'))
751
class TestWeave(TestCaseWithTransport, VersionedFileTestMixIn):
753
def get_file(self, name='foo'):
754
return WeaveFile(name, get_transport(self.get_url('.')), create=True)
756
def get_file_corrupted_text(self):
757
w = WeaveFile('foo', get_transport(self.get_url('.')), create=True)
758
w.add_lines('v1', [], ['hello\n'])
759
w.add_lines('v2', ['v1'], ['hello\n', 'there\n'])
761
# We are going to invasively corrupt the text
762
# Make sure the internals of weave are the same
763
self.assertEqual([('{', 0)
771
self.assertEqual(['f572d396fae9206628714fb2ce00f72e94f2258f'
772
, '90f265c6e75f1c8f9ab76dcf85528352c5f215ef'
777
w._weave[4] = 'There\n'
780
def get_file_corrupted_checksum(self):
781
w = self.get_file_corrupted_text()
783
w._weave[4] = 'there\n'
784
self.assertEqual('hello\nthere\n', w.get_text('v2'))
786
#Invalid checksum, first digit changed
787
w._sha1s[1] = 'f0f265c6e75f1c8f9ab76dcf85528352c5f215ef'
790
def reopen_file(self, name='foo', create=False):
791
return WeaveFile(name, get_transport(self.get_url('.')), create=create)
793
def test_no_implicit_create(self):
794
self.assertRaises(errors.NoSuchFile,
797
get_transport(self.get_url('.')))
799
def get_factory(self):
803
class TestKnit(TestCaseWithTransport, VersionedFileTestMixIn):
805
def get_file(self, name='foo'):
806
return KnitVersionedFile(name, get_transport(self.get_url('.')),
807
delta=True, create=True)
809
def get_factory(self):
810
return KnitVersionedFile
812
def get_file_corrupted_text(self):
813
knit = self.get_file()
814
knit.add_lines('v1', [], ['hello\n'])
815
knit.add_lines('v2', ['v1'], ['hello\n', 'there\n'])
818
def reopen_file(self, name='foo', create=False):
819
return KnitVersionedFile(name, get_transport(self.get_url('.')),
823
def test_detection(self):
824
knit = self.get_file()
827
def test_no_implicit_create(self):
828
self.assertRaises(errors.NoSuchFile,
831
get_transport(self.get_url('.')))
834
class InterString(versionedfile.InterVersionedFile):
835
"""An inter-versionedfile optimised code path for strings.
837
This is for use during testing where we use strings as versionedfiles
838
so that none of the default regsitered interversionedfile classes will
839
match - which lets us test the match logic.
843
def is_compatible(source, target):
844
"""InterString is compatible with strings-as-versionedfiles."""
845
return isinstance(source, str) and isinstance(target, str)
848
# TODO this and the InterRepository core logic should be consolidatable
849
# if we make the registry a separate class though we still need to
850
# test the behaviour in the active registry to catch failure-to-handle-
852
class TestInterVersionedFile(TestCaseWithTransport):
854
def test_get_default_inter_versionedfile(self):
855
# test that the InterVersionedFile.get(a, b) probes
856
# for a class where is_compatible(a, b) returns
857
# true and returns a default interversionedfile otherwise.
858
# This also tests that the default registered optimised interversionedfile
859
# classes do not barf inappropriately when a surprising versionedfile type
861
dummy_a = "VersionedFile 1."
862
dummy_b = "VersionedFile 2."
863
self.assertGetsDefaultInterVersionedFile(dummy_a, dummy_b)
865
def assertGetsDefaultInterVersionedFile(self, a, b):
866
"""Asserts that InterVersionedFile.get(a, b) -> the default."""
867
inter = versionedfile.InterVersionedFile.get(a, b)
868
self.assertEqual(versionedfile.InterVersionedFile,
870
self.assertEqual(a, inter.source)
871
self.assertEqual(b, inter.target)
873
def test_register_inter_versionedfile_class(self):
874
# test that a optimised code path provider - a
875
# InterVersionedFile subclass can be registered and unregistered
876
# and that it is correctly selected when given a versionedfile
877
# pair that it returns true on for the is_compatible static method
879
dummy_a = "VersionedFile 1."
880
dummy_b = "VersionedFile 2."
881
versionedfile.InterVersionedFile.register_optimiser(InterString)
883
# we should get the default for something InterString returns False
885
self.assertFalse(InterString.is_compatible(dummy_a, None))
886
self.assertGetsDefaultInterVersionedFile(dummy_a, None)
887
# and we should get an InterString for a pair it 'likes'
888
self.assertTrue(InterString.is_compatible(dummy_a, dummy_b))
889
inter = versionedfile.InterVersionedFile.get(dummy_a, dummy_b)
890
self.assertEqual(InterString, inter.__class__)
891
self.assertEqual(dummy_a, inter.source)
892
self.assertEqual(dummy_b, inter.target)
894
versionedfile.InterVersionedFile.unregister_optimiser(InterString)
895
# now we should get the default InterVersionedFile object again.
896
self.assertGetsDefaultInterVersionedFile(dummy_a, dummy_b)
899
class TestReadonlyHttpMixin(object):
901
def test_readonly_http_works(self):
902
# we should be able to read from http with a versioned file.
904
# try an empty file access
905
readonly_vf = self.get_factory()('foo', get_transport(self.get_readonly_url('.')))
906
self.assertEqual([], readonly_vf.versions())
908
vf.add_lines('1', [], ['a\n'])
909
vf.add_lines('2', ['1'], ['b\n', 'a\n'])
910
readonly_vf = self.get_factory()('foo', get_transport(self.get_readonly_url('.')))
911
self.assertEqual(['1', '2'], vf.versions())
912
for version in readonly_vf.versions():
913
readonly_vf.get_lines(version)
916
class TestWeaveHTTP(TestCaseWithWebserver, TestReadonlyHttpMixin):
919
return WeaveFile('foo', get_transport(self.get_url('.')), create=True)
921
def get_factory(self):
925
class TestKnitHTTP(TestCaseWithWebserver, TestReadonlyHttpMixin):
928
return KnitVersionedFile('foo', get_transport(self.get_url('.')),
929
delta=True, create=True)
931
def get_factory(self):
932
return KnitVersionedFile
935
class MergeCasesMixin(object):
937
def doMerge(self, base, a, b, mp):
938
from cStringIO import StringIO
939
from textwrap import dedent
945
w.add_lines('text0', [], map(addcrlf, base))
946
w.add_lines('text1', ['text0'], map(addcrlf, a))
947
w.add_lines('text2', ['text0'], map(addcrlf, b))
951
self.log('merge plan:')
952
p = list(w.plan_merge('text1', 'text2'))
953
for state, line in p:
955
self.log('%12s | %s' % (state, line[:-1]))
959
mt.writelines(w.weave_merge(p))
961
self.log(mt.getvalue())
963
mp = map(addcrlf, mp)
964
self.assertEqual(mt.readlines(), mp)
967
def testOneInsert(self):
973
def testSeparateInserts(self):
974
self.doMerge(['aaa', 'bbb', 'ccc'],
975
['aaa', 'xxx', 'bbb', 'ccc'],
976
['aaa', 'bbb', 'yyy', 'ccc'],
977
['aaa', 'xxx', 'bbb', 'yyy', 'ccc'])
979
def testSameInsert(self):
980
self.doMerge(['aaa', 'bbb', 'ccc'],
981
['aaa', 'xxx', 'bbb', 'ccc'],
982
['aaa', 'xxx', 'bbb', 'yyy', 'ccc'],
983
['aaa', 'xxx', 'bbb', 'yyy', 'ccc'])
984
overlappedInsertExpected = ['aaa', 'xxx', 'yyy', 'bbb']
985
def testOverlappedInsert(self):
986
self.doMerge(['aaa', 'bbb'],
987
['aaa', 'xxx', 'yyy', 'bbb'],
988
['aaa', 'xxx', 'bbb'], self.overlappedInsertExpected)
990
# really it ought to reduce this to
991
# ['aaa', 'xxx', 'yyy', 'bbb']
994
def testClashReplace(self):
995
self.doMerge(['aaa'],
998
['<<<<<<< ', 'xxx', '=======', 'yyy', 'zzz',
1001
def testNonClashInsert1(self):
1002
self.doMerge(['aaa'],
1005
['<<<<<<< ', 'xxx', 'aaa', '=======', 'yyy', 'zzz',
1008
def testNonClashInsert2(self):
1009
self.doMerge(['aaa'],
1015
def testDeleteAndModify(self):
1016
"""Clashing delete and modification.
1018
If one side modifies a region and the other deletes it then
1019
there should be a conflict with one side blank.
1022
#######################################
1023
# skippd, not working yet
1026
self.doMerge(['aaa', 'bbb', 'ccc'],
1027
['aaa', 'ddd', 'ccc'],
1029
['<<<<<<<< ', 'aaa', '=======', '>>>>>>> ', 'ccc'])
1031
def _test_merge_from_strings(self, base, a, b, expected):
1033
w.add_lines('text0', [], base.splitlines(True))
1034
w.add_lines('text1', ['text0'], a.splitlines(True))
1035
w.add_lines('text2', ['text0'], b.splitlines(True))
1036
self.log('merge plan:')
1037
p = list(w.plan_merge('text1', 'text2'))
1038
for state, line in p:
1040
self.log('%12s | %s' % (state, line[:-1]))
1041
self.log('merge result:')
1042
result_text = ''.join(w.weave_merge(p))
1043
self.log(result_text)
1044
self.assertEqualDiff(result_text, expected)
1046
def test_weave_merge_conflicts(self):
1047
# does weave merge properly handle plans that end with unchanged?
1048
result = ''.join(self.get_file().weave_merge([('new-a', 'hello\n')]))
1049
self.assertEqual(result, 'hello\n')
1051
def test_deletion_extended(self):
1052
"""One side deletes, the other deletes more.
1069
self._test_merge_from_strings(base, a, b, result)
1071
def test_deletion_overlap(self):
1072
"""Delete overlapping regions with no other conflict.
1074
Arguably it'd be better to treat these as agreement, rather than
1075
conflict, but for now conflict is safer.
1103
self._test_merge_from_strings(base, a, b, result)
1105
def test_agreement_deletion(self):
1106
"""Agree to delete some lines, without conflicts."""
1128
self._test_merge_from_strings(base, a, b, result)
1130
def test_sync_on_deletion(self):
1131
"""Specific case of merge where we can synchronize incorrectly.
1133
A previous version of the weave merge concluded that the two versions
1134
agreed on deleting line 2, and this could be a synchronization point.
1135
Line 1 was then considered in isolation, and thought to be deleted on
1138
It's better to consider the whole thing as a disagreement region.
1149
a's replacement line 2
1162
a's replacement line 2
1169
self._test_merge_from_strings(base, a, b, result)
1172
class TestKnitMerge(TestCaseWithTransport, MergeCasesMixin):
1174
def get_file(self, name='foo'):
1175
return KnitVersionedFile(name, get_transport(self.get_url('.')),
1176
delta=True, create=True)
1178
def log_contents(self, w):
1182
class TestWeaveMerge(TestCaseWithTransport, MergeCasesMixin):
1184
def get_file(self, name='foo'):
1185
return WeaveFile(name, get_transport(self.get_url('.')), create=True)
1187
def log_contents(self, w):
1188
self.log('weave is:')
1190
write_weave(w, tmpf)
1191
self.log(tmpf.getvalue())
1193
overlappedInsertExpected = ['aaa', '<<<<<<< ', 'xxx', 'yyy', '=======',
1194
'xxx', '>>>>>>> ', 'bbb']
added
removed
bzrlib/tests/test_versionedfile.py
