1
# Copyright (C) 2005, 2006, 2007 Canonical Ltd
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17
"""Tests for Knit data structure"""
19
from cStringIO import StringIO
31
from bzrlib.errors import (
32
RevisionAlreadyPresent,
37
from bzrlib.index import *
38
from bzrlib.knit import (
58
from bzrlib.osutils import split_lines
59
from bzrlib.tests import (
62
TestCaseWithMemoryTransport,
63
TestCaseWithTransport,
65
from bzrlib.transport import get_transport
66
from bzrlib.transport.memory import MemoryTransport
67
from bzrlib.tuned_gzip import GzipFile
68
from bzrlib.util import bencode
69
from bzrlib.weave import Weave
72
class _CompiledKnitFeature(Feature):
76
import bzrlib._knit_load_data_c
81
def feature_name(self):
82
return 'bzrlib._knit_load_data_c'
84
CompiledKnitFeature = _CompiledKnitFeature()
87
class KnitContentTestsMixin(object):
89
def test_constructor(self):
90
content = self._make_content([])
93
content = self._make_content([])
94
self.assertEqual(content.text(), [])
96
content = self._make_content([("origin1", "text1"), ("origin2", "text2")])
97
self.assertEqual(content.text(), ["text1", "text2"])
100
content = self._make_content([("origin1", "text1"), ("origin2", "text2")])
101
copy = content.copy()
102
self.assertIsInstance(copy, content.__class__)
103
self.assertEqual(copy.annotate(), content.annotate())
105
def assertDerivedBlocksEqual(self, source, target, noeol=False):
106
"""Assert that the derived matching blocks match real output"""
107
source_lines = source.splitlines(True)
108
target_lines = target.splitlines(True)
110
if noeol and not line.endswith('\n'):
114
source_content = self._make_content([(None, nl(l)) for l in source_lines])
115
target_content = self._make_content([(None, nl(l)) for l in target_lines])
116
line_delta = source_content.line_delta(target_content)
117
delta_blocks = list(KnitContent.get_line_delta_blocks(line_delta,
118
source_lines, target_lines))
119
matcher = KnitSequenceMatcher(None, source_lines, target_lines)
120
matcher_blocks = list(list(matcher.get_matching_blocks()))
121
self.assertEqual(matcher_blocks, delta_blocks)
123
def test_get_line_delta_blocks(self):
124
self.assertDerivedBlocksEqual('a\nb\nc\n', 'q\nc\n')
125
self.assertDerivedBlocksEqual(TEXT_1, TEXT_1)
126
self.assertDerivedBlocksEqual(TEXT_1, TEXT_1A)
127
self.assertDerivedBlocksEqual(TEXT_1, TEXT_1B)
128
self.assertDerivedBlocksEqual(TEXT_1B, TEXT_1A)
129
self.assertDerivedBlocksEqual(TEXT_1A, TEXT_1B)
130
self.assertDerivedBlocksEqual(TEXT_1A, '')
131
self.assertDerivedBlocksEqual('', TEXT_1A)
132
self.assertDerivedBlocksEqual('', '')
133
self.assertDerivedBlocksEqual('a\nb\nc', 'a\nb\nc\nd')
135
def test_get_line_delta_blocks_noeol(self):
136
"""Handle historical knit deltas safely
138
Some existing knit deltas don't consider the last line to differ
139
when the only difference whether it has a final newline.
141
New knit deltas appear to always consider the last line to differ
144
self.assertDerivedBlocksEqual('a\nb\nc', 'a\nb\nc\nd\n', noeol=True)
145
self.assertDerivedBlocksEqual('a\nb\nc\nd\n', 'a\nb\nc', noeol=True)
146
self.assertDerivedBlocksEqual('a\nb\nc\n', 'a\nb\nc', noeol=True)
147
self.assertDerivedBlocksEqual('a\nb\nc', 'a\nb\nc\n', noeol=True)
150
class TestPlainKnitContent(TestCase, KnitContentTestsMixin):
152
def _make_content(self, lines):
153
annotated_content = AnnotatedKnitContent(lines)
154
return PlainKnitContent(annotated_content.text(), 'bogus')
156
def test_annotate(self):
157
content = self._make_content([])
158
self.assertEqual(content.annotate(), [])
160
content = self._make_content([("origin1", "text1"), ("origin2", "text2")])
161
self.assertEqual(content.annotate(),
162
[("bogus", "text1"), ("bogus", "text2")])
164
def test_annotate_iter(self):
165
content = self._make_content([])
166
it = content.annotate_iter()
167
self.assertRaises(StopIteration, it.next)
169
content = self._make_content([("bogus", "text1"), ("bogus", "text2")])
170
it = content.annotate_iter()
171
self.assertEqual(it.next(), ("bogus", "text1"))
172
self.assertEqual(it.next(), ("bogus", "text2"))
173
self.assertRaises(StopIteration, it.next)
175
def test_line_delta(self):
176
content1 = self._make_content([("", "a"), ("", "b")])
177
content2 = self._make_content([("", "a"), ("", "a"), ("", "c")])
178
self.assertEqual(content1.line_delta(content2),
179
[(1, 2, 2, ["a", "c"])])
181
def test_line_delta_iter(self):
182
content1 = self._make_content([("", "a"), ("", "b")])
183
content2 = self._make_content([("", "a"), ("", "a"), ("", "c")])
184
it = content1.line_delta_iter(content2)
185
self.assertEqual(it.next(), (1, 2, 2, ["a", "c"]))
186
self.assertRaises(StopIteration, it.next)
189
class TestAnnotatedKnitContent(TestCase, KnitContentTestsMixin):
191
def _make_content(self, lines):
192
return AnnotatedKnitContent(lines)
194
def test_annotate(self):
195
content = self._make_content([])
196
self.assertEqual(content.annotate(), [])
198
content = self._make_content([("origin1", "text1"), ("origin2", "text2")])
199
self.assertEqual(content.annotate(),
200
[("origin1", "text1"), ("origin2", "text2")])
202
def test_annotate_iter(self):
203
content = self._make_content([])
204
it = content.annotate_iter()
205
self.assertRaises(StopIteration, it.next)
207
content = self._make_content([("origin1", "text1"), ("origin2", "text2")])
208
it = content.annotate_iter()
209
self.assertEqual(it.next(), ("origin1", "text1"))
210
self.assertEqual(it.next(), ("origin2", "text2"))
211
self.assertRaises(StopIteration, it.next)
213
def test_line_delta(self):
214
content1 = self._make_content([("", "a"), ("", "b")])
215
content2 = self._make_content([("", "a"), ("", "a"), ("", "c")])
216
self.assertEqual(content1.line_delta(content2),
217
[(1, 2, 2, [("", "a"), ("", "c")])])
219
def test_line_delta_iter(self):
220
content1 = self._make_content([("", "a"), ("", "b")])
221
content2 = self._make_content([("", "a"), ("", "a"), ("", "c")])
222
it = content1.line_delta_iter(content2)
223
self.assertEqual(it.next(), (1, 2, 2, [("", "a"), ("", "c")]))
224
self.assertRaises(StopIteration, it.next)
227
class MockTransport(object):
229
def __init__(self, file_lines=None):
230
self.file_lines = file_lines
232
# We have no base directory for the MockTransport
235
def get(self, filename):
236
if self.file_lines is None:
237
raise NoSuchFile(filename)
239
return StringIO("\n".join(self.file_lines))
241
def readv(self, relpath, offsets):
242
fp = self.get(relpath)
243
for offset, size in offsets:
245
yield offset, fp.read(size)
247
def __getattr__(self, name):
248
def queue_call(*args, **kwargs):
249
self.calls.append((name, args, kwargs))
253
class KnitRecordAccessTestsMixin(object):
254
"""Tests for getting and putting knit records."""
256
def assertAccessExists(self, access):
257
"""Ensure the data area for access has been initialised/exists."""
258
raise NotImplementedError(self.assertAccessExists)
260
def test_add_raw_records(self):
261
"""Add_raw_records adds records retrievable later."""
262
access = self.get_access()
263
memos = access.add_raw_records([10], '1234567890')
264
self.assertEqual(['1234567890'], list(access.get_raw_records(memos)))
266
def test_add_several_raw_records(self):
267
"""add_raw_records with many records and read some back."""
268
access = self.get_access()
269
memos = access.add_raw_records([10, 2, 5], '12345678901234567')
270
self.assertEqual(['1234567890', '12', '34567'],
271
list(access.get_raw_records(memos)))
272
self.assertEqual(['1234567890'],
273
list(access.get_raw_records(memos[0:1])))
274
self.assertEqual(['12'],
275
list(access.get_raw_records(memos[1:2])))
276
self.assertEqual(['34567'],
277
list(access.get_raw_records(memos[2:3])))
278
self.assertEqual(['1234567890', '34567'],
279
list(access.get_raw_records(memos[0:1] + memos[2:3])))
281
def test_create(self):
282
"""create() should make a file on disk."""
283
access = self.get_access()
285
self.assertAccessExists(access)
287
def test_open_file(self):
288
"""open_file never errors."""
289
access = self.get_access()
293
class TestKnitKnitAccess(TestCaseWithMemoryTransport, KnitRecordAccessTestsMixin):
294
"""Tests for the .kndx implementation."""
296
def assertAccessExists(self, access):
297
self.assertNotEqual(None, access.open_file())
299
def get_access(self):
300
"""Get a .knit style access instance."""
301
access = _KnitAccess(self.get_transport(), "foo.knit", None, None,
306
class TestPackKnitAccess(TestCaseWithMemoryTransport, KnitRecordAccessTestsMixin):
307
"""Tests for the pack based access."""
309
def assertAccessExists(self, access):
310
# as pack based access has no backing unless an index maps data, this
314
def get_access(self):
315
return self._get_access()[0]
317
def _get_access(self, packname='packfile', index='FOO'):
318
transport = self.get_transport()
319
def write_data(bytes):
320
transport.append_bytes(packname, bytes)
321
writer = pack.ContainerWriter(write_data)
323
indices = {index:(transport, packname)}
324
access = _PackAccess(indices, writer=(writer, index))
325
return access, writer
327
def test_read_from_several_packs(self):
328
access, writer = self._get_access()
330
memos.extend(access.add_raw_records([10], '1234567890'))
332
access, writer = self._get_access('pack2', 'FOOBAR')
333
memos.extend(access.add_raw_records([5], '12345'))
335
access, writer = self._get_access('pack3', 'BAZ')
336
memos.extend(access.add_raw_records([5], 'alpha'))
338
transport = self.get_transport()
339
access = _PackAccess({"FOO":(transport, 'packfile'),
340
"FOOBAR":(transport, 'pack2'),
341
"BAZ":(transport, 'pack3')})
342
self.assertEqual(['1234567890', '12345', 'alpha'],
343
list(access.get_raw_records(memos)))
344
self.assertEqual(['1234567890'],
345
list(access.get_raw_records(memos[0:1])))
346
self.assertEqual(['12345'],
347
list(access.get_raw_records(memos[1:2])))
348
self.assertEqual(['alpha'],
349
list(access.get_raw_records(memos[2:3])))
350
self.assertEqual(['1234567890', 'alpha'],
351
list(access.get_raw_records(memos[0:1] + memos[2:3])))
353
def test_set_writer(self):
354
"""The writer should be settable post construction."""
355
access = _PackAccess({})
356
transport = self.get_transport()
357
packname = 'packfile'
359
def write_data(bytes):
360
transport.append_bytes(packname, bytes)
361
writer = pack.ContainerWriter(write_data)
363
access.set_writer(writer, index, (transport, packname))
364
memos = access.add_raw_records([10], '1234567890')
366
self.assertEqual(['1234567890'], list(access.get_raw_records(memos)))
369
class LowLevelKnitDataTests(TestCase):
371
def create_gz_content(self, text):
373
gz_file = gzip.GzipFile(mode='wb', fileobj=sio)
376
return sio.getvalue()
378
def test_valid_knit_data(self):
379
sha1sum = sha.new('foo\nbar\n').hexdigest()
380
gz_txt = self.create_gz_content('version rev-id-1 2 %s\n'
385
transport = MockTransport([gz_txt])
386
access = _KnitAccess(transport, 'filename', None, None, False, False)
387
data = _KnitData(access=access)
388
records = [('rev-id-1', (None, 0, len(gz_txt)))]
390
contents = data.read_records(records)
391
self.assertEqual({'rev-id-1':(['foo\n', 'bar\n'], sha1sum)}, contents)
393
raw_contents = list(data.read_records_iter_raw(records))
394
self.assertEqual([('rev-id-1', gz_txt)], raw_contents)
396
def test_not_enough_lines(self):
397
sha1sum = sha.new('foo\n').hexdigest()
398
# record says 2 lines data says 1
399
gz_txt = self.create_gz_content('version rev-id-1 2 %s\n'
403
transport = MockTransport([gz_txt])
404
access = _KnitAccess(transport, 'filename', None, None, False, False)
405
data = _KnitData(access=access)
406
records = [('rev-id-1', (None, 0, len(gz_txt)))]
407
self.assertRaises(errors.KnitCorrupt, data.read_records, records)
409
# read_records_iter_raw won't detect that sort of mismatch/corruption
410
raw_contents = list(data.read_records_iter_raw(records))
411
self.assertEqual([('rev-id-1', gz_txt)], raw_contents)
413
def test_too_many_lines(self):
414
sha1sum = sha.new('foo\nbar\n').hexdigest()
415
# record says 1 lines data says 2
416
gz_txt = self.create_gz_content('version rev-id-1 1 %s\n'
421
transport = MockTransport([gz_txt])
422
access = _KnitAccess(transport, 'filename', None, None, False, False)
423
data = _KnitData(access=access)
424
records = [('rev-id-1', (None, 0, len(gz_txt)))]
425
self.assertRaises(errors.KnitCorrupt, data.read_records, records)
427
# read_records_iter_raw won't detect that sort of mismatch/corruption
428
raw_contents = list(data.read_records_iter_raw(records))
429
self.assertEqual([('rev-id-1', gz_txt)], raw_contents)
431
def test_mismatched_version_id(self):
432
sha1sum = sha.new('foo\nbar\n').hexdigest()
433
gz_txt = self.create_gz_content('version rev-id-1 2 %s\n'
438
transport = MockTransport([gz_txt])
439
access = _KnitAccess(transport, 'filename', None, None, False, False)
440
data = _KnitData(access=access)
441
# We are asking for rev-id-2, but the data is rev-id-1
442
records = [('rev-id-2', (None, 0, len(gz_txt)))]
443
self.assertRaises(errors.KnitCorrupt, data.read_records, records)
445
# read_records_iter_raw will notice if we request the wrong version.
446
self.assertRaises(errors.KnitCorrupt, list,
447
data.read_records_iter_raw(records))
449
def test_uncompressed_data(self):
450
sha1sum = sha.new('foo\nbar\n').hexdigest()
451
txt = ('version rev-id-1 2 %s\n'
456
transport = MockTransport([txt])
457
access = _KnitAccess(transport, 'filename', None, None, False, False)
458
data = _KnitData(access=access)
459
records = [('rev-id-1', (None, 0, len(txt)))]
461
# We don't have valid gzip data ==> corrupt
462
self.assertRaises(errors.KnitCorrupt, data.read_records, records)
464
# read_records_iter_raw will notice the bad data
465
self.assertRaises(errors.KnitCorrupt, list,
466
data.read_records_iter_raw(records))
468
def test_corrupted_data(self):
469
sha1sum = sha.new('foo\nbar\n').hexdigest()
470
gz_txt = self.create_gz_content('version rev-id-1 2 %s\n'
475
# Change 2 bytes in the middle to \xff
476
gz_txt = gz_txt[:10] + '\xff\xff' + gz_txt[12:]
477
transport = MockTransport([gz_txt])
478
access = _KnitAccess(transport, 'filename', None, None, False, False)
479
data = _KnitData(access=access)
480
records = [('rev-id-1', (None, 0, len(gz_txt)))]
482
self.assertRaises(errors.KnitCorrupt, data.read_records, records)
484
# read_records_iter_raw will notice if we request the wrong version.
485
self.assertRaises(errors.KnitCorrupt, list,
486
data.read_records_iter_raw(records))
489
class LowLevelKnitIndexTests(TestCase):
491
def get_knit_index(self, *args, **kwargs):
492
orig = knit._load_data
494
knit._load_data = orig
495
self.addCleanup(reset)
496
from bzrlib._knit_load_data_py import _load_data_py
497
knit._load_data = _load_data_py
498
return _KnitIndex(get_scope=lambda:None, *args, **kwargs)
500
def test_no_such_file(self):
501
transport = MockTransport()
503
self.assertRaises(NoSuchFile, self.get_knit_index,
504
transport, "filename", "r")
505
self.assertRaises(NoSuchFile, self.get_knit_index,
506
transport, "filename", "w", create=False)
508
def test_create_file(self):
509
transport = MockTransport()
511
index = self.get_knit_index(transport, "filename", "w",
512
file_mode="wb", create=True)
514
("put_bytes_non_atomic",
515
("filename", index.HEADER), {"mode": "wb"}),
516
transport.calls.pop(0))
518
def test_delay_create_file(self):
519
transport = MockTransport()
521
index = self.get_knit_index(transport, "filename", "w",
522
create=True, file_mode="wb", create_parent_dir=True,
523
delay_create=True, dir_mode=0777)
524
self.assertEqual([], transport.calls)
526
index.add_versions([])
527
name, (filename, f), kwargs = transport.calls.pop(0)
528
self.assertEqual("put_file_non_atomic", name)
530
{"dir_mode": 0777, "create_parent_dir": True, "mode": "wb"},
532
self.assertEqual("filename", filename)
533
self.assertEqual(index.HEADER, f.read())
535
index.add_versions([])
536
self.assertEqual(("append_bytes", ("filename", ""), {}),
537
transport.calls.pop(0))
539
def test_read_utf8_version_id(self):
540
unicode_revision_id = u"version-\N{CYRILLIC CAPITAL LETTER A}"
541
utf8_revision_id = unicode_revision_id.encode('utf-8')
542
transport = MockTransport([
544
'%s option 0 1 :' % (utf8_revision_id,)
546
index = self.get_knit_index(transport, "filename", "r")
547
# _KnitIndex is a private class, and deals in utf8 revision_ids, not
548
# Unicode revision_ids.
549
self.assertTrue(index.has_version(utf8_revision_id))
550
self.assertFalse(index.has_version(unicode_revision_id))
552
def test_read_utf8_parents(self):
553
unicode_revision_id = u"version-\N{CYRILLIC CAPITAL LETTER A}"
554
utf8_revision_id = unicode_revision_id.encode('utf-8')
555
transport = MockTransport([
557
"version option 0 1 .%s :" % (utf8_revision_id,)
559
index = self.get_knit_index(transport, "filename", "r")
560
self.assertEqual((utf8_revision_id,),
561
index.get_parents_with_ghosts("version"))
563
def test_read_ignore_corrupted_lines(self):
564
transport = MockTransport([
567
"corrupted options 0 1 .b .c ",
568
"version options 0 1 :"
570
index = self.get_knit_index(transport, "filename", "r")
571
self.assertEqual(1, index.num_versions())
572
self.assertTrue(index.has_version("version"))
574
def test_read_corrupted_header(self):
575
transport = MockTransport(['not a bzr knit index header\n'])
576
self.assertRaises(KnitHeaderError,
577
self.get_knit_index, transport, "filename", "r")
579
def test_read_duplicate_entries(self):
580
transport = MockTransport([
582
"parent options 0 1 :",
583
"version options1 0 1 0 :",
584
"version options2 1 2 .other :",
585
"version options3 3 4 0 .other :"
587
index = self.get_knit_index(transport, "filename", "r")
588
self.assertEqual(2, index.num_versions())
589
# check that the index used is the first one written. (Specific
590
# to KnitIndex style indices.
591
self.assertEqual("1", index._version_list_to_index(["version"]))
592
self.assertEqual((None, 3, 4), index.get_position("version"))
593
self.assertEqual(["options3"], index.get_options("version"))
594
self.assertEqual(("parent", "other"),
595
index.get_parents_with_ghosts("version"))
597
def test_read_compressed_parents(self):
598
transport = MockTransport([
602
"c option 0 1 1 0 :",
604
index = self.get_knit_index(transport, "filename", "r")
605
self.assertEqual({"b":("a",), "c":("b", "a")},
606
index.get_parent_map(["b", "c"]))
608
def test_write_utf8_version_id(self):
609
unicode_revision_id = u"version-\N{CYRILLIC CAPITAL LETTER A}"
610
utf8_revision_id = unicode_revision_id.encode('utf-8')
611
transport = MockTransport([
614
index = self.get_knit_index(transport, "filename", "r")
615
index.add_version(utf8_revision_id, ["option"], (None, 0, 1), [])
616
self.assertEqual(("append_bytes", ("filename",
617
"\n%s option 0 1 :" % (utf8_revision_id,)),
619
transport.calls.pop(0))
621
def test_write_utf8_parents(self):
622
unicode_revision_id = u"version-\N{CYRILLIC CAPITAL LETTER A}"
623
utf8_revision_id = unicode_revision_id.encode('utf-8')
624
transport = MockTransport([
627
index = self.get_knit_index(transport, "filename", "r")
628
index.add_version("version", ["option"], (None, 0, 1), [utf8_revision_id])
629
self.assertEqual(("append_bytes", ("filename",
630
"\nversion option 0 1 .%s :" % (utf8_revision_id,)),
632
transport.calls.pop(0))
634
def test_get_ancestry(self):
635
transport = MockTransport([
638
"b option 0 1 0 .e :",
639
"c option 0 1 1 0 :",
640
"d option 0 1 2 .f :"
642
index = self.get_knit_index(transport, "filename", "r")
644
self.assertEqual([], index.get_ancestry([]))
645
self.assertEqual(["a"], index.get_ancestry(["a"]))
646
self.assertEqual(["a", "b"], index.get_ancestry(["b"]))
647
self.assertEqual(["a", "b", "c"], index.get_ancestry(["c"]))
648
self.assertEqual(["a", "b", "c", "d"], index.get_ancestry(["d"]))
649
self.assertEqual(["a", "b"], index.get_ancestry(["a", "b"]))
650
self.assertEqual(["a", "b", "c"], index.get_ancestry(["a", "c"]))
652
self.assertRaises(RevisionNotPresent, index.get_ancestry, ["e"])
654
def test_get_ancestry_with_ghosts(self):
655
transport = MockTransport([
658
"b option 0 1 0 .e :",
659
"c option 0 1 0 .f .g :",
660
"d option 0 1 2 .h .j .k :"
662
index = self.get_knit_index(transport, "filename", "r")
664
self.assertEqual([], index.get_ancestry_with_ghosts([]))
665
self.assertEqual(["a"], index.get_ancestry_with_ghosts(["a"]))
666
self.assertEqual(["a", "e", "b"],
667
index.get_ancestry_with_ghosts(["b"]))
668
self.assertEqual(["a", "g", "f", "c"],
669
index.get_ancestry_with_ghosts(["c"]))
670
self.assertEqual(["a", "g", "f", "c", "k", "j", "h", "d"],
671
index.get_ancestry_with_ghosts(["d"]))
672
self.assertEqual(["a", "e", "b"],
673
index.get_ancestry_with_ghosts(["a", "b"]))
674
self.assertEqual(["a", "g", "f", "c"],
675
index.get_ancestry_with_ghosts(["a", "c"]))
677
["a", "g", "f", "c", "e", "b", "k", "j", "h", "d"],
678
index.get_ancestry_with_ghosts(["b", "d"]))
680
self.assertRaises(RevisionNotPresent,
681
index.get_ancestry_with_ghosts, ["e"])
683
def test_num_versions(self):
684
transport = MockTransport([
687
index = self.get_knit_index(transport, "filename", "r")
689
self.assertEqual(0, index.num_versions())
690
self.assertEqual(0, len(index))
692
index.add_version("a", ["option"], (None, 0, 1), [])
693
self.assertEqual(1, index.num_versions())
694
self.assertEqual(1, len(index))
696
index.add_version("a", ["option2"], (None, 1, 2), [])
697
self.assertEqual(1, index.num_versions())
698
self.assertEqual(1, len(index))
700
index.add_version("b", ["option"], (None, 0, 1), [])
701
self.assertEqual(2, index.num_versions())
702
self.assertEqual(2, len(index))
704
def test_get_versions(self):
705
transport = MockTransport([
708
index = self.get_knit_index(transport, "filename", "r")
710
self.assertEqual([], index.get_versions())
712
index.add_version("a", ["option"], (None, 0, 1), [])
713
self.assertEqual(["a"], index.get_versions())
715
index.add_version("a", ["option"], (None, 0, 1), [])
716
self.assertEqual(["a"], index.get_versions())
718
index.add_version("b", ["option"], (None, 0, 1), [])
719
self.assertEqual(["a", "b"], index.get_versions())
721
def test_add_version(self):
722
transport = MockTransport([
725
index = self.get_knit_index(transport, "filename", "r")
727
index.add_version("a", ["option"], (None, 0, 1), ["b"])
728
self.assertEqual(("append_bytes",
729
("filename", "\na option 0 1 .b :"),
730
{}), transport.calls.pop(0))
731
self.assertTrue(index.has_version("a"))
732
self.assertEqual(1, index.num_versions())
733
self.assertEqual((None, 0, 1), index.get_position("a"))
734
self.assertEqual(["option"], index.get_options("a"))
735
self.assertEqual(("b",), index.get_parents_with_ghosts("a"))
737
index.add_version("a", ["opt"], (None, 1, 2), ["c"])
738
self.assertEqual(("append_bytes",
739
("filename", "\na opt 1 2 .c :"),
740
{}), transport.calls.pop(0))
741
self.assertTrue(index.has_version("a"))
742
self.assertEqual(1, index.num_versions())
743
self.assertEqual((None, 1, 2), index.get_position("a"))
744
self.assertEqual(["opt"], index.get_options("a"))
745
self.assertEqual(("c",), index.get_parents_with_ghosts("a"))
747
index.add_version("b", ["option"], (None, 2, 3), ["a"])
748
self.assertEqual(("append_bytes",
749
("filename", "\nb option 2 3 0 :"),
750
{}), transport.calls.pop(0))
751
self.assertTrue(index.has_version("b"))
752
self.assertEqual(2, index.num_versions())
753
self.assertEqual((None, 2, 3), index.get_position("b"))
754
self.assertEqual(["option"], index.get_options("b"))
755
self.assertEqual(("a",), index.get_parents_with_ghosts("b"))
757
def test_add_versions(self):
758
transport = MockTransport([
761
index = self.get_knit_index(transport, "filename", "r")
764
("a", ["option"], (None, 0, 1), ["b"]),
765
("a", ["opt"], (None, 1, 2), ["c"]),
766
("b", ["option"], (None, 2, 3), ["a"])
768
self.assertEqual(("append_bytes", ("filename",
769
"\na option 0 1 .b :"
772
), {}), transport.calls.pop(0))
773
self.assertTrue(index.has_version("a"))
774
self.assertTrue(index.has_version("b"))
775
self.assertEqual(2, index.num_versions())
776
self.assertEqual((None, 1, 2), index.get_position("a"))
777
self.assertEqual((None, 2, 3), index.get_position("b"))
778
self.assertEqual(["opt"], index.get_options("a"))
779
self.assertEqual(["option"], index.get_options("b"))
780
self.assertEqual(("c",), index.get_parents_with_ghosts("a"))
781
self.assertEqual(("a",), index.get_parents_with_ghosts("b"))
783
def test_add_versions_random_id_is_accepted(self):
784
transport = MockTransport([
787
index = self.get_knit_index(transport, "filename", "r")
790
("a", ["option"], (None, 0, 1), ["b"]),
791
("a", ["opt"], (None, 1, 2), ["c"]),
792
("b", ["option"], (None, 2, 3), ["a"])
795
def test_delay_create_and_add_versions(self):
796
transport = MockTransport()
798
index = self.get_knit_index(transport, "filename", "w",
799
create=True, file_mode="wb", create_parent_dir=True,
800
delay_create=True, dir_mode=0777)
801
self.assertEqual([], transport.calls)
804
("a", ["option"], (None, 0, 1), ["b"]),
805
("a", ["opt"], (None, 1, 2), ["c"]),
806
("b", ["option"], (None, 2, 3), ["a"])
808
name, (filename, f), kwargs = transport.calls.pop(0)
809
self.assertEqual("put_file_non_atomic", name)
811
{"dir_mode": 0777, "create_parent_dir": True, "mode": "wb"},
813
self.assertEqual("filename", filename)
816
"\na option 0 1 .b :"
818
"\nb option 2 3 0 :",
821
def test_has_version(self):
822
transport = MockTransport([
826
index = self.get_knit_index(transport, "filename", "r")
828
self.assertTrue(index.has_version("a"))
829
self.assertFalse(index.has_version("b"))
831
def test_get_position(self):
832
transport = MockTransport([
837
index = self.get_knit_index(transport, "filename", "r")
839
self.assertEqual((None, 0, 1), index.get_position("a"))
840
self.assertEqual((None, 1, 2), index.get_position("b"))
842
def test_get_method(self):
843
transport = MockTransport([
845
"a fulltext,unknown 0 1 :",
846
"b unknown,line-delta 1 2 :",
849
index = self.get_knit_index(transport, "filename", "r")
851
self.assertEqual("fulltext", index.get_method("a"))
852
self.assertEqual("line-delta", index.get_method("b"))
853
self.assertRaises(errors.KnitIndexUnknownMethod, index.get_method, "c")
855
def test_get_options(self):
856
transport = MockTransport([
861
index = self.get_knit_index(transport, "filename", "r")
863
self.assertEqual(["opt1"], index.get_options("a"))
864
self.assertEqual(["opt2", "opt3"], index.get_options("b"))
866
def test_get_parent_map(self):
867
transport = MockTransport([
870
"b option 1 2 0 .c :",
871
"c option 1 2 1 0 .e :"
873
index = self.get_knit_index(transport, "filename", "r")
879
}, index.get_parent_map(["a", "b", "c"]))
881
def test_get_parents_with_ghosts(self):
882
transport = MockTransport([
885
"b option 1 2 0 .c :",
886
"c option 1 2 1 0 .e :"
888
index = self.get_knit_index(transport, "filename", "r")
890
self.assertEqual((), index.get_parents_with_ghosts("a"))
891
self.assertEqual(("a", "c"), index.get_parents_with_ghosts("b"))
892
self.assertEqual(("b", "a", "e"),
893
index.get_parents_with_ghosts("c"))
895
def test_check_versions_present(self):
896
transport = MockTransport([
901
index = self.get_knit_index(transport, "filename", "r")
903
check = index.check_versions_present
909
self.assertRaises(RevisionNotPresent, check, ["c"])
910
self.assertRaises(RevisionNotPresent, check, ["a", "b", "c"])
912
def test_impossible_parent(self):
913
"""Test we get KnitCorrupt if the parent couldn't possibly exist."""
914
transport = MockTransport([
917
"b option 0 1 4 :" # We don't have a 4th record
920
self.assertRaises(errors.KnitCorrupt,
921
self.get_knit_index, transport, 'filename', 'r')
923
if (str(e) == ('exceptions must be strings, classes, or instances,'
924
' not exceptions.IndexError')
925
and sys.version_info[0:2] >= (2,5)):
926
self.knownFailure('Pyrex <0.9.5 fails with TypeError when'
927
' raising new style exceptions with python'
932
def test_corrupted_parent(self):
933
transport = MockTransport([
937
"c option 0 1 1v :", # Can't have a parent of '1v'
940
self.assertRaises(errors.KnitCorrupt,
941
self.get_knit_index, transport, 'filename', 'r')
943
if (str(e) == ('exceptions must be strings, classes, or instances,'
944
' not exceptions.ValueError')
945
and sys.version_info[0:2] >= (2,5)):
946
self.knownFailure('Pyrex <0.9.5 fails with TypeError when'
947
' raising new style exceptions with python'
952
def test_corrupted_parent_in_list(self):
953
transport = MockTransport([
957
"c option 0 1 1 v :", # Can't have a parent of 'v'
960
self.assertRaises(errors.KnitCorrupt,
961
self.get_knit_index, transport, 'filename', 'r')
963
if (str(e) == ('exceptions must be strings, classes, or instances,'
964
' not exceptions.ValueError')
965
and sys.version_info[0:2] >= (2,5)):
966
self.knownFailure('Pyrex <0.9.5 fails with TypeError when'
967
' raising new style exceptions with python'
972
def test_invalid_position(self):
973
transport = MockTransport([
978
self.assertRaises(errors.KnitCorrupt,
979
self.get_knit_index, transport, 'filename', 'r')
981
if (str(e) == ('exceptions must be strings, classes, or instances,'
982
' not exceptions.ValueError')
983
and sys.version_info[0:2] >= (2,5)):
984
self.knownFailure('Pyrex <0.9.5 fails with TypeError when'
985
' raising new style exceptions with python'
990
def test_invalid_size(self):
991
transport = MockTransport([
996
self.assertRaises(errors.KnitCorrupt,
997
self.get_knit_index, transport, 'filename', 'r')
999
if (str(e) == ('exceptions must be strings, classes, or instances,'
1000
' not exceptions.ValueError')
1001
and sys.version_info[0:2] >= (2,5)):
1002
self.knownFailure('Pyrex <0.9.5 fails with TypeError when'
1003
' raising new style exceptions with python'
1008
def test_short_line(self):
1009
transport = MockTransport([
1012
"b option 10 10 0", # This line isn't terminated, ignored
1014
index = self.get_knit_index(transport, "filename", "r")
1015
self.assertEqual(['a'], index.get_versions())
1017
def test_skip_incomplete_record(self):
1018
# A line with bogus data should just be skipped
1019
transport = MockTransport([
1022
"b option 10 10 0", # This line isn't terminated, ignored
1023
"c option 20 10 0 :", # Properly terminated, and starts with '\n'
1025
index = self.get_knit_index(transport, "filename", "r")
1026
self.assertEqual(['a', 'c'], index.get_versions())
1028
def test_trailing_characters(self):
1029
# A line with bogus data should just be skipped
1030
transport = MockTransport([
1033
"b option 10 10 0 :a", # This line has extra trailing characters
1034
"c option 20 10 0 :", # Properly terminated, and starts with '\n'
1036
index = self.get_knit_index(transport, "filename", "r")
1037
self.assertEqual(['a', 'c'], index.get_versions())
1040
class LowLevelKnitIndexTests_c(LowLevelKnitIndexTests):
1042
_test_needs_features = [CompiledKnitFeature]
1044
def get_knit_index(self, *args, **kwargs):
1045
orig = knit._load_data
1047
knit._load_data = orig
1048
self.addCleanup(reset)
1049
from bzrlib._knit_load_data_c import _load_data_c
1050
knit._load_data = _load_data_c
1051
return _KnitIndex(get_scope=lambda:None, *args, **kwargs)
1054
class KnitTests(TestCaseWithTransport):
1055
"""Class containing knit test helper routines."""
1057
def make_test_knit(self, annotate=False, delay_create=False, index=None,
1058
name='test', delta=True, access_mode='w'):
1060
factory = KnitPlainFactory()
1064
index = _KnitIndex(get_transport('.'), name + INDEX_SUFFIX,
1065
access_mode, create=True, file_mode=None,
1066
create_parent_dir=False, delay_create=delay_create,
1067
dir_mode=None, get_scope=lambda:None)
1068
access = _KnitAccess(get_transport('.'), name + DATA_SUFFIX, None,
1069
None, delay_create, False)
1070
return KnitVersionedFile(name, get_transport('.'), factory=factory,
1071
create=True, delay_create=delay_create, index=index,
1072
access_method=access, delta=delta)
1074
def assertRecordContentEqual(self, knit, version_id, candidate_content):
1075
"""Assert that some raw record content matches the raw record content
1076
for a particular version_id in the given knit.
1078
index_memo = knit._index.get_position(version_id)
1079
record = (version_id, index_memo)
1080
[(_, expected_content)] = list(knit._data.read_records_iter_raw([record]))
1081
self.assertEqual(expected_content, candidate_content)
1084
class BasicKnitTests(KnitTests):
1086
def add_stock_one_and_one_a(self, k):
1087
k.add_lines('text-1', [], split_lines(TEXT_1))
1088
k.add_lines('text-1a', ['text-1'], split_lines(TEXT_1A))
1090
def test_knit_constructor(self):
1091
"""Construct empty k"""
1092
self.make_test_knit()
1094
def test_make_explicit_index(self):
1095
"""We can supply an index to use."""
1096
knit = KnitVersionedFile('test', get_transport('.'),
1097
index='strangelove', access_method="a")
1098
self.assertEqual(knit._index, 'strangelove')
1100
def test_knit_add(self):
1101
"""Store one text in knit and retrieve"""
1102
k = self.make_test_knit()
1103
k.add_lines('text-1', [], split_lines(TEXT_1))
1104
self.assertTrue(k.has_version('text-1'))
1105
self.assertEqualDiff(''.join(k.get_lines('text-1')), TEXT_1)
1107
def test_newline_empty_lines(self):
1108
# ensure that ["\n"] round trips ok.
1109
knit = self.make_test_knit()
1110
knit.add_lines('a', [], ["\n"])
1111
knit.add_lines_with_ghosts('b', [], ["\n"])
1112
self.assertEqual(["\n"], knit.get_lines('a'))
1113
self.assertEqual(["\n"], knit.get_lines('b'))
1114
self.assertEqual(['fulltext'], knit._index.get_options('a'))
1115
self.assertEqual(['fulltext'], knit._index.get_options('b'))
1116
knit.add_lines('c', ['a'], ["\n"])
1117
knit.add_lines_with_ghosts('d', ['b'], ["\n"])
1118
self.assertEqual(["\n"], knit.get_lines('c'))
1119
self.assertEqual(["\n"], knit.get_lines('d'))
1120
self.assertEqual(['line-delta'], knit._index.get_options('c'))
1121
self.assertEqual(['line-delta'], knit._index.get_options('d'))
1123
def test_empty_lines(self):
1124
# bizarrely, [] is not listed as having no-eol.
1125
knit = self.make_test_knit()
1126
knit.add_lines('a', [], [])
1127
knit.add_lines_with_ghosts('b', [], [])
1128
self.assertEqual([], knit.get_lines('a'))
1129
self.assertEqual([], knit.get_lines('b'))
1130
self.assertEqual(['fulltext'], knit._index.get_options('a'))
1131
self.assertEqual(['fulltext'], knit._index.get_options('b'))
1132
knit.add_lines('c', ['a'], [])
1133
knit.add_lines_with_ghosts('d', ['b'], [])
1134
self.assertEqual([], knit.get_lines('c'))
1135
self.assertEqual([], knit.get_lines('d'))
1136
self.assertEqual(['line-delta'], knit._index.get_options('c'))
1137
self.assertEqual(['line-delta'], knit._index.get_options('d'))
1139
def test_knit_reload(self):
1140
# test that the content in a reloaded knit is correct
1141
k = self.make_test_knit()
1142
k.add_lines('text-1', [], split_lines(TEXT_1))
1144
k2 = make_file_knit('test', get_transport('.'), access_mode='r',
1145
factory=KnitPlainFactory(), create=True)
1146
self.assertTrue(k2.has_version('text-1'))
1147
self.assertEqualDiff(''.join(k2.get_lines('text-1')), TEXT_1)
1149
def test_knit_several(self):
1150
"""Store several texts in a knit"""
1151
k = self.make_test_knit()
1152
k.add_lines('text-1', [], split_lines(TEXT_1))
1153
k.add_lines('text-2', [], split_lines(TEXT_2))
1154
self.assertEqualDiff(''.join(k.get_lines('text-1')), TEXT_1)
1155
self.assertEqualDiff(''.join(k.get_lines('text-2')), TEXT_2)
1157
def test_repeated_add(self):
1158
"""Knit traps attempt to replace existing version"""
1159
k = self.make_test_knit()
1160
k.add_lines('text-1', [], split_lines(TEXT_1))
1161
self.assertRaises(RevisionAlreadyPresent,
1163
'text-1', [], split_lines(TEXT_1))
1165
def test_empty(self):
1166
k = self.make_test_knit(True)
1167
k.add_lines('text-1', [], [])
1168
self.assertEquals(k.get_lines('text-1'), [])
1170
def test_incomplete(self):
1171
"""Test if texts without a ending line-end can be inserted and
1173
k = make_file_knit('test', get_transport('.'), delta=False, create=True)
1174
k.add_lines('text-1', [], ['a\n', 'b' ])
1175
k.add_lines('text-2', ['text-1'], ['a\rb\n', 'b\n'])
1176
# reopening ensures maximum room for confusion
1177
k = make_file_knit('test', get_transport('.'), delta=False, create=True)
1178
self.assertEquals(k.get_lines('text-1'), ['a\n', 'b' ])
1179
self.assertEquals(k.get_lines('text-2'), ['a\rb\n', 'b\n'])
1181
def test_delta(self):
1182
"""Expression of knit delta as lines"""
1183
k = self.make_test_knit()
1184
td = list(line_delta(TEXT_1.splitlines(True),
1185
TEXT_1A.splitlines(True)))
1186
self.assertEqualDiff(''.join(td), delta_1_1a)
1187
out = apply_line_delta(TEXT_1.splitlines(True), td)
1188
self.assertEqualDiff(''.join(out), TEXT_1A)
1190
def test_add_with_parents(self):
1191
"""Store in knit with parents"""
1192
k = self.make_test_knit()
1193
self.add_stock_one_and_one_a(k)
1194
self.assertEqual({'text-1':(), 'text-1a':('text-1',)},
1195
k.get_parent_map(['text-1', 'text-1a']))
1197
def test_ancestry(self):
1198
"""Store in knit with parents"""
1199
k = self.make_test_knit()
1200
self.add_stock_one_and_one_a(k)
1201
self.assertEquals(set(k.get_ancestry(['text-1a'])), set(['text-1a', 'text-1']))
1203
def test_add_delta(self):
1204
"""Store in knit with parents"""
1205
k = self.make_test_knit(annotate=False)
1206
self.add_stock_one_and_one_a(k)
1208
self.assertEqualDiff(''.join(k.get_lines('text-1a')), TEXT_1A)
1210
def test_add_delta_knit_graph_index(self):
1211
"""Does adding work with a KnitGraphIndex."""
1212
index = InMemoryGraphIndex(2)
1213
knit_index = KnitGraphIndex(index, add_callback=index.add_nodes,
1215
k = self.make_test_knit(annotate=True, index=knit_index)
1216
self.add_stock_one_and_one_a(k)
1218
self.assertEqualDiff(''.join(k.get_lines('text-1a')), TEXT_1A)
1219
# check the index had the right data added.
1220
self.assertEqual(set([
1221
(index, ('text-1', ), ' 0 127', ((), ())),
1222
(index, ('text-1a', ), ' 127 140', ((('text-1', ),), (('text-1', ),))),
1223
]), set(index.iter_all_entries()))
1224
# we should not have a .kndx file
1225
self.assertFalse(get_transport('.').has('test.kndx'))
1227
def test_annotate(self):
1229
k = self.make_test_knit(annotate=True, name='knit')
1230
self.insert_and_test_small_annotate(k)
1232
def insert_and_test_small_annotate(self, k):
1233
"""test annotation with k works correctly."""
1234
k.add_lines('text-1', [], ['a\n', 'b\n'])
1235
k.add_lines('text-2', ['text-1'], ['a\n', 'c\n'])
1237
origins = k.annotate('text-2')
1238
self.assertEquals(origins[0], ('text-1', 'a\n'))
1239
self.assertEquals(origins[1], ('text-2', 'c\n'))
1241
def test_annotate_fulltext(self):
1243
k = self.make_test_knit(annotate=True, name='knit', delta=False)
1244
self.insert_and_test_small_annotate(k)
1246
def test_annotate_merge_1(self):
1247
k = self.make_test_knit(True)
1248
k.add_lines('text-a1', [], ['a\n', 'b\n'])
1249
k.add_lines('text-a2', [], ['d\n', 'c\n'])
1250
k.add_lines('text-am', ['text-a1', 'text-a2'], ['d\n', 'b\n'])
1251
origins = k.annotate('text-am')
1252
self.assertEquals(origins[0], ('text-a2', 'd\n'))
1253
self.assertEquals(origins[1], ('text-a1', 'b\n'))
1255
def test_annotate_merge_2(self):
1256
k = self.make_test_knit(True)
1257
k.add_lines('text-a1', [], ['a\n', 'b\n', 'c\n'])
1258
k.add_lines('text-a2', [], ['x\n', 'y\n', 'z\n'])
1259
k.add_lines('text-am', ['text-a1', 'text-a2'], ['a\n', 'y\n', 'c\n'])
1260
origins = k.annotate('text-am')
1261
self.assertEquals(origins[0], ('text-a1', 'a\n'))
1262
self.assertEquals(origins[1], ('text-a2', 'y\n'))
1263
self.assertEquals(origins[2], ('text-a1', 'c\n'))
1265
def test_annotate_merge_9(self):
1266
k = self.make_test_knit(True)
1267
k.add_lines('text-a1', [], ['a\n', 'b\n', 'c\n'])
1268
k.add_lines('text-a2', [], ['x\n', 'y\n', 'z\n'])
1269
k.add_lines('text-am', ['text-a1', 'text-a2'], ['k\n', 'y\n', 'c\n'])
1270
origins = k.annotate('text-am')
1271
self.assertEquals(origins[0], ('text-am', 'k\n'))
1272
self.assertEquals(origins[1], ('text-a2', 'y\n'))
1273
self.assertEquals(origins[2], ('text-a1', 'c\n'))
1275
def test_annotate_merge_3(self):
1276
k = self.make_test_knit(True)
1277
k.add_lines('text-a1', [], ['a\n', 'b\n', 'c\n'])
1278
k.add_lines('text-a2', [] ,['x\n', 'y\n', 'z\n'])
1279
k.add_lines('text-am', ['text-a1', 'text-a2'], ['k\n', 'y\n', 'z\n'])
1280
origins = k.annotate('text-am')
1281
self.assertEquals(origins[0], ('text-am', 'k\n'))
1282
self.assertEquals(origins[1], ('text-a2', 'y\n'))
1283
self.assertEquals(origins[2], ('text-a2', 'z\n'))
1285
def test_annotate_merge_4(self):
1286
k = self.make_test_knit(True)
1287
k.add_lines('text-a1', [], ['a\n', 'b\n', 'c\n'])
1288
k.add_lines('text-a2', [], ['x\n', 'y\n', 'z\n'])
1289
k.add_lines('text-a3', ['text-a1'], ['a\n', 'b\n', 'p\n'])
1290
k.add_lines('text-am', ['text-a2', 'text-a3'], ['a\n', 'b\n', 'z\n'])
1291
origins = k.annotate('text-am')
1292
self.assertEquals(origins[0], ('text-a1', 'a\n'))
1293
self.assertEquals(origins[1], ('text-a1', 'b\n'))
1294
self.assertEquals(origins[2], ('text-a2', 'z\n'))
1296
def test_annotate_merge_5(self):
1297
k = self.make_test_knit(True)
1298
k.add_lines('text-a1', [], ['a\n', 'b\n', 'c\n'])
1299
k.add_lines('text-a2', [], ['d\n', 'e\n', 'f\n'])
1300
k.add_lines('text-a3', [], ['x\n', 'y\n', 'z\n'])
1301
k.add_lines('text-am',
1302
['text-a1', 'text-a2', 'text-a3'],
1303
['a\n', 'e\n', 'z\n'])
1304
origins = k.annotate('text-am')
1305
self.assertEquals(origins[0], ('text-a1', 'a\n'))
1306
self.assertEquals(origins[1], ('text-a2', 'e\n'))
1307
self.assertEquals(origins[2], ('text-a3', 'z\n'))
1309
def test_annotate_file_cherry_pick(self):
1310
k = self.make_test_knit(True)
1311
k.add_lines('text-1', [], ['a\n', 'b\n', 'c\n'])
1312
k.add_lines('text-2', ['text-1'], ['d\n', 'e\n', 'f\n'])
1313
k.add_lines('text-3', ['text-2', 'text-1'], ['a\n', 'b\n', 'c\n'])
1314
origins = k.annotate('text-3')
1315
self.assertEquals(origins[0], ('text-1', 'a\n'))
1316
self.assertEquals(origins[1], ('text-1', 'b\n'))
1317
self.assertEquals(origins[2], ('text-1', 'c\n'))
1319
def _test_join_with_factories(self, k1_factory, k2_factory):
1320
k1 = make_file_knit('test1', get_transport('.'), factory=k1_factory, create=True)
1321
k1.add_lines('text-a', [], ['a1\n', 'a2\n', 'a3\n'])
1322
k1.add_lines('text-b', ['text-a'], ['a1\n', 'b2\n', 'a3\n'])
1323
k1.add_lines('text-c', [], ['c1\n', 'c2\n', 'c3\n'])
1324
k1.add_lines('text-d', ['text-c'], ['c1\n', 'd2\n', 'd3\n'])
1325
k1.add_lines('text-m', ['text-b', 'text-d'], ['a1\n', 'b2\n', 'd3\n'])
1326
k2 = make_file_knit('test2', get_transport('.'), factory=k2_factory, create=True)
1327
count = k2.join(k1, version_ids=['text-m'])
1328
self.assertEquals(count, 5)
1329
self.assertTrue(k2.has_version('text-a'))
1330
self.assertTrue(k2.has_version('text-c'))
1331
origins = k2.annotate('text-m')
1332
self.assertEquals(origins[0], ('text-a', 'a1\n'))
1333
self.assertEquals(origins[1], ('text-b', 'b2\n'))
1334
self.assertEquals(origins[2], ('text-d', 'd3\n'))
1336
def test_knit_join_plain_to_plain(self):
1337
"""Test joining a plain knit with a plain knit."""
1338
self._test_join_with_factories(KnitPlainFactory(), KnitPlainFactory())
1340
def test_knit_join_anno_to_anno(self):
1341
"""Test joining an annotated knit with an annotated knit."""
1342
self._test_join_with_factories(None, None)
1344
def test_knit_join_anno_to_plain(self):
1345
"""Test joining an annotated knit with a plain knit."""
1346
self._test_join_with_factories(None, KnitPlainFactory())
1348
def test_knit_join_plain_to_anno(self):
1349
"""Test joining a plain knit with an annotated knit."""
1350
self._test_join_with_factories(KnitPlainFactory(), None)
1352
def test_reannotate(self):
1353
k1 = make_file_knit('knit1', get_transport('.'),
1354
factory=KnitAnnotateFactory(), create=True)
1356
k1.add_lines('text-a', [], ['a\n', 'b\n'])
1358
k1.add_lines('text-b', ['text-a'], ['a\n', 'c\n'])
1360
k2 = make_file_knit('test2', get_transport('.'),
1361
factory=KnitAnnotateFactory(), create=True)
1362
k2.join(k1, version_ids=['text-b'])
1365
k1.add_lines('text-X', ['text-b'], ['a\n', 'b\n'])
1367
k2.add_lines('text-c', ['text-b'], ['z\n', 'c\n'])
1369
k2.add_lines('text-Y', ['text-b'], ['b\n', 'c\n'])
1371
# test-c will have index 3
1372
k1.join(k2, version_ids=['text-c'])
1374
lines = k1.get_lines('text-c')
1375
self.assertEquals(lines, ['z\n', 'c\n'])
1377
origins = k1.annotate('text-c')
1378
self.assertEquals(origins[0], ('text-c', 'z\n'))
1379
self.assertEquals(origins[1], ('text-b', 'c\n'))
1381
def test_get_line_delta_texts(self):
1382
"""Make sure we can call get_texts on text with reused line deltas"""
1383
k1 = make_file_knit('test1', get_transport('.'),
1384
factory=KnitPlainFactory(), create=True)
1389
parents = ['%d' % (t-1)]
1390
k1.add_lines('%d' % t, parents, ['hello\n'] * t)
1391
k1.get_texts(('%d' % t) for t in range(3))
1393
def test_iter_lines_reads_in_order(self):
1394
instrumented_t = get_transport('trace+memory:///')
1395
k1 = make_file_knit('id', instrumented_t, create=True, delta=True)
1396
self.assertEqual([('get', 'id.kndx',)], instrumented_t._activity)
1397
# add texts with no required ordering
1398
k1.add_lines('base', [], ['text\n'])
1399
k1.add_lines('base2', [], ['text2\n'])
1401
# clear the logged activity, but preserve the list instance in case of
1402
# clones pointing at it.
1403
del instrumented_t._activity[:]
1404
# request a last-first iteration
1405
results = list(k1.iter_lines_added_or_present_in_versions(
1408
[('readv', 'id.knit', [(0, 87), (87, 89)], False, None)],
1409
instrumented_t._activity)
1410
self.assertEqual([('text\n', 'base'), ('text2\n', 'base2')], results)
1412
def test_knit_format(self):
1413
# this tests that a new knit index file has the expected content
1414
# and that is writes the data we expect as records are added.
1415
knit = self.make_test_knit(True)
1416
# Now knit files are not created until we first add data to them
1417
self.assertFileEqual("# bzr knit index 8\n", 'test.kndx')
1418
knit.add_lines_with_ghosts('revid', ['a_ghost'], ['a\n'])
1419
self.assertFileEqual(
1420
"# bzr knit index 8\n"
1422
"revid fulltext 0 84 .a_ghost :",
1424
knit.add_lines_with_ghosts('revid2', ['revid'], ['a\n'])
1425
self.assertFileEqual(
1426
"# bzr knit index 8\n"
1427
"\nrevid fulltext 0 84 .a_ghost :"
1428
"\nrevid2 line-delta 84 82 0 :",
1430
# we should be able to load this file again
1431
knit = make_file_knit('test', get_transport('.'), access_mode='r')
1432
self.assertEqual(['revid', 'revid2'], knit.versions())
1433
# write a short write to the file and ensure that its ignored
1434
indexfile = file('test.kndx', 'ab')
1435
indexfile.write('\nrevid3 line-delta 166 82 1 2 3 4 5 .phwoar:demo ')
1437
# we should be able to load this file again
1438
knit = make_file_knit('test', get_transport('.'), access_mode='w')
1439
self.assertEqual(['revid', 'revid2'], knit.versions())
1440
# and add a revision with the same id the failed write had
1441
knit.add_lines('revid3', ['revid2'], ['a\n'])
1442
# and when reading it revid3 should now appear.
1443
knit = make_file_knit('test', get_transport('.'), access_mode='r')
1444
self.assertEqual(['revid', 'revid2', 'revid3'], knit.versions())
1445
self.assertEqual({'revid3':('revid2',)}, knit.get_parent_map(['revid3']))
1447
def test_delay_create(self):
1448
"""Test that passing delay_create=True creates files late"""
1449
knit = self.make_test_knit(annotate=True, delay_create=True)
1450
self.failIfExists('test.knit')
1451
self.failIfExists('test.kndx')
1452
knit.add_lines_with_ghosts('revid', ['a_ghost'], ['a\n'])
1453
self.failUnlessExists('test.knit')
1454
self.assertFileEqual(
1455
"# bzr knit index 8\n"
1457
"revid fulltext 0 84 .a_ghost :",
1460
def test_create_parent_dir(self):
1461
"""create_parent_dir can create knits in nonexistant dirs"""
1462
# Has no effect if we don't set 'delay_create'
1463
trans = get_transport('.')
1464
self.assertRaises(NoSuchFile, make_file_knit, 'dir/test',
1465
trans, access_mode='w', factory=None,
1466
create=True, create_parent_dir=True)
1467
# Nothing should have changed yet
1468
knit = make_file_knit('dir/test', trans, access_mode='w',
1469
factory=None, create=True,
1470
create_parent_dir=True,
1472
self.failIfExists('dir/test.knit')
1473
self.failIfExists('dir/test.kndx')
1474
self.failIfExists('dir')
1475
knit.add_lines('revid', [], ['a\n'])
1476
self.failUnlessExists('dir')
1477
self.failUnlessExists('dir/test.knit')
1478
self.assertFileEqual(
1479
"# bzr knit index 8\n"
1481
"revid fulltext 0 84 :",
1484
def test_create_mode_700(self):
1485
trans = get_transport('.')
1486
if not trans._can_roundtrip_unix_modebits():
1487
# Can't roundtrip, so no need to run this test
1489
knit = make_file_knit('dir/test', trans, access_mode='w', factory=None,
1490
create=True, create_parent_dir=True, delay_create=True,
1491
file_mode=0600, dir_mode=0700)
1492
knit.add_lines('revid', [], ['a\n'])
1493
self.assertTransportMode(trans, 'dir', 0700)
1494
self.assertTransportMode(trans, 'dir/test.knit', 0600)
1495
self.assertTransportMode(trans, 'dir/test.kndx', 0600)
1497
def test_create_mode_770(self):
1498
trans = get_transport('.')
1499
if not trans._can_roundtrip_unix_modebits():
1500
# Can't roundtrip, so no need to run this test
1502
knit = make_file_knit('dir/test', trans, access_mode='w', factory=None,
1503
create=True, create_parent_dir=True, delay_create=True,
1504
file_mode=0660, dir_mode=0770)
1505
knit.add_lines('revid', [], ['a\n'])
1506
self.assertTransportMode(trans, 'dir', 0770)
1507
self.assertTransportMode(trans, 'dir/test.knit', 0660)
1508
self.assertTransportMode(trans, 'dir/test.kndx', 0660)
1510
def test_create_mode_777(self):
1511
trans = get_transport('.')
1512
if not trans._can_roundtrip_unix_modebits():
1513
# Can't roundtrip, so no need to run this test
1515
knit = make_file_knit('dir/test', trans, access_mode='w', factory=None,
1516
create=True, create_parent_dir=True, delay_create=True,
1517
file_mode=0666, dir_mode=0777)
1518
knit.add_lines('revid', [], ['a\n'])
1519
self.assertTransportMode(trans, 'dir', 0777)
1520
self.assertTransportMode(trans, 'dir/test.knit', 0666)
1521
self.assertTransportMode(trans, 'dir/test.kndx', 0666)
1523
def test_plan_merge(self):
1524
my_knit = self.make_test_knit(annotate=True)
1525
my_knit.add_lines('text1', [], split_lines(TEXT_1))
1526
my_knit.add_lines('text1a', ['text1'], split_lines(TEXT_1A))
1527
my_knit.add_lines('text1b', ['text1'], split_lines(TEXT_1B))
1528
plan = list(my_knit.plan_merge('text1a', 'text1b'))
1529
for plan_line, expected_line in zip(plan, AB_MERGE):
1530
self.assertEqual(plan_line, expected_line)
1532
def test_get_stream_empty(self):
1533
"""Get a data stream for an empty knit file."""
1534
k1 = self.make_test_knit()
1535
format, data_list, reader_callable = k1.get_data_stream([])
1536
self.assertEqual('knit-plain', format)
1537
self.assertEqual([], data_list)
1538
content = reader_callable(None)
1539
self.assertEqual('', content)
1540
self.assertIsInstance(content, str)
1542
def test_get_stream_one_version(self):
1543
"""Get a data stream for a single record out of a knit containing just
1546
k1 = self.make_test_knit()
1548
('text-a', [], TEXT_1),
1550
expected_data_list = [
1551
# version, options, length, parents
1552
('text-a', ['fulltext'], 122, ()),
1554
for version_id, parents, lines in test_data:
1555
k1.add_lines(version_id, parents, split_lines(lines))
1557
format, data_list, reader_callable = k1.get_data_stream(['text-a'])
1558
self.assertEqual('knit-plain', format)
1559
self.assertEqual(expected_data_list, data_list)
1560
# There's only one record in the knit, so the content should be the
1561
# entire knit data file's contents.
1562
self.assertEqual(k1.transport.get_bytes(k1._data._access._filename),
1563
reader_callable(None))
1565
def test_get_stream_get_one_version_of_many(self):
1566
"""Get a data stream for just one version out of a knit containing many
1569
k1 = self.make_test_knit()
1570
# Insert the same data as test_knit_join, as they seem to cover a range
1571
# of cases (no parents, one parent, multiple parents).
1573
('text-a', [], TEXT_1),
1574
('text-b', ['text-a'], TEXT_1),
1575
('text-c', [], TEXT_1),
1576
('text-d', ['text-c'], TEXT_1),
1577
('text-m', ['text-b', 'text-d'], TEXT_1),
1579
expected_data_list = [
1580
# version, options, length, parents
1581
('text-m', ['line-delta'], 84, ('text-b', 'text-d')),
1583
for version_id, parents, lines in test_data:
1584
k1.add_lines(version_id, parents, split_lines(lines))
1586
format, data_list, reader_callable = k1.get_data_stream(['text-m'])
1587
self.assertEqual('knit-plain', format)
1588
self.assertEqual(expected_data_list, data_list)
1589
self.assertRecordContentEqual(k1, 'text-m', reader_callable(None))
1591
def test_get_data_stream_unordered_index(self):
1592
"""Get a data stream when the knit index reports versions out of order.
1594
https://bugs.launchpad.net/bzr/+bug/164637
1596
k1 = self.make_test_knit()
1598
('text-a', [], TEXT_1),
1599
('text-b', ['text-a'], TEXT_1),
1600
('text-c', [], TEXT_1),
1601
('text-d', ['text-c'], TEXT_1),
1602
('text-m', ['text-b', 'text-d'], TEXT_1),
1604
for version_id, parents, lines in test_data:
1605
k1.add_lines(version_id, parents, split_lines(lines))
1606
# monkey-patch versions method to return out of order, as if coming
1607
# from multiple independently indexed packs
1608
original_versions = k1.versions
1609
k1.versions = lambda: reversed(original_versions())
1610
expected_data_list = [
1611
('text-a', ['fulltext'], 122, ()),
1612
('text-b', ['line-delta'], 84, ('text-a',))]
1613
# now check the fulltext is first and the delta second
1614
format, data_list, _ = k1.get_data_stream(['text-a', 'text-b'])
1615
self.assertEqual('knit-plain', format)
1616
self.assertEqual(expected_data_list, data_list)
1617
# and that's true if we ask for them in the opposite order too
1618
format, data_list, _ = k1.get_data_stream(['text-b', 'text-a'])
1619
self.assertEqual(expected_data_list, data_list)
1620
# also try requesting more versions
1621
format, data_list, _ = k1.get_data_stream([
1622
'text-m', 'text-b', 'text-a'])
1624
('text-a', ['fulltext'], 122, ()),
1625
('text-b', ['line-delta'], 84, ('text-a',)),
1626
('text-m', ['line-delta'], 84, ('text-b', 'text-d')),
1629
def test_get_stream_ghost_parent(self):
1630
"""Get a data stream for a version with a ghost parent."""
1631
k1 = self.make_test_knit()
1633
k1.add_lines('text-a', [], split_lines(TEXT_1))
1634
k1.add_lines_with_ghosts('text-b', ['text-a', 'text-ghost'],
1635
split_lines(TEXT_1))
1637
expected_data_list = [
1638
# version, options, length, parents
1639
('text-b', ['line-delta'], 84, ('text-a', 'text-ghost')),
1642
format, data_list, reader_callable = k1.get_data_stream(['text-b'])
1643
self.assertEqual('knit-plain', format)
1644
self.assertEqual(expected_data_list, data_list)
1645
self.assertRecordContentEqual(k1, 'text-b', reader_callable(None))
1647
def test_get_stream_get_multiple_records(self):
1648
"""Get a stream for multiple records of a knit."""
1649
k1 = self.make_test_knit()
1650
# Insert the same data as test_knit_join, as they seem to cover a range
1651
# of cases (no parents, one parent, multiple parents).
1653
('text-a', [], TEXT_1),
1654
('text-b', ['text-a'], TEXT_1),
1655
('text-c', [], TEXT_1),
1656
('text-d', ['text-c'], TEXT_1),
1657
('text-m', ['text-b', 'text-d'], TEXT_1),
1659
for version_id, parents, lines in test_data:
1660
k1.add_lines(version_id, parents, split_lines(lines))
1662
# This test is actually a bit strict as the order in which they're
1663
# returned is not defined. This matches the current (deterministic)
1665
expected_data_list = [
1666
# version, options, length, parents
1667
('text-d', ['line-delta'], 84, ('text-c',)),
1668
('text-b', ['line-delta'], 84, ('text-a',)),
1670
# Note that even though we request the revision IDs in a particular
1671
# order, the data stream may return them in any order it likes. In this
1672
# case, they'll be in the order they were inserted into the knit.
1673
format, data_list, reader_callable = k1.get_data_stream(
1674
['text-d', 'text-b'])
1675
self.assertEqual('knit-plain', format)
1676
self.assertEqual(expected_data_list, data_list)
1677
# must match order they're returned
1678
self.assertRecordContentEqual(k1, 'text-d', reader_callable(84))
1679
self.assertRecordContentEqual(k1, 'text-b', reader_callable(84))
1680
self.assertEqual('', reader_callable(None),
1681
"There should be no more bytes left to read.")
1683
def test_get_stream_all(self):
1684
"""Get a data stream for all the records in a knit.
1686
This exercises fulltext records, line-delta records, records with
1687
various numbers of parents, and reading multiple records out of the
1688
callable. These cases ought to all be exercised individually by the
1689
other test_get_stream_* tests; this test is basically just paranoia.
1691
k1 = self.make_test_knit()
1692
# Insert the same data as test_knit_join, as they seem to cover a range
1693
# of cases (no parents, one parent, multiple parents).
1695
('text-a', [], TEXT_1),
1696
('text-b', ['text-a'], TEXT_1),
1697
('text-c', [], TEXT_1),
1698
('text-d', ['text-c'], TEXT_1),
1699
('text-m', ['text-b', 'text-d'], TEXT_1),
1701
for version_id, parents, lines in test_data:
1702
k1.add_lines(version_id, parents, split_lines(lines))
1704
# This test is actually a bit strict as the order in which they're
1705
# returned is not defined. This matches the current (deterministic)
1707
expected_data_list = [
1708
# version, options, length, parents
1709
('text-a', ['fulltext'], 122, ()),
1710
('text-b', ['line-delta'], 84, ('text-a',)),
1711
('text-m', ['line-delta'], 84, ('text-b', 'text-d')),
1712
('text-c', ['fulltext'], 121, ()),
1713
('text-d', ['line-delta'], 84, ('text-c',)),
1715
format, data_list, reader_callable = k1.get_data_stream(
1716
['text-a', 'text-b', 'text-c', 'text-d', 'text-m'])
1717
self.assertEqual('knit-plain', format)
1718
self.assertEqual(expected_data_list, data_list)
1719
for version_id, options, length, parents in expected_data_list:
1720
bytes = reader_callable(length)
1721
self.assertRecordContentEqual(k1, version_id, bytes)
1723
def assertKnitFilesEqual(self, knit1, knit2):
1724
"""Assert that the contents of the index and data files of two knits are
1728
knit1.transport.get_bytes(knit1._data._access._filename),
1729
knit2.transport.get_bytes(knit2._data._access._filename))
1731
knit1.transport.get_bytes(knit1._index._filename),
1732
knit2.transport.get_bytes(knit2._index._filename))
1734
def assertKnitValuesEqual(self, left, right):
1735
"""Assert that the texts, annotations and graph of left and right are
1738
self.assertEqual(set(left.versions()), set(right.versions()))
1739
for version in left.versions():
1740
self.assertEqual(left.get_parents_with_ghosts(version),
1741
right.get_parents_with_ghosts(version))
1742
self.assertEqual(left.get_lines(version),
1743
right.get_lines(version))
1744
self.assertEqual(left.annotate(version),
1745
right.annotate(version))
1747
def test_insert_data_stream_empty(self):
1748
"""Inserting a data stream with no records should not put any data into
1751
k1 = self.make_test_knit()
1752
k1.insert_data_stream(
1753
(k1.get_format_signature(), [], lambda ignored: ''))
1754
self.assertEqual('', k1.transport.get_bytes(k1._data._access._filename),
1755
"The .knit should be completely empty.")
1756
self.assertEqual(k1._index.HEADER,
1757
k1.transport.get_bytes(k1._index._filename),
1758
"The .kndx should have nothing apart from the header.")
1760
def test_insert_data_stream_one_record(self):
1761
"""Inserting a data stream with one record from a knit with one record
1762
results in byte-identical files.
1764
source = self.make_test_knit(name='source')
1765
source.add_lines('text-a', [], split_lines(TEXT_1))
1766
data_stream = source.get_data_stream(['text-a'])
1767
target = self.make_test_knit(name='target')
1768
target.insert_data_stream(data_stream)
1769
self.assertKnitFilesEqual(source, target)
1771
def test_insert_data_stream_annotated_unannotated(self):
1772
"""Inserting an annotated datastream to an unannotated knit works."""
1773
# case one - full texts.
1774
source = self.make_test_knit(name='source', annotate=True)
1775
target = self.make_test_knit(name='target', annotate=False)
1776
source.add_lines('text-a', [], split_lines(TEXT_1))
1777
target.insert_data_stream(source.get_data_stream(['text-a']))
1778
self.assertKnitValuesEqual(source, target)
1779
# case two - deltas.
1780
source.add_lines('text-b', ['text-a'], split_lines(TEXT_2))
1781
target.insert_data_stream(source.get_data_stream(['text-b']))
1782
self.assertKnitValuesEqual(source, target)
1784
def test_insert_data_stream_unannotated_annotated(self):
1785
"""Inserting an unannotated datastream to an annotated knit works."""
1786
# case one - full texts.
1787
source = self.make_test_knit(name='source', annotate=False)
1788
target = self.make_test_knit(name='target', annotate=True)
1789
source.add_lines('text-a', [], split_lines(TEXT_1))
1790
target.insert_data_stream(source.get_data_stream(['text-a']))
1791
self.assertKnitValuesEqual(source, target)
1792
# case two - deltas.
1793
source.add_lines('text-b', ['text-a'], split_lines(TEXT_2))
1794
target.insert_data_stream(source.get_data_stream(['text-b']))
1795
self.assertKnitValuesEqual(source, target)
1797
def test_insert_data_stream_records_already_present(self):
1798
"""Insert a data stream where some records are alreday present in the
1799
target, and some not. Only the new records are inserted.
1801
source = self.make_test_knit(name='source')
1802
target = self.make_test_knit(name='target')
1803
# Insert 'text-a' into both source and target
1804
source.add_lines('text-a', [], split_lines(TEXT_1))
1805
target.insert_data_stream(source.get_data_stream(['text-a']))
1806
# Insert 'text-b' into just the source.
1807
source.add_lines('text-b', ['text-a'], split_lines(TEXT_1))
1808
# Get a data stream of both text-a and text-b, and insert it.
1809
data_stream = source.get_data_stream(['text-a', 'text-b'])
1810
target.insert_data_stream(data_stream)
1811
# The source and target will now be identical. This means the text-a
1812
# record was not added a second time.
1813
self.assertKnitFilesEqual(source, target)
1815
def test_insert_data_stream_multiple_records(self):
1816
"""Inserting a data stream of all records from a knit with multiple
1817
records results in byte-identical files.
1819
source = self.make_test_knit(name='source')
1820
source.add_lines('text-a', [], split_lines(TEXT_1))
1821
source.add_lines('text-b', ['text-a'], split_lines(TEXT_1))
1822
source.add_lines('text-c', [], split_lines(TEXT_1))
1823
data_stream = source.get_data_stream(['text-a', 'text-b', 'text-c'])
1825
target = self.make_test_knit(name='target')
1826
target.insert_data_stream(data_stream)
1828
self.assertKnitFilesEqual(source, target)
1830
def test_insert_data_stream_ghost_parent(self):
1831
"""Insert a data stream with a record that has a ghost parent."""
1832
# Make a knit with a record, text-a, that has a ghost parent.
1833
source = self.make_test_knit(name='source')
1834
source.add_lines_with_ghosts('text-a', ['text-ghost'],
1835
split_lines(TEXT_1))
1836
data_stream = source.get_data_stream(['text-a'])
1838
target = self.make_test_knit(name='target')
1839
target.insert_data_stream(data_stream)
1841
self.assertKnitFilesEqual(source, target)
1843
# The target knit object is in a consistent state, i.e. the record we
1844
# just added is immediately visible.
1845
self.assertTrue(target.has_version('text-a'))
1846
self.assertFalse(target.has_version('text-ghost'))
1847
self.assertEqual({'text-a':('text-ghost',)},
1848
target.get_parent_map(['text-a', 'text-ghost']))
1849
self.assertEqual(split_lines(TEXT_1), target.get_lines('text-a'))
1851
def test_insert_data_stream_inconsistent_version_lines(self):
1852
"""Inserting a data stream which has different content for a version_id
1853
than already exists in the knit will raise KnitCorrupt.
1855
source = self.make_test_knit(name='source')
1856
target = self.make_test_knit(name='target')
1857
# Insert a different 'text-a' into both source and target
1858
source.add_lines('text-a', [], split_lines(TEXT_1))
1859
target.add_lines('text-a', [], split_lines(TEXT_2))
1860
# Insert a data stream with conflicting content into the target
1861
data_stream = source.get_data_stream(['text-a'])
1863
errors.KnitCorrupt, target.insert_data_stream, data_stream)
1865
def test_insert_data_stream_inconsistent_version_parents(self):
1866
"""Inserting a data stream which has different parents for a version_id
1867
than already exists in the knit will raise KnitCorrupt.
1869
source = self.make_test_knit(name='source')
1870
target = self.make_test_knit(name='target')
1871
# Insert a different 'text-a' into both source and target. They differ
1872
# only by the parents list, the content is the same.
1873
source.add_lines_with_ghosts('text-a', [], split_lines(TEXT_1))
1874
target.add_lines_with_ghosts('text-a', ['a-ghost'], split_lines(TEXT_1))
1875
# Insert a data stream with conflicting content into the target
1876
data_stream = source.get_data_stream(['text-a'])
1878
errors.KnitCorrupt, target.insert_data_stream, data_stream)
1880
def test_insert_data_stream_unknown_format(self):
1881
"""A data stream in a different format to the target knit cannot be
1884
It will raise KnitDataStreamUnknown because the fallback code will fail
1885
to make a knit. In future we may need KnitDataStreamIncompatible again,
1886
for more exotic cases.
1888
data_stream = ('fake-format-signature', [], lambda _: '')
1889
target = self.make_test_knit(name='target')
1891
errors.KnitDataStreamUnknown,
1892
target.insert_data_stream, data_stream)
1894
# * test that a stream of "already present version, then new version"
1895
# inserts correctly.
1898
def assertMadeStreamKnit(self, source_knit, versions, target_knit):
1899
"""Assert that a knit made from a stream is as expected."""
1900
a_stream = source_knit.get_data_stream(versions)
1901
expected_data = a_stream[2](None)
1902
a_stream = source_knit.get_data_stream(versions)
1903
a_knit = target_knit._knit_from_datastream(a_stream)
1904
self.assertEqual(source_knit.factory.__class__,
1905
a_knit.factory.__class__)
1906
self.assertIsInstance(a_knit._data._access, _StreamAccess)
1907
self.assertIsInstance(a_knit._index, _StreamIndex)
1908
self.assertEqual(a_knit._index.data_list, a_stream[1])
1909
self.assertEqual(a_knit._data._access.data, expected_data)
1910
self.assertEqual(a_knit.filename, target_knit.filename)
1911
self.assertEqual(a_knit.transport, target_knit.transport)
1912
self.assertEqual(a_knit._index, a_knit._data._access.stream_index)
1913
self.assertEqual(target_knit, a_knit._data._access.backing_knit)
1914
self.assertIsInstance(a_knit._data._access.orig_factory,
1915
source_knit.factory.__class__)
1917
def test__knit_from_data_stream_empty(self):
1918
"""Create a knit object from a datastream."""
1919
annotated = self.make_test_knit(name='source', annotate=True)
1920
plain = self.make_test_knit(name='target', annotate=False)
1921
# case 1: annotated source
1922
self.assertMadeStreamKnit(annotated, [], annotated)
1923
self.assertMadeStreamKnit(annotated, [], plain)
1924
# case 2: plain source
1925
self.assertMadeStreamKnit(plain, [], annotated)
1926
self.assertMadeStreamKnit(plain, [], plain)
1928
def test__knit_from_data_stream_unknown_format(self):
1929
annotated = self.make_test_knit(name='source', annotate=True)
1930
self.assertRaises(errors.KnitDataStreamUnknown,
1931
annotated._knit_from_datastream, ("unknown", None, None))
1943
Banana cup cake recipe
1949
- self-raising flour
1953
Banana cup cake recipe
1955
- bananas (do not use plantains!!!)
1962
Banana cup cake recipe
1965
- self-raising flour
1978
AB_MERGE_TEXT="""unchanged|Banana cup cake recipe
1983
new-b|- bananas (do not use plantains!!!)
1984
unchanged|- broken tea cups
1985
new-a|- self-raising flour
1988
AB_MERGE=[tuple(l.split('|')) for l in AB_MERGE_TEXT.splitlines(True)]
1991
def line_delta(from_lines, to_lines):
1992
"""Generate line-based delta from one text to another"""
1993
s = difflib.SequenceMatcher(None, from_lines, to_lines)
1994
for op in s.get_opcodes():
1995
if op[0] == 'equal':
1997
yield '%d,%d,%d\n' % (op[1], op[2], op[4]-op[3])
1998
for i in range(op[3], op[4]):
2002
def apply_line_delta(basis_lines, delta_lines):
2003
"""Apply a line-based perfect diff
2005
basis_lines -- text to apply the patch to
2006
delta_lines -- diff instructions and content
2008
out = basis_lines[:]
2011
while i < len(delta_lines):
2013
a, b, c = map(long, l.split(','))
2015
out[offset+a:offset+b] = delta_lines[i:i+c]
2017
offset = offset + (b - a) + c
2021
class TestWeaveToKnit(KnitTests):
2023
def test_weave_to_knit_matches(self):
2024
# check that the WeaveToKnit is_compatible function
2025
# registers True for a Weave to a Knit.
2026
w = Weave(get_scope=lambda:None)
2027
k = self.make_test_knit()
2028
self.failUnless(WeaveToKnit.is_compatible(w, k))
2029
self.failIf(WeaveToKnit.is_compatible(k, w))
2030
self.failIf(WeaveToKnit.is_compatible(w, w))
2031
self.failIf(WeaveToKnit.is_compatible(k, k))
2034
class TestKnitCaching(KnitTests):
2036
def create_knit(self):
2037
k = self.make_test_knit(True)
2038
k.add_lines('text-1', [], split_lines(TEXT_1))
2039
k.add_lines('text-2', [], split_lines(TEXT_2))
2042
def test_no_caching(self):
2043
k = self.create_knit()
2044
# Nothing should be cached without setting 'enable_cache'
2045
self.assertEqual({}, k._data._cache)
2047
def test_cache_data_read_raw(self):
2048
k = self.create_knit()
2050
# Now cache and read
2053
def read_one_raw(version):
2054
pos_map = k._get_components_positions([version])
2055
method, index_memo, next = pos_map[version]
2056
lst = list(k._data.read_records_iter_raw([(version, index_memo)]))
2057
self.assertEqual(1, len(lst))
2060
val = read_one_raw('text-1')
2061
self.assertEqual({'text-1':val[1]}, k._data._cache)
2064
# After clear, new reads are not cached
2065
self.assertEqual({}, k._data._cache)
2067
val2 = read_one_raw('text-1')
2068
self.assertEqual(val, val2)
2069
self.assertEqual({}, k._data._cache)
2071
def test_cache_data_read(self):
2072
k = self.create_knit()
2074
def read_one(version):
2075
pos_map = k._get_components_positions([version])
2076
method, index_memo, next = pos_map[version]
2077
lst = list(k._data.read_records_iter([(version, index_memo)]))
2078
self.assertEqual(1, len(lst))
2081
# Now cache and read
2084
val = read_one('text-2')
2085
self.assertEqual(['text-2'], k._data._cache.keys())
2086
self.assertEqual('text-2', val[0])
2087
content, digest = k._data._parse_record('text-2',
2088
k._data._cache['text-2'])
2089
self.assertEqual(content, val[1])
2090
self.assertEqual(digest, val[2])
2093
self.assertEqual({}, k._data._cache)
2095
val2 = read_one('text-2')
2096
self.assertEqual(val, val2)
2097
self.assertEqual({}, k._data._cache)
2099
def test_cache_read(self):
2100
k = self.create_knit()
2103
text = k.get_text('text-1')
2104
self.assertEqual(TEXT_1, text)
2105
self.assertEqual(['text-1'], k._data._cache.keys())
2108
self.assertEqual({}, k._data._cache)
2110
text = k.get_text('text-1')
2111
self.assertEqual(TEXT_1, text)
2112
self.assertEqual({}, k._data._cache)
2115
class TestKnitIndex(KnitTests):
2117
def test_add_versions_dictionary_compresses(self):
2118
"""Adding versions to the index should update the lookup dict"""
2119
knit = self.make_test_knit()
2121
idx.add_version('a-1', ['fulltext'], (None, 0, 0), [])
2122
self.check_file_contents('test.kndx',
2123
'# bzr knit index 8\n'
2125
'a-1 fulltext 0 0 :'
2127
idx.add_versions([('a-2', ['fulltext'], (None, 0, 0), ['a-1']),
2128
('a-3', ['fulltext'], (None, 0, 0), ['a-2']),
2130
self.check_file_contents('test.kndx',
2131
'# bzr knit index 8\n'
2133
'a-1 fulltext 0 0 :\n'
2134
'a-2 fulltext 0 0 0 :\n'
2135
'a-3 fulltext 0 0 1 :'
2137
self.assertEqual(['a-1', 'a-2', 'a-3'], idx._history)
2138
self.assertEqual({'a-1':('a-1', ['fulltext'], 0, 0, (), 0),
2139
'a-2':('a-2', ['fulltext'], 0, 0, ('a-1',), 1),
2140
'a-3':('a-3', ['fulltext'], 0, 0, ('a-2',), 2),
2143
def test_add_versions_fails_clean(self):
2144
"""If add_versions fails in the middle, it restores a pristine state.
2146
Any modifications that are made to the index are reset if all versions
2149
# This cheats a little bit by passing in a generator which will
2150
# raise an exception before the processing finishes
2151
# Other possibilities would be to have an version with the wrong number
2152
# of entries, or to make the backing transport unable to write any
2155
knit = self.make_test_knit()
2157
idx.add_version('a-1', ['fulltext'], (None, 0, 0), [])
2159
class StopEarly(Exception):
2162
def generate_failure():
2163
"""Add some entries and then raise an exception"""
2164
yield ('a-2', ['fulltext'], (None, 0, 0), ('a-1',))
2165
yield ('a-3', ['fulltext'], (None, 0, 0), ('a-2',))
2168
# Assert the pre-condition
2169
self.assertEqual(['a-1'], idx._history)
2170
self.assertEqual({'a-1':('a-1', ['fulltext'], 0, 0, (), 0)}, idx._cache)
2172
self.assertRaises(StopEarly, idx.add_versions, generate_failure())
2174
# And it shouldn't be modified
2175
self.assertEqual(['a-1'], idx._history)
2176
self.assertEqual({'a-1':('a-1', ['fulltext'], 0, 0, (), 0)}, idx._cache)
2178
def test_knit_index_ignores_empty_files(self):
2179
# There was a race condition in older bzr, where a ^C at the right time
2180
# could leave an empty .kndx file, which bzr would later claim was a
2181
# corrupted file since the header was not present. In reality, the file
2182
# just wasn't created, so it should be ignored.
2183
t = get_transport('.')
2184
t.put_bytes('test.kndx', '')
2186
knit = self.make_test_knit()
2188
def test_knit_index_checks_header(self):
2189
t = get_transport('.')
2190
t.put_bytes('test.kndx', '# not really a knit header\n\n')
2192
self.assertRaises(KnitHeaderError, self.make_test_knit)
2195
class TestGraphIndexKnit(KnitTests):
2196
"""Tests for knits using a GraphIndex rather than a KnitIndex."""
2198
def make_g_index(self, name, ref_lists=0, nodes=[]):
2199
builder = GraphIndexBuilder(ref_lists)
2200
for node, references, value in nodes:
2201
builder.add_node(node, references, value)
2202
stream = builder.finish()
2203
trans = self.get_transport()
2204
size = trans.put_file(name, stream)
2205
return GraphIndex(trans, name, size)
2207
def two_graph_index(self, deltas=False, catch_adds=False):
2208
"""Build a two-graph index.
2210
:param deltas: If true, use underlying indices with two node-ref
2211
lists and 'parent' set to a delta-compressed against tail.
2213
# build a complex graph across several indices.
2215
# delta compression inn the index
2216
index1 = self.make_g_index('1', 2, [
2217
(('tip', ), 'N0 100', ([('parent', )], [], )),
2218
(('tail', ), '', ([], []))])
2219
index2 = self.make_g_index('2', 2, [
2220
(('parent', ), ' 100 78', ([('tail', ), ('ghost', )], [('tail', )])),
2221
(('separate', ), '', ([], []))])
2223
# just blob location and graph in the index.
2224
index1 = self.make_g_index('1', 1, [
2225
(('tip', ), 'N0 100', ([('parent', )], )),
2226
(('tail', ), '', ([], ))])
2227
index2 = self.make_g_index('2', 1, [
2228
(('parent', ), ' 100 78', ([('tail', ), ('ghost', )], )),
2229
(('separate', ), '', ([], ))])
2230
combined_index = CombinedGraphIndex([index1, index2])
2232
self.combined_index = combined_index
2233
self.caught_entries = []
2234
add_callback = self.catch_add
2237
return KnitGraphIndex(combined_index, deltas=deltas,
2238
add_callback=add_callback)
2240
def test_get_ancestry(self):
2241
# get_ancestry is defined as eliding ghosts, not erroring.
2242
index = self.two_graph_index()
2243
self.assertEqual([], index.get_ancestry([]))
2244
self.assertEqual(['separate'], index.get_ancestry(['separate']))
2245
self.assertEqual(['tail'], index.get_ancestry(['tail']))
2246
self.assertEqual(['tail', 'parent'], index.get_ancestry(['parent']))
2247
self.assertEqual(['tail', 'parent', 'tip'], index.get_ancestry(['tip']))
2248
self.assertTrue(index.get_ancestry(['tip', 'separate']) in
2249
(['tail', 'parent', 'tip', 'separate'],
2250
['separate', 'tail', 'parent', 'tip'],
2252
# and without topo_sort
2253
self.assertEqual(set(['separate']),
2254
set(index.get_ancestry(['separate'], topo_sorted=False)))
2255
self.assertEqual(set(['tail']),
2256
set(index.get_ancestry(['tail'], topo_sorted=False)))
2257
self.assertEqual(set(['tail', 'parent']),
2258
set(index.get_ancestry(['parent'], topo_sorted=False)))
2259
self.assertEqual(set(['tail', 'parent', 'tip']),
2260
set(index.get_ancestry(['tip'], topo_sorted=False)))
2261
self.assertEqual(set(['separate', 'tail', 'parent', 'tip']),
2262
set(index.get_ancestry(['tip', 'separate'])))
2263
# asking for a ghost makes it go boom.
2264
self.assertRaises(errors.RevisionNotPresent, index.get_ancestry, ['ghost'])
2266
def test_get_ancestry_with_ghosts(self):
2267
index = self.two_graph_index()
2268
self.assertEqual([], index.get_ancestry_with_ghosts([]))
2269
self.assertEqual(['separate'], index.get_ancestry_with_ghosts(['separate']))
2270
self.assertEqual(['tail'], index.get_ancestry_with_ghosts(['tail']))
2271
self.assertTrue(index.get_ancestry_with_ghosts(['parent']) in
2272
(['tail', 'ghost', 'parent'],
2273
['ghost', 'tail', 'parent'],
2275
self.assertTrue(index.get_ancestry_with_ghosts(['tip']) in
2276
(['tail', 'ghost', 'parent', 'tip'],
2277
['ghost', 'tail', 'parent', 'tip'],
2279
self.assertTrue(index.get_ancestry_with_ghosts(['tip', 'separate']) in
2280
(['tail', 'ghost', 'parent', 'tip', 'separate'],
2281
['ghost', 'tail', 'parent', 'tip', 'separate'],
2282
['separate', 'tail', 'ghost', 'parent', 'tip'],
2283
['separate', 'ghost', 'tail', 'parent', 'tip'],
2285
# asking for a ghost makes it go boom.
2286
self.assertRaises(errors.RevisionNotPresent, index.get_ancestry_with_ghosts, ['ghost'])
2288
def test_num_versions(self):
2289
index = self.two_graph_index()
2290
self.assertEqual(4, index.num_versions())
2292
def test_get_versions(self):
2293
index = self.two_graph_index()
2294
self.assertEqual(set(['tail', 'tip', 'parent', 'separate']),
2295
set(index.get_versions()))
2297
def test_has_version(self):
2298
index = self.two_graph_index()
2299
self.assertTrue(index.has_version('tail'))
2300
self.assertFalse(index.has_version('ghost'))
2302
def test_get_position(self):
2303
index = self.two_graph_index()
2304
self.assertEqual((index._graph_index._indices[0], 0, 100), index.get_position('tip'))
2305
self.assertEqual((index._graph_index._indices[1], 100, 78), index.get_position('parent'))
2307
def test_get_method_deltas(self):
2308
index = self.two_graph_index(deltas=True)
2309
self.assertEqual('fulltext', index.get_method('tip'))
2310
self.assertEqual('line-delta', index.get_method('parent'))
2312
def test_get_method_no_deltas(self):
2313
# check that the parent-history lookup is ignored with deltas=False.
2314
index = self.two_graph_index(deltas=False)
2315
self.assertEqual('fulltext', index.get_method('tip'))
2316
self.assertEqual('fulltext', index.get_method('parent'))
2318
def test_get_options_deltas(self):
2319
index = self.two_graph_index(deltas=True)
2320
self.assertEqual(['fulltext', 'no-eol'], index.get_options('tip'))
2321
self.assertEqual(['line-delta'], index.get_options('parent'))
2323
def test_get_options_no_deltas(self):
2324
# check that the parent-history lookup is ignored with deltas=False.
2325
index = self.two_graph_index(deltas=False)
2326
self.assertEqual(['fulltext', 'no-eol'], index.get_options('tip'))
2327
self.assertEqual(['fulltext'], index.get_options('parent'))
2329
def test_get_parents_with_ghosts(self):
2330
index = self.two_graph_index()
2331
self.assertEqual(('tail', 'ghost'), index.get_parents_with_ghosts('parent'))
2332
# and errors on ghosts.
2333
self.assertRaises(errors.RevisionNotPresent,
2334
index.get_parents_with_ghosts, 'ghost')
2336
def test_check_versions_present(self):
2337
# ghosts should not be considered present
2338
index = self.two_graph_index()
2339
self.assertRaises(RevisionNotPresent, index.check_versions_present,
2341
self.assertRaises(RevisionNotPresent, index.check_versions_present,
2343
index.check_versions_present(['tail', 'separate'])
2345
def catch_add(self, entries):
2346
self.caught_entries.append(entries)
2348
def test_add_no_callback_errors(self):
2349
index = self.two_graph_index()
2350
self.assertRaises(errors.ReadOnlyError, index.add_version,
2351
'new', 'fulltext,no-eol', (None, 50, 60), ['separate'])
2353
def test_add_version_smoke(self):
2354
index = self.two_graph_index(catch_adds=True)
2355
index.add_version('new', 'fulltext,no-eol', (None, 50, 60), ['separate'])
2356
self.assertEqual([[(('new', ), 'N50 60', ((('separate',),),))]],
2357
self.caught_entries)
2359
def test_add_version_delta_not_delta_index(self):
2360
index = self.two_graph_index(catch_adds=True)
2361
self.assertRaises(errors.KnitCorrupt, index.add_version,
2362
'new', 'no-eol,line-delta', (None, 0, 100), ['parent'])
2363
self.assertEqual([], self.caught_entries)
2365
def test_add_version_same_dup(self):
2366
index = self.two_graph_index(catch_adds=True)
2367
# options can be spelt two different ways
2368
index.add_version('tip', 'fulltext,no-eol', (None, 0, 100), ['parent'])
2369
index.add_version('tip', 'no-eol,fulltext', (None, 0, 100), ['parent'])
2370
# but neither should have added data.
2371
self.assertEqual([[], []], self.caught_entries)
2373
def test_add_version_different_dup(self):
2374
index = self.two_graph_index(deltas=True, catch_adds=True)
2376
self.assertRaises(errors.KnitCorrupt, index.add_version,
2377
'tip', 'no-eol,line-delta', (None, 0, 100), ['parent'])
2378
self.assertRaises(errors.KnitCorrupt, index.add_version,
2379
'tip', 'line-delta,no-eol', (None, 0, 100), ['parent'])
2380
self.assertRaises(errors.KnitCorrupt, index.add_version,
2381
'tip', 'fulltext', (None, 0, 100), ['parent'])
2383
self.assertRaises(errors.KnitCorrupt, index.add_version,
2384
'tip', 'fulltext,no-eol', (None, 50, 100), ['parent'])
2385
self.assertRaises(errors.KnitCorrupt, index.add_version,
2386
'tip', 'fulltext,no-eol', (None, 0, 1000), ['parent'])
2388
self.assertRaises(errors.KnitCorrupt, index.add_version,
2389
'tip', 'fulltext,no-eol', (None, 0, 100), [])
2390
self.assertEqual([], self.caught_entries)
2392
def test_add_versions_nodeltas(self):
2393
index = self.two_graph_index(catch_adds=True)
2394
index.add_versions([
2395
('new', 'fulltext,no-eol', (None, 50, 60), ['separate']),
2396
('new2', 'fulltext', (None, 0, 6), ['new']),
2398
self.assertEqual([(('new', ), 'N50 60', ((('separate',),),)),
2399
(('new2', ), ' 0 6', ((('new',),),))],
2400
sorted(self.caught_entries[0]))
2401
self.assertEqual(1, len(self.caught_entries))
2403
def test_add_versions_deltas(self):
2404
index = self.two_graph_index(deltas=True, catch_adds=True)
2405
index.add_versions([
2406
('new', 'fulltext,no-eol', (None, 50, 60), ['separate']),
2407
('new2', 'line-delta', (None, 0, 6), ['new']),
2409
self.assertEqual([(('new', ), 'N50 60', ((('separate',),), ())),
2410
(('new2', ), ' 0 6', ((('new',),), (('new',),), ))],
2411
sorted(self.caught_entries[0]))
2412
self.assertEqual(1, len(self.caught_entries))
2414
def test_add_versions_delta_not_delta_index(self):
2415
index = self.two_graph_index(catch_adds=True)
2416
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2417
[('new', 'no-eol,line-delta', (None, 0, 100), ['parent'])])
2418
self.assertEqual([], self.caught_entries)
2420
def test_add_versions_random_id_accepted(self):
2421
index = self.two_graph_index(catch_adds=True)
2422
index.add_versions([], random_id=True)
2424
def test_add_versions_same_dup(self):
2425
index = self.two_graph_index(catch_adds=True)
2426
# options can be spelt two different ways
2427
index.add_versions([('tip', 'fulltext,no-eol', (None, 0, 100), ['parent'])])
2428
index.add_versions([('tip', 'no-eol,fulltext', (None, 0, 100), ['parent'])])
2429
# but neither should have added data.
2430
self.assertEqual([[], []], self.caught_entries)
2432
def test_add_versions_different_dup(self):
2433
index = self.two_graph_index(deltas=True, catch_adds=True)
2435
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2436
[('tip', 'no-eol,line-delta', (None, 0, 100), ['parent'])])
2437
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2438
[('tip', 'line-delta,no-eol', (None, 0, 100), ['parent'])])
2439
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2440
[('tip', 'fulltext', (None, 0, 100), ['parent'])])
2442
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2443
[('tip', 'fulltext,no-eol', (None, 50, 100), ['parent'])])
2444
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2445
[('tip', 'fulltext,no-eol', (None, 0, 1000), ['parent'])])
2447
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2448
[('tip', 'fulltext,no-eol', (None, 0, 100), [])])
2449
# change options in the second record
2450
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2451
[('tip', 'fulltext,no-eol', (None, 0, 100), ['parent']),
2452
('tip', 'no-eol,line-delta', (None, 0, 100), ['parent'])])
2453
self.assertEqual([], self.caught_entries)
2455
class TestNoParentsGraphIndexKnit(KnitTests):
2456
"""Tests for knits using KnitGraphIndex with no parents."""
2458
def make_g_index(self, name, ref_lists=0, nodes=[]):
2459
builder = GraphIndexBuilder(ref_lists)
2460
for node, references in nodes:
2461
builder.add_node(node, references)
2462
stream = builder.finish()
2463
trans = self.get_transport()
2464
size = trans.put_file(name, stream)
2465
return GraphIndex(trans, name, size)
2467
def test_parents_deltas_incompatible(self):
2468
index = CombinedGraphIndex([])
2469
self.assertRaises(errors.KnitError, KnitGraphIndex, index,
2470
deltas=True, parents=False)
2472
def two_graph_index(self, catch_adds=False):
2473
"""Build a two-graph index.
2475
:param deltas: If true, use underlying indices with two node-ref
2476
lists and 'parent' set to a delta-compressed against tail.
2478
# put several versions in the index.
2479
index1 = self.make_g_index('1', 0, [
2480
(('tip', ), 'N0 100'),
2482
index2 = self.make_g_index('2', 0, [
2483
(('parent', ), ' 100 78'),
2484
(('separate', ), '')])
2485
combined_index = CombinedGraphIndex([index1, index2])
2487
self.combined_index = combined_index
2488
self.caught_entries = []
2489
add_callback = self.catch_add
2492
return KnitGraphIndex(combined_index, parents=False,
2493
add_callback=add_callback)
2495
def test_get_ancestry(self):
2496
# with no parents, ancestry is always just the key.
2497
index = self.two_graph_index()
2498
self.assertEqual([], index.get_ancestry([]))
2499
self.assertEqual(['separate'], index.get_ancestry(['separate']))
2500
self.assertEqual(['tail'], index.get_ancestry(['tail']))
2501
self.assertEqual(['parent'], index.get_ancestry(['parent']))
2502
self.assertEqual(['tip'], index.get_ancestry(['tip']))
2503
self.assertTrue(index.get_ancestry(['tip', 'separate']) in
2504
(['tip', 'separate'],
2505
['separate', 'tip'],
2507
# asking for a ghost makes it go boom.
2508
self.assertRaises(errors.RevisionNotPresent, index.get_ancestry, ['ghost'])
2510
def test_get_ancestry_with_ghosts(self):
2511
index = self.two_graph_index()
2512
self.assertEqual([], index.get_ancestry_with_ghosts([]))
2513
self.assertEqual(['separate'], index.get_ancestry_with_ghosts(['separate']))
2514
self.assertEqual(['tail'], index.get_ancestry_with_ghosts(['tail']))
2515
self.assertEqual(['parent'], index.get_ancestry_with_ghosts(['parent']))
2516
self.assertEqual(['tip'], index.get_ancestry_with_ghosts(['tip']))
2517
self.assertTrue(index.get_ancestry_with_ghosts(['tip', 'separate']) in
2518
(['tip', 'separate'],
2519
['separate', 'tip'],
2521
# asking for a ghost makes it go boom.
2522
self.assertRaises(errors.RevisionNotPresent, index.get_ancestry_with_ghosts, ['ghost'])
2524
def test_num_versions(self):
2525
index = self.two_graph_index()
2526
self.assertEqual(4, index.num_versions())
2528
def test_get_versions(self):
2529
index = self.two_graph_index()
2530
self.assertEqual(set(['tail', 'tip', 'parent', 'separate']),
2531
set(index.get_versions()))
2533
def test_has_version(self):
2534
index = self.two_graph_index()
2535
self.assertTrue(index.has_version('tail'))
2536
self.assertFalse(index.has_version('ghost'))
2538
def test_get_position(self):
2539
index = self.two_graph_index()
2540
self.assertEqual((index._graph_index._indices[0], 0, 100), index.get_position('tip'))
2541
self.assertEqual((index._graph_index._indices[1], 100, 78), index.get_position('parent'))
2543
def test_get_method(self):
2544
index = self.two_graph_index()
2545
self.assertEqual('fulltext', index.get_method('tip'))
2546
self.assertEqual(['fulltext'], index.get_options('parent'))
2548
def test_get_options(self):
2549
index = self.two_graph_index()
2550
self.assertEqual(['fulltext', 'no-eol'], index.get_options('tip'))
2551
self.assertEqual(['fulltext'], index.get_options('parent'))
2553
def test_get_parents_with_ghosts(self):
2554
index = self.two_graph_index()
2555
self.assertEqual((), index.get_parents_with_ghosts('parent'))
2556
# and errors on ghosts.
2557
self.assertRaises(errors.RevisionNotPresent,
2558
index.get_parents_with_ghosts, 'ghost')
2560
def test_check_versions_present(self):
2561
index = self.two_graph_index()
2562
self.assertRaises(RevisionNotPresent, index.check_versions_present,
2564
self.assertRaises(RevisionNotPresent, index.check_versions_present,
2565
['tail', 'missing'])
2566
index.check_versions_present(['tail', 'separate'])
2568
def catch_add(self, entries):
2569
self.caught_entries.append(entries)
2571
def test_add_no_callback_errors(self):
2572
index = self.two_graph_index()
2573
self.assertRaises(errors.ReadOnlyError, index.add_version,
2574
'new', 'fulltext,no-eol', (None, 50, 60), ['separate'])
2576
def test_add_version_smoke(self):
2577
index = self.two_graph_index(catch_adds=True)
2578
index.add_version('new', 'fulltext,no-eol', (None, 50, 60), [])
2579
self.assertEqual([[(('new', ), 'N50 60')]],
2580
self.caught_entries)
2582
def test_add_version_delta_not_delta_index(self):
2583
index = self.two_graph_index(catch_adds=True)
2584
self.assertRaises(errors.KnitCorrupt, index.add_version,
2585
'new', 'no-eol,line-delta', (None, 0, 100), [])
2586
self.assertEqual([], self.caught_entries)
2588
def test_add_version_same_dup(self):
2589
index = self.two_graph_index(catch_adds=True)
2590
# options can be spelt two different ways
2591
index.add_version('tip', 'fulltext,no-eol', (None, 0, 100), [])
2592
index.add_version('tip', 'no-eol,fulltext', (None, 0, 100), [])
2593
# but neither should have added data.
2594
self.assertEqual([[], []], self.caught_entries)
2596
def test_add_version_different_dup(self):
2597
index = self.two_graph_index(catch_adds=True)
2599
self.assertRaises(errors.KnitCorrupt, index.add_version,
2600
'tip', 'no-eol,line-delta', (None, 0, 100), [])
2601
self.assertRaises(errors.KnitCorrupt, index.add_version,
2602
'tip', 'line-delta,no-eol', (None, 0, 100), [])
2603
self.assertRaises(errors.KnitCorrupt, index.add_version,
2604
'tip', 'fulltext', (None, 0, 100), [])
2606
self.assertRaises(errors.KnitCorrupt, index.add_version,
2607
'tip', 'fulltext,no-eol', (None, 50, 100), [])
2608
self.assertRaises(errors.KnitCorrupt, index.add_version,
2609
'tip', 'fulltext,no-eol', (None, 0, 1000), [])
2611
self.assertRaises(errors.KnitCorrupt, index.add_version,
2612
'tip', 'fulltext,no-eol', (None, 0, 100), ['parent'])
2613
self.assertEqual([], self.caught_entries)
2615
def test_add_versions(self):
2616
index = self.two_graph_index(catch_adds=True)
2617
index.add_versions([
2618
('new', 'fulltext,no-eol', (None, 50, 60), []),
2619
('new2', 'fulltext', (None, 0, 6), []),
2621
self.assertEqual([(('new', ), 'N50 60'), (('new2', ), ' 0 6')],
2622
sorted(self.caught_entries[0]))
2623
self.assertEqual(1, len(self.caught_entries))
2625
def test_add_versions_delta_not_delta_index(self):
2626
index = self.two_graph_index(catch_adds=True)
2627
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2628
[('new', 'no-eol,line-delta', (None, 0, 100), ['parent'])])
2629
self.assertEqual([], self.caught_entries)
2631
def test_add_versions_parents_not_parents_index(self):
2632
index = self.two_graph_index(catch_adds=True)
2633
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2634
[('new', 'no-eol,fulltext', (None, 0, 100), ['parent'])])
2635
self.assertEqual([], self.caught_entries)
2637
def test_add_versions_random_id_accepted(self):
2638
index = self.two_graph_index(catch_adds=True)
2639
index.add_versions([], random_id=True)
2641
def test_add_versions_same_dup(self):
2642
index = self.two_graph_index(catch_adds=True)
2643
# options can be spelt two different ways
2644
index.add_versions([('tip', 'fulltext,no-eol', (None, 0, 100), [])])
2645
index.add_versions([('tip', 'no-eol,fulltext', (None, 0, 100), [])])
2646
# but neither should have added data.
2647
self.assertEqual([[], []], self.caught_entries)
2649
def test_add_versions_different_dup(self):
2650
index = self.two_graph_index(catch_adds=True)
2652
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2653
[('tip', 'no-eol,line-delta', (None, 0, 100), [])])
2654
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2655
[('tip', 'line-delta,no-eol', (None, 0, 100), [])])
2656
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2657
[('tip', 'fulltext', (None, 0, 100), [])])
2659
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2660
[('tip', 'fulltext,no-eol', (None, 50, 100), [])])
2661
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2662
[('tip', 'fulltext,no-eol', (None, 0, 1000), [])])
2664
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2665
[('tip', 'fulltext,no-eol', (None, 0, 100), ['parent'])])
2666
# change options in the second record
2667
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2668
[('tip', 'fulltext,no-eol', (None, 0, 100), []),
2669
('tip', 'no-eol,line-delta', (None, 0, 100), [])])
2670
self.assertEqual([], self.caught_entries)
2672
class TestPackKnits(KnitTests):
2673
"""Tests that use a _PackAccess and KnitGraphIndex."""
2675
def test_get_data_stream_packs_ignores_pack_overhead(self):
2676
# Packs have an encoding overhead that should not be included in the
2677
# 'size' field of a data stream, because it is not returned by the
2678
# raw_reading functions - it is why index_memo's are opaque, and
2679
# get_data_stream was abusing this.
2680
packname = 'test.pack'
2681
transport = self.get_transport()
2682
def write_data(bytes):
2683
transport.append_bytes(packname, bytes)
2684
writer = pack.ContainerWriter(write_data)
2686
index = InMemoryGraphIndex(2)
2687
knit_index = KnitGraphIndex(index, add_callback=index.add_nodes,
2689
indices = {index:(transport, packname)}
2690
access = _PackAccess(indices, writer=(writer, index))
2691
k = KnitVersionedFile('test', get_transport('.'),
2692
delta=True, create=True, index=knit_index, access_method=access)
2693
# insert something into the knit
2694
k.add_lines('text-1', [], ["foo\n"])
2695
# get a data stream for it
2696
stream = k.get_data_stream(['text-1'])
2697
# if the stream has been incorrectly assembled, we will get a short read
2698
# reading from the stream (as streams have no trailer)
2699
expected_length = stream[1][0][2]
2700
# we use -1 to do the read, so that if a trailer is added this test
2701
# will fail and we'll adjust it to handle that case correctly, rather
2702
# than allowing an over-read that is bogus.
2703
self.assertEqual(expected_length, len(stream[2](-1)))
2706
class Test_StreamIndex(KnitTests):
2708
def get_index(self, knit, stream):
2709
"""Get a _StreamIndex from knit and stream."""
2710
return knit._knit_from_datastream(stream)._index
2712
def assertIndexVersions(self, knit, versions):
2713
"""Check that the _StreamIndex versions are those of the stream."""
2714
index = self.get_index(knit, knit.get_data_stream(versions))
2715
self.assertEqual(set(index.get_versions()), set(versions))
2716
# check we didn't get duplicates
2717
self.assertEqual(len(index.get_versions()), len(versions))
2719
def assertIndexAncestry(self, knit, ancestry_versions, versions, result):
2720
"""Check the result of a get_ancestry call on knit."""
2721
index = self.get_index(knit, knit.get_data_stream(versions))
2724
set(index.get_ancestry(ancestry_versions, False)))
2726
def assertGetMethod(self, knit, versions, version, result):
2727
index = self.get_index(knit, knit.get_data_stream(versions))
2728
self.assertEqual(result, index.get_method(version))
2730
def assertGetOptions(self, knit, version, options):
2731
index = self.get_index(knit, knit.get_data_stream(version))
2732
self.assertEqual(options, index.get_options(version))
2734
def assertGetPosition(self, knit, versions, version, result):
2735
index = self.get_index(knit, knit.get_data_stream(versions))
2736
if result[1] is None:
2737
result = (result[0], index, result[2], result[3])
2738
self.assertEqual(result, index.get_position(version))
2740
def assertGetParentsWithGhosts(self, knit, versions, version, parents):
2741
index = self.get_index(knit, knit.get_data_stream(versions))
2742
self.assertEqual(parents, index.get_parents_with_ghosts(version))
2744
def make_knit_with_4_versions_2_dags(self):
2745
knit = self.make_test_knit()
2746
knit.add_lines('a', [], ["foo"])
2747
knit.add_lines('b', [], [])
2748
knit.add_lines('c', ['b', 'a'], [])
2749
knit.add_lines_with_ghosts('d', ['e', 'f'], [])
2752
def test_versions(self):
2753
"""The versions of a StreamIndex are those of the datastream."""
2754
knit = self.make_knit_with_4_versions_2_dags()
2755
# ask for most permutations, which catches bugs like falling back to the
2756
# target knit, or showing ghosts, etc.
2757
self.assertIndexVersions(knit, [])
2758
self.assertIndexVersions(knit, ['a'])
2759
self.assertIndexVersions(knit, ['b'])
2760
self.assertIndexVersions(knit, ['c'])
2761
self.assertIndexVersions(knit, ['d'])
2762
self.assertIndexVersions(knit, ['a', 'b'])
2763
self.assertIndexVersions(knit, ['b', 'c'])
2764
self.assertIndexVersions(knit, ['a', 'c'])
2765
self.assertIndexVersions(knit, ['a', 'b', 'c'])
2766
self.assertIndexVersions(knit, ['a', 'b', 'c', 'd'])
2768
def test_construct(self):
2769
"""Constructing a StreamIndex generates index data."""
2770
data_list = [('text-a', ['fulltext'], 127, []),
2771
('text-b', ['option'], 128, ['text-c'])]
2772
index = _StreamIndex(data_list, None)
2773
self.assertEqual({'text-a':(['fulltext'], (0, 127), []),
2774
'text-b':(['option'], (127, 127 + 128), ['text-c'])},
2777
def test_get_ancestry(self):
2778
knit = self.make_knit_with_4_versions_2_dags()
2779
self.assertIndexAncestry(knit, ['a'], ['a'], ['a'])
2780
self.assertIndexAncestry(knit, ['b'], ['b'], ['b'])
2781
self.assertIndexAncestry(knit, ['c'], ['c'], ['c'])
2782
self.assertIndexAncestry(knit, ['c'], ['a', 'b', 'c'],
2783
set(['a', 'b', 'c']))
2784
self.assertIndexAncestry(knit, ['c', 'd'], ['a', 'b', 'c', 'd'],
2785
set(['a', 'b', 'c', 'd']))
2787
def test_get_method(self):
2788
knit = self.make_knit_with_4_versions_2_dags()
2789
self.assertGetMethod(knit, ['a'], 'a', 'fulltext')
2790
self.assertGetMethod(knit, ['c'], 'c', 'line-delta')
2791
# get_method on a basis that is not in the datastream (but in the
2792
# backing knit) returns 'fulltext', because thats what we'll create as
2794
self.assertGetMethod(knit, ['c'], 'b', 'fulltext')
2796
def test_get_options(self):
2797
knit = self.make_knit_with_4_versions_2_dags()
2798
self.assertGetOptions(knit, 'a', ['no-eol', 'fulltext'])
2799
self.assertGetOptions(knit, 'c', ['line-delta'])
2801
def test_get_parents_with_ghosts(self):
2802
knit = self.make_knit_with_4_versions_2_dags()
2803
self.assertGetParentsWithGhosts(knit, ['a'], 'a', ())
2804
self.assertGetParentsWithGhosts(knit, ['c'], 'c', ('b', 'a'))
2805
self.assertGetParentsWithGhosts(knit, ['d'], 'd', ('e', 'f'))
2807
def test_get_position(self):
2808
knit = self.make_knit_with_4_versions_2_dags()
2809
# get_position returns (thunk_flag, index(can be None), start, end) for
2810
# _StreamAccess to use.
2811
self.assertGetPosition(knit, ['a'], 'a', (False, None, 0, 78))
2812
self.assertGetPosition(knit, ['a', 'c'], 'c', (False, None, 78, 156))
2813
# get_position on a text that is not in the datastream (but in the
2814
# backing knit) returns (True, 'versionid', None, None) - and then the
2815
# access object can construct the relevant data as needed.
2816
self.assertGetPosition(knit, ['a', 'c'], 'b', (True, 'b', None, None))
2819
class Test_StreamAccess(KnitTests):
2821
def get_index_access(self, knit, stream):
2822
"""Get a _StreamAccess from knit and stream."""
2823
knit = knit._knit_from_datastream(stream)
2824
return knit._index, knit._data._access
2826
def assertGetRawRecords(self, knit, versions):
2827
index, access = self.get_index_access(knit,
2828
knit.get_data_stream(versions))
2829
# check that every version asked for can be obtained from the resulting
2833
for version in versions:
2834
memos.append(knit._index.get_position(version))
2836
for version, data in zip(
2837
versions, knit._data._access.get_raw_records(memos)):
2838
original[version] = data
2840
for version in versions:
2841
memos.append(index.get_position(version))
2843
for version, data in zip(versions, access.get_raw_records(memos)):
2844
streamed[version] = data
2845
self.assertEqual(original, streamed)
2847
for version in versions:
2848
data = list(access.get_raw_records(
2849
[index.get_position(version)]))[0]
2850
self.assertEqual(original[version], data)
2852
def make_knit_with_two_versions(self):
2853
knit = self.make_test_knit()
2854
knit.add_lines('a', [], ["foo"])
2855
knit.add_lines('b', [], ["bar"])
2858
def test_get_raw_records(self):
2859
knit = self.make_knit_with_two_versions()
2860
self.assertGetRawRecords(knit, ['a', 'b'])
2861
self.assertGetRawRecords(knit, ['a'])
2862
self.assertGetRawRecords(knit, ['b'])
2864
def test_get_raw_record_from_backing_knit(self):
2865
# the thunk layer should create an artificial A on-demand when needed.
2866
source_knit = self.make_test_knit(name='plain', annotate=False)
2867
target_knit = self.make_test_knit(name='annotated', annotate=True)
2868
source_knit.add_lines("A", [], ["Foo\n"])
2869
# Give the target A, so we can try to thunk across to it.
2870
target_knit.join(source_knit)
2871
index, access = self.get_index_access(target_knit,
2872
source_knit.get_data_stream([]))
2873
raw_data = list(access.get_raw_records([(True, "A", None, None)]))[0]
2874
df = GzipFile(mode='rb', fileobj=StringIO(raw_data))
2876
'version A 1 5d36b88bb697a2d778f024048bafabd443d74503\n'
2880
def test_asking_for_thunk_stream_is_not_plain_errors(self):
2881
knit = self.make_test_knit(name='annotated', annotate=True)
2882
knit.add_lines("A", [], ["Foo\n"])
2883
index, access = self.get_index_access(knit,
2884
knit.get_data_stream([]))
2885
self.assertRaises(errors.KnitCorrupt,
2886
list, access.get_raw_records([(True, "A", None, None)]))
2889
class TestFormatSignatures(KnitTests):
2891
def test_knit_format_signatures(self):
2892
"""Different formats of knit have different signature strings."""
2893
knit = self.make_test_knit(name='a', annotate=True)
2894
self.assertEqual('knit-annotated', knit.get_format_signature())
2895
knit = self.make_test_knit(name='p', annotate=False)
2896
self.assertEqual('knit-plain', knit.get_format_signature())
added
removed
bzrlib/tests/test_knit.py
