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
f = self.reopen_file()
72
def test_adds_with_parent_texts(self):
75
parent_texts['r0'] = f.add_lines('r0', [], ['a\n', 'b\n'])
77
parent_texts['r1'] = f.add_lines_with_ghosts('r1',
80
parent_texts=parent_texts)
81
except NotImplementedError:
82
# if the format doesn't support ghosts, just add normally.
83
parent_texts['r1'] = f.add_lines('r1',
86
parent_texts=parent_texts)
87
f.add_lines('r2', ['r1'], ['c\n', 'd\n'], parent_texts=parent_texts)
88
self.assertNotEqual(None, parent_texts['r0'])
89
self.assertNotEqual(None, parent_texts['r1'])
91
versions = f.versions()
92
self.assertTrue('r0' in versions)
93
self.assertTrue('r1' in versions)
94
self.assertTrue('r2' in versions)
95
self.assertEquals(f.get_lines('r0'), ['a\n', 'b\n'])
96
self.assertEquals(f.get_lines('r1'), ['b\n', 'c\n'])
97
self.assertEquals(f.get_lines('r2'), ['c\n', 'd\n'])
98
self.assertEqual(3, f.num_versions())
99
origins = f.annotate('r1')
100
self.assertEquals(origins[0][0], 'r0')
101
self.assertEquals(origins[1][0], 'r1')
102
origins = f.annotate('r2')
103
self.assertEquals(origins[0][0], 'r1')
104
self.assertEquals(origins[1][0], 'r2')
107
f = self.reopen_file()
110
def test_get_delta(self):
112
sha1s = self._setup_for_deltas(f)
113
expected_delta = (None, '6bfa09d82ce3e898ad4641ae13dd4fdb9cf0d76b', False,
114
[(0, 0, 1, [('base', 'line\n')])])
115
self.assertEqual(expected_delta, f.get_delta('base'))
117
text_name = 'chain1-'
118
for depth in range(26):
119
new_version = text_name + '%s' % depth
120
expected_delta = (next_parent, sha1s[depth],
122
[(depth + 1, depth + 1, 1, [(new_version, 'line\n')])])
123
self.assertEqual(expected_delta, f.get_delta(new_version))
124
next_parent = new_version
126
text_name = 'chain2-'
127
for depth in range(26):
128
new_version = text_name + '%s' % depth
129
expected_delta = (next_parent, sha1s[depth], False,
130
[(depth + 1, depth + 1, 1, [(new_version, 'line\n')])])
131
self.assertEqual(expected_delta, f.get_delta(new_version))
132
next_parent = new_version
133
# smoke test for eol support
134
expected_delta = ('base', '264f39cab871e4cfd65b3a002f7255888bb5ed97', True, [])
135
self.assertEqual(['line'], f.get_lines('noeol'))
136
self.assertEqual(expected_delta, f.get_delta('noeol'))
138
def test_get_deltas(self):
140
sha1s = self._setup_for_deltas(f)
141
deltas = f.get_deltas(f.versions())
142
expected_delta = (None, '6bfa09d82ce3e898ad4641ae13dd4fdb9cf0d76b', False,
143
[(0, 0, 1, [('base', 'line\n')])])
144
self.assertEqual(expected_delta, deltas['base'])
146
text_name = 'chain1-'
147
for depth in range(26):
148
new_version = text_name + '%s' % depth
149
expected_delta = (next_parent, sha1s[depth],
151
[(depth + 1, depth + 1, 1, [(new_version, 'line\n')])])
152
self.assertEqual(expected_delta, deltas[new_version])
153
next_parent = new_version
155
text_name = 'chain2-'
156
for depth in range(26):
157
new_version = text_name + '%s' % depth
158
expected_delta = (next_parent, sha1s[depth], False,
159
[(depth + 1, depth + 1, 1, [(new_version, 'line\n')])])
160
self.assertEqual(expected_delta, deltas[new_version])
161
next_parent = new_version
162
# smoke tests for eol support
163
expected_delta = ('base', '264f39cab871e4cfd65b3a002f7255888bb5ed97', True, [])
164
self.assertEqual(['line'], f.get_lines('noeol'))
165
self.assertEqual(expected_delta, deltas['noeol'])
166
# smoke tests for eol support - two noeol in a row same content
167
expected_deltas = (('noeol', '3ad7ee82dbd8f29ecba073f96e43e414b3f70a4d', True,
168
[(0, 1, 2, [(u'noeolsecond', 'line\n'), (u'noeolsecond', 'line\n')])]),
169
('noeol', '3ad7ee82dbd8f29ecba073f96e43e414b3f70a4d', True,
170
[(0, 0, 1, [('noeolsecond', 'line\n')]), (1, 1, 0, [])]))
171
self.assertEqual(['line\n', 'line'], f.get_lines('noeolsecond'))
172
self.assertTrue(deltas['noeolsecond'] in expected_deltas)
173
# two no-eol in a row, different content
174
expected_delta = ('noeolsecond', '8bb553a84e019ef1149db082d65f3133b195223b', True,
175
[(1, 2, 1, [(u'noeolnotshared', 'phone\n')])])
176
self.assertEqual(['line\n', 'phone'], f.get_lines('noeolnotshared'))
177
self.assertEqual(expected_delta, deltas['noeolnotshared'])
178
# eol folling a no-eol with content change
179
expected_delta = ('noeol', 'a61f6fb6cfc4596e8d88c34a308d1e724caf8977', False,
180
[(0, 1, 1, [(u'eol', 'phone\n')])])
181
self.assertEqual(['phone\n'], f.get_lines('eol'))
182
self.assertEqual(expected_delta, deltas['eol'])
183
# eol folling a no-eol with content change
184
expected_delta = ('noeol', '6bfa09d82ce3e898ad4641ae13dd4fdb9cf0d76b', False,
185
[(0, 1, 1, [(u'eolline', 'line\n')])])
186
self.assertEqual(['line\n'], f.get_lines('eolline'))
187
self.assertEqual(expected_delta, deltas['eolline'])
188
# eol with no parents
189
expected_delta = (None, '264f39cab871e4cfd65b3a002f7255888bb5ed97', True,
190
[(0, 0, 1, [(u'noeolbase', 'line\n')])])
191
self.assertEqual(['line'], f.get_lines('noeolbase'))
192
self.assertEqual(expected_delta, deltas['noeolbase'])
193
# eol with two parents, in inverse insertion order
194
expected_deltas = (('noeolbase', '264f39cab871e4cfd65b3a002f7255888bb5ed97', True,
195
[(0, 1, 1, [(u'eolbeforefirstparent', 'line\n')])]),
196
('noeolbase', '264f39cab871e4cfd65b3a002f7255888bb5ed97', True,
197
[(0, 1, 1, [(u'eolbeforefirstparent', 'line\n')])]))
198
self.assertEqual(['line'], f.get_lines('eolbeforefirstparent'))
199
#self.assertTrue(deltas['eolbeforefirstparent'] in expected_deltas)
201
def _setup_for_deltas(self, f):
202
self.assertRaises(errors.RevisionNotPresent, f.get_delta, 'base')
203
# add texts that should trip the knit maximum delta chain threshold
204
# as well as doing parallel chains of data in knits.
205
# this is done by two chains of 25 insertions
206
f.add_lines('base', [], ['line\n'])
207
f.add_lines('noeol', ['base'], ['line'])
208
# detailed eol tests:
209
# shared last line with parent no-eol
210
f.add_lines('noeolsecond', ['noeol'], ['line\n', 'line'])
211
# differing last line with parent, both no-eol
212
f.add_lines('noeolnotshared', ['noeolsecond'], ['line\n', 'phone'])
213
# add eol following a noneol parent, change content
214
f.add_lines('eol', ['noeol'], ['phone\n'])
215
# add eol following a noneol parent, no change content
216
f.add_lines('eolline', ['noeol'], ['line\n'])
217
# noeol with no parents:
218
f.add_lines('noeolbase', [], ['line'])
219
# noeol preceeding its leftmost parent in the output:
220
# this is done by making it a merge of two parents with no common
221
# anestry: noeolbase and noeol with the
222
# later-inserted parent the leftmost.
223
f.add_lines('eolbeforefirstparent', ['noeolbase', 'noeol'], ['line'])
224
# two identical eol texts
225
f.add_lines('noeoldup', ['noeol'], ['line'])
227
text_name = 'chain1-'
229
sha1s = {0 :'da6d3141cb4a5e6f464bf6e0518042ddc7bfd079',
230
1 :'45e21ea146a81ea44a821737acdb4f9791c8abe7',
231
2 :'e1f11570edf3e2a070052366c582837a4fe4e9fa',
232
3 :'26b4b8626da827088c514b8f9bbe4ebf181edda1',
233
4 :'e28a5510be25ba84d31121cff00956f9970ae6f6',
234
5 :'d63ec0ce22e11dcf65a931b69255d3ac747a318d',
235
6 :'2c2888d288cb5e1d98009d822fedfe6019c6a4ea',
236
7 :'95c14da9cafbf828e3e74a6f016d87926ba234ab',
237
8 :'779e9a0b28f9f832528d4b21e17e168c67697272',
238
9 :'1f8ff4e5c6ff78ac106fcfe6b1e8cb8740ff9a8f',
239
10:'131a2ae712cf51ed62f143e3fbac3d4206c25a05',
240
11:'c5a9d6f520d2515e1ec401a8f8a67e6c3c89f199',
241
12:'31a2286267f24d8bedaa43355f8ad7129509ea85',
242
13:'dc2a7fe80e8ec5cae920973973a8ee28b2da5e0a',
243
14:'2c4b1736566b8ca6051e668de68650686a3922f2',
244
15:'5912e4ecd9b0c07be4d013e7e2bdcf9323276cde',
245
16:'b0d2e18d3559a00580f6b49804c23fea500feab3',
246
17:'8e1d43ad72f7562d7cb8f57ee584e20eb1a69fc7',
247
18:'5cf64a3459ae28efa60239e44b20312d25b253f3',
248
19:'1ebed371807ba5935958ad0884595126e8c4e823',
249
20:'2aa62a8b06fb3b3b892a3292a068ade69d5ee0d3',
250
21:'01edc447978004f6e4e962b417a4ae1955b6fe5d',
251
22:'d8d8dc49c4bf0bab401e0298bb5ad827768618bb',
252
23:'c21f62b1c482862983a8ffb2b0c64b3451876e3f',
253
24:'c0593fe795e00dff6b3c0fe857a074364d5f04fc',
254
25:'dd1a1cf2ba9cc225c3aff729953e6364bf1d1855',
256
for depth in range(26):
257
new_version = text_name + '%s' % depth
258
text = text + ['line\n']
259
f.add_lines(new_version, [next_parent], text)
260
next_parent = new_version
262
text_name = 'chain2-'
264
for depth in range(26):
265
new_version = text_name + '%s' % depth
266
text = text + ['line\n']
267
f.add_lines(new_version, [next_parent], text)
268
next_parent = new_version
271
def test_add_delta(self):
272
# tests for the add-delta facility.
273
# at this point, optimising for speed, we assume no checks when deltas are inserted.
274
# this may need to be revisited.
275
source = self.get_file('source')
276
source.add_lines('base', [], ['line\n'])
278
text_name = 'chain1-'
280
for depth in range(26):
281
new_version = text_name + '%s' % depth
282
text = text + ['line\n']
283
source.add_lines(new_version, [next_parent], text)
284
next_parent = new_version
286
text_name = 'chain2-'
288
for depth in range(26):
289
new_version = text_name + '%s' % depth
290
text = text + ['line\n']
291
source.add_lines(new_version, [next_parent], text)
292
next_parent = new_version
293
source.add_lines('noeol', ['base'], ['line'])
295
target = self.get_file('target')
296
for version in source.versions():
297
parent, sha1, noeol, delta = source.get_delta(version)
298
target.add_delta(version,
299
source.get_parents(version),
304
self.assertRaises(RevisionAlreadyPresent,
305
target.add_delta, 'base', [], None, '', False, [])
306
for version in source.versions():
307
self.assertEqual(source.get_lines(version),
308
target.get_lines(version))
310
def test_ancestry(self):
312
self.assertEqual([], f.get_ancestry([]))
313
f.add_lines('r0', [], ['a\n', 'b\n'])
314
f.add_lines('r1', ['r0'], ['b\n', 'c\n'])
315
f.add_lines('r2', ['r0'], ['b\n', 'c\n'])
316
f.add_lines('r3', ['r2'], ['b\n', 'c\n'])
317
f.add_lines('rM', ['r1', 'r2'], ['b\n', 'c\n'])
318
self.assertEqual([], f.get_ancestry([]))
319
versions = f.get_ancestry(['rM'])
320
# there are some possibilities:
324
# so we check indexes
325
r0 = versions.index('r0')
326
r1 = versions.index('r1')
327
r2 = versions.index('r2')
328
self.assertFalse('r3' in versions)
329
rM = versions.index('rM')
330
self.assertTrue(r0 < r1)
331
self.assertTrue(r0 < r2)
332
self.assertTrue(r1 < rM)
333
self.assertTrue(r2 < rM)
335
self.assertRaises(RevisionNotPresent,
336
f.get_ancestry, ['rM', 'rX'])
338
def test_mutate_after_finish(self):
340
f.transaction_finished()
341
self.assertRaises(errors.OutSideTransaction, f.add_delta, '', [], '', '', False, [])
342
self.assertRaises(errors.OutSideTransaction, f.add_lines, '', [], [])
343
self.assertRaises(errors.OutSideTransaction, f.add_lines_with_ghosts, '', [], [])
344
self.assertRaises(errors.OutSideTransaction, f.fix_parents, '', [])
345
self.assertRaises(errors.OutSideTransaction, f.join, '')
346
self.assertRaises(errors.OutSideTransaction, f.clone_text, 'base', 'bar', ['foo'])
348
def test_clear_cache(self):
350
# on a new file it should not error
352
# and after adding content, doing a clear_cache and a get should work.
353
f.add_lines('0', [], ['a'])
355
self.assertEqual(['a'], f.get_lines('0'))
357
def test_clone_text(self):
359
f.add_lines('r0', [], ['a\n', 'b\n'])
360
f.clone_text('r1', 'r0', ['r0'])
362
self.assertEquals(f.get_lines('r1'), f.get_lines('r0'))
363
self.assertEquals(f.get_lines('r1'), ['a\n', 'b\n'])
364
self.assertEquals(f.get_parents('r1'), ['r0'])
366
self.assertRaises(RevisionNotPresent,
367
f.clone_text, 'r2', 'rX', [])
368
self.assertRaises(RevisionAlreadyPresent,
369
f.clone_text, 'r1', 'r0', [])
371
verify_file(self.reopen_file())
373
def test_create_empty(self):
375
f.add_lines('0', [], ['a\n'])
376
new_f = f.create_empty('t', MemoryTransport())
377
# smoke test, specific types should check it is honoured correctly for
378
# non type attributes
379
self.assertEqual([], new_f.versions())
380
self.assertTrue(isinstance(new_f, f.__class__))
382
def test_copy_to(self):
384
f.add_lines('0', [], ['a\n'])
385
t = MemoryTransport()
387
for suffix in f.__class__.get_suffixes():
388
self.assertTrue(t.has('foo' + suffix))
390
def test_get_suffixes(self):
393
self.assertEqual(f.__class__.get_suffixes(), f.__class__.get_suffixes())
394
# and should be a list
395
self.assertTrue(isinstance(f.__class__.get_suffixes(), list))
397
def test_get_graph(self):
399
f.add_lines('v1', [], ['hello\n'])
400
f.add_lines('v2', ['v1'], ['hello\n', 'world\n'])
401
f.add_lines('v3', ['v2'], ['hello\n', 'cruel\n', 'world\n'])
402
self.assertEqual({'v1': [],
407
def test_get_parents(self):
409
f.add_lines('r0', [], ['a\n', 'b\n'])
410
f.add_lines('r1', [], ['a\n', 'b\n'])
411
f.add_lines('r2', [], ['a\n', 'b\n'])
412
f.add_lines('r3', [], ['a\n', 'b\n'])
413
f.add_lines('m', ['r0', 'r1', 'r2', 'r3'], ['a\n', 'b\n'])
414
self.assertEquals(f.get_parents('m'), ['r0', 'r1', 'r2', 'r3'])
416
self.assertRaises(RevisionNotPresent,
419
def test_annotate(self):
421
f.add_lines('r0', [], ['a\n', 'b\n'])
422
f.add_lines('r1', ['r0'], ['c\n', 'b\n'])
423
origins = f.annotate('r1')
424
self.assertEquals(origins[0][0], 'r1')
425
self.assertEquals(origins[1][0], 'r0')
427
self.assertRaises(RevisionNotPresent,
431
# tests that walk returns all the inclusions for the requested
432
# revisions as well as the revisions changes themselves.
433
f = self.get_file('1')
434
f.add_lines('r0', [], ['a\n', 'b\n'])
435
f.add_lines('r1', ['r0'], ['c\n', 'b\n'])
436
f.add_lines('rX', ['r1'], ['d\n', 'b\n'])
437
f.add_lines('rY', ['r1'], ['c\n', 'e\n'])
440
for lineno, insert, dset, text in f.walk(['rX', 'rY']):
441
lines[text] = (insert, dset)
443
self.assertTrue(lines['a\n'], ('r0', set(['r1'])))
444
self.assertTrue(lines['b\n'], ('r0', set(['rY'])))
445
self.assertTrue(lines['c\n'], ('r1', set(['rX'])))
446
self.assertTrue(lines['d\n'], ('rX', set([])))
447
self.assertTrue(lines['e\n'], ('rY', set([])))
449
def test_detection(self):
450
# Test weaves detect corruption.
452
# Weaves contain a checksum of their texts.
453
# When a text is extracted, this checksum should be
456
w = self.get_file_corrupted_text()
458
self.assertEqual('hello\n', w.get_text('v1'))
459
self.assertRaises(errors.WeaveInvalidChecksum, w.get_text, 'v2')
460
self.assertRaises(errors.WeaveInvalidChecksum, w.get_lines, 'v2')
461
self.assertRaises(errors.WeaveInvalidChecksum, w.check)
463
w = self.get_file_corrupted_checksum()
465
self.assertEqual('hello\n', w.get_text('v1'))
466
self.assertRaises(errors.WeaveInvalidChecksum, w.get_text, 'v2')
467
self.assertRaises(errors.WeaveInvalidChecksum, w.get_lines, 'v2')
468
self.assertRaises(errors.WeaveInvalidChecksum, w.check)
470
def get_file_corrupted_text(self):
471
"""Return a versioned file with corrupt text but valid metadata."""
472
raise NotImplementedError(self.get_file_corrupted_text)
474
def reopen_file(self, name='foo'):
475
"""Open the versioned file from disk again."""
476
raise NotImplementedError(self.reopen_file)
478
def test_iter_lines_added_or_present_in_versions(self):
479
# test that we get at least an equalset of the lines added by
480
# versions in the weave
481
# the ordering here is to make a tree so that dumb searches have
482
# more changes to muck up.
484
# add a base to get included
485
vf.add_lines('base', [], ['base\n'])
486
# add a ancestor to be included on one side
487
vf.add_lines('lancestor', [], ['lancestor\n'])
488
# add a ancestor to be included on the other side
489
vf.add_lines('rancestor', ['base'], ['rancestor\n'])
490
# add a child of rancestor with no eofile-nl
491
vf.add_lines('child', ['rancestor'], ['base\n', 'child\n'])
492
# add a child of lancestor and base to join the two roots
493
vf.add_lines('otherchild',
494
['lancestor', 'base'],
495
['base\n', 'lancestor\n', 'otherchild\n'])
496
def iter_with_versions(versions):
497
# now we need to see what lines are returned, and how often.
504
# iterate over the lines
505
for line in vf.iter_lines_added_or_present_in_versions(versions):
508
lines = iter_with_versions(['child', 'otherchild'])
509
# we must see child and otherchild
510
self.assertTrue(lines['child\n'] > 0)
511
self.assertTrue(lines['otherchild\n'] > 0)
512
# we dont care if we got more than that.
515
lines = iter_with_versions(None)
516
# all lines must be seen at least once
517
self.assertTrue(lines['base\n'] > 0)
518
self.assertTrue(lines['lancestor\n'] > 0)
519
self.assertTrue(lines['rancestor\n'] > 0)
520
self.assertTrue(lines['child\n'] > 0)
521
self.assertTrue(lines['otherchild\n'] > 0)
523
def test_fix_parents(self):
524
# some versioned files allow incorrect parents to be corrected after
525
# insertion - this may not fix ancestry..
526
# if they do not supported, they just do not implement it.
527
# we test this as an interface test to ensure that those that *do*
528
# implementent it get it right.
530
vf.add_lines('notbase', [], [])
531
vf.add_lines('base', [], [])
533
vf.fix_parents('notbase', ['base'])
534
except NotImplementedError:
536
self.assertEqual(['base'], vf.get_parents('notbase'))
537
# open again, check it stuck.
539
self.assertEqual(['base'], vf.get_parents('notbase'))
541
def test_fix_parents_with_ghosts(self):
542
# when fixing parents, ghosts that are listed should not be ghosts
547
vf.add_lines_with_ghosts('notbase', ['base', 'stillghost'], [])
548
except NotImplementedError:
550
vf.add_lines('base', [], [])
551
vf.fix_parents('notbase', ['base', 'stillghost'])
552
self.assertEqual(['base'], vf.get_parents('notbase'))
553
# open again, check it stuck.
555
self.assertEqual(['base'], vf.get_parents('notbase'))
556
# and check the ghosts
557
self.assertEqual(['base', 'stillghost'],
558
vf.get_parents_with_ghosts('notbase'))
560
def test_add_lines_with_ghosts(self):
561
# some versioned file formats allow lines to be added with parent
562
# information that is > than that in the format. Formats that do
563
# not support this need to raise NotImplementedError on the
564
# add_lines_with_ghosts api.
566
# add a revision with ghost parents
568
vf.add_lines_with_ghosts(u'notbxbfse', [u'b\xbfse'], [])
569
except NotImplementedError:
570
# check the other ghost apis are also not implemented
571
self.assertRaises(NotImplementedError, vf.has_ghost, 'foo')
572
self.assertRaises(NotImplementedError, vf.get_ancestry_with_ghosts, ['foo'])
573
self.assertRaises(NotImplementedError, vf.get_parents_with_ghosts, 'foo')
574
self.assertRaises(NotImplementedError, vf.get_graph_with_ghosts)
576
# test key graph related apis: getncestry, _graph, get_parents
578
# - these are ghost unaware and must not be reflect ghosts
579
self.assertEqual([u'notbxbfse'], vf.get_ancestry(u'notbxbfse'))
580
self.assertEqual([], vf.get_parents(u'notbxbfse'))
581
self.assertEqual({u'notbxbfse':[]}, vf.get_graph())
582
self.assertFalse(vf.has_version(u'b\xbfse'))
583
# we have _with_ghost apis to give us ghost information.
584
self.assertEqual([u'b\xbfse', u'notbxbfse'], vf.get_ancestry_with_ghosts([u'notbxbfse']))
585
self.assertEqual([u'b\xbfse'], vf.get_parents_with_ghosts(u'notbxbfse'))
586
self.assertEqual({u'notbxbfse':[u'b\xbfse']}, vf.get_graph_with_ghosts())
587
self.assertTrue(vf.has_ghost(u'b\xbfse'))
588
# if we add something that is a ghost of another, it should correct the
589
# results of the prior apis
590
vf.add_lines(u'b\xbfse', [], [])
591
self.assertEqual([u'b\xbfse', u'notbxbfse'], vf.get_ancestry([u'notbxbfse']))
592
self.assertEqual([u'b\xbfse'], vf.get_parents(u'notbxbfse'))
593
self.assertEqual({u'b\xbfse':[],
594
u'notbxbfse':[u'b\xbfse'],
597
self.assertTrue(vf.has_version(u'b\xbfse'))
598
# we have _with_ghost apis to give us ghost information.
599
self.assertEqual([u'b\xbfse', u'notbxbfse'], vf.get_ancestry_with_ghosts([u'notbxbfse']))
600
self.assertEqual([u'b\xbfse'], vf.get_parents_with_ghosts(u'notbxbfse'))
601
self.assertEqual({u'b\xbfse':[],
602
u'notbxbfse':[u'b\xbfse'],
604
vf.get_graph_with_ghosts())
605
self.assertFalse(vf.has_ghost(u'b\xbfse'))
607
def test_add_lines_with_ghosts_after_normal_revs(self):
608
# some versioned file formats allow lines to be added with parent
609
# information that is > than that in the format. Formats that do
610
# not support this need to raise NotImplementedError on the
611
# add_lines_with_ghosts api.
613
# probe for ghost support
616
except NotImplementedError:
618
vf.add_lines_with_ghosts('base', [], ['line\n', 'line_b\n'])
619
vf.add_lines_with_ghosts('references_ghost',
621
['line\n', 'line_b\n', 'line_c\n'])
622
origins = vf.annotate('references_ghost')
623
self.assertEquals(('base', 'line\n'), origins[0])
624
self.assertEquals(('base', 'line_b\n'), origins[1])
625
self.assertEquals(('references_ghost', 'line_c\n'), origins[2])
627
def test_readonly_mode(self):
628
transport = get_transport(self.get_url('.'))
629
factory = self.get_factory()
630
vf = factory('id', transport, 0777, create=True, access_mode='w')
631
vf = factory('id', transport, access_mode='r')
632
self.assertRaises(errors.ReadOnlyError, vf.add_delta, '', [], '', '', False, [])
633
self.assertRaises(errors.ReadOnlyError, vf.add_lines, 'base', [], [])
634
self.assertRaises(errors.ReadOnlyError,
635
vf.add_lines_with_ghosts,
639
self.assertRaises(errors.ReadOnlyError, vf.fix_parents, 'base', [])
640
self.assertRaises(errors.ReadOnlyError, vf.join, 'base')
641
self.assertRaises(errors.ReadOnlyError, vf.clone_text, 'base', 'bar', ['foo'])
644
class TestWeave(TestCaseWithTransport, VersionedFileTestMixIn):
646
def get_file(self, name='foo'):
647
return WeaveFile(name, get_transport(self.get_url('.')), create=True)
649
def get_file_corrupted_text(self):
650
w = WeaveFile('foo', get_transport(self.get_url('.')), create=True)
651
w.add_lines('v1', [], ['hello\n'])
652
w.add_lines('v2', ['v1'], ['hello\n', 'there\n'])
654
# We are going to invasively corrupt the text
655
# Make sure the internals of weave are the same
656
self.assertEqual([('{', 0)
664
self.assertEqual(['f572d396fae9206628714fb2ce00f72e94f2258f'
665
, '90f265c6e75f1c8f9ab76dcf85528352c5f215ef'
670
w._weave[4] = 'There\n'
673
def get_file_corrupted_checksum(self):
674
w = self.get_file_corrupted_text()
676
w._weave[4] = 'there\n'
677
self.assertEqual('hello\nthere\n', w.get_text('v2'))
679
#Invalid checksum, first digit changed
680
w._sha1s[1] = 'f0f265c6e75f1c8f9ab76dcf85528352c5f215ef'
683
def reopen_file(self, name='foo'):
684
return WeaveFile(name, get_transport(self.get_url('.')))
686
def test_no_implicit_create(self):
687
self.assertRaises(errors.NoSuchFile,
690
get_transport(self.get_url('.')))
692
def get_factory(self):
696
class TestKnit(TestCaseWithTransport, VersionedFileTestMixIn):
698
def get_file(self, name='foo'):
699
return KnitVersionedFile(name, get_transport(self.get_url('.')),
700
delta=True, create=True)
702
def get_factory(self):
703
return KnitVersionedFile
705
def get_file_corrupted_text(self):
706
knit = self.get_file()
707
knit.add_lines('v1', [], ['hello\n'])
708
knit.add_lines('v2', ['v1'], ['hello\n', 'there\n'])
711
def reopen_file(self, name='foo'):
712
return KnitVersionedFile(name, get_transport(self.get_url('.')), delta=True)
714
def test_detection(self):
715
print "TODO for merging: create a corrupted knit."
716
knit = self.get_file()
719
def test_no_implicit_create(self):
720
self.assertRaises(errors.NoSuchFile,
723
get_transport(self.get_url('.')))
726
class InterString(versionedfile.InterVersionedFile):
727
"""An inter-versionedfile optimised code path for strings.
729
This is for use during testing where we use strings as versionedfiles
730
so that none of the default regsitered interversionedfile classes will
731
match - which lets us test the match logic.
735
def is_compatible(source, target):
736
"""InterString is compatible with strings-as-versionedfiles."""
737
return isinstance(source, str) and isinstance(target, str)
740
# TODO this and the InterRepository core logic should be consolidatable
741
# if we make the registry a separate class though we still need to
742
# test the behaviour in the active registry to catch failure-to-handle-
744
class TestInterVersionedFile(TestCaseWithTransport):
746
def test_get_default_inter_versionedfile(self):
747
# test that the InterVersionedFile.get(a, b) probes
748
# for a class where is_compatible(a, b) returns
749
# true and returns a default interversionedfile otherwise.
750
# This also tests that the default registered optimised interversionedfile
751
# classes do not barf inappropriately when a surprising versionedfile type
753
dummy_a = "VersionedFile 1."
754
dummy_b = "VersionedFile 2."
755
self.assertGetsDefaultInterVersionedFile(dummy_a, dummy_b)
757
def assertGetsDefaultInterVersionedFile(self, a, b):
758
"""Asserts that InterVersionedFile.get(a, b) -> the default."""
759
inter = versionedfile.InterVersionedFile.get(a, b)
760
self.assertEqual(versionedfile.InterVersionedFile,
762
self.assertEqual(a, inter.source)
763
self.assertEqual(b, inter.target)
765
def test_register_inter_versionedfile_class(self):
766
# test that a optimised code path provider - a
767
# InterVersionedFile subclass can be registered and unregistered
768
# and that it is correctly selected when given a versionedfile
769
# pair that it returns true on for the is_compatible static method
771
dummy_a = "VersionedFile 1."
772
dummy_b = "VersionedFile 2."
773
versionedfile.InterVersionedFile.register_optimiser(InterString)
775
# we should get the default for something InterString returns False
777
self.assertFalse(InterString.is_compatible(dummy_a, None))
778
self.assertGetsDefaultInterVersionedFile(dummy_a, None)
779
# and we should get an InterString for a pair it 'likes'
780
self.assertTrue(InterString.is_compatible(dummy_a, dummy_b))
781
inter = versionedfile.InterVersionedFile.get(dummy_a, dummy_b)
782
self.assertEqual(InterString, inter.__class__)
783
self.assertEqual(dummy_a, inter.source)
784
self.assertEqual(dummy_b, inter.target)
786
versionedfile.InterVersionedFile.unregister_optimiser(InterString)
787
# now we should get the default InterVersionedFile object again.
788
self.assertGetsDefaultInterVersionedFile(dummy_a, dummy_b)
791
class TestReadonlyHttpMixin(object):
793
def test_readonly_http_works(self):
794
# we should be able to read from http with a versioned file.
796
vf.add_lines('1', [], ['a\n'])
797
vf.add_lines('2', ['1'], ['b\n', 'a\n'])
798
readonly_vf = self.get_factory()('foo', get_transport(self.get_readonly_url('.')))
799
for version in readonly_vf.versions():
800
readonly_vf.get_lines(version)
803
class TestWeaveHTTP(TestCaseWithWebserver, TestReadonlyHttpMixin):
806
return WeaveFile('foo', get_transport(self.get_url('.')), create=True)
808
def get_factory(self):
812
class TestKnitHTTP(TestCaseWithWebserver, TestReadonlyHttpMixin):
815
return KnitVersionedFile('foo', get_transport(self.get_url('.')),
816
delta=True, create=True)
818
def get_factory(self):
819
return KnitVersionedFile
822
class MergeCasesMixin(object):
824
def doMerge(self, base, a, b, mp):
825
from cStringIO import StringIO
826
from textwrap import dedent
832
w.add_lines('text0', [], map(addcrlf, base))
833
w.add_lines('text1', ['text0'], map(addcrlf, a))
834
w.add_lines('text2', ['text0'], map(addcrlf, b))
838
self.log('merge plan:')
839
p = list(w.plan_merge('text1', 'text2'))
840
for state, line in p:
842
self.log('%12s | %s' % (state, line[:-1]))
846
mt.writelines(w.weave_merge(p))
848
self.log(mt.getvalue())
850
mp = map(addcrlf, mp)
851
self.assertEqual(mt.readlines(), mp)
854
def testOneInsert(self):
860
def testSeparateInserts(self):
861
self.doMerge(['aaa', 'bbb', 'ccc'],
862
['aaa', 'xxx', 'bbb', 'ccc'],
863
['aaa', 'bbb', 'yyy', 'ccc'],
864
['aaa', 'xxx', 'bbb', 'yyy', 'ccc'])
866
def testSameInsert(self):
867
self.doMerge(['aaa', 'bbb', 'ccc'],
868
['aaa', 'xxx', 'bbb', 'ccc'],
869
['aaa', 'xxx', 'bbb', 'yyy', 'ccc'],
870
['aaa', 'xxx', 'bbb', 'yyy', 'ccc'])
871
overlappedInsertExpected = ['aaa', 'xxx', 'yyy', 'bbb']
872
def testOverlappedInsert(self):
873
self.doMerge(['aaa', 'bbb'],
874
['aaa', 'xxx', 'yyy', 'bbb'],
875
['aaa', 'xxx', 'bbb'], self.overlappedInsertExpected)
877
# really it ought to reduce this to
878
# ['aaa', 'xxx', 'yyy', 'bbb']
881
def testClashReplace(self):
882
self.doMerge(['aaa'],
885
['<<<<<<< ', 'xxx', '=======', 'yyy', 'zzz',
888
def testNonClashInsert1(self):
889
self.doMerge(['aaa'],
892
['<<<<<<< ', 'xxx', 'aaa', '=======', 'yyy', 'zzz',
895
def testNonClashInsert2(self):
896
self.doMerge(['aaa'],
902
def testDeleteAndModify(self):
903
"""Clashing delete and modification.
905
If one side modifies a region and the other deletes it then
906
there should be a conflict with one side blank.
909
#######################################
910
# skippd, not working yet
913
self.doMerge(['aaa', 'bbb', 'ccc'],
914
['aaa', 'ddd', 'ccc'],
916
['<<<<<<<< ', 'aaa', '=======', '>>>>>>> ', 'ccc'])
918
def _test_merge_from_strings(self, base, a, b, expected):
920
w.add_lines('text0', [], base.splitlines(True))
921
w.add_lines('text1', ['text0'], a.splitlines(True))
922
w.add_lines('text2', ['text0'], b.splitlines(True))
923
self.log('merge plan:')
924
p = list(w.plan_merge('text1', 'text2'))
925
for state, line in p:
927
self.log('%12s | %s' % (state, line[:-1]))
928
self.log('merge result:')
929
result_text = ''.join(w.weave_merge(p))
930
self.log(result_text)
931
self.assertEqualDiff(result_text, expected)
933
def test_weave_merge_conflicts(self):
934
# does weave merge properly handle plans that end with unchanged?
935
result = ''.join(self.get_file().weave_merge([('new-a', 'hello\n')]))
936
self.assertEqual(result, 'hello\n')
938
def test_deletion_extended(self):
939
"""One side deletes, the other deletes more.
956
self._test_merge_from_strings(base, a, b, result)
958
def test_deletion_overlap(self):
959
"""Delete overlapping regions with no other conflict.
961
Arguably it'd be better to treat these as agreement, rather than
962
conflict, but for now conflict is safer.
990
self._test_merge_from_strings(base, a, b, result)
992
def test_agreement_deletion(self):
993
"""Agree to delete some lines, without conflicts."""
1015
self._test_merge_from_strings(base, a, b, result)
1017
def test_sync_on_deletion(self):
1018
"""Specific case of merge where we can synchronize incorrectly.
1020
A previous version of the weave merge concluded that the two versions
1021
agreed on deleting line 2, and this could be a synchronization point.
1022
Line 1 was then considered in isolation, and thought to be deleted on
1025
It's better to consider the whole thing as a disagreement region.
1036
a's replacement line 2
1049
a's replacement line 2
1056
self._test_merge_from_strings(base, a, b, result)
1059
class TestKnitMerge(TestCaseWithTransport, MergeCasesMixin):
1061
def get_file(self, name='foo'):
1062
return KnitVersionedFile(name, get_transport(self.get_url('.')),
1063
delta=True, create=True)
1065
def log_contents(self, w):
1069
class TestWeaveMerge(TestCaseWithTransport, MergeCasesMixin):
1071
def get_file(self, name='foo'):
1072
return WeaveFile(name, get_transport(self.get_url('.')), create=True)
1074
def log_contents(self, w):
1075
self.log('weave is:')
1077
write_weave(w, tmpf)
1078
self.log(tmpf.getvalue())
1080
overlappedInsertExpected = ['aaa', '<<<<<<< ', 'xxx', 'yyy', '=======',
1081
'xxx', '>>>>>>> ', 'bbb']
added
removed
bzrlib/tests/test_versionedfile.py
