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
from StringIO import StringIO
24
import bzrlib.errors as errors
25
from bzrlib.errors import (
27
RevisionAlreadyPresent,
30
from bzrlib.knit import KnitVersionedFile, \
32
from bzrlib.tests import TestCaseWithTransport
33
from bzrlib.tests.HTTPTestUtil import TestCaseWithWebserver
34
from bzrlib.trace import mutter
35
from bzrlib.transport import get_transport
36
from bzrlib.transport.memory import MemoryTransport
37
import bzrlib.versionedfile as versionedfile
38
from bzrlib.weave import WeaveFile
39
from bzrlib.weavefile import read_weave, write_weave
42
class VersionedFileTestMixIn(object):
43
"""A mixin test class for testing VersionedFiles.
45
This is not an adaptor-style test at this point because
46
theres no dynamic substitution of versioned file implementations,
47
they are strictly controlled by their owning repositories.
52
f.add_lines('r0', [], ['a\n', 'b\n'])
53
f.add_lines('r1', ['r0'], ['b\n', 'c\n'])
55
versions = f.versions()
56
self.assertTrue('r0' in versions)
57
self.assertTrue('r1' in versions)
58
self.assertEquals(f.get_lines('r0'), ['a\n', 'b\n'])
59
self.assertEquals(f.get_text('r0'), 'a\nb\n')
60
self.assertEquals(f.get_lines('r1'), ['b\n', 'c\n'])
61
self.assertEqual(2, len(f))
62
self.assertEqual(2, f.num_versions())
64
self.assertRaises(RevisionNotPresent,
65
f.add_lines, 'r2', ['foo'], [])
66
self.assertRaises(RevisionAlreadyPresent,
67
f.add_lines, 'r1', [], [])
69
# this checks that reopen with create=True does not break anything.
70
f = self.reopen_file(create=True)
73
def test_adds_with_parent_texts(self):
76
parent_texts['r0'] = f.add_lines('r0', [], ['a\n', 'b\n'])
78
parent_texts['r1'] = f.add_lines_with_ghosts('r1',
81
parent_texts=parent_texts)
82
except NotImplementedError:
83
# if the format doesn't support ghosts, just add normally.
84
parent_texts['r1'] = f.add_lines('r1',
87
parent_texts=parent_texts)
88
f.add_lines('r2', ['r1'], ['c\n', 'd\n'], parent_texts=parent_texts)
89
self.assertNotEqual(None, parent_texts['r0'])
90
self.assertNotEqual(None, parent_texts['r1'])
92
versions = f.versions()
93
self.assertTrue('r0' in versions)
94
self.assertTrue('r1' in versions)
95
self.assertTrue('r2' in versions)
96
self.assertEquals(f.get_lines('r0'), ['a\n', 'b\n'])
97
self.assertEquals(f.get_lines('r1'), ['b\n', 'c\n'])
98
self.assertEquals(f.get_lines('r2'), ['c\n', 'd\n'])
99
self.assertEqual(3, f.num_versions())
100
origins = f.annotate('r1')
101
self.assertEquals(origins[0][0], 'r0')
102
self.assertEquals(origins[1][0], 'r1')
103
origins = f.annotate('r2')
104
self.assertEquals(origins[0][0], 'r1')
105
self.assertEquals(origins[1][0], 'r2')
108
f = self.reopen_file()
111
def test_add_unicode_content(self):
112
# unicode content is not permitted in versioned files.
113
# versioned files version sequences of bytes only.
115
self.assertRaises(errors.BzrBadParameterUnicode,
116
vf.add_lines, 'a', [], ['a\n', u'b\n', 'c\n'])
118
(errors.BzrBadParameterUnicode, NotImplementedError),
119
vf.add_lines_with_ghosts, 'a', [], ['a\n', u'b\n', 'c\n'])
121
def test_inline_newline_throws(self):
122
# \r characters are not permitted in lines being added
124
self.assertRaises(errors.BzrBadParameterContainsNewline,
125
vf.add_lines, 'a', [], ['a\n\n'])
127
(errors.BzrBadParameterContainsNewline, NotImplementedError),
128
vf.add_lines_with_ghosts, 'a', [], ['a\n\n'])
129
# but inline CR's are allowed
130
vf.add_lines('a', [], ['a\r\n'])
132
vf.add_lines_with_ghosts('b', [], ['a\r\n'])
133
except NotImplementedError:
136
def test_get_delta(self):
138
sha1s = self._setup_for_deltas(f)
139
expected_delta = (None, '6bfa09d82ce3e898ad4641ae13dd4fdb9cf0d76b', False,
140
[(0, 0, 1, [('base', 'line\n')])])
141
self.assertEqual(expected_delta, f.get_delta('base'))
143
text_name = 'chain1-'
144
for depth in range(26):
145
new_version = text_name + '%s' % depth
146
expected_delta = (next_parent, sha1s[depth],
148
[(depth + 1, depth + 1, 1, [(new_version, 'line\n')])])
149
self.assertEqual(expected_delta, f.get_delta(new_version))
150
next_parent = new_version
152
text_name = 'chain2-'
153
for depth in range(26):
154
new_version = text_name + '%s' % depth
155
expected_delta = (next_parent, sha1s[depth], False,
156
[(depth + 1, depth + 1, 1, [(new_version, 'line\n')])])
157
self.assertEqual(expected_delta, f.get_delta(new_version))
158
next_parent = new_version
159
# smoke test for eol support
160
expected_delta = ('base', '264f39cab871e4cfd65b3a002f7255888bb5ed97', True, [])
161
self.assertEqual(['line'], f.get_lines('noeol'))
162
self.assertEqual(expected_delta, f.get_delta('noeol'))
164
def test_get_deltas(self):
166
sha1s = self._setup_for_deltas(f)
167
deltas = f.get_deltas(f.versions())
168
expected_delta = (None, '6bfa09d82ce3e898ad4641ae13dd4fdb9cf0d76b', False,
169
[(0, 0, 1, [('base', 'line\n')])])
170
self.assertEqual(expected_delta, deltas['base'])
172
text_name = 'chain1-'
173
for depth in range(26):
174
new_version = text_name + '%s' % depth
175
expected_delta = (next_parent, sha1s[depth],
177
[(depth + 1, depth + 1, 1, [(new_version, 'line\n')])])
178
self.assertEqual(expected_delta, deltas[new_version])
179
next_parent = new_version
181
text_name = 'chain2-'
182
for depth in range(26):
183
new_version = text_name + '%s' % depth
184
expected_delta = (next_parent, sha1s[depth], False,
185
[(depth + 1, depth + 1, 1, [(new_version, 'line\n')])])
186
self.assertEqual(expected_delta, deltas[new_version])
187
next_parent = new_version
188
# smoke tests for eol support
189
expected_delta = ('base', '264f39cab871e4cfd65b3a002f7255888bb5ed97', True, [])
190
self.assertEqual(['line'], f.get_lines('noeol'))
191
self.assertEqual(expected_delta, deltas['noeol'])
192
# smoke tests for eol support - two noeol in a row same content
193
expected_deltas = (('noeol', '3ad7ee82dbd8f29ecba073f96e43e414b3f70a4d', True,
194
[(0, 1, 2, [(u'noeolsecond', 'line\n'), (u'noeolsecond', 'line\n')])]),
195
('noeol', '3ad7ee82dbd8f29ecba073f96e43e414b3f70a4d', True,
196
[(0, 0, 1, [('noeolsecond', 'line\n')]), (1, 1, 0, [])]))
197
self.assertEqual(['line\n', 'line'], f.get_lines('noeolsecond'))
198
self.assertTrue(deltas['noeolsecond'] in expected_deltas)
199
# two no-eol in a row, different content
200
expected_delta = ('noeolsecond', '8bb553a84e019ef1149db082d65f3133b195223b', True,
201
[(1, 2, 1, [(u'noeolnotshared', 'phone\n')])])
202
self.assertEqual(['line\n', 'phone'], f.get_lines('noeolnotshared'))
203
self.assertEqual(expected_delta, deltas['noeolnotshared'])
204
# eol folling a no-eol with content change
205
expected_delta = ('noeol', 'a61f6fb6cfc4596e8d88c34a308d1e724caf8977', False,
206
[(0, 1, 1, [(u'eol', 'phone\n')])])
207
self.assertEqual(['phone\n'], f.get_lines('eol'))
208
self.assertEqual(expected_delta, deltas['eol'])
209
# eol folling a no-eol with content change
210
expected_delta = ('noeol', '6bfa09d82ce3e898ad4641ae13dd4fdb9cf0d76b', False,
211
[(0, 1, 1, [(u'eolline', 'line\n')])])
212
self.assertEqual(['line\n'], f.get_lines('eolline'))
213
self.assertEqual(expected_delta, deltas['eolline'])
214
# eol with no parents
215
expected_delta = (None, '264f39cab871e4cfd65b3a002f7255888bb5ed97', True,
216
[(0, 0, 1, [(u'noeolbase', 'line\n')])])
217
self.assertEqual(['line'], f.get_lines('noeolbase'))
218
self.assertEqual(expected_delta, deltas['noeolbase'])
219
# eol with two parents, in inverse insertion order
220
expected_deltas = (('noeolbase', '264f39cab871e4cfd65b3a002f7255888bb5ed97', True,
221
[(0, 1, 1, [(u'eolbeforefirstparent', 'line\n')])]),
222
('noeolbase', '264f39cab871e4cfd65b3a002f7255888bb5ed97', True,
223
[(0, 1, 1, [(u'eolbeforefirstparent', 'line\n')])]))
224
self.assertEqual(['line'], f.get_lines('eolbeforefirstparent'))
225
#self.assertTrue(deltas['eolbeforefirstparent'] in expected_deltas)
227
def _setup_for_deltas(self, f):
228
self.assertRaises(errors.RevisionNotPresent, f.get_delta, 'base')
229
# add texts that should trip the knit maximum delta chain threshold
230
# as well as doing parallel chains of data in knits.
231
# this is done by two chains of 25 insertions
232
f.add_lines('base', [], ['line\n'])
233
f.add_lines('noeol', ['base'], ['line'])
234
# detailed eol tests:
235
# shared last line with parent no-eol
236
f.add_lines('noeolsecond', ['noeol'], ['line\n', 'line'])
237
# differing last line with parent, both no-eol
238
f.add_lines('noeolnotshared', ['noeolsecond'], ['line\n', 'phone'])
239
# add eol following a noneol parent, change content
240
f.add_lines('eol', ['noeol'], ['phone\n'])
241
# add eol following a noneol parent, no change content
242
f.add_lines('eolline', ['noeol'], ['line\n'])
243
# noeol with no parents:
244
f.add_lines('noeolbase', [], ['line'])
245
# noeol preceeding its leftmost parent in the output:
246
# this is done by making it a merge of two parents with no common
247
# anestry: noeolbase and noeol with the
248
# later-inserted parent the leftmost.
249
f.add_lines('eolbeforefirstparent', ['noeolbase', 'noeol'], ['line'])
250
# two identical eol texts
251
f.add_lines('noeoldup', ['noeol'], ['line'])
253
text_name = 'chain1-'
255
sha1s = {0 :'da6d3141cb4a5e6f464bf6e0518042ddc7bfd079',
256
1 :'45e21ea146a81ea44a821737acdb4f9791c8abe7',
257
2 :'e1f11570edf3e2a070052366c582837a4fe4e9fa',
258
3 :'26b4b8626da827088c514b8f9bbe4ebf181edda1',
259
4 :'e28a5510be25ba84d31121cff00956f9970ae6f6',
260
5 :'d63ec0ce22e11dcf65a931b69255d3ac747a318d',
261
6 :'2c2888d288cb5e1d98009d822fedfe6019c6a4ea',
262
7 :'95c14da9cafbf828e3e74a6f016d87926ba234ab',
263
8 :'779e9a0b28f9f832528d4b21e17e168c67697272',
264
9 :'1f8ff4e5c6ff78ac106fcfe6b1e8cb8740ff9a8f',
265
10:'131a2ae712cf51ed62f143e3fbac3d4206c25a05',
266
11:'c5a9d6f520d2515e1ec401a8f8a67e6c3c89f199',
267
12:'31a2286267f24d8bedaa43355f8ad7129509ea85',
268
13:'dc2a7fe80e8ec5cae920973973a8ee28b2da5e0a',
269
14:'2c4b1736566b8ca6051e668de68650686a3922f2',
270
15:'5912e4ecd9b0c07be4d013e7e2bdcf9323276cde',
271
16:'b0d2e18d3559a00580f6b49804c23fea500feab3',
272
17:'8e1d43ad72f7562d7cb8f57ee584e20eb1a69fc7',
273
18:'5cf64a3459ae28efa60239e44b20312d25b253f3',
274
19:'1ebed371807ba5935958ad0884595126e8c4e823',
275
20:'2aa62a8b06fb3b3b892a3292a068ade69d5ee0d3',
276
21:'01edc447978004f6e4e962b417a4ae1955b6fe5d',
277
22:'d8d8dc49c4bf0bab401e0298bb5ad827768618bb',
278
23:'c21f62b1c482862983a8ffb2b0c64b3451876e3f',
279
24:'c0593fe795e00dff6b3c0fe857a074364d5f04fc',
280
25:'dd1a1cf2ba9cc225c3aff729953e6364bf1d1855',
282
for depth in range(26):
283
new_version = text_name + '%s' % depth
284
text = text + ['line\n']
285
f.add_lines(new_version, [next_parent], text)
286
next_parent = new_version
288
text_name = 'chain2-'
290
for depth in range(26):
291
new_version = text_name + '%s' % depth
292
text = text + ['line\n']
293
f.add_lines(new_version, [next_parent], text)
294
next_parent = new_version
297
def test_add_delta(self):
298
# tests for the add-delta facility.
299
# at this point, optimising for speed, we assume no checks when deltas are inserted.
300
# this may need to be revisited.
301
source = self.get_file('source')
302
source.add_lines('base', [], ['line\n'])
304
text_name = 'chain1-'
306
for depth in range(26):
307
new_version = text_name + '%s' % depth
308
text = text + ['line\n']
309
source.add_lines(new_version, [next_parent], text)
310
next_parent = new_version
312
text_name = 'chain2-'
314
for depth in range(26):
315
new_version = text_name + '%s' % depth
316
text = text + ['line\n']
317
source.add_lines(new_version, [next_parent], text)
318
next_parent = new_version
319
source.add_lines('noeol', ['base'], ['line'])
321
target = self.get_file('target')
322
for version in source.versions():
323
parent, sha1, noeol, delta = source.get_delta(version)
324
target.add_delta(version,
325
source.get_parents(version),
330
self.assertRaises(RevisionAlreadyPresent,
331
target.add_delta, 'base', [], None, '', False, [])
332
for version in source.versions():
333
self.assertEqual(source.get_lines(version),
334
target.get_lines(version))
336
def test_ancestry(self):
338
self.assertEqual([], f.get_ancestry([]))
339
f.add_lines('r0', [], ['a\n', 'b\n'])
340
f.add_lines('r1', ['r0'], ['b\n', 'c\n'])
341
f.add_lines('r2', ['r0'], ['b\n', 'c\n'])
342
f.add_lines('r3', ['r2'], ['b\n', 'c\n'])
343
f.add_lines('rM', ['r1', 'r2'], ['b\n', 'c\n'])
344
self.assertEqual([], f.get_ancestry([]))
345
versions = f.get_ancestry(['rM'])
346
# there are some possibilities:
350
# so we check indexes
351
r0 = versions.index('r0')
352
r1 = versions.index('r1')
353
r2 = versions.index('r2')
354
self.assertFalse('r3' in versions)
355
rM = versions.index('rM')
356
self.assertTrue(r0 < r1)
357
self.assertTrue(r0 < r2)
358
self.assertTrue(r1 < rM)
359
self.assertTrue(r2 < rM)
361
self.assertRaises(RevisionNotPresent,
362
f.get_ancestry, ['rM', 'rX'])
364
def test_mutate_after_finish(self):
366
f.transaction_finished()
367
self.assertRaises(errors.OutSideTransaction, f.add_delta, '', [], '', '', False, [])
368
self.assertRaises(errors.OutSideTransaction, f.add_lines, '', [], [])
369
self.assertRaises(errors.OutSideTransaction, f.add_lines_with_ghosts, '', [], [])
370
self.assertRaises(errors.OutSideTransaction, f.fix_parents, '', [])
371
self.assertRaises(errors.OutSideTransaction, f.join, '')
372
self.assertRaises(errors.OutSideTransaction, f.clone_text, 'base', 'bar', ['foo'])
374
def test_clear_cache(self):
376
# on a new file it should not error
378
# and after adding content, doing a clear_cache and a get should work.
379
f.add_lines('0', [], ['a'])
381
self.assertEqual(['a'], f.get_lines('0'))
383
def test_clone_text(self):
385
f.add_lines('r0', [], ['a\n', 'b\n'])
386
f.clone_text('r1', 'r0', ['r0'])
388
self.assertEquals(f.get_lines('r1'), f.get_lines('r0'))
389
self.assertEquals(f.get_lines('r1'), ['a\n', 'b\n'])
390
self.assertEquals(f.get_parents('r1'), ['r0'])
392
self.assertRaises(RevisionNotPresent,
393
f.clone_text, 'r2', 'rX', [])
394
self.assertRaises(RevisionAlreadyPresent,
395
f.clone_text, 'r1', 'r0', [])
397
verify_file(self.reopen_file())
399
def test_create_empty(self):
401
f.add_lines('0', [], ['a\n'])
402
new_f = f.create_empty('t', MemoryTransport())
403
# smoke test, specific types should check it is honoured correctly for
404
# non type attributes
405
self.assertEqual([], new_f.versions())
406
self.assertTrue(isinstance(new_f, f.__class__))
408
def test_copy_to(self):
410
f.add_lines('0', [], ['a\n'])
411
t = MemoryTransport()
413
for suffix in f.__class__.get_suffixes():
414
self.assertTrue(t.has('foo' + suffix))
416
def test_get_suffixes(self):
419
self.assertEqual(f.__class__.get_suffixes(), f.__class__.get_suffixes())
420
# and should be a list
421
self.assertTrue(isinstance(f.__class__.get_suffixes(), list))
423
def test_get_graph(self):
425
f.add_lines('v1', [], ['hello\n'])
426
f.add_lines('v2', ['v1'], ['hello\n', 'world\n'])
427
f.add_lines('v3', ['v2'], ['hello\n', 'cruel\n', 'world\n'])
428
self.assertEqual({'v1': [],
433
def test_get_parents(self):
435
f.add_lines('r0', [], ['a\n', 'b\n'])
436
f.add_lines('r1', [], ['a\n', 'b\n'])
437
f.add_lines('r2', [], ['a\n', 'b\n'])
438
f.add_lines('r3', [], ['a\n', 'b\n'])
439
f.add_lines('m', ['r0', 'r1', 'r2', 'r3'], ['a\n', 'b\n'])
440
self.assertEquals(f.get_parents('m'), ['r0', 'r1', 'r2', 'r3'])
442
self.assertRaises(RevisionNotPresent,
445
def test_annotate(self):
447
f.add_lines('r0', [], ['a\n', 'b\n'])
448
f.add_lines('r1', ['r0'], ['c\n', 'b\n'])
449
origins = f.annotate('r1')
450
self.assertEquals(origins[0][0], 'r1')
451
self.assertEquals(origins[1][0], 'r0')
453
self.assertRaises(RevisionNotPresent,
457
# tests that walk returns all the inclusions for the requested
458
# revisions as well as the revisions changes themselves.
459
f = self.get_file('1')
460
f.add_lines('r0', [], ['a\n', 'b\n'])
461
f.add_lines('r1', ['r0'], ['c\n', 'b\n'])
462
f.add_lines('rX', ['r1'], ['d\n', 'b\n'])
463
f.add_lines('rY', ['r1'], ['c\n', 'e\n'])
466
for lineno, insert, dset, text in f.walk(['rX', 'rY']):
467
lines[text] = (insert, dset)
469
self.assertTrue(lines['a\n'], ('r0', set(['r1'])))
470
self.assertTrue(lines['b\n'], ('r0', set(['rY'])))
471
self.assertTrue(lines['c\n'], ('r1', set(['rX'])))
472
self.assertTrue(lines['d\n'], ('rX', set([])))
473
self.assertTrue(lines['e\n'], ('rY', set([])))
475
def test_detection(self):
476
# Test weaves detect corruption.
478
# Weaves contain a checksum of their texts.
479
# When a text is extracted, this checksum should be
482
w = self.get_file_corrupted_text()
484
self.assertEqual('hello\n', w.get_text('v1'))
485
self.assertRaises(errors.WeaveInvalidChecksum, w.get_text, 'v2')
486
self.assertRaises(errors.WeaveInvalidChecksum, w.get_lines, 'v2')
487
self.assertRaises(errors.WeaveInvalidChecksum, w.check)
489
w = self.get_file_corrupted_checksum()
491
self.assertEqual('hello\n', w.get_text('v1'))
492
self.assertRaises(errors.WeaveInvalidChecksum, w.get_text, 'v2')
493
self.assertRaises(errors.WeaveInvalidChecksum, w.get_lines, 'v2')
494
self.assertRaises(errors.WeaveInvalidChecksum, w.check)
496
def get_file_corrupted_text(self):
497
"""Return a versioned file with corrupt text but valid metadata."""
498
raise NotImplementedError(self.get_file_corrupted_text)
500
def reopen_file(self, name='foo'):
501
"""Open the versioned file from disk again."""
502
raise NotImplementedError(self.reopen_file)
504
def test_iter_lines_added_or_present_in_versions(self):
505
# test that we get at least an equalset of the lines added by
506
# versions in the weave
507
# the ordering here is to make a tree so that dumb searches have
508
# more changes to muck up.
510
# add a base to get included
511
vf.add_lines('base', [], ['base\n'])
512
# add a ancestor to be included on one side
513
vf.add_lines('lancestor', [], ['lancestor\n'])
514
# add a ancestor to be included on the other side
515
vf.add_lines('rancestor', ['base'], ['rancestor\n'])
516
# add a child of rancestor with no eofile-nl
517
vf.add_lines('child', ['rancestor'], ['base\n', 'child\n'])
518
# add a child of lancestor and base to join the two roots
519
vf.add_lines('otherchild',
520
['lancestor', 'base'],
521
['base\n', 'lancestor\n', 'otherchild\n'])
522
def iter_with_versions(versions):
523
# now we need to see what lines are returned, and how often.
530
# iterate over the lines
531
for line in vf.iter_lines_added_or_present_in_versions(versions):
534
lines = iter_with_versions(['child', 'otherchild'])
535
# we must see child and otherchild
536
self.assertTrue(lines['child\n'] > 0)
537
self.assertTrue(lines['otherchild\n'] > 0)
538
# we dont care if we got more than that.
541
lines = iter_with_versions(None)
542
# all lines must be seen at least once
543
self.assertTrue(lines['base\n'] > 0)
544
self.assertTrue(lines['lancestor\n'] > 0)
545
self.assertTrue(lines['rancestor\n'] > 0)
546
self.assertTrue(lines['child\n'] > 0)
547
self.assertTrue(lines['otherchild\n'] > 0)
549
def test_fix_parents(self):
550
# some versioned files allow incorrect parents to be corrected after
551
# insertion - this may not fix ancestry..
552
# if they do not supported, they just do not implement it.
553
# we test this as an interface test to ensure that those that *do*
554
# implementent it get it right.
556
vf.add_lines('notbase', [], [])
557
vf.add_lines('base', [], [])
559
vf.fix_parents('notbase', ['base'])
560
except NotImplementedError:
562
self.assertEqual(['base'], vf.get_parents('notbase'))
563
# open again, check it stuck.
565
self.assertEqual(['base'], vf.get_parents('notbase'))
567
def test_fix_parents_with_ghosts(self):
568
# when fixing parents, ghosts that are listed should not be ghosts
573
vf.add_lines_with_ghosts('notbase', ['base', 'stillghost'], [])
574
except NotImplementedError:
576
vf.add_lines('base', [], [])
577
vf.fix_parents('notbase', ['base', 'stillghost'])
578
self.assertEqual(['base'], vf.get_parents('notbase'))
579
# open again, check it stuck.
581
self.assertEqual(['base'], vf.get_parents('notbase'))
582
# and check the ghosts
583
self.assertEqual(['base', 'stillghost'],
584
vf.get_parents_with_ghosts('notbase'))
586
def test_add_lines_with_ghosts(self):
587
# some versioned file formats allow lines to be added with parent
588
# information that is > than that in the format. Formats that do
589
# not support this need to raise NotImplementedError on the
590
# add_lines_with_ghosts api.
592
# add a revision with ghost parents
594
vf.add_lines_with_ghosts(u'notbxbfse', [u'b\xbfse'], [])
595
except NotImplementedError:
596
# check the other ghost apis are also not implemented
597
self.assertRaises(NotImplementedError, vf.has_ghost, 'foo')
598
self.assertRaises(NotImplementedError, vf.get_ancestry_with_ghosts, ['foo'])
599
self.assertRaises(NotImplementedError, vf.get_parents_with_ghosts, 'foo')
600
self.assertRaises(NotImplementedError, vf.get_graph_with_ghosts)
602
# test key graph related apis: getncestry, _graph, get_parents
604
# - these are ghost unaware and must not be reflect ghosts
605
self.assertEqual([u'notbxbfse'], vf.get_ancestry(u'notbxbfse'))
606
self.assertEqual([], vf.get_parents(u'notbxbfse'))
607
self.assertEqual({u'notbxbfse':[]}, vf.get_graph())
608
self.assertFalse(vf.has_version(u'b\xbfse'))
609
# we have _with_ghost apis to give us ghost information.
610
self.assertEqual([u'b\xbfse', u'notbxbfse'], vf.get_ancestry_with_ghosts([u'notbxbfse']))
611
self.assertEqual([u'b\xbfse'], vf.get_parents_with_ghosts(u'notbxbfse'))
612
self.assertEqual({u'notbxbfse':[u'b\xbfse']}, vf.get_graph_with_ghosts())
613
self.assertTrue(vf.has_ghost(u'b\xbfse'))
614
# if we add something that is a ghost of another, it should correct the
615
# results of the prior apis
616
vf.add_lines(u'b\xbfse', [], [])
617
self.assertEqual([u'b\xbfse', u'notbxbfse'], vf.get_ancestry([u'notbxbfse']))
618
self.assertEqual([u'b\xbfse'], vf.get_parents(u'notbxbfse'))
619
self.assertEqual({u'b\xbfse':[],
620
u'notbxbfse':[u'b\xbfse'],
623
self.assertTrue(vf.has_version(u'b\xbfse'))
624
# we have _with_ghost apis to give us ghost information.
625
self.assertEqual([u'b\xbfse', u'notbxbfse'], vf.get_ancestry_with_ghosts([u'notbxbfse']))
626
self.assertEqual([u'b\xbfse'], vf.get_parents_with_ghosts(u'notbxbfse'))
627
self.assertEqual({u'b\xbfse':[],
628
u'notbxbfse':[u'b\xbfse'],
630
vf.get_graph_with_ghosts())
631
self.assertFalse(vf.has_ghost(u'b\xbfse'))
633
def test_add_lines_with_ghosts_after_normal_revs(self):
634
# some versioned file formats allow lines to be added with parent
635
# information that is > than that in the format. Formats that do
636
# not support this need to raise NotImplementedError on the
637
# add_lines_with_ghosts api.
639
# probe for ghost support
642
except NotImplementedError:
644
vf.add_lines_with_ghosts('base', [], ['line\n', 'line_b\n'])
645
vf.add_lines_with_ghosts('references_ghost',
647
['line\n', 'line_b\n', 'line_c\n'])
648
origins = vf.annotate('references_ghost')
649
self.assertEquals(('base', 'line\n'), origins[0])
650
self.assertEquals(('base', 'line_b\n'), origins[1])
651
self.assertEquals(('references_ghost', 'line_c\n'), origins[2])
653
def test_readonly_mode(self):
654
transport = get_transport(self.get_url('.'))
655
factory = self.get_factory()
656
vf = factory('id', transport, 0777, create=True, access_mode='w')
657
vf = factory('id', transport, access_mode='r')
658
self.assertRaises(errors.ReadOnlyError, vf.add_delta, '', [], '', '', False, [])
659
self.assertRaises(errors.ReadOnlyError, vf.add_lines, 'base', [], [])
660
self.assertRaises(errors.ReadOnlyError,
661
vf.add_lines_with_ghosts,
665
self.assertRaises(errors.ReadOnlyError, vf.fix_parents, 'base', [])
666
self.assertRaises(errors.ReadOnlyError, vf.join, 'base')
667
self.assertRaises(errors.ReadOnlyError, vf.clone_text, 'base', 'bar', ['foo'])
669
def test_get_sha1(self):
670
# check the sha1 data is available
673
vf.add_lines('a', [], ['a\n'])
674
# the same file, different metadata
675
vf.add_lines('b', ['a'], ['a\n'])
676
# a file differing only in last newline.
677
vf.add_lines('c', [], ['a'])
679
'3f786850e387550fdab836ed7e6dc881de23001b', vf.get_sha1('a'))
681
'3f786850e387550fdab836ed7e6dc881de23001b', vf.get_sha1('b'))
683
'86f7e437faa5a7fce15d1ddcb9eaeaea377667b8', vf.get_sha1('c'))
686
class TestWeave(TestCaseWithTransport, VersionedFileTestMixIn):
688
def get_file(self, name='foo'):
689
return WeaveFile(name, get_transport(self.get_url('.')), create=True)
691
def get_file_corrupted_text(self):
692
w = WeaveFile('foo', get_transport(self.get_url('.')), create=True)
693
w.add_lines('v1', [], ['hello\n'])
694
w.add_lines('v2', ['v1'], ['hello\n', 'there\n'])
696
# We are going to invasively corrupt the text
697
# Make sure the internals of weave are the same
698
self.assertEqual([('{', 0)
706
self.assertEqual(['f572d396fae9206628714fb2ce00f72e94f2258f'
707
, '90f265c6e75f1c8f9ab76dcf85528352c5f215ef'
712
w._weave[4] = 'There\n'
715
def get_file_corrupted_checksum(self):
716
w = self.get_file_corrupted_text()
718
w._weave[4] = 'there\n'
719
self.assertEqual('hello\nthere\n', w.get_text('v2'))
721
#Invalid checksum, first digit changed
722
w._sha1s[1] = 'f0f265c6e75f1c8f9ab76dcf85528352c5f215ef'
725
def reopen_file(self, name='foo', create=False):
726
return WeaveFile(name, get_transport(self.get_url('.')), create=create)
728
def test_no_implicit_create(self):
729
self.assertRaises(errors.NoSuchFile,
732
get_transport(self.get_url('.')))
734
def get_factory(self):
738
class TestKnit(TestCaseWithTransport, VersionedFileTestMixIn):
740
def get_file(self, name='foo'):
741
return KnitVersionedFile(name, get_transport(self.get_url('.')),
742
delta=True, create=True)
744
def get_factory(self):
745
return KnitVersionedFile
747
def get_file_corrupted_text(self):
748
knit = self.get_file()
749
knit.add_lines('v1', [], ['hello\n'])
750
knit.add_lines('v2', ['v1'], ['hello\n', 'there\n'])
753
def reopen_file(self, name='foo', create=False):
754
return KnitVersionedFile(name, get_transport(self.get_url('.')),
758
def test_detection(self):
759
print "TODO for merging: create a corrupted knit."
760
knit = self.get_file()
763
def test_no_implicit_create(self):
764
self.assertRaises(errors.NoSuchFile,
767
get_transport(self.get_url('.')))
770
class InterString(versionedfile.InterVersionedFile):
771
"""An inter-versionedfile optimised code path for strings.
773
This is for use during testing where we use strings as versionedfiles
774
so that none of the default regsitered interversionedfile classes will
775
match - which lets us test the match logic.
779
def is_compatible(source, target):
780
"""InterString is compatible with strings-as-versionedfiles."""
781
return isinstance(source, str) and isinstance(target, str)
784
# TODO this and the InterRepository core logic should be consolidatable
785
# if we make the registry a separate class though we still need to
786
# test the behaviour in the active registry to catch failure-to-handle-
788
class TestInterVersionedFile(TestCaseWithTransport):
790
def test_get_default_inter_versionedfile(self):
791
# test that the InterVersionedFile.get(a, b) probes
792
# for a class where is_compatible(a, b) returns
793
# true and returns a default interversionedfile otherwise.
794
# This also tests that the default registered optimised interversionedfile
795
# classes do not barf inappropriately when a surprising versionedfile type
797
dummy_a = "VersionedFile 1."
798
dummy_b = "VersionedFile 2."
799
self.assertGetsDefaultInterVersionedFile(dummy_a, dummy_b)
801
def assertGetsDefaultInterVersionedFile(self, a, b):
802
"""Asserts that InterVersionedFile.get(a, b) -> the default."""
803
inter = versionedfile.InterVersionedFile.get(a, b)
804
self.assertEqual(versionedfile.InterVersionedFile,
806
self.assertEqual(a, inter.source)
807
self.assertEqual(b, inter.target)
809
def test_register_inter_versionedfile_class(self):
810
# test that a optimised code path provider - a
811
# InterVersionedFile subclass can be registered and unregistered
812
# and that it is correctly selected when given a versionedfile
813
# pair that it returns true on for the is_compatible static method
815
dummy_a = "VersionedFile 1."
816
dummy_b = "VersionedFile 2."
817
versionedfile.InterVersionedFile.register_optimiser(InterString)
819
# we should get the default for something InterString returns False
821
self.assertFalse(InterString.is_compatible(dummy_a, None))
822
self.assertGetsDefaultInterVersionedFile(dummy_a, None)
823
# and we should get an InterString for a pair it 'likes'
824
self.assertTrue(InterString.is_compatible(dummy_a, dummy_b))
825
inter = versionedfile.InterVersionedFile.get(dummy_a, dummy_b)
826
self.assertEqual(InterString, inter.__class__)
827
self.assertEqual(dummy_a, inter.source)
828
self.assertEqual(dummy_b, inter.target)
830
versionedfile.InterVersionedFile.unregister_optimiser(InterString)
831
# now we should get the default InterVersionedFile object again.
832
self.assertGetsDefaultInterVersionedFile(dummy_a, dummy_b)
835
class TestReadonlyHttpMixin(object):
837
def test_readonly_http_works(self):
838
# we should be able to read from http with a versioned file.
840
# try an empty file access
841
readonly_vf = self.get_factory()('foo', get_transport(self.get_readonly_url('.')))
842
self.assertEqual([], readonly_vf.versions())
844
vf.add_lines('1', [], ['a\n'])
845
vf.add_lines('2', ['1'], ['b\n', 'a\n'])
846
readonly_vf = self.get_factory()('foo', get_transport(self.get_readonly_url('.')))
847
self.assertEqual(['1', '2'], vf.versions())
848
for version in readonly_vf.versions():
849
readonly_vf.get_lines(version)
852
class TestWeaveHTTP(TestCaseWithWebserver, TestReadonlyHttpMixin):
855
return WeaveFile('foo', get_transport(self.get_url('.')), create=True)
857
def get_factory(self):
861
class TestKnitHTTP(TestCaseWithWebserver, TestReadonlyHttpMixin):
864
return KnitVersionedFile('foo', get_transport(self.get_url('.')),
865
delta=True, create=True)
867
def get_factory(self):
868
return KnitVersionedFile
871
class MergeCasesMixin(object):
873
def doMerge(self, base, a, b, mp):
874
from cStringIO import StringIO
875
from textwrap import dedent
881
w.add_lines('text0', [], map(addcrlf, base))
882
w.add_lines('text1', ['text0'], map(addcrlf, a))
883
w.add_lines('text2', ['text0'], map(addcrlf, b))
887
self.log('merge plan:')
888
p = list(w.plan_merge('text1', 'text2'))
889
for state, line in p:
891
self.log('%12s | %s' % (state, line[:-1]))
895
mt.writelines(w.weave_merge(p))
897
self.log(mt.getvalue())
899
mp = map(addcrlf, mp)
900
self.assertEqual(mt.readlines(), mp)
903
def testOneInsert(self):
909
def testSeparateInserts(self):
910
self.doMerge(['aaa', 'bbb', 'ccc'],
911
['aaa', 'xxx', 'bbb', 'ccc'],
912
['aaa', 'bbb', 'yyy', 'ccc'],
913
['aaa', 'xxx', 'bbb', 'yyy', 'ccc'])
915
def testSameInsert(self):
916
self.doMerge(['aaa', 'bbb', 'ccc'],
917
['aaa', 'xxx', 'bbb', 'ccc'],
918
['aaa', 'xxx', 'bbb', 'yyy', 'ccc'],
919
['aaa', 'xxx', 'bbb', 'yyy', 'ccc'])
920
overlappedInsertExpected = ['aaa', 'xxx', 'yyy', 'bbb']
921
def testOverlappedInsert(self):
922
self.doMerge(['aaa', 'bbb'],
923
['aaa', 'xxx', 'yyy', 'bbb'],
924
['aaa', 'xxx', 'bbb'], self.overlappedInsertExpected)
926
# really it ought to reduce this to
927
# ['aaa', 'xxx', 'yyy', 'bbb']
930
def testClashReplace(self):
931
self.doMerge(['aaa'],
934
['<<<<<<< ', 'xxx', '=======', 'yyy', 'zzz',
937
def testNonClashInsert1(self):
938
self.doMerge(['aaa'],
941
['<<<<<<< ', 'xxx', 'aaa', '=======', 'yyy', 'zzz',
944
def testNonClashInsert2(self):
945
self.doMerge(['aaa'],
951
def testDeleteAndModify(self):
952
"""Clashing delete and modification.
954
If one side modifies a region and the other deletes it then
955
there should be a conflict with one side blank.
958
#######################################
959
# skippd, not working yet
962
self.doMerge(['aaa', 'bbb', 'ccc'],
963
['aaa', 'ddd', 'ccc'],
965
['<<<<<<<< ', 'aaa', '=======', '>>>>>>> ', 'ccc'])
967
def _test_merge_from_strings(self, base, a, b, expected):
969
w.add_lines('text0', [], base.splitlines(True))
970
w.add_lines('text1', ['text0'], a.splitlines(True))
971
w.add_lines('text2', ['text0'], b.splitlines(True))
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]))
977
self.log('merge result:')
978
result_text = ''.join(w.weave_merge(p))
979
self.log(result_text)
980
self.assertEqualDiff(result_text, expected)
982
def test_weave_merge_conflicts(self):
983
# does weave merge properly handle plans that end with unchanged?
984
result = ''.join(self.get_file().weave_merge([('new-a', 'hello\n')]))
985
self.assertEqual(result, 'hello\n')
987
def test_deletion_extended(self):
988
"""One side deletes, the other deletes more.
1005
self._test_merge_from_strings(base, a, b, result)
1007
def test_deletion_overlap(self):
1008
"""Delete overlapping regions with no other conflict.
1010
Arguably it'd be better to treat these as agreement, rather than
1011
conflict, but for now conflict is safer.
1039
self._test_merge_from_strings(base, a, b, result)
1041
def test_agreement_deletion(self):
1042
"""Agree to delete some lines, without conflicts."""
1064
self._test_merge_from_strings(base, a, b, result)
1066
def test_sync_on_deletion(self):
1067
"""Specific case of merge where we can synchronize incorrectly.
1069
A previous version of the weave merge concluded that the two versions
1070
agreed on deleting line 2, and this could be a synchronization point.
1071
Line 1 was then considered in isolation, and thought to be deleted on
1074
It's better to consider the whole thing as a disagreement region.
1085
a's replacement line 2
1098
a's replacement line 2
1105
self._test_merge_from_strings(base, a, b, result)
1108
class TestKnitMerge(TestCaseWithTransport, MergeCasesMixin):
1110
def get_file(self, name='foo'):
1111
return KnitVersionedFile(name, get_transport(self.get_url('.')),
1112
delta=True, create=True)
1114
def log_contents(self, w):
1118
class TestWeaveMerge(TestCaseWithTransport, MergeCasesMixin):
1120
def get_file(self, name='foo'):
1121
return WeaveFile(name, get_transport(self.get_url('.')), create=True)
1123
def log_contents(self, w):
1124
self.log('weave is:')
1126
write_weave(w, tmpf)
1127
self.log(tmpf.getvalue())
1129
overlappedInsertExpected = ['aaa', '<<<<<<< ', 'xxx', 'yyy', '=======',
1130
'xxx', '>>>>>>> ', 'bbb']
added
removed
bzrlib/tests/test_versionedfile.py
