1
# Copyright (C) 2005 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_add_reserved(self):
145
self.assertRaises(errors.ReservedId,
146
vf.add_lines, 'a:', [], ['a\n', 'b\n', 'c\n'])
148
self.assertRaises(errors.ReservedId,
149
vf.add_delta, 'a:', [], None, 'sha1', False, ((0, 0, 0, []),))
151
def test_get_reserved(self):
153
self.assertRaises(errors.ReservedId, vf.get_delta, 'b:')
154
self.assertRaises(errors.ReservedId, vf.get_texts, ['b:'])
155
self.assertRaises(errors.ReservedId, vf.get_lines, 'b:')
156
self.assertRaises(errors.ReservedId, vf.get_text, 'b:')
158
def test_get_delta(self):
160
sha1s = self._setup_for_deltas(f)
161
expected_delta = (None, '6bfa09d82ce3e898ad4641ae13dd4fdb9cf0d76b', False,
162
[(0, 0, 1, [('base', 'line\n')])])
163
self.assertEqual(expected_delta, f.get_delta('base'))
165
text_name = 'chain1-'
166
for depth in range(26):
167
new_version = text_name + '%s' % depth
168
expected_delta = (next_parent, sha1s[depth],
170
[(depth + 1, depth + 1, 1, [(new_version, 'line\n')])])
171
self.assertEqual(expected_delta, f.get_delta(new_version))
172
next_parent = new_version
174
text_name = 'chain2-'
175
for depth in range(26):
176
new_version = text_name + '%s' % depth
177
expected_delta = (next_parent, sha1s[depth], False,
178
[(depth + 1, depth + 1, 1, [(new_version, 'line\n')])])
179
self.assertEqual(expected_delta, f.get_delta(new_version))
180
next_parent = new_version
181
# smoke test for eol support
182
expected_delta = ('base', '264f39cab871e4cfd65b3a002f7255888bb5ed97', True, [])
183
self.assertEqual(['line'], f.get_lines('noeol'))
184
self.assertEqual(expected_delta, f.get_delta('noeol'))
186
def test_get_deltas(self):
188
sha1s = self._setup_for_deltas(f)
189
deltas = f.get_deltas(f.versions())
190
expected_delta = (None, '6bfa09d82ce3e898ad4641ae13dd4fdb9cf0d76b', False,
191
[(0, 0, 1, [('base', 'line\n')])])
192
self.assertEqual(expected_delta, deltas['base'])
194
text_name = 'chain1-'
195
for depth in range(26):
196
new_version = text_name + '%s' % depth
197
expected_delta = (next_parent, sha1s[depth],
199
[(depth + 1, depth + 1, 1, [(new_version, 'line\n')])])
200
self.assertEqual(expected_delta, deltas[new_version])
201
next_parent = new_version
203
text_name = 'chain2-'
204
for depth in range(26):
205
new_version = text_name + '%s' % depth
206
expected_delta = (next_parent, sha1s[depth], False,
207
[(depth + 1, depth + 1, 1, [(new_version, 'line\n')])])
208
self.assertEqual(expected_delta, deltas[new_version])
209
next_parent = new_version
210
# smoke tests for eol support
211
expected_delta = ('base', '264f39cab871e4cfd65b3a002f7255888bb5ed97', True, [])
212
self.assertEqual(['line'], f.get_lines('noeol'))
213
self.assertEqual(expected_delta, deltas['noeol'])
214
# smoke tests for eol support - two noeol in a row same content
215
expected_deltas = (('noeol', '3ad7ee82dbd8f29ecba073f96e43e414b3f70a4d', True,
216
[(0, 1, 2, [('noeolsecond', 'line\n'), ('noeolsecond', 'line\n')])]),
217
('noeol', '3ad7ee82dbd8f29ecba073f96e43e414b3f70a4d', True,
218
[(0, 0, 1, [('noeolsecond', 'line\n')]), (1, 1, 0, [])]))
219
self.assertEqual(['line\n', 'line'], f.get_lines('noeolsecond'))
220
self.assertTrue(deltas['noeolsecond'] in expected_deltas)
221
# two no-eol in a row, different content
222
expected_delta = ('noeolsecond', '8bb553a84e019ef1149db082d65f3133b195223b', True,
223
[(1, 2, 1, [('noeolnotshared', 'phone\n')])])
224
self.assertEqual(['line\n', 'phone'], f.get_lines('noeolnotshared'))
225
self.assertEqual(expected_delta, deltas['noeolnotshared'])
226
# eol folling a no-eol with content change
227
expected_delta = ('noeol', 'a61f6fb6cfc4596e8d88c34a308d1e724caf8977', False,
228
[(0, 1, 1, [('eol', 'phone\n')])])
229
self.assertEqual(['phone\n'], f.get_lines('eol'))
230
self.assertEqual(expected_delta, deltas['eol'])
231
# eol folling a no-eol with content change
232
expected_delta = ('noeol', '6bfa09d82ce3e898ad4641ae13dd4fdb9cf0d76b', False,
233
[(0, 1, 1, [('eolline', 'line\n')])])
234
self.assertEqual(['line\n'], f.get_lines('eolline'))
235
self.assertEqual(expected_delta, deltas['eolline'])
236
# eol with no parents
237
expected_delta = (None, '264f39cab871e4cfd65b3a002f7255888bb5ed97', True,
238
[(0, 0, 1, [('noeolbase', 'line\n')])])
239
self.assertEqual(['line'], f.get_lines('noeolbase'))
240
self.assertEqual(expected_delta, deltas['noeolbase'])
241
# eol with two parents, in inverse insertion order
242
expected_deltas = (('noeolbase', '264f39cab871e4cfd65b3a002f7255888bb5ed97', True,
243
[(0, 1, 1, [('eolbeforefirstparent', 'line\n')])]),
244
('noeolbase', '264f39cab871e4cfd65b3a002f7255888bb5ed97', True,
245
[(0, 1, 1, [('eolbeforefirstparent', 'line\n')])]))
246
self.assertEqual(['line'], f.get_lines('eolbeforefirstparent'))
247
#self.assertTrue(deltas['eolbeforefirstparent'] in expected_deltas)
249
def _setup_for_deltas(self, f):
250
self.assertRaises(errors.RevisionNotPresent, f.get_delta, 'base')
251
# add texts that should trip the knit maximum delta chain threshold
252
# as well as doing parallel chains of data in knits.
253
# this is done by two chains of 25 insertions
254
f.add_lines('base', [], ['line\n'])
255
f.add_lines('noeol', ['base'], ['line'])
256
# detailed eol tests:
257
# shared last line with parent no-eol
258
f.add_lines('noeolsecond', ['noeol'], ['line\n', 'line'])
259
# differing last line with parent, both no-eol
260
f.add_lines('noeolnotshared', ['noeolsecond'], ['line\n', 'phone'])
261
# add eol following a noneol parent, change content
262
f.add_lines('eol', ['noeol'], ['phone\n'])
263
# add eol following a noneol parent, no change content
264
f.add_lines('eolline', ['noeol'], ['line\n'])
265
# noeol with no parents:
266
f.add_lines('noeolbase', [], ['line'])
267
# noeol preceeding its leftmost parent in the output:
268
# this is done by making it a merge of two parents with no common
269
# anestry: noeolbase and noeol with the
270
# later-inserted parent the leftmost.
271
f.add_lines('eolbeforefirstparent', ['noeolbase', 'noeol'], ['line'])
272
# two identical eol texts
273
f.add_lines('noeoldup', ['noeol'], ['line'])
275
text_name = 'chain1-'
277
sha1s = {0 :'da6d3141cb4a5e6f464bf6e0518042ddc7bfd079',
278
1 :'45e21ea146a81ea44a821737acdb4f9791c8abe7',
279
2 :'e1f11570edf3e2a070052366c582837a4fe4e9fa',
280
3 :'26b4b8626da827088c514b8f9bbe4ebf181edda1',
281
4 :'e28a5510be25ba84d31121cff00956f9970ae6f6',
282
5 :'d63ec0ce22e11dcf65a931b69255d3ac747a318d',
283
6 :'2c2888d288cb5e1d98009d822fedfe6019c6a4ea',
284
7 :'95c14da9cafbf828e3e74a6f016d87926ba234ab',
285
8 :'779e9a0b28f9f832528d4b21e17e168c67697272',
286
9 :'1f8ff4e5c6ff78ac106fcfe6b1e8cb8740ff9a8f',
287
10:'131a2ae712cf51ed62f143e3fbac3d4206c25a05',
288
11:'c5a9d6f520d2515e1ec401a8f8a67e6c3c89f199',
289
12:'31a2286267f24d8bedaa43355f8ad7129509ea85',
290
13:'dc2a7fe80e8ec5cae920973973a8ee28b2da5e0a',
291
14:'2c4b1736566b8ca6051e668de68650686a3922f2',
292
15:'5912e4ecd9b0c07be4d013e7e2bdcf9323276cde',
293
16:'b0d2e18d3559a00580f6b49804c23fea500feab3',
294
17:'8e1d43ad72f7562d7cb8f57ee584e20eb1a69fc7',
295
18:'5cf64a3459ae28efa60239e44b20312d25b253f3',
296
19:'1ebed371807ba5935958ad0884595126e8c4e823',
297
20:'2aa62a8b06fb3b3b892a3292a068ade69d5ee0d3',
298
21:'01edc447978004f6e4e962b417a4ae1955b6fe5d',
299
22:'d8d8dc49c4bf0bab401e0298bb5ad827768618bb',
300
23:'c21f62b1c482862983a8ffb2b0c64b3451876e3f',
301
24:'c0593fe795e00dff6b3c0fe857a074364d5f04fc',
302
25:'dd1a1cf2ba9cc225c3aff729953e6364bf1d1855',
304
for depth in range(26):
305
new_version = text_name + '%s' % depth
306
text = text + ['line\n']
307
f.add_lines(new_version, [next_parent], text)
308
next_parent = new_version
310
text_name = 'chain2-'
312
for depth in range(26):
313
new_version = text_name + '%s' % depth
314
text = text + ['line\n']
315
f.add_lines(new_version, [next_parent], text)
316
next_parent = new_version
319
def test_add_delta(self):
320
# tests for the add-delta facility.
321
# at this point, optimising for speed, we assume no checks when deltas are inserted.
322
# this may need to be revisited.
323
source = self.get_file('source')
324
source.add_lines('base', [], ['line\n'])
326
text_name = 'chain1-'
328
for depth in range(26):
329
new_version = text_name + '%s' % depth
330
text = text + ['line\n']
331
source.add_lines(new_version, [next_parent], text)
332
next_parent = new_version
334
text_name = 'chain2-'
336
for depth in range(26):
337
new_version = text_name + '%s' % depth
338
text = text + ['line\n']
339
source.add_lines(new_version, [next_parent], text)
340
next_parent = new_version
341
source.add_lines('noeol', ['base'], ['line'])
343
target = self.get_file('target')
344
for version in source.versions():
345
parent, sha1, noeol, delta = source.get_delta(version)
346
target.add_delta(version,
347
source.get_parents(version),
352
self.assertRaises(RevisionAlreadyPresent,
353
target.add_delta, 'base', [], None, '', False, [])
354
for version in source.versions():
355
self.assertEqual(source.get_lines(version),
356
target.get_lines(version))
358
def test_ancestry(self):
360
self.assertEqual([], f.get_ancestry([]))
361
f.add_lines('r0', [], ['a\n', 'b\n'])
362
f.add_lines('r1', ['r0'], ['b\n', 'c\n'])
363
f.add_lines('r2', ['r0'], ['b\n', 'c\n'])
364
f.add_lines('r3', ['r2'], ['b\n', 'c\n'])
365
f.add_lines('rM', ['r1', 'r2'], ['b\n', 'c\n'])
366
self.assertEqual([], f.get_ancestry([]))
367
versions = f.get_ancestry(['rM'])
368
# there are some possibilities:
372
# so we check indexes
373
r0 = versions.index('r0')
374
r1 = versions.index('r1')
375
r2 = versions.index('r2')
376
self.assertFalse('r3' in versions)
377
rM = versions.index('rM')
378
self.assertTrue(r0 < r1)
379
self.assertTrue(r0 < r2)
380
self.assertTrue(r1 < rM)
381
self.assertTrue(r2 < rM)
383
self.assertRaises(RevisionNotPresent,
384
f.get_ancestry, ['rM', 'rX'])
386
def test_mutate_after_finish(self):
388
f.transaction_finished()
389
self.assertRaises(errors.OutSideTransaction, f.add_delta, '', [], '', '', False, [])
390
self.assertRaises(errors.OutSideTransaction, f.add_lines, '', [], [])
391
self.assertRaises(errors.OutSideTransaction, f.add_lines_with_ghosts, '', [], [])
392
self.assertRaises(errors.OutSideTransaction, f.fix_parents, '', [])
393
self.assertRaises(errors.OutSideTransaction, f.join, '')
394
self.assertRaises(errors.OutSideTransaction, f.clone_text, 'base', 'bar', ['foo'])
396
def test_clear_cache(self):
398
# on a new file it should not error
400
# and after adding content, doing a clear_cache and a get should work.
401
f.add_lines('0', [], ['a'])
403
self.assertEqual(['a'], f.get_lines('0'))
405
def test_clone_text(self):
407
f.add_lines('r0', [], ['a\n', 'b\n'])
408
f.clone_text('r1', 'r0', ['r0'])
410
self.assertEquals(f.get_lines('r1'), f.get_lines('r0'))
411
self.assertEquals(f.get_lines('r1'), ['a\n', 'b\n'])
412
self.assertEquals(f.get_parents('r1'), ['r0'])
414
self.assertRaises(RevisionNotPresent,
415
f.clone_text, 'r2', 'rX', [])
416
self.assertRaises(RevisionAlreadyPresent,
417
f.clone_text, 'r1', 'r0', [])
419
verify_file(self.reopen_file())
421
def test_create_empty(self):
423
f.add_lines('0', [], ['a\n'])
424
new_f = f.create_empty('t', MemoryTransport())
425
# smoke test, specific types should check it is honoured correctly for
426
# non type attributes
427
self.assertEqual([], new_f.versions())
428
self.assertTrue(isinstance(new_f, f.__class__))
430
def test_copy_to(self):
432
f.add_lines('0', [], ['a\n'])
433
t = MemoryTransport()
435
for suffix in f.__class__.get_suffixes():
436
self.assertTrue(t.has('foo' + suffix))
438
def test_get_suffixes(self):
441
self.assertEqual(f.__class__.get_suffixes(), f.__class__.get_suffixes())
442
# and should be a list
443
self.assertTrue(isinstance(f.__class__.get_suffixes(), list))
445
def build_graph(self, file, graph):
446
for node in topo_sort(graph.items()):
447
file.add_lines(node, graph[node], [])
449
def test_get_graph(self):
455
self.build_graph(f, graph)
456
self.assertEqual(graph, f.get_graph())
458
def test_get_graph_partial(self):
466
complex_graph.update(simple_a)
471
complex_graph.update(simple_b)
478
complex_graph.update(simple_gam)
480
simple_b_gam.update(simple_gam)
481
simple_b_gam.update(simple_b)
482
self.build_graph(f, complex_graph)
483
self.assertEqual(simple_a, f.get_graph(['a']))
484
self.assertEqual(simple_b, f.get_graph(['b']))
485
self.assertEqual(simple_gam, f.get_graph(['gam']))
486
self.assertEqual(simple_b_gam, f.get_graph(['b', 'gam']))
488
def test_get_parents(self):
490
f.add_lines('r0', [], ['a\n', 'b\n'])
491
f.add_lines('r1', [], ['a\n', 'b\n'])
492
f.add_lines('r2', [], ['a\n', 'b\n'])
493
f.add_lines('r3', [], ['a\n', 'b\n'])
494
f.add_lines('m', ['r0', 'r1', 'r2', 'r3'], ['a\n', 'b\n'])
495
self.assertEquals(f.get_parents('m'), ['r0', 'r1', 'r2', 'r3'])
497
self.assertRaises(RevisionNotPresent,
500
def test_annotate(self):
502
f.add_lines('r0', [], ['a\n', 'b\n'])
503
f.add_lines('r1', ['r0'], ['c\n', 'b\n'])
504
origins = f.annotate('r1')
505
self.assertEquals(origins[0][0], 'r1')
506
self.assertEquals(origins[1][0], 'r0')
508
self.assertRaises(RevisionNotPresent,
512
# tests that walk returns all the inclusions for the requested
513
# revisions as well as the revisions changes themselves.
514
f = self.get_file('1')
515
f.add_lines('r0', [], ['a\n', 'b\n'])
516
f.add_lines('r1', ['r0'], ['c\n', 'b\n'])
517
f.add_lines('rX', ['r1'], ['d\n', 'b\n'])
518
f.add_lines('rY', ['r1'], ['c\n', 'e\n'])
521
for lineno, insert, dset, text in f.walk(['rX', 'rY']):
522
lines[text] = (insert, dset)
524
self.assertTrue(lines['a\n'], ('r0', set(['r1'])))
525
self.assertTrue(lines['b\n'], ('r0', set(['rY'])))
526
self.assertTrue(lines['c\n'], ('r1', set(['rX'])))
527
self.assertTrue(lines['d\n'], ('rX', set([])))
528
self.assertTrue(lines['e\n'], ('rY', set([])))
530
def test_detection(self):
531
# Test weaves detect corruption.
533
# Weaves contain a checksum of their texts.
534
# When a text is extracted, this checksum should be
537
w = self.get_file_corrupted_text()
539
self.assertEqual('hello\n', w.get_text('v1'))
540
self.assertRaises(errors.WeaveInvalidChecksum, w.get_text, 'v2')
541
self.assertRaises(errors.WeaveInvalidChecksum, w.get_lines, 'v2')
542
self.assertRaises(errors.WeaveInvalidChecksum, w.check)
544
w = self.get_file_corrupted_checksum()
546
self.assertEqual('hello\n', w.get_text('v1'))
547
self.assertRaises(errors.WeaveInvalidChecksum, w.get_text, 'v2')
548
self.assertRaises(errors.WeaveInvalidChecksum, w.get_lines, 'v2')
549
self.assertRaises(errors.WeaveInvalidChecksum, w.check)
551
def get_file_corrupted_text(self):
552
"""Return a versioned file with corrupt text but valid metadata."""
553
raise NotImplementedError(self.get_file_corrupted_text)
555
def reopen_file(self, name='foo'):
556
"""Open the versioned file from disk again."""
557
raise NotImplementedError(self.reopen_file)
559
def test_iter_lines_added_or_present_in_versions(self):
560
# test that we get at least an equalset of the lines added by
561
# versions in the weave
562
# the ordering here is to make a tree so that dumb searches have
563
# more changes to muck up.
565
class InstrumentedProgress(progress.DummyProgress):
569
progress.DummyProgress.__init__(self)
572
def update(self, msg=None, current=None, total=None):
573
self.updates.append((msg, current, total))
576
# add a base to get included
577
vf.add_lines('base', [], ['base\n'])
578
# add a ancestor to be included on one side
579
vf.add_lines('lancestor', [], ['lancestor\n'])
580
# add a ancestor to be included on the other side
581
vf.add_lines('rancestor', ['base'], ['rancestor\n'])
582
# add a child of rancestor with no eofile-nl
583
vf.add_lines('child', ['rancestor'], ['base\n', 'child\n'])
584
# add a child of lancestor and base to join the two roots
585
vf.add_lines('otherchild',
586
['lancestor', 'base'],
587
['base\n', 'lancestor\n', 'otherchild\n'])
588
def iter_with_versions(versions, expected):
589
# now we need to see what lines are returned, and how often.
596
progress = InstrumentedProgress()
597
# iterate over the lines
598
for line in vf.iter_lines_added_or_present_in_versions(versions,
601
if []!= progress.updates:
602
self.assertEqual(expected, progress.updates)
604
lines = iter_with_versions(['child', 'otherchild'],
605
[('Walking content.', 0, 2),
606
('Walking content.', 1, 2),
607
('Walking content.', 2, 2)])
608
# we must see child and otherchild
609
self.assertTrue(lines['child\n'] > 0)
610
self.assertTrue(lines['otherchild\n'] > 0)
611
# we dont care if we got more than that.
614
lines = iter_with_versions(None, [('Walking content.', 0, 5),
615
('Walking content.', 1, 5),
616
('Walking content.', 2, 5),
617
('Walking content.', 3, 5),
618
('Walking content.', 4, 5),
619
('Walking content.', 5, 5)])
620
# all lines must be seen at least once
621
self.assertTrue(lines['base\n'] > 0)
622
self.assertTrue(lines['lancestor\n'] > 0)
623
self.assertTrue(lines['rancestor\n'] > 0)
624
self.assertTrue(lines['child\n'] > 0)
625
self.assertTrue(lines['otherchild\n'] > 0)
627
def test_fix_parents(self):
628
# some versioned files allow incorrect parents to be corrected after
629
# insertion - this may not fix ancestry..
630
# if they do not supported, they just do not implement it.
631
# we test this as an interface test to ensure that those that *do*
632
# implementent it get it right.
634
vf.add_lines('notbase', [], [])
635
vf.add_lines('base', [], [])
637
vf.fix_parents('notbase', ['base'])
638
except NotImplementedError:
640
self.assertEqual(['base'], vf.get_parents('notbase'))
641
# open again, check it stuck.
643
self.assertEqual(['base'], vf.get_parents('notbase'))
645
def test_fix_parents_with_ghosts(self):
646
# when fixing parents, ghosts that are listed should not be ghosts
651
vf.add_lines_with_ghosts('notbase', ['base', 'stillghost'], [])
652
except NotImplementedError:
654
vf.add_lines('base', [], [])
655
vf.fix_parents('notbase', ['base', 'stillghost'])
656
self.assertEqual(['base'], vf.get_parents('notbase'))
657
# open again, check it stuck.
659
self.assertEqual(['base'], vf.get_parents('notbase'))
660
# and check the ghosts
661
self.assertEqual(['base', 'stillghost'],
662
vf.get_parents_with_ghosts('notbase'))
664
def test_add_lines_with_ghosts(self):
665
# some versioned file formats allow lines to be added with parent
666
# information that is > than that in the format. Formats that do
667
# not support this need to raise NotImplementedError on the
668
# add_lines_with_ghosts api.
670
# add a revision with ghost parents
671
# The preferred form is utf8, but we should translate when needed
672
parent_id_unicode = u'b\xbfse'
673
parent_id_utf8 = parent_id_unicode.encode('utf8')
675
vf.add_lines_with_ghosts(u'notbxbfse', [parent_id_utf8], [])
676
except NotImplementedError:
677
# check the other ghost apis are also not implemented
678
self.assertRaises(NotImplementedError, vf.has_ghost, 'foo')
679
self.assertRaises(NotImplementedError, vf.get_ancestry_with_ghosts, ['foo'])
680
self.assertRaises(NotImplementedError, vf.get_parents_with_ghosts, 'foo')
681
self.assertRaises(NotImplementedError, vf.get_graph_with_ghosts)
683
vf = self.reopen_file()
684
# test key graph related apis: getncestry, _graph, get_parents
686
# - these are ghost unaware and must not be reflect ghosts
687
self.assertEqual(['notbxbfse'], vf.get_ancestry('notbxbfse'))
688
self.assertEqual([], vf.get_parents('notbxbfse'))
689
self.assertEqual({'notbxbfse':[]}, vf.get_graph())
690
self.assertFalse(vf.has_version(parent_id_unicode))
691
self.assertFalse(vf.has_version(parent_id_utf8))
692
# we have _with_ghost apis to give us ghost information.
693
self.assertEqual([parent_id_utf8, 'notbxbfse'], vf.get_ancestry_with_ghosts(['notbxbfse']))
694
self.assertEqual([parent_id_utf8], vf.get_parents_with_ghosts('notbxbfse'))
695
self.assertEqual({'notbxbfse':[parent_id_utf8]}, vf.get_graph_with_ghosts())
696
self.assertTrue(vf.has_ghost(parent_id_unicode))
697
self.assertTrue(vf.has_ghost(parent_id_utf8))
698
# if we add something that is a ghost of another, it should correct the
699
# results of the prior apis
700
vf.add_lines(parent_id_unicode, [], [])
701
self.assertEqual([parent_id_utf8, 'notbxbfse'], vf.get_ancestry(['notbxbfse']))
702
self.assertEqual([parent_id_utf8], vf.get_parents('notbxbfse'))
703
self.assertEqual({parent_id_utf8:[],
704
'notbxbfse':[parent_id_utf8],
707
self.assertTrue(vf.has_version(parent_id_unicode))
708
self.assertTrue(vf.has_version(parent_id_utf8))
709
# we have _with_ghost apis to give us ghost information.
710
self.assertEqual([parent_id_utf8, 'notbxbfse'], vf.get_ancestry_with_ghosts(['notbxbfse']))
711
self.assertEqual([parent_id_utf8], vf.get_parents_with_ghosts('notbxbfse'))
712
self.assertEqual({parent_id_utf8:[],
713
'notbxbfse':[parent_id_utf8],
715
vf.get_graph_with_ghosts())
716
self.assertFalse(vf.has_ghost(parent_id_unicode))
717
self.assertFalse(vf.has_ghost(parent_id_utf8))
719
def test_add_lines_with_ghosts_after_normal_revs(self):
720
# some versioned file formats allow lines to be added with parent
721
# information that is > than that in the format. Formats that do
722
# not support this need to raise NotImplementedError on the
723
# add_lines_with_ghosts api.
725
# probe for ghost support
728
except NotImplementedError:
730
vf.add_lines_with_ghosts('base', [], ['line\n', 'line_b\n'])
731
vf.add_lines_with_ghosts('references_ghost',
733
['line\n', 'line_b\n', 'line_c\n'])
734
origins = vf.annotate('references_ghost')
735
self.assertEquals(('base', 'line\n'), origins[0])
736
self.assertEquals(('base', 'line_b\n'), origins[1])
737
self.assertEquals(('references_ghost', 'line_c\n'), origins[2])
739
def test_readonly_mode(self):
740
transport = get_transport(self.get_url('.'))
741
factory = self.get_factory()
742
vf = factory('id', transport, 0777, create=True, access_mode='w')
743
vf = factory('id', transport, access_mode='r')
744
self.assertRaises(errors.ReadOnlyError, vf.add_delta, '', [], '', '', False, [])
745
self.assertRaises(errors.ReadOnlyError, vf.add_lines, 'base', [], [])
746
self.assertRaises(errors.ReadOnlyError,
747
vf.add_lines_with_ghosts,
751
self.assertRaises(errors.ReadOnlyError, vf.fix_parents, 'base', [])
752
self.assertRaises(errors.ReadOnlyError, vf.join, 'base')
753
self.assertRaises(errors.ReadOnlyError, vf.clone_text, 'base', 'bar', ['foo'])
755
def test_get_sha1(self):
756
# check the sha1 data is available
759
vf.add_lines('a', [], ['a\n'])
760
# the same file, different metadata
761
vf.add_lines('b', ['a'], ['a\n'])
762
# a file differing only in last newline.
763
vf.add_lines('c', [], ['a'])
765
'3f786850e387550fdab836ed7e6dc881de23001b', vf.get_sha1('a'))
767
'3f786850e387550fdab836ed7e6dc881de23001b', vf.get_sha1('b'))
769
'86f7e437faa5a7fce15d1ddcb9eaeaea377667b8', vf.get_sha1('c'))
772
class TestWeave(TestCaseWithTransport, VersionedFileTestMixIn):
774
def get_file(self, name='foo'):
775
return WeaveFile(name, get_transport(self.get_url('.')), create=True)
777
def get_file_corrupted_text(self):
778
w = WeaveFile('foo', get_transport(self.get_url('.')), create=True)
779
w.add_lines('v1', [], ['hello\n'])
780
w.add_lines('v2', ['v1'], ['hello\n', 'there\n'])
782
# We are going to invasively corrupt the text
783
# Make sure the internals of weave are the same
784
self.assertEqual([('{', 0)
792
self.assertEqual(['f572d396fae9206628714fb2ce00f72e94f2258f'
793
, '90f265c6e75f1c8f9ab76dcf85528352c5f215ef'
798
w._weave[4] = 'There\n'
801
def get_file_corrupted_checksum(self):
802
w = self.get_file_corrupted_text()
804
w._weave[4] = 'there\n'
805
self.assertEqual('hello\nthere\n', w.get_text('v2'))
807
#Invalid checksum, first digit changed
808
w._sha1s[1] = 'f0f265c6e75f1c8f9ab76dcf85528352c5f215ef'
811
def reopen_file(self, name='foo', create=False):
812
return WeaveFile(name, get_transport(self.get_url('.')), create=create)
814
def test_no_implicit_create(self):
815
self.assertRaises(errors.NoSuchFile,
818
get_transport(self.get_url('.')))
820
def get_factory(self):
824
class TestKnit(TestCaseWithTransport, VersionedFileTestMixIn):
826
def get_file(self, name='foo'):
827
return KnitVersionedFile(name, get_transport(self.get_url('.')),
828
delta=True, create=True)
830
def get_factory(self):
831
return KnitVersionedFile
833
def get_file_corrupted_text(self):
834
knit = self.get_file()
835
knit.add_lines('v1', [], ['hello\n'])
836
knit.add_lines('v2', ['v1'], ['hello\n', 'there\n'])
839
def reopen_file(self, name='foo', create=False):
840
return KnitVersionedFile(name, get_transport(self.get_url('.')),
844
def test_detection(self):
845
knit = self.get_file()
848
def test_no_implicit_create(self):
849
self.assertRaises(errors.NoSuchFile,
852
get_transport(self.get_url('.')))
855
class InterString(versionedfile.InterVersionedFile):
856
"""An inter-versionedfile optimised code path for strings.
858
This is for use during testing where we use strings as versionedfiles
859
so that none of the default regsitered interversionedfile classes will
860
match - which lets us test the match logic.
864
def is_compatible(source, target):
865
"""InterString is compatible with strings-as-versionedfiles."""
866
return isinstance(source, str) and isinstance(target, str)
869
# TODO this and the InterRepository core logic should be consolidatable
870
# if we make the registry a separate class though we still need to
871
# test the behaviour in the active registry to catch failure-to-handle-
873
class TestInterVersionedFile(TestCaseWithTransport):
875
def test_get_default_inter_versionedfile(self):
876
# test that the InterVersionedFile.get(a, b) probes
877
# for a class where is_compatible(a, b) returns
878
# true and returns a default interversionedfile otherwise.
879
# This also tests that the default registered optimised interversionedfile
880
# classes do not barf inappropriately when a surprising versionedfile type
882
dummy_a = "VersionedFile 1."
883
dummy_b = "VersionedFile 2."
884
self.assertGetsDefaultInterVersionedFile(dummy_a, dummy_b)
886
def assertGetsDefaultInterVersionedFile(self, a, b):
887
"""Asserts that InterVersionedFile.get(a, b) -> the default."""
888
inter = versionedfile.InterVersionedFile.get(a, b)
889
self.assertEqual(versionedfile.InterVersionedFile,
891
self.assertEqual(a, inter.source)
892
self.assertEqual(b, inter.target)
894
def test_register_inter_versionedfile_class(self):
895
# test that a optimised code path provider - a
896
# InterVersionedFile subclass can be registered and unregistered
897
# and that it is correctly selected when given a versionedfile
898
# pair that it returns true on for the is_compatible static method
900
dummy_a = "VersionedFile 1."
901
dummy_b = "VersionedFile 2."
902
versionedfile.InterVersionedFile.register_optimiser(InterString)
904
# we should get the default for something InterString returns False
906
self.assertFalse(InterString.is_compatible(dummy_a, None))
907
self.assertGetsDefaultInterVersionedFile(dummy_a, None)
908
# and we should get an InterString for a pair it 'likes'
909
self.assertTrue(InterString.is_compatible(dummy_a, dummy_b))
910
inter = versionedfile.InterVersionedFile.get(dummy_a, dummy_b)
911
self.assertEqual(InterString, inter.__class__)
912
self.assertEqual(dummy_a, inter.source)
913
self.assertEqual(dummy_b, inter.target)
915
versionedfile.InterVersionedFile.unregister_optimiser(InterString)
916
# now we should get the default InterVersionedFile object again.
917
self.assertGetsDefaultInterVersionedFile(dummy_a, dummy_b)
920
class TestReadonlyHttpMixin(object):
922
def test_readonly_http_works(self):
923
# we should be able to read from http with a versioned file.
925
# try an empty file access
926
readonly_vf = self.get_factory()('foo', get_transport(self.get_readonly_url('.')))
927
self.assertEqual([], readonly_vf.versions())
929
vf.add_lines('1', [], ['a\n'])
930
vf.add_lines('2', ['1'], ['b\n', 'a\n'])
931
readonly_vf = self.get_factory()('foo', get_transport(self.get_readonly_url('.')))
932
self.assertEqual(['1', '2'], vf.versions())
933
for version in readonly_vf.versions():
934
readonly_vf.get_lines(version)
937
class TestWeaveHTTP(TestCaseWithWebserver, TestReadonlyHttpMixin):
940
return WeaveFile('foo', get_transport(self.get_url('.')), create=True)
942
def get_factory(self):
946
class TestKnitHTTP(TestCaseWithWebserver, TestReadonlyHttpMixin):
949
return KnitVersionedFile('foo', get_transport(self.get_url('.')),
950
delta=True, create=True)
952
def get_factory(self):
953
return KnitVersionedFile
956
class MergeCasesMixin(object):
958
def doMerge(self, base, a, b, mp):
959
from cStringIO import StringIO
960
from textwrap import dedent
966
w.add_lines('text0', [], map(addcrlf, base))
967
w.add_lines('text1', ['text0'], map(addcrlf, a))
968
w.add_lines('text2', ['text0'], map(addcrlf, b))
972
self.log('merge plan:')
973
p = list(w.plan_merge('text1', 'text2'))
974
for state, line in p:
976
self.log('%12s | %s' % (state, line[:-1]))
980
mt.writelines(w.weave_merge(p))
982
self.log(mt.getvalue())
984
mp = map(addcrlf, mp)
985
self.assertEqual(mt.readlines(), mp)
988
def testOneInsert(self):
994
def testSeparateInserts(self):
995
self.doMerge(['aaa', 'bbb', 'ccc'],
996
['aaa', 'xxx', 'bbb', 'ccc'],
997
['aaa', 'bbb', 'yyy', 'ccc'],
998
['aaa', 'xxx', 'bbb', 'yyy', 'ccc'])
1000
def testSameInsert(self):
1001
self.doMerge(['aaa', 'bbb', 'ccc'],
1002
['aaa', 'xxx', 'bbb', 'ccc'],
1003
['aaa', 'xxx', 'bbb', 'yyy', 'ccc'],
1004
['aaa', 'xxx', 'bbb', 'yyy', 'ccc'])
1005
overlappedInsertExpected = ['aaa', 'xxx', 'yyy', 'bbb']
1006
def testOverlappedInsert(self):
1007
self.doMerge(['aaa', 'bbb'],
1008
['aaa', 'xxx', 'yyy', 'bbb'],
1009
['aaa', 'xxx', 'bbb'], self.overlappedInsertExpected)
1011
# really it ought to reduce this to
1012
# ['aaa', 'xxx', 'yyy', 'bbb']
1015
def testClashReplace(self):
1016
self.doMerge(['aaa'],
1019
['<<<<<<< ', 'xxx', '=======', 'yyy', 'zzz',
1022
def testNonClashInsert1(self):
1023
self.doMerge(['aaa'],
1026
['<<<<<<< ', 'xxx', 'aaa', '=======', 'yyy', 'zzz',
1029
def testNonClashInsert2(self):
1030
self.doMerge(['aaa'],
1036
def testDeleteAndModify(self):
1037
"""Clashing delete and modification.
1039
If one side modifies a region and the other deletes it then
1040
there should be a conflict with one side blank.
1043
#######################################
1044
# skippd, not working yet
1047
self.doMerge(['aaa', 'bbb', 'ccc'],
1048
['aaa', 'ddd', 'ccc'],
1050
['<<<<<<<< ', 'aaa', '=======', '>>>>>>> ', 'ccc'])
1052
def _test_merge_from_strings(self, base, a, b, expected):
1054
w.add_lines('text0', [], base.splitlines(True))
1055
w.add_lines('text1', ['text0'], a.splitlines(True))
1056
w.add_lines('text2', ['text0'], b.splitlines(True))
1057
self.log('merge plan:')
1058
p = list(w.plan_merge('text1', 'text2'))
1059
for state, line in p:
1061
self.log('%12s | %s' % (state, line[:-1]))
1062
self.log('merge result:')
1063
result_text = ''.join(w.weave_merge(p))
1064
self.log(result_text)
1065
self.assertEqualDiff(result_text, expected)
1067
def test_weave_merge_conflicts(self):
1068
# does weave merge properly handle plans that end with unchanged?
1069
result = ''.join(self.get_file().weave_merge([('new-a', 'hello\n')]))
1070
self.assertEqual(result, 'hello\n')
1072
def test_deletion_extended(self):
1073
"""One side deletes, the other deletes more.
1090
self._test_merge_from_strings(base, a, b, result)
1092
def test_deletion_overlap(self):
1093
"""Delete overlapping regions with no other conflict.
1095
Arguably it'd be better to treat these as agreement, rather than
1096
conflict, but for now conflict is safer.
1124
self._test_merge_from_strings(base, a, b, result)
1126
def test_agreement_deletion(self):
1127
"""Agree to delete some lines, without conflicts."""
1149
self._test_merge_from_strings(base, a, b, result)
1151
def test_sync_on_deletion(self):
1152
"""Specific case of merge where we can synchronize incorrectly.
1154
A previous version of the weave merge concluded that the two versions
1155
agreed on deleting line 2, and this could be a synchronization point.
1156
Line 1 was then considered in isolation, and thought to be deleted on
1159
It's better to consider the whole thing as a disagreement region.
1170
a's replacement line 2
1183
a's replacement line 2
1190
self._test_merge_from_strings(base, a, b, result)
1193
class TestKnitMerge(TestCaseWithTransport, MergeCasesMixin):
1195
def get_file(self, name='foo'):
1196
return KnitVersionedFile(name, get_transport(self.get_url('.')),
1197
delta=True, create=True)
1199
def log_contents(self, w):
1203
class TestWeaveMerge(TestCaseWithTransport, MergeCasesMixin):
1205
def get_file(self, name='foo'):
1206
return WeaveFile(name, get_transport(self.get_url('.')), create=True)
1208
def log_contents(self, w):
1209
self.log('weave is:')
1211
write_weave(w, tmpf)
1212
self.log(tmpf.getvalue())
1214
overlappedInsertExpected = ['aaa', '<<<<<<< ', 'xxx', 'yyy', '=======',
1215
'xxx', '>>>>>>> ', 'bbb']
added
removed
bzrlib/tests/test_versionedfile.py
