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 (
53
from bzrlib.osutils import split_lines
54
from bzrlib.tests import (
57
TestCaseWithMemoryTransport,
58
TestCaseWithTransport,
60
from bzrlib.transport import TransportLogger, get_transport
61
from bzrlib.transport.memory import MemoryTransport
62
from bzrlib.util import bencode
63
from bzrlib.weave import Weave
66
class _CompiledKnitFeature(Feature):
70
import bzrlib._knit_load_data_c
75
def feature_name(self):
76
return 'bzrlib._knit_load_data_c'
78
CompiledKnitFeature = _CompiledKnitFeature()
81
class KnitContentTestsMixin(object):
83
def test_constructor(self):
84
content = self._make_content([])
87
content = self._make_content([])
88
self.assertEqual(content.text(), [])
90
content = self._make_content([("origin1", "text1"), ("origin2", "text2")])
91
self.assertEqual(content.text(), ["text1", "text2"])
94
content = self._make_content([("origin1", "text1"), ("origin2", "text2")])
96
self.assertIsInstance(copy, content.__class__)
97
self.assertEqual(copy.annotate(), content.annotate())
99
def assertDerivedBlocksEqual(self, source, target, noeol=False):
100
"""Assert that the derived matching blocks match real output"""
101
source_lines = source.splitlines(True)
102
target_lines = target.splitlines(True)
104
if noeol and not line.endswith('\n'):
108
source_content = self._make_content([(None, nl(l)) for l in source_lines])
109
target_content = self._make_content([(None, nl(l)) for l in target_lines])
110
line_delta = source_content.line_delta(target_content)
111
delta_blocks = list(KnitContent.get_line_delta_blocks(line_delta,
112
source_lines, target_lines))
113
matcher = KnitSequenceMatcher(None, source_lines, target_lines)
114
matcher_blocks = list(list(matcher.get_matching_blocks()))
115
self.assertEqual(matcher_blocks, delta_blocks)
117
def test_get_line_delta_blocks(self):
118
self.assertDerivedBlocksEqual('a\nb\nc\n', 'q\nc\n')
119
self.assertDerivedBlocksEqual(TEXT_1, TEXT_1)
120
self.assertDerivedBlocksEqual(TEXT_1, TEXT_1A)
121
self.assertDerivedBlocksEqual(TEXT_1, TEXT_1B)
122
self.assertDerivedBlocksEqual(TEXT_1B, TEXT_1A)
123
self.assertDerivedBlocksEqual(TEXT_1A, TEXT_1B)
124
self.assertDerivedBlocksEqual(TEXT_1A, '')
125
self.assertDerivedBlocksEqual('', TEXT_1A)
126
self.assertDerivedBlocksEqual('', '')
127
self.assertDerivedBlocksEqual('a\nb\nc', 'a\nb\nc\nd')
129
def test_get_line_delta_blocks_noeol(self):
130
"""Handle historical knit deltas safely
132
Some existing knit deltas don't consider the last line to differ
133
when the only difference whether it has a final newline.
135
New knit deltas appear to always consider the last line to differ
138
self.assertDerivedBlocksEqual('a\nb\nc', 'a\nb\nc\nd\n', noeol=True)
139
self.assertDerivedBlocksEqual('a\nb\nc\nd\n', 'a\nb\nc', noeol=True)
140
self.assertDerivedBlocksEqual('a\nb\nc\n', 'a\nb\nc', noeol=True)
141
self.assertDerivedBlocksEqual('a\nb\nc', 'a\nb\nc\n', noeol=True)
144
class TestPlainKnitContent(TestCase, KnitContentTestsMixin):
146
def _make_content(self, lines):
147
annotated_content = AnnotatedKnitContent(lines)
148
return PlainKnitContent(annotated_content.text(), 'bogus')
150
def test_annotate(self):
151
content = self._make_content([])
152
self.assertEqual(content.annotate(), [])
154
content = self._make_content([("origin1", "text1"), ("origin2", "text2")])
155
self.assertEqual(content.annotate(),
156
[("bogus", "text1"), ("bogus", "text2")])
158
def test_annotate_iter(self):
159
content = self._make_content([])
160
it = content.annotate_iter()
161
self.assertRaises(StopIteration, it.next)
163
content = self._make_content([("bogus", "text1"), ("bogus", "text2")])
164
it = content.annotate_iter()
165
self.assertEqual(it.next(), ("bogus", "text1"))
166
self.assertEqual(it.next(), ("bogus", "text2"))
167
self.assertRaises(StopIteration, it.next)
169
def test_line_delta(self):
170
content1 = self._make_content([("", "a"), ("", "b")])
171
content2 = self._make_content([("", "a"), ("", "a"), ("", "c")])
172
self.assertEqual(content1.line_delta(content2),
173
[(1, 2, 2, ["a", "c"])])
175
def test_line_delta_iter(self):
176
content1 = self._make_content([("", "a"), ("", "b")])
177
content2 = self._make_content([("", "a"), ("", "a"), ("", "c")])
178
it = content1.line_delta_iter(content2)
179
self.assertEqual(it.next(), (1, 2, 2, ["a", "c"]))
180
self.assertRaises(StopIteration, it.next)
183
class TestAnnotatedKnitContent(TestCase, KnitContentTestsMixin):
185
def _make_content(self, lines):
186
return AnnotatedKnitContent(lines)
188
def test_annotate(self):
189
content = self._make_content([])
190
self.assertEqual(content.annotate(), [])
192
content = self._make_content([("origin1", "text1"), ("origin2", "text2")])
193
self.assertEqual(content.annotate(),
194
[("origin1", "text1"), ("origin2", "text2")])
196
def test_annotate_iter(self):
197
content = self._make_content([])
198
it = content.annotate_iter()
199
self.assertRaises(StopIteration, it.next)
201
content = self._make_content([("origin1", "text1"), ("origin2", "text2")])
202
it = content.annotate_iter()
203
self.assertEqual(it.next(), ("origin1", "text1"))
204
self.assertEqual(it.next(), ("origin2", "text2"))
205
self.assertRaises(StopIteration, it.next)
207
def test_line_delta(self):
208
content1 = self._make_content([("", "a"), ("", "b")])
209
content2 = self._make_content([("", "a"), ("", "a"), ("", "c")])
210
self.assertEqual(content1.line_delta(content2),
211
[(1, 2, 2, [("", "a"), ("", "c")])])
213
def test_line_delta_iter(self):
214
content1 = self._make_content([("", "a"), ("", "b")])
215
content2 = self._make_content([("", "a"), ("", "a"), ("", "c")])
216
it = content1.line_delta_iter(content2)
217
self.assertEqual(it.next(), (1, 2, 2, [("", "a"), ("", "c")]))
218
self.assertRaises(StopIteration, it.next)
221
class MockTransport(object):
223
def __init__(self, file_lines=None):
224
self.file_lines = file_lines
226
# We have no base directory for the MockTransport
229
def get(self, filename):
230
if self.file_lines is None:
231
raise NoSuchFile(filename)
233
return StringIO("\n".join(self.file_lines))
235
def readv(self, relpath, offsets):
236
fp = self.get(relpath)
237
for offset, size in offsets:
239
yield offset, fp.read(size)
241
def __getattr__(self, name):
242
def queue_call(*args, **kwargs):
243
self.calls.append((name, args, kwargs))
247
class KnitRecordAccessTestsMixin(object):
248
"""Tests for getting and putting knit records."""
250
def assertAccessExists(self, access):
251
"""Ensure the data area for access has been initialised/exists."""
252
raise NotImplementedError(self.assertAccessExists)
254
def test_add_raw_records(self):
255
"""Add_raw_records adds records retrievable later."""
256
access = self.get_access()
257
memos = access.add_raw_records([10], '1234567890')
258
self.assertEqual(['1234567890'], list(access.get_raw_records(memos)))
260
def test_add_several_raw_records(self):
261
"""add_raw_records with many records and read some back."""
262
access = self.get_access()
263
memos = access.add_raw_records([10, 2, 5], '12345678901234567')
264
self.assertEqual(['1234567890', '12', '34567'],
265
list(access.get_raw_records(memos)))
266
self.assertEqual(['1234567890'],
267
list(access.get_raw_records(memos[0:1])))
268
self.assertEqual(['12'],
269
list(access.get_raw_records(memos[1:2])))
270
self.assertEqual(['34567'],
271
list(access.get_raw_records(memos[2:3])))
272
self.assertEqual(['1234567890', '34567'],
273
list(access.get_raw_records(memos[0:1] + memos[2:3])))
275
def test_create(self):
276
"""create() should make a file on disk."""
277
access = self.get_access()
279
self.assertAccessExists(access)
281
def test_open_file(self):
282
"""open_file never errors."""
283
access = self.get_access()
287
class TestKnitKnitAccess(TestCaseWithMemoryTransport, KnitRecordAccessTestsMixin):
288
"""Tests for the .kndx implementation."""
290
def assertAccessExists(self, access):
291
self.assertNotEqual(None, access.open_file())
293
def get_access(self):
294
"""Get a .knit style access instance."""
295
access = _KnitAccess(self.get_transport(), "foo.knit", None, None,
300
class TestPackKnitAccess(TestCaseWithMemoryTransport, KnitRecordAccessTestsMixin):
301
"""Tests for the pack based access."""
303
def assertAccessExists(self, access):
304
# as pack based access has no backing unless an index maps data, this
308
def get_access(self):
309
return self._get_access()[0]
311
def _get_access(self, packname='packfile', index='FOO'):
312
transport = self.get_transport()
313
def write_data(bytes):
314
transport.append_bytes(packname, bytes)
315
writer = pack.ContainerWriter(write_data)
317
indices = {index:(transport, packname)}
318
access = _PackAccess(indices, writer=(writer, index))
319
return access, writer
321
def test_read_from_several_packs(self):
322
access, writer = self._get_access()
324
memos.extend(access.add_raw_records([10], '1234567890'))
326
access, writer = self._get_access('pack2', 'FOOBAR')
327
memos.extend(access.add_raw_records([5], '12345'))
329
access, writer = self._get_access('pack3', 'BAZ')
330
memos.extend(access.add_raw_records([5], 'alpha'))
332
transport = self.get_transport()
333
access = _PackAccess({"FOO":(transport, 'packfile'),
334
"FOOBAR":(transport, 'pack2'),
335
"BAZ":(transport, 'pack3')})
336
self.assertEqual(['1234567890', '12345', 'alpha'],
337
list(access.get_raw_records(memos)))
338
self.assertEqual(['1234567890'],
339
list(access.get_raw_records(memos[0:1])))
340
self.assertEqual(['12345'],
341
list(access.get_raw_records(memos[1:2])))
342
self.assertEqual(['alpha'],
343
list(access.get_raw_records(memos[2:3])))
344
self.assertEqual(['1234567890', 'alpha'],
345
list(access.get_raw_records(memos[0:1] + memos[2:3])))
347
def test_set_writer(self):
348
"""The writer should be settable post construction."""
349
access = _PackAccess({})
350
transport = self.get_transport()
351
packname = 'packfile'
353
def write_data(bytes):
354
transport.append_bytes(packname, bytes)
355
writer = pack.ContainerWriter(write_data)
357
access.set_writer(writer, index, (transport, packname))
358
memos = access.add_raw_records([10], '1234567890')
360
self.assertEqual(['1234567890'], list(access.get_raw_records(memos)))
363
class LowLevelKnitDataTests(TestCase):
365
def create_gz_content(self, text):
367
gz_file = gzip.GzipFile(mode='wb', fileobj=sio)
370
return sio.getvalue()
372
def test_valid_knit_data(self):
373
sha1sum = sha.new('foo\nbar\n').hexdigest()
374
gz_txt = self.create_gz_content('version rev-id-1 2 %s\n'
379
transport = MockTransport([gz_txt])
380
access = _KnitAccess(transport, 'filename', None, None, False, False)
381
data = _KnitData(access=access)
382
records = [('rev-id-1', (None, 0, len(gz_txt)))]
384
contents = data.read_records(records)
385
self.assertEqual({'rev-id-1':(['foo\n', 'bar\n'], sha1sum)}, contents)
387
raw_contents = list(data.read_records_iter_raw(records))
388
self.assertEqual([('rev-id-1', gz_txt)], raw_contents)
390
def test_not_enough_lines(self):
391
sha1sum = sha.new('foo\n').hexdigest()
392
# record says 2 lines data says 1
393
gz_txt = self.create_gz_content('version rev-id-1 2 %s\n'
397
transport = MockTransport([gz_txt])
398
access = _KnitAccess(transport, 'filename', None, None, False, False)
399
data = _KnitData(access=access)
400
records = [('rev-id-1', (None, 0, len(gz_txt)))]
401
self.assertRaises(errors.KnitCorrupt, data.read_records, records)
403
# read_records_iter_raw won't detect that sort of mismatch/corruption
404
raw_contents = list(data.read_records_iter_raw(records))
405
self.assertEqual([('rev-id-1', gz_txt)], raw_contents)
407
def test_too_many_lines(self):
408
sha1sum = sha.new('foo\nbar\n').hexdigest()
409
# record says 1 lines data says 2
410
gz_txt = self.create_gz_content('version rev-id-1 1 %s\n'
415
transport = MockTransport([gz_txt])
416
access = _KnitAccess(transport, 'filename', None, None, False, False)
417
data = _KnitData(access=access)
418
records = [('rev-id-1', (None, 0, len(gz_txt)))]
419
self.assertRaises(errors.KnitCorrupt, data.read_records, records)
421
# read_records_iter_raw won't detect that sort of mismatch/corruption
422
raw_contents = list(data.read_records_iter_raw(records))
423
self.assertEqual([('rev-id-1', gz_txt)], raw_contents)
425
def test_mismatched_version_id(self):
426
sha1sum = sha.new('foo\nbar\n').hexdigest()
427
gz_txt = self.create_gz_content('version rev-id-1 2 %s\n'
432
transport = MockTransport([gz_txt])
433
access = _KnitAccess(transport, 'filename', None, None, False, False)
434
data = _KnitData(access=access)
435
# We are asking for rev-id-2, but the data is rev-id-1
436
records = [('rev-id-2', (None, 0, len(gz_txt)))]
437
self.assertRaises(errors.KnitCorrupt, data.read_records, records)
439
# read_records_iter_raw will notice if we request the wrong version.
440
self.assertRaises(errors.KnitCorrupt, list,
441
data.read_records_iter_raw(records))
443
def test_uncompressed_data(self):
444
sha1sum = sha.new('foo\nbar\n').hexdigest()
445
txt = ('version rev-id-1 2 %s\n'
450
transport = MockTransport([txt])
451
access = _KnitAccess(transport, 'filename', None, None, False, False)
452
data = _KnitData(access=access)
453
records = [('rev-id-1', (None, 0, len(txt)))]
455
# We don't have valid gzip data ==> corrupt
456
self.assertRaises(errors.KnitCorrupt, data.read_records, records)
458
# read_records_iter_raw will notice the bad data
459
self.assertRaises(errors.KnitCorrupt, list,
460
data.read_records_iter_raw(records))
462
def test_corrupted_data(self):
463
sha1sum = sha.new('foo\nbar\n').hexdigest()
464
gz_txt = self.create_gz_content('version rev-id-1 2 %s\n'
469
# Change 2 bytes in the middle to \xff
470
gz_txt = gz_txt[:10] + '\xff\xff' + gz_txt[12:]
471
transport = MockTransport([gz_txt])
472
access = _KnitAccess(transport, 'filename', None, None, False, False)
473
data = _KnitData(access=access)
474
records = [('rev-id-1', (None, 0, len(gz_txt)))]
476
self.assertRaises(errors.KnitCorrupt, data.read_records, records)
478
# read_records_iter_raw will notice if we request the wrong version.
479
self.assertRaises(errors.KnitCorrupt, list,
480
data.read_records_iter_raw(records))
483
class LowLevelKnitIndexTests(TestCase):
485
def get_knit_index(self, *args, **kwargs):
486
orig = knit._load_data
488
knit._load_data = orig
489
self.addCleanup(reset)
490
from bzrlib._knit_load_data_py import _load_data_py
491
knit._load_data = _load_data_py
492
return _KnitIndex(*args, **kwargs)
494
def test_no_such_file(self):
495
transport = MockTransport()
497
self.assertRaises(NoSuchFile, self.get_knit_index,
498
transport, "filename", "r")
499
self.assertRaises(NoSuchFile, self.get_knit_index,
500
transport, "filename", "w", create=False)
502
def test_create_file(self):
503
transport = MockTransport()
505
index = self.get_knit_index(transport, "filename", "w",
506
file_mode="wb", create=True)
508
("put_bytes_non_atomic",
509
("filename", index.HEADER), {"mode": "wb"}),
510
transport.calls.pop(0))
512
def test_delay_create_file(self):
513
transport = MockTransport()
515
index = self.get_knit_index(transport, "filename", "w",
516
create=True, file_mode="wb", create_parent_dir=True,
517
delay_create=True, dir_mode=0777)
518
self.assertEqual([], transport.calls)
520
index.add_versions([])
521
name, (filename, f), kwargs = transport.calls.pop(0)
522
self.assertEqual("put_file_non_atomic", name)
524
{"dir_mode": 0777, "create_parent_dir": True, "mode": "wb"},
526
self.assertEqual("filename", filename)
527
self.assertEqual(index.HEADER, f.read())
529
index.add_versions([])
530
self.assertEqual(("append_bytes", ("filename", ""), {}),
531
transport.calls.pop(0))
533
def test_read_utf8_version_id(self):
534
unicode_revision_id = u"version-\N{CYRILLIC CAPITAL LETTER A}"
535
utf8_revision_id = unicode_revision_id.encode('utf-8')
536
transport = MockTransport([
538
'%s option 0 1 :' % (utf8_revision_id,)
540
index = self.get_knit_index(transport, "filename", "r")
541
# _KnitIndex is a private class, and deals in utf8 revision_ids, not
542
# Unicode revision_ids.
543
self.assertTrue(index.has_version(utf8_revision_id))
544
self.assertFalse(index.has_version(unicode_revision_id))
546
def test_read_utf8_parents(self):
547
unicode_revision_id = u"version-\N{CYRILLIC CAPITAL LETTER A}"
548
utf8_revision_id = unicode_revision_id.encode('utf-8')
549
transport = MockTransport([
551
"version option 0 1 .%s :" % (utf8_revision_id,)
553
index = self.get_knit_index(transport, "filename", "r")
554
self.assertEqual([utf8_revision_id],
555
index.get_parents_with_ghosts("version"))
557
def test_read_ignore_corrupted_lines(self):
558
transport = MockTransport([
561
"corrupted options 0 1 .b .c ",
562
"version options 0 1 :"
564
index = self.get_knit_index(transport, "filename", "r")
565
self.assertEqual(1, index.num_versions())
566
self.assertTrue(index.has_version("version"))
568
def test_read_corrupted_header(self):
569
transport = MockTransport(['not a bzr knit index header\n'])
570
self.assertRaises(KnitHeaderError,
571
self.get_knit_index, transport, "filename", "r")
573
def test_read_duplicate_entries(self):
574
transport = MockTransport([
576
"parent options 0 1 :",
577
"version options1 0 1 0 :",
578
"version options2 1 2 .other :",
579
"version options3 3 4 0 .other :"
581
index = self.get_knit_index(transport, "filename", "r")
582
self.assertEqual(2, index.num_versions())
583
# check that the index used is the first one written. (Specific
584
# to KnitIndex style indices.
585
self.assertEqual("1", index._version_list_to_index(["version"]))
586
self.assertEqual((None, 3, 4), index.get_position("version"))
587
self.assertEqual(["options3"], index.get_options("version"))
588
self.assertEqual(["parent", "other"],
589
index.get_parents_with_ghosts("version"))
591
def test_read_compressed_parents(self):
592
transport = MockTransport([
596
"c option 0 1 1 0 :",
598
index = self.get_knit_index(transport, "filename", "r")
599
self.assertEqual(["a"], index.get_parents("b"))
600
self.assertEqual(["b", "a"], index.get_parents("c"))
602
def test_write_utf8_version_id(self):
603
unicode_revision_id = u"version-\N{CYRILLIC CAPITAL LETTER A}"
604
utf8_revision_id = unicode_revision_id.encode('utf-8')
605
transport = MockTransport([
608
index = self.get_knit_index(transport, "filename", "r")
609
index.add_version(utf8_revision_id, ["option"], (None, 0, 1), [])
610
self.assertEqual(("append_bytes", ("filename",
611
"\n%s option 0 1 :" % (utf8_revision_id,)),
613
transport.calls.pop(0))
615
def test_write_utf8_parents(self):
616
unicode_revision_id = u"version-\N{CYRILLIC CAPITAL LETTER A}"
617
utf8_revision_id = unicode_revision_id.encode('utf-8')
618
transport = MockTransport([
621
index = self.get_knit_index(transport, "filename", "r")
622
index.add_version("version", ["option"], (None, 0, 1), [utf8_revision_id])
623
self.assertEqual(("append_bytes", ("filename",
624
"\nversion option 0 1 .%s :" % (utf8_revision_id,)),
626
transport.calls.pop(0))
628
def test_get_graph(self):
629
transport = MockTransport()
630
index = self.get_knit_index(transport, "filename", "w", create=True)
631
self.assertEqual([], index.get_graph())
633
index.add_version("a", ["option"], (None, 0, 1), ["b"])
634
self.assertEqual([("a", ["b"])], index.get_graph())
636
index.add_version("c", ["option"], (None, 0, 1), ["d"])
637
self.assertEqual([("a", ["b"]), ("c", ["d"])],
638
sorted(index.get_graph()))
640
def test_get_ancestry(self):
641
transport = MockTransport([
644
"b option 0 1 0 .e :",
645
"c option 0 1 1 0 :",
646
"d option 0 1 2 .f :"
648
index = self.get_knit_index(transport, "filename", "r")
650
self.assertEqual([], index.get_ancestry([]))
651
self.assertEqual(["a"], index.get_ancestry(["a"]))
652
self.assertEqual(["a", "b"], index.get_ancestry(["b"]))
653
self.assertEqual(["a", "b", "c"], index.get_ancestry(["c"]))
654
self.assertEqual(["a", "b", "c", "d"], index.get_ancestry(["d"]))
655
self.assertEqual(["a", "b"], index.get_ancestry(["a", "b"]))
656
self.assertEqual(["a", "b", "c"], index.get_ancestry(["a", "c"]))
658
self.assertRaises(RevisionNotPresent, index.get_ancestry, ["e"])
660
def test_get_ancestry_with_ghosts(self):
661
transport = MockTransport([
664
"b option 0 1 0 .e :",
665
"c option 0 1 0 .f .g :",
666
"d option 0 1 2 .h .j .k :"
668
index = self.get_knit_index(transport, "filename", "r")
670
self.assertEqual([], index.get_ancestry_with_ghosts([]))
671
self.assertEqual(["a"], index.get_ancestry_with_ghosts(["a"]))
672
self.assertEqual(["a", "e", "b"],
673
index.get_ancestry_with_ghosts(["b"]))
674
self.assertEqual(["a", "g", "f", "c"],
675
index.get_ancestry_with_ghosts(["c"]))
676
self.assertEqual(["a", "g", "f", "c", "k", "j", "h", "d"],
677
index.get_ancestry_with_ghosts(["d"]))
678
self.assertEqual(["a", "e", "b"],
679
index.get_ancestry_with_ghosts(["a", "b"]))
680
self.assertEqual(["a", "g", "f", "c"],
681
index.get_ancestry_with_ghosts(["a", "c"]))
683
["a", "g", "f", "c", "e", "b", "k", "j", "h", "d"],
684
index.get_ancestry_with_ghosts(["b", "d"]))
686
self.assertRaises(RevisionNotPresent,
687
index.get_ancestry_with_ghosts, ["e"])
689
def test_iter_parents(self):
690
transport = MockTransport()
691
index = self.get_knit_index(transport, "filename", "w", create=True)
693
index.add_version('r0', ['option'], (None, 0, 1), [])
695
index.add_version('r1', ['option'], (None, 0, 1), ['r0'])
697
index.add_version('r2', ['option'], (None, 0, 1), ['r1', 'r0'])
699
# cases: each sample data individually:
700
self.assertEqual(set([('r0', ())]),
701
set(index.iter_parents(['r0'])))
702
self.assertEqual(set([('r1', ('r0', ))]),
703
set(index.iter_parents(['r1'])))
704
self.assertEqual(set([('r2', ('r1', 'r0'))]),
705
set(index.iter_parents(['r2'])))
706
# no nodes returned for a missing node
707
self.assertEqual(set(),
708
set(index.iter_parents(['missing'])))
709
# 1 node returned with missing nodes skipped
710
self.assertEqual(set([('r1', ('r0', ))]),
711
set(index.iter_parents(['ghost1', 'r1', 'ghost'])))
713
self.assertEqual(set([('r0', ()), ('r1', ('r0', ))]),
714
set(index.iter_parents(['r0', 'r1'])))
715
# 2 nodes returned, missing skipped
716
self.assertEqual(set([('r0', ()), ('r1', ('r0', ))]),
717
set(index.iter_parents(['a', 'r0', 'b', 'r1', 'c'])))
719
def test_num_versions(self):
720
transport = MockTransport([
723
index = self.get_knit_index(transport, "filename", "r")
725
self.assertEqual(0, index.num_versions())
726
self.assertEqual(0, len(index))
728
index.add_version("a", ["option"], (None, 0, 1), [])
729
self.assertEqual(1, index.num_versions())
730
self.assertEqual(1, len(index))
732
index.add_version("a", ["option2"], (None, 1, 2), [])
733
self.assertEqual(1, index.num_versions())
734
self.assertEqual(1, len(index))
736
index.add_version("b", ["option"], (None, 0, 1), [])
737
self.assertEqual(2, index.num_versions())
738
self.assertEqual(2, len(index))
740
def test_get_versions(self):
741
transport = MockTransport([
744
index = self.get_knit_index(transport, "filename", "r")
746
self.assertEqual([], index.get_versions())
748
index.add_version("a", ["option"], (None, 0, 1), [])
749
self.assertEqual(["a"], index.get_versions())
751
index.add_version("a", ["option"], (None, 0, 1), [])
752
self.assertEqual(["a"], index.get_versions())
754
index.add_version("b", ["option"], (None, 0, 1), [])
755
self.assertEqual(["a", "b"], index.get_versions())
757
def test_add_version(self):
758
transport = MockTransport([
761
index = self.get_knit_index(transport, "filename", "r")
763
index.add_version("a", ["option"], (None, 0, 1), ["b"])
764
self.assertEqual(("append_bytes",
765
("filename", "\na option 0 1 .b :"),
766
{}), transport.calls.pop(0))
767
self.assertTrue(index.has_version("a"))
768
self.assertEqual(1, index.num_versions())
769
self.assertEqual((None, 0, 1), index.get_position("a"))
770
self.assertEqual(["option"], index.get_options("a"))
771
self.assertEqual(["b"], index.get_parents_with_ghosts("a"))
773
index.add_version("a", ["opt"], (None, 1, 2), ["c"])
774
self.assertEqual(("append_bytes",
775
("filename", "\na opt 1 2 .c :"),
776
{}), transport.calls.pop(0))
777
self.assertTrue(index.has_version("a"))
778
self.assertEqual(1, index.num_versions())
779
self.assertEqual((None, 1, 2), index.get_position("a"))
780
self.assertEqual(["opt"], index.get_options("a"))
781
self.assertEqual(["c"], index.get_parents_with_ghosts("a"))
783
index.add_version("b", ["option"], (None, 2, 3), ["a"])
784
self.assertEqual(("append_bytes",
785
("filename", "\nb option 2 3 0 :"),
786
{}), transport.calls.pop(0))
787
self.assertTrue(index.has_version("b"))
788
self.assertEqual(2, index.num_versions())
789
self.assertEqual((None, 2, 3), index.get_position("b"))
790
self.assertEqual(["option"], index.get_options("b"))
791
self.assertEqual(["a"], index.get_parents_with_ghosts("b"))
793
def test_add_versions(self):
794
transport = MockTransport([
797
index = self.get_knit_index(transport, "filename", "r")
800
("a", ["option"], (None, 0, 1), ["b"]),
801
("a", ["opt"], (None, 1, 2), ["c"]),
802
("b", ["option"], (None, 2, 3), ["a"])
804
self.assertEqual(("append_bytes", ("filename",
805
"\na option 0 1 .b :"
808
), {}), transport.calls.pop(0))
809
self.assertTrue(index.has_version("a"))
810
self.assertTrue(index.has_version("b"))
811
self.assertEqual(2, index.num_versions())
812
self.assertEqual((None, 1, 2), index.get_position("a"))
813
self.assertEqual((None, 2, 3), index.get_position("b"))
814
self.assertEqual(["opt"], index.get_options("a"))
815
self.assertEqual(["option"], index.get_options("b"))
816
self.assertEqual(["c"], index.get_parents_with_ghosts("a"))
817
self.assertEqual(["a"], index.get_parents_with_ghosts("b"))
819
def test_delay_create_and_add_versions(self):
820
transport = MockTransport()
822
index = self.get_knit_index(transport, "filename", "w",
823
create=True, file_mode="wb", create_parent_dir=True,
824
delay_create=True, dir_mode=0777)
825
self.assertEqual([], transport.calls)
828
("a", ["option"], (None, 0, 1), ["b"]),
829
("a", ["opt"], (None, 1, 2), ["c"]),
830
("b", ["option"], (None, 2, 3), ["a"])
832
name, (filename, f), kwargs = transport.calls.pop(0)
833
self.assertEqual("put_file_non_atomic", name)
835
{"dir_mode": 0777, "create_parent_dir": True, "mode": "wb"},
837
self.assertEqual("filename", filename)
840
"\na option 0 1 .b :"
842
"\nb option 2 3 0 :",
845
def test_has_version(self):
846
transport = MockTransport([
850
index = self.get_knit_index(transport, "filename", "r")
852
self.assertTrue(index.has_version("a"))
853
self.assertFalse(index.has_version("b"))
855
def test_get_position(self):
856
transport = MockTransport([
861
index = self.get_knit_index(transport, "filename", "r")
863
self.assertEqual((None, 0, 1), index.get_position("a"))
864
self.assertEqual((None, 1, 2), index.get_position("b"))
866
def test_get_method(self):
867
transport = MockTransport([
869
"a fulltext,unknown 0 1 :",
870
"b unknown,line-delta 1 2 :",
873
index = self.get_knit_index(transport, "filename", "r")
875
self.assertEqual("fulltext", index.get_method("a"))
876
self.assertEqual("line-delta", index.get_method("b"))
877
self.assertRaises(errors.KnitIndexUnknownMethod, index.get_method, "c")
879
def test_get_options(self):
880
transport = MockTransport([
885
index = self.get_knit_index(transport, "filename", "r")
887
self.assertEqual(["opt1"], index.get_options("a"))
888
self.assertEqual(["opt2", "opt3"], index.get_options("b"))
890
def test_get_parents(self):
891
transport = MockTransport([
894
"b option 1 2 0 .c :",
895
"c option 1 2 1 0 .e :"
897
index = self.get_knit_index(transport, "filename", "r")
899
self.assertEqual([], index.get_parents("a"))
900
self.assertEqual(["a", "c"], index.get_parents("b"))
901
self.assertEqual(["b", "a"], index.get_parents("c"))
903
def test_get_parents_with_ghosts(self):
904
transport = MockTransport([
907
"b option 1 2 0 .c :",
908
"c option 1 2 1 0 .e :"
910
index = self.get_knit_index(transport, "filename", "r")
912
self.assertEqual([], index.get_parents_with_ghosts("a"))
913
self.assertEqual(["a", "c"], index.get_parents_with_ghosts("b"))
914
self.assertEqual(["b", "a", "e"],
915
index.get_parents_with_ghosts("c"))
917
def test_check_versions_present(self):
918
transport = MockTransport([
923
index = self.get_knit_index(transport, "filename", "r")
925
check = index.check_versions_present
931
self.assertRaises(RevisionNotPresent, check, ["c"])
932
self.assertRaises(RevisionNotPresent, check, ["a", "b", "c"])
934
def test_impossible_parent(self):
935
"""Test we get KnitCorrupt if the parent couldn't possibly exist."""
936
transport = MockTransport([
939
"b option 0 1 4 :" # We don't have a 4th record
942
self.assertRaises(errors.KnitCorrupt,
943
self.get_knit_index, transport, 'filename', 'r')
945
if (str(e) == ('exceptions must be strings, classes, or instances,'
946
' not exceptions.IndexError')
947
and sys.version_info[0:2] >= (2,5)):
948
self.knownFailure('Pyrex <0.9.5 fails with TypeError when'
949
' raising new style exceptions with python'
954
def test_corrupted_parent(self):
955
transport = MockTransport([
959
"c option 0 1 1v :", # Can't have a parent of '1v'
962
self.assertRaises(errors.KnitCorrupt,
963
self.get_knit_index, transport, 'filename', 'r')
965
if (str(e) == ('exceptions must be strings, classes, or instances,'
966
' not exceptions.ValueError')
967
and sys.version_info[0:2] >= (2,5)):
968
self.knownFailure('Pyrex <0.9.5 fails with TypeError when'
969
' raising new style exceptions with python'
974
def test_corrupted_parent_in_list(self):
975
transport = MockTransport([
979
"c option 0 1 1 v :", # Can't have a parent of 'v'
982
self.assertRaises(errors.KnitCorrupt,
983
self.get_knit_index, transport, 'filename', 'r')
985
if (str(e) == ('exceptions must be strings, classes, or instances,'
986
' not exceptions.ValueError')
987
and sys.version_info[0:2] >= (2,5)):
988
self.knownFailure('Pyrex <0.9.5 fails with TypeError when'
989
' raising new style exceptions with python'
994
def test_invalid_position(self):
995
transport = MockTransport([
1000
self.assertRaises(errors.KnitCorrupt,
1001
self.get_knit_index, transport, 'filename', 'r')
1002
except TypeError, e:
1003
if (str(e) == ('exceptions must be strings, classes, or instances,'
1004
' not exceptions.ValueError')
1005
and sys.version_info[0:2] >= (2,5)):
1006
self.knownFailure('Pyrex <0.9.5 fails with TypeError when'
1007
' raising new style exceptions with python'
1012
def test_invalid_size(self):
1013
transport = MockTransport([
1018
self.assertRaises(errors.KnitCorrupt,
1019
self.get_knit_index, transport, 'filename', 'r')
1020
except TypeError, e:
1021
if (str(e) == ('exceptions must be strings, classes, or instances,'
1022
' not exceptions.ValueError')
1023
and sys.version_info[0:2] >= (2,5)):
1024
self.knownFailure('Pyrex <0.9.5 fails with TypeError when'
1025
' raising new style exceptions with python'
1030
def test_short_line(self):
1031
transport = MockTransport([
1034
"b option 10 10 0", # This line isn't terminated, ignored
1036
index = self.get_knit_index(transport, "filename", "r")
1037
self.assertEqual(['a'], index.get_versions())
1039
def test_skip_incomplete_record(self):
1040
# A line with bogus data should just be skipped
1041
transport = MockTransport([
1044
"b option 10 10 0", # This line isn't terminated, ignored
1045
"c option 20 10 0 :", # Properly terminated, and starts with '\n'
1047
index = self.get_knit_index(transport, "filename", "r")
1048
self.assertEqual(['a', 'c'], index.get_versions())
1050
def test_trailing_characters(self):
1051
# A line with bogus data should just be skipped
1052
transport = MockTransport([
1055
"b option 10 10 0 :a", # This line has extra trailing characters
1056
"c option 20 10 0 :", # Properly terminated, and starts with '\n'
1058
index = self.get_knit_index(transport, "filename", "r")
1059
self.assertEqual(['a', 'c'], index.get_versions())
1062
class LowLevelKnitIndexTests_c(LowLevelKnitIndexTests):
1064
_test_needs_features = [CompiledKnitFeature]
1066
def get_knit_index(self, *args, **kwargs):
1067
orig = knit._load_data
1069
knit._load_data = orig
1070
self.addCleanup(reset)
1071
from bzrlib._knit_load_data_c import _load_data_c
1072
knit._load_data = _load_data_c
1073
return _KnitIndex(*args, **kwargs)
1077
class KnitTests(TestCaseWithTransport):
1078
"""Class containing knit test helper routines."""
1080
def make_test_knit(self, annotate=False, delay_create=False, index=None,
1083
factory = KnitPlainFactory()
1086
return KnitVersionedFile(name, get_transport('.'), access_mode='w',
1087
factory=factory, create=True,
1088
delay_create=delay_create, index=index)
1090
def assertRecordContentEqual(self, knit, version_id, candidate_content):
1091
"""Assert that some raw record content matches the raw record content
1092
for a particular version_id in the given knit.
1094
index_memo = knit._index.get_position(version_id)
1095
record = (version_id, index_memo)
1096
[(_, expected_content)] = list(knit._data.read_records_iter_raw([record]))
1097
self.assertEqual(expected_content, candidate_content)
1100
class BasicKnitTests(KnitTests):
1102
def add_stock_one_and_one_a(self, k):
1103
k.add_lines('text-1', [], split_lines(TEXT_1))
1104
k.add_lines('text-1a', ['text-1'], split_lines(TEXT_1A))
1106
def test_knit_constructor(self):
1107
"""Construct empty k"""
1108
self.make_test_knit()
1110
def test_make_explicit_index(self):
1111
"""We can supply an index to use."""
1112
knit = KnitVersionedFile('test', get_transport('.'),
1113
index='strangelove')
1114
self.assertEqual(knit._index, 'strangelove')
1116
def test_knit_add(self):
1117
"""Store one text in knit and retrieve"""
1118
k = self.make_test_knit()
1119
k.add_lines('text-1', [], split_lines(TEXT_1))
1120
self.assertTrue(k.has_version('text-1'))
1121
self.assertEqualDiff(''.join(k.get_lines('text-1')), TEXT_1)
1123
def test_knit_reload(self):
1124
# test that the content in a reloaded knit is correct
1125
k = self.make_test_knit()
1126
k.add_lines('text-1', [], split_lines(TEXT_1))
1128
k2 = KnitVersionedFile('test', get_transport('.'), access_mode='r', factory=KnitPlainFactory(), create=True)
1129
self.assertTrue(k2.has_version('text-1'))
1130
self.assertEqualDiff(''.join(k2.get_lines('text-1')), TEXT_1)
1132
def test_knit_several(self):
1133
"""Store several texts in a knit"""
1134
k = self.make_test_knit()
1135
k.add_lines('text-1', [], split_lines(TEXT_1))
1136
k.add_lines('text-2', [], split_lines(TEXT_2))
1137
self.assertEqualDiff(''.join(k.get_lines('text-1')), TEXT_1)
1138
self.assertEqualDiff(''.join(k.get_lines('text-2')), TEXT_2)
1140
def test_repeated_add(self):
1141
"""Knit traps attempt to replace existing version"""
1142
k = self.make_test_knit()
1143
k.add_lines('text-1', [], split_lines(TEXT_1))
1144
self.assertRaises(RevisionAlreadyPresent,
1146
'text-1', [], split_lines(TEXT_1))
1148
def test_empty(self):
1149
k = self.make_test_knit(True)
1150
k.add_lines('text-1', [], [])
1151
self.assertEquals(k.get_lines('text-1'), [])
1153
def test_incomplete(self):
1154
"""Test if texts without a ending line-end can be inserted and
1156
k = KnitVersionedFile('test', get_transport('.'), delta=False, create=True)
1157
k.add_lines('text-1', [], ['a\n', 'b' ])
1158
k.add_lines('text-2', ['text-1'], ['a\rb\n', 'b\n'])
1159
# reopening ensures maximum room for confusion
1160
k = KnitVersionedFile('test', get_transport('.'), delta=False, create=True)
1161
self.assertEquals(k.get_lines('text-1'), ['a\n', 'b' ])
1162
self.assertEquals(k.get_lines('text-2'), ['a\rb\n', 'b\n'])
1164
def test_delta(self):
1165
"""Expression of knit delta as lines"""
1166
k = self.make_test_knit()
1167
td = list(line_delta(TEXT_1.splitlines(True),
1168
TEXT_1A.splitlines(True)))
1169
self.assertEqualDiff(''.join(td), delta_1_1a)
1170
out = apply_line_delta(TEXT_1.splitlines(True), td)
1171
self.assertEqualDiff(''.join(out), TEXT_1A)
1173
def test_add_with_parents(self):
1174
"""Store in knit with parents"""
1175
k = self.make_test_knit()
1176
self.add_stock_one_and_one_a(k)
1177
self.assertEquals(k.get_parents('text-1'), [])
1178
self.assertEquals(k.get_parents('text-1a'), ['text-1'])
1180
def test_ancestry(self):
1181
"""Store in knit with parents"""
1182
k = self.make_test_knit()
1183
self.add_stock_one_and_one_a(k)
1184
self.assertEquals(set(k.get_ancestry(['text-1a'])), set(['text-1a', 'text-1']))
1186
def test_add_delta(self):
1187
"""Store in knit with parents"""
1188
k = KnitVersionedFile('test', get_transport('.'), factory=KnitPlainFactory(),
1189
delta=True, create=True)
1190
self.add_stock_one_and_one_a(k)
1192
self.assertEqualDiff(''.join(k.get_lines('text-1a')), TEXT_1A)
1194
def test_add_delta_knit_graph_index(self):
1195
"""Does adding work with a KnitGraphIndex."""
1196
index = InMemoryGraphIndex(2)
1197
knit_index = KnitGraphIndex(index, add_callback=index.add_nodes,
1199
k = KnitVersionedFile('test', get_transport('.'),
1200
delta=True, create=True, index=knit_index)
1201
self.add_stock_one_and_one_a(k)
1203
self.assertEqualDiff(''.join(k.get_lines('text-1a')), TEXT_1A)
1204
# check the index had the right data added.
1205
self.assertEqual(set([
1206
(index, ('text-1', ), ' 0 127', ((), ())),
1207
(index, ('text-1a', ), ' 127 140', ((('text-1', ),), (('text-1', ),))),
1208
]), set(index.iter_all_entries()))
1209
# we should not have a .kndx file
1210
self.assertFalse(get_transport('.').has('test.kndx'))
1212
def test_annotate(self):
1214
k = KnitVersionedFile('knit', get_transport('.'), factory=KnitAnnotateFactory(),
1215
delta=True, create=True)
1216
self.insert_and_test_small_annotate(k)
1218
def insert_and_test_small_annotate(self, k):
1219
"""test annotation with k works correctly."""
1220
k.add_lines('text-1', [], ['a\n', 'b\n'])
1221
k.add_lines('text-2', ['text-1'], ['a\n', 'c\n'])
1223
origins = k.annotate('text-2')
1224
self.assertEquals(origins[0], ('text-1', 'a\n'))
1225
self.assertEquals(origins[1], ('text-2', 'c\n'))
1227
def test_annotate_fulltext(self):
1229
k = KnitVersionedFile('knit', get_transport('.'), factory=KnitAnnotateFactory(),
1230
delta=False, create=True)
1231
self.insert_and_test_small_annotate(k)
1233
def test_annotate_merge_1(self):
1234
k = self.make_test_knit(True)
1235
k.add_lines('text-a1', [], ['a\n', 'b\n'])
1236
k.add_lines('text-a2', [], ['d\n', 'c\n'])
1237
k.add_lines('text-am', ['text-a1', 'text-a2'], ['d\n', 'b\n'])
1238
origins = k.annotate('text-am')
1239
self.assertEquals(origins[0], ('text-a2', 'd\n'))
1240
self.assertEquals(origins[1], ('text-a1', 'b\n'))
1242
def test_annotate_merge_2(self):
1243
k = self.make_test_knit(True)
1244
k.add_lines('text-a1', [], ['a\n', 'b\n', 'c\n'])
1245
k.add_lines('text-a2', [], ['x\n', 'y\n', 'z\n'])
1246
k.add_lines('text-am', ['text-a1', 'text-a2'], ['a\n', 'y\n', 'c\n'])
1247
origins = k.annotate('text-am')
1248
self.assertEquals(origins[0], ('text-a1', 'a\n'))
1249
self.assertEquals(origins[1], ('text-a2', 'y\n'))
1250
self.assertEquals(origins[2], ('text-a1', 'c\n'))
1252
def test_annotate_merge_9(self):
1253
k = self.make_test_knit(True)
1254
k.add_lines('text-a1', [], ['a\n', 'b\n', 'c\n'])
1255
k.add_lines('text-a2', [], ['x\n', 'y\n', 'z\n'])
1256
k.add_lines('text-am', ['text-a1', 'text-a2'], ['k\n', 'y\n', 'c\n'])
1257
origins = k.annotate('text-am')
1258
self.assertEquals(origins[0], ('text-am', 'k\n'))
1259
self.assertEquals(origins[1], ('text-a2', 'y\n'))
1260
self.assertEquals(origins[2], ('text-a1', 'c\n'))
1262
def test_annotate_merge_3(self):
1263
k = self.make_test_knit(True)
1264
k.add_lines('text-a1', [], ['a\n', 'b\n', 'c\n'])
1265
k.add_lines('text-a2', [] ,['x\n', 'y\n', 'z\n'])
1266
k.add_lines('text-am', ['text-a1', 'text-a2'], ['k\n', 'y\n', 'z\n'])
1267
origins = k.annotate('text-am')
1268
self.assertEquals(origins[0], ('text-am', 'k\n'))
1269
self.assertEquals(origins[1], ('text-a2', 'y\n'))
1270
self.assertEquals(origins[2], ('text-a2', 'z\n'))
1272
def test_annotate_merge_4(self):
1273
k = self.make_test_knit(True)
1274
k.add_lines('text-a1', [], ['a\n', 'b\n', 'c\n'])
1275
k.add_lines('text-a2', [], ['x\n', 'y\n', 'z\n'])
1276
k.add_lines('text-a3', ['text-a1'], ['a\n', 'b\n', 'p\n'])
1277
k.add_lines('text-am', ['text-a2', 'text-a3'], ['a\n', 'b\n', 'z\n'])
1278
origins = k.annotate('text-am')
1279
self.assertEquals(origins[0], ('text-a1', 'a\n'))
1280
self.assertEquals(origins[1], ('text-a1', 'b\n'))
1281
self.assertEquals(origins[2], ('text-a2', 'z\n'))
1283
def test_annotate_merge_5(self):
1284
k = self.make_test_knit(True)
1285
k.add_lines('text-a1', [], ['a\n', 'b\n', 'c\n'])
1286
k.add_lines('text-a2', [], ['d\n', 'e\n', 'f\n'])
1287
k.add_lines('text-a3', [], ['x\n', 'y\n', 'z\n'])
1288
k.add_lines('text-am',
1289
['text-a1', 'text-a2', 'text-a3'],
1290
['a\n', 'e\n', 'z\n'])
1291
origins = k.annotate('text-am')
1292
self.assertEquals(origins[0], ('text-a1', 'a\n'))
1293
self.assertEquals(origins[1], ('text-a2', 'e\n'))
1294
self.assertEquals(origins[2], ('text-a3', 'z\n'))
1296
def test_annotate_file_cherry_pick(self):
1297
k = self.make_test_knit(True)
1298
k.add_lines('text-1', [], ['a\n', 'b\n', 'c\n'])
1299
k.add_lines('text-2', ['text-1'], ['d\n', 'e\n', 'f\n'])
1300
k.add_lines('text-3', ['text-2', 'text-1'], ['a\n', 'b\n', 'c\n'])
1301
origins = k.annotate('text-3')
1302
self.assertEquals(origins[0], ('text-1', 'a\n'))
1303
self.assertEquals(origins[1], ('text-1', 'b\n'))
1304
self.assertEquals(origins[2], ('text-1', 'c\n'))
1306
def test_knit_join(self):
1307
"""Store in knit with parents"""
1308
k1 = KnitVersionedFile('test1', get_transport('.'), factory=KnitPlainFactory(), create=True)
1309
k1.add_lines('text-a', [], split_lines(TEXT_1))
1310
k1.add_lines('text-b', ['text-a'], split_lines(TEXT_1))
1312
k1.add_lines('text-c', [], split_lines(TEXT_1))
1313
k1.add_lines('text-d', ['text-c'], split_lines(TEXT_1))
1315
k1.add_lines('text-m', ['text-b', 'text-d'], split_lines(TEXT_1))
1317
k2 = KnitVersionedFile('test2', get_transport('.'), factory=KnitPlainFactory(), create=True)
1318
count = k2.join(k1, version_ids=['text-m'])
1319
self.assertEquals(count, 5)
1320
self.assertTrue(k2.has_version('text-a'))
1321
self.assertTrue(k2.has_version('text-c'))
1323
def test_reannotate(self):
1324
k1 = KnitVersionedFile('knit1', get_transport('.'),
1325
factory=KnitAnnotateFactory(), create=True)
1327
k1.add_lines('text-a', [], ['a\n', 'b\n'])
1329
k1.add_lines('text-b', ['text-a'], ['a\n', 'c\n'])
1331
k2 = KnitVersionedFile('test2', get_transport('.'),
1332
factory=KnitAnnotateFactory(), create=True)
1333
k2.join(k1, version_ids=['text-b'])
1336
k1.add_lines('text-X', ['text-b'], ['a\n', 'b\n'])
1338
k2.add_lines('text-c', ['text-b'], ['z\n', 'c\n'])
1340
k2.add_lines('text-Y', ['text-b'], ['b\n', 'c\n'])
1342
# test-c will have index 3
1343
k1.join(k2, version_ids=['text-c'])
1345
lines = k1.get_lines('text-c')
1346
self.assertEquals(lines, ['z\n', 'c\n'])
1348
origins = k1.annotate('text-c')
1349
self.assertEquals(origins[0], ('text-c', 'z\n'))
1350
self.assertEquals(origins[1], ('text-b', 'c\n'))
1352
def test_get_line_delta_texts(self):
1353
"""Make sure we can call get_texts on text with reused line deltas"""
1354
k1 = KnitVersionedFile('test1', get_transport('.'),
1355
factory=KnitPlainFactory(), create=True)
1360
parents = ['%d' % (t-1)]
1361
k1.add_lines('%d' % t, parents, ['hello\n'] * t)
1362
k1.get_texts(('%d' % t) for t in range(3))
1364
def test_iter_lines_reads_in_order(self):
1365
t = MemoryTransport()
1366
instrumented_t = TransportLogger(t)
1367
k1 = KnitVersionedFile('id', instrumented_t, create=True, delta=True)
1368
self.assertEqual([('id.kndx',)], instrumented_t._calls)
1369
# add texts with no required ordering
1370
k1.add_lines('base', [], ['text\n'])
1371
k1.add_lines('base2', [], ['text2\n'])
1373
instrumented_t._calls = []
1374
# request a last-first iteration
1375
results = list(k1.iter_lines_added_or_present_in_versions(['base2', 'base']))
1376
self.assertEqual([('id.knit', [(0, 87), (87, 89)])], instrumented_t._calls)
1377
self.assertEqual(['text\n', 'text2\n'], results)
1379
def test_create_empty_annotated(self):
1380
k1 = self.make_test_knit(True)
1382
k1.add_lines('text-a', [], ['a\n', 'b\n'])
1383
k2 = k1.create_empty('t', MemoryTransport())
1384
self.assertTrue(isinstance(k2.factory, KnitAnnotateFactory))
1385
self.assertEqual(k1.delta, k2.delta)
1386
# the generic test checks for empty content and file class
1388
def test_knit_format(self):
1389
# this tests that a new knit index file has the expected content
1390
# and that is writes the data we expect as records are added.
1391
knit = self.make_test_knit(True)
1392
# Now knit files are not created until we first add data to them
1393
self.assertFileEqual("# bzr knit index 8\n", 'test.kndx')
1394
knit.add_lines_with_ghosts('revid', ['a_ghost'], ['a\n'])
1395
self.assertFileEqual(
1396
"# bzr knit index 8\n"
1398
"revid fulltext 0 84 .a_ghost :",
1400
knit.add_lines_with_ghosts('revid2', ['revid'], ['a\n'])
1401
self.assertFileEqual(
1402
"# bzr knit index 8\n"
1403
"\nrevid fulltext 0 84 .a_ghost :"
1404
"\nrevid2 line-delta 84 82 0 :",
1406
# we should be able to load this file again
1407
knit = KnitVersionedFile('test', get_transport('.'), access_mode='r')
1408
self.assertEqual(['revid', 'revid2'], knit.versions())
1409
# write a short write to the file and ensure that its ignored
1410
indexfile = file('test.kndx', 'ab')
1411
indexfile.write('\nrevid3 line-delta 166 82 1 2 3 4 5 .phwoar:demo ')
1413
# we should be able to load this file again
1414
knit = KnitVersionedFile('test', get_transport('.'), access_mode='w')
1415
self.assertEqual(['revid', 'revid2'], knit.versions())
1416
# and add a revision with the same id the failed write had
1417
knit.add_lines('revid3', ['revid2'], ['a\n'])
1418
# and when reading it revid3 should now appear.
1419
knit = KnitVersionedFile('test', get_transport('.'), access_mode='r')
1420
self.assertEqual(['revid', 'revid2', 'revid3'], knit.versions())
1421
self.assertEqual(['revid2'], knit.get_parents('revid3'))
1423
def test_delay_create(self):
1424
"""Test that passing delay_create=True creates files late"""
1425
knit = self.make_test_knit(annotate=True, delay_create=True)
1426
self.failIfExists('test.knit')
1427
self.failIfExists('test.kndx')
1428
knit.add_lines_with_ghosts('revid', ['a_ghost'], ['a\n'])
1429
self.failUnlessExists('test.knit')
1430
self.assertFileEqual(
1431
"# bzr knit index 8\n"
1433
"revid fulltext 0 84 .a_ghost :",
1436
def test_create_parent_dir(self):
1437
"""create_parent_dir can create knits in nonexistant dirs"""
1438
# Has no effect if we don't set 'delay_create'
1439
trans = get_transport('.')
1440
self.assertRaises(NoSuchFile, KnitVersionedFile, 'dir/test',
1441
trans, access_mode='w', factory=None,
1442
create=True, create_parent_dir=True)
1443
# Nothing should have changed yet
1444
knit = KnitVersionedFile('dir/test', trans, access_mode='w',
1445
factory=None, create=True,
1446
create_parent_dir=True,
1448
self.failIfExists('dir/test.knit')
1449
self.failIfExists('dir/test.kndx')
1450
self.failIfExists('dir')
1451
knit.add_lines('revid', [], ['a\n'])
1452
self.failUnlessExists('dir')
1453
self.failUnlessExists('dir/test.knit')
1454
self.assertFileEqual(
1455
"# bzr knit index 8\n"
1457
"revid fulltext 0 84 :",
1460
def test_create_mode_700(self):
1461
trans = get_transport('.')
1462
if not trans._can_roundtrip_unix_modebits():
1463
# Can't roundtrip, so no need to run this test
1465
knit = KnitVersionedFile('dir/test', trans, access_mode='w',
1466
factory=None, create=True,
1467
create_parent_dir=True,
1471
knit.add_lines('revid', [], ['a\n'])
1472
self.assertTransportMode(trans, 'dir', 0700)
1473
self.assertTransportMode(trans, 'dir/test.knit', 0600)
1474
self.assertTransportMode(trans, 'dir/test.kndx', 0600)
1476
def test_create_mode_770(self):
1477
trans = get_transport('.')
1478
if not trans._can_roundtrip_unix_modebits():
1479
# Can't roundtrip, so no need to run this test
1481
knit = KnitVersionedFile('dir/test', trans, access_mode='w',
1482
factory=None, create=True,
1483
create_parent_dir=True,
1487
knit.add_lines('revid', [], ['a\n'])
1488
self.assertTransportMode(trans, 'dir', 0770)
1489
self.assertTransportMode(trans, 'dir/test.knit', 0660)
1490
self.assertTransportMode(trans, 'dir/test.kndx', 0660)
1492
def test_create_mode_777(self):
1493
trans = get_transport('.')
1494
if not trans._can_roundtrip_unix_modebits():
1495
# Can't roundtrip, so no need to run this test
1497
knit = KnitVersionedFile('dir/test', trans, access_mode='w',
1498
factory=None, create=True,
1499
create_parent_dir=True,
1503
knit.add_lines('revid', [], ['a\n'])
1504
self.assertTransportMode(trans, 'dir', 0777)
1505
self.assertTransportMode(trans, 'dir/test.knit', 0666)
1506
self.assertTransportMode(trans, 'dir/test.kndx', 0666)
1508
def test_plan_merge(self):
1509
my_knit = self.make_test_knit(annotate=True)
1510
my_knit.add_lines('text1', [], split_lines(TEXT_1))
1511
my_knit.add_lines('text1a', ['text1'], split_lines(TEXT_1A))
1512
my_knit.add_lines('text1b', ['text1'], split_lines(TEXT_1B))
1513
plan = list(my_knit.plan_merge('text1a', 'text1b'))
1514
for plan_line, expected_line in zip(plan, AB_MERGE):
1515
self.assertEqual(plan_line, expected_line)
1517
def test_get_stream_empty(self):
1518
"""Get a data stream for an empty knit file."""
1519
k1 = self.make_test_knit()
1520
format, data_list, reader_callable = k1.get_data_stream([])
1521
self.assertEqual('knit-plain', format)
1522
self.assertEqual([], data_list)
1523
content = reader_callable(None)
1524
self.assertEqual('', content)
1525
self.assertIsInstance(content, str)
1527
def test_get_stream_one_version(self):
1528
"""Get a data stream for a single record out of a knit containing just
1531
k1 = self.make_test_knit()
1533
('text-a', [], TEXT_1),
1535
expected_data_list = [
1536
# version, options, length, parents
1537
('text-a', ['fulltext'], 122, []),
1539
for version_id, parents, lines in test_data:
1540
k1.add_lines(version_id, parents, split_lines(lines))
1542
format, data_list, reader_callable = k1.get_data_stream(['text-a'])
1543
self.assertEqual('knit-plain', format)
1544
self.assertEqual(expected_data_list, data_list)
1545
# There's only one record in the knit, so the content should be the
1546
# entire knit data file's contents.
1547
self.assertEqual(k1.transport.get_bytes(k1._data._access._filename),
1548
reader_callable(None))
1550
def test_get_stream_get_one_version_of_many(self):
1551
"""Get a data stream for just one version out of a knit containing many
1554
k1 = self.make_test_knit()
1555
# Insert the same data as test_knit_join, as they seem to cover a range
1556
# of cases (no parents, one parent, multiple parents).
1558
('text-a', [], TEXT_1),
1559
('text-b', ['text-a'], TEXT_1),
1560
('text-c', [], TEXT_1),
1561
('text-d', ['text-c'], TEXT_1),
1562
('text-m', ['text-b', 'text-d'], TEXT_1),
1564
expected_data_list = [
1565
# version, options, length, parents
1566
('text-m', ['line-delta'], 84, ['text-b', 'text-d']),
1568
for version_id, parents, lines in test_data:
1569
k1.add_lines(version_id, parents, split_lines(lines))
1571
format, data_list, reader_callable = k1.get_data_stream(['text-m'])
1572
self.assertEqual('knit-plain', format)
1573
self.assertEqual(expected_data_list, data_list)
1574
self.assertRecordContentEqual(k1, 'text-m', reader_callable(None))
1576
def test_get_stream_ghost_parent(self):
1577
"""Get a data stream for a version with a ghost parent."""
1578
k1 = self.make_test_knit()
1580
k1.add_lines('text-a', [], split_lines(TEXT_1))
1581
k1.add_lines_with_ghosts('text-b', ['text-a', 'text-ghost'],
1582
split_lines(TEXT_1))
1584
expected_data_list = [
1585
# version, options, length, parents
1586
('text-b', ['line-delta'], 84, ['text-a', 'text-ghost']),
1589
format, data_list, reader_callable = k1.get_data_stream(['text-b'])
1590
self.assertEqual('knit-plain', format)
1591
self.assertEqual(expected_data_list, data_list)
1592
self.assertRecordContentEqual(k1, 'text-b', reader_callable(None))
1594
def test_get_stream_get_multiple_records(self):
1595
"""Get a stream for multiple records of a knit."""
1596
k1 = self.make_test_knit()
1597
# Insert the same data as test_knit_join, as they seem to cover a range
1598
# of cases (no parents, one parent, multiple parents).
1600
('text-a', [], TEXT_1),
1601
('text-b', ['text-a'], TEXT_1),
1602
('text-c', [], TEXT_1),
1603
('text-d', ['text-c'], TEXT_1),
1604
('text-m', ['text-b', 'text-d'], TEXT_1),
1606
expected_data_list = [
1607
# version, options, length, parents
1608
('text-b', ['line-delta'], 84, ['text-a']),
1609
('text-d', ['line-delta'], 84, ['text-c']),
1611
for version_id, parents, lines in test_data:
1612
k1.add_lines(version_id, parents, split_lines(lines))
1614
# Note that even though we request the revision IDs in a particular
1615
# order, the data stream may return them in any order it likes. In this
1616
# case, they'll be in the order they were inserted into the knit.
1617
format, data_list, reader_callable = k1.get_data_stream(
1618
['text-d', 'text-b'])
1619
self.assertEqual('knit-plain', format)
1620
self.assertEqual(expected_data_list, data_list)
1621
self.assertRecordContentEqual(k1, 'text-b', reader_callable(84))
1622
self.assertRecordContentEqual(k1, 'text-d', reader_callable(84))
1623
self.assertEqual('', reader_callable(None),
1624
"There should be no more bytes left to read.")
1626
def test_get_stream_all(self):
1627
"""Get a data stream for all the records in a knit.
1629
This exercises fulltext records, line-delta records, records with
1630
various numbers of parents, and reading multiple records out of the
1631
callable. These cases ought to all be exercised individually by the
1632
other test_get_stream_* tests; this test is basically just paranoia.
1634
k1 = self.make_test_knit()
1635
# Insert the same data as test_knit_join, as they seem to cover a range
1636
# of cases (no parents, one parent, multiple parents).
1638
('text-a', [], TEXT_1),
1639
('text-b', ['text-a'], TEXT_1),
1640
('text-c', [], TEXT_1),
1641
('text-d', ['text-c'], TEXT_1),
1642
('text-m', ['text-b', 'text-d'], TEXT_1),
1644
expected_data_list = [
1645
# version, options, length, parents
1646
('text-a', ['fulltext'], 122, []),
1647
('text-b', ['line-delta'], 84, ['text-a']),
1648
('text-c', ['fulltext'], 121, []),
1649
('text-d', ['line-delta'], 84, ['text-c']),
1650
('text-m', ['line-delta'], 84, ['text-b', 'text-d']),
1652
for version_id, parents, lines in test_data:
1653
k1.add_lines(version_id, parents, split_lines(lines))
1655
format, data_list, reader_callable = k1.get_data_stream(
1656
['text-a', 'text-b', 'text-c', 'text-d', 'text-m'])
1657
self.assertEqual('knit-plain', format)
1658
self.assertEqual(expected_data_list, data_list)
1659
for version_id, options, length, parents in expected_data_list:
1660
bytes = reader_callable(length)
1661
self.assertRecordContentEqual(k1, version_id, bytes)
1663
def assertKnitFilesEqual(self, knit1, knit2):
1664
"""Assert that the contents of the index and data files of two knits are
1668
knit1.transport.get_bytes(knit1._data._access._filename),
1669
knit2.transport.get_bytes(knit2._data._access._filename))
1671
knit1.transport.get_bytes(knit1._index._filename),
1672
knit2.transport.get_bytes(knit2._index._filename))
1674
def test_insert_data_stream_empty(self):
1675
"""Inserting a data stream with no records should not put any data into
1678
k1 = self.make_test_knit()
1679
k1.insert_data_stream(
1680
(k1.get_format_signature(), [], lambda ignored: ''))
1681
self.assertEqual('', k1.transport.get_bytes(k1._data._access._filename),
1682
"The .knit should be completely empty.")
1683
self.assertEqual(k1._index.HEADER,
1684
k1.transport.get_bytes(k1._index._filename),
1685
"The .kndx should have nothing apart from the header.")
1687
def test_insert_data_stream_one_record(self):
1688
"""Inserting a data stream with one record from a knit with one record
1689
results in byte-identical files.
1691
source = self.make_test_knit(name='source')
1692
source.add_lines('text-a', [], split_lines(TEXT_1))
1693
data_stream = source.get_data_stream(['text-a'])
1695
target = self.make_test_knit(name='target')
1696
target.insert_data_stream(data_stream)
1698
self.assertKnitFilesEqual(source, target)
1700
def test_insert_data_stream_records_already_present(self):
1701
"""Insert a data stream where some records are alreday present in the
1702
target, and some not. Only the new records are inserted.
1704
source = self.make_test_knit(name='source')
1705
target = self.make_test_knit(name='target')
1706
# Insert 'text-a' into both source and target
1707
source.add_lines('text-a', [], split_lines(TEXT_1))
1708
target.insert_data_stream(source.get_data_stream(['text-a']))
1709
# Insert 'text-b' into just the source.
1710
source.add_lines('text-b', ['text-a'], split_lines(TEXT_1))
1711
# Get a data stream of both text-a and text-b, and insert it.
1712
data_stream = source.get_data_stream(['text-a', 'text-b'])
1713
target.insert_data_stream(data_stream)
1714
# The source and target will now be identical. This means the text-a
1715
# record was not added a second time.
1716
self.assertKnitFilesEqual(source, target)
1718
def test_insert_data_stream_multiple_records(self):
1719
"""Inserting a data stream of all records from a knit with multiple
1720
records results in byte-identical files.
1722
source = self.make_test_knit(name='source')
1723
source.add_lines('text-a', [], split_lines(TEXT_1))
1724
source.add_lines('text-b', ['text-a'], split_lines(TEXT_1))
1725
source.add_lines('text-c', [], split_lines(TEXT_1))
1726
data_stream = source.get_data_stream(['text-a', 'text-b', 'text-c'])
1728
target = self.make_test_knit(name='target')
1729
target.insert_data_stream(data_stream)
1731
self.assertKnitFilesEqual(source, target)
1733
def test_insert_data_stream_ghost_parent(self):
1734
"""Insert a data stream with a record that has a ghost parent."""
1735
# Make a knit with a record, text-a, that has a ghost parent.
1736
source = self.make_test_knit(name='source')
1737
source.add_lines_with_ghosts('text-a', ['text-ghost'],
1738
split_lines(TEXT_1))
1739
data_stream = source.get_data_stream(['text-a'])
1741
target = self.make_test_knit(name='target')
1742
target.insert_data_stream(data_stream)
1744
self.assertKnitFilesEqual(source, target)
1746
# The target knit object is in a consistent state, i.e. the record we
1747
# just added is immediately visible.
1748
self.assertTrue(target.has_version('text-a'))
1749
self.assertTrue(target.has_ghost('text-ghost'))
1750
self.assertEqual(split_lines(TEXT_1), target.get_lines('text-a'))
1752
def test_insert_data_stream_inconsistent_version_lines(self):
1753
"""Inserting a data stream which has different content for a version_id
1754
than already exists in the knit will raise KnitCorrupt.
1756
source = self.make_test_knit(name='source')
1757
target = self.make_test_knit(name='target')
1758
# Insert a different 'text-a' into both source and target
1759
source.add_lines('text-a', [], split_lines(TEXT_1))
1760
target.add_lines('text-a', [], split_lines(TEXT_2))
1761
# Insert a data stream with conflicting content into the target
1762
data_stream = source.get_data_stream(['text-a'])
1764
errors.KnitCorrupt, target.insert_data_stream, data_stream)
1766
def test_insert_data_stream_inconsistent_version_parents(self):
1767
"""Inserting a data stream which has different parents for a version_id
1768
than already exists in the knit will raise KnitCorrupt.
1770
source = self.make_test_knit(name='source')
1771
target = self.make_test_knit(name='target')
1772
# Insert a different 'text-a' into both source and target. They differ
1773
# only by the parents list, the content is the same.
1774
source.add_lines_with_ghosts('text-a', [], split_lines(TEXT_1))
1775
target.add_lines_with_ghosts('text-a', ['a-ghost'], split_lines(TEXT_1))
1776
# Insert a data stream with conflicting content into the target
1777
data_stream = source.get_data_stream(['text-a'])
1779
errors.KnitCorrupt, target.insert_data_stream, data_stream)
1781
def test_insert_data_stream_incompatible_format(self):
1782
"""A data stream in a different format to the target knit cannot be
1785
It will raise KnitDataStreamIncompatible.
1787
data_stream = ('fake-format-signature', [], lambda _: '')
1788
target = self.make_test_knit(name='target')
1790
errors.KnitDataStreamIncompatible,
1791
target.insert_data_stream, data_stream)
1793
# * test that a stream of "already present version, then new version"
1794
# inserts correctly.
1805
Banana cup cake recipe
1811
- self-raising flour
1815
Banana cup cake recipe
1817
- bananas (do not use plantains!!!)
1824
Banana cup cake recipe
1827
- self-raising flour
1840
AB_MERGE_TEXT="""unchanged|Banana cup cake recipe
1845
new-b|- bananas (do not use plantains!!!)
1846
unchanged|- broken tea cups
1847
new-a|- self-raising flour
1850
AB_MERGE=[tuple(l.split('|')) for l in AB_MERGE_TEXT.splitlines(True)]
1853
def line_delta(from_lines, to_lines):
1854
"""Generate line-based delta from one text to another"""
1855
s = difflib.SequenceMatcher(None, from_lines, to_lines)
1856
for op in s.get_opcodes():
1857
if op[0] == 'equal':
1859
yield '%d,%d,%d\n' % (op[1], op[2], op[4]-op[3])
1860
for i in range(op[3], op[4]):
1864
def apply_line_delta(basis_lines, delta_lines):
1865
"""Apply a line-based perfect diff
1867
basis_lines -- text to apply the patch to
1868
delta_lines -- diff instructions and content
1870
out = basis_lines[:]
1873
while i < len(delta_lines):
1875
a, b, c = map(long, l.split(','))
1877
out[offset+a:offset+b] = delta_lines[i:i+c]
1879
offset = offset + (b - a) + c
1883
class TestWeaveToKnit(KnitTests):
1885
def test_weave_to_knit_matches(self):
1886
# check that the WeaveToKnit is_compatible function
1887
# registers True for a Weave to a Knit.
1889
k = self.make_test_knit()
1890
self.failUnless(WeaveToKnit.is_compatible(w, k))
1891
self.failIf(WeaveToKnit.is_compatible(k, w))
1892
self.failIf(WeaveToKnit.is_compatible(w, w))
1893
self.failIf(WeaveToKnit.is_compatible(k, k))
1896
class TestKnitCaching(KnitTests):
1898
def create_knit(self, cache_add=False):
1899
k = self.make_test_knit(True)
1903
k.add_lines('text-1', [], split_lines(TEXT_1))
1904
k.add_lines('text-2', [], split_lines(TEXT_2))
1907
def test_no_caching(self):
1908
k = self.create_knit()
1909
# Nothing should be cached without setting 'enable_cache'
1910
self.assertEqual({}, k._data._cache)
1912
def test_cache_add_and_clear(self):
1913
k = self.create_knit(True)
1915
self.assertEqual(['text-1', 'text-2'], sorted(k._data._cache.keys()))
1918
self.assertEqual({}, k._data._cache)
1920
def test_cache_data_read_raw(self):
1921
k = self.create_knit()
1923
# Now cache and read
1926
def read_one_raw(version):
1927
pos_map = k._get_components_positions([version])
1928
method, index_memo, next = pos_map[version]
1929
lst = list(k._data.read_records_iter_raw([(version, index_memo)]))
1930
self.assertEqual(1, len(lst))
1933
val = read_one_raw('text-1')
1934
self.assertEqual({'text-1':val[1]}, k._data._cache)
1937
# After clear, new reads are not cached
1938
self.assertEqual({}, k._data._cache)
1940
val2 = read_one_raw('text-1')
1941
self.assertEqual(val, val2)
1942
self.assertEqual({}, k._data._cache)
1944
def test_cache_data_read(self):
1945
k = self.create_knit()
1947
def read_one(version):
1948
pos_map = k._get_components_positions([version])
1949
method, index_memo, next = pos_map[version]
1950
lst = list(k._data.read_records_iter([(version, index_memo)]))
1951
self.assertEqual(1, len(lst))
1954
# Now cache and read
1957
val = read_one('text-2')
1958
self.assertEqual(['text-2'], k._data._cache.keys())
1959
self.assertEqual('text-2', val[0])
1960
content, digest = k._data._parse_record('text-2',
1961
k._data._cache['text-2'])
1962
self.assertEqual(content, val[1])
1963
self.assertEqual(digest, val[2])
1966
self.assertEqual({}, k._data._cache)
1968
val2 = read_one('text-2')
1969
self.assertEqual(val, val2)
1970
self.assertEqual({}, k._data._cache)
1972
def test_cache_read(self):
1973
k = self.create_knit()
1976
text = k.get_text('text-1')
1977
self.assertEqual(TEXT_1, text)
1978
self.assertEqual(['text-1'], k._data._cache.keys())
1981
self.assertEqual({}, k._data._cache)
1983
text = k.get_text('text-1')
1984
self.assertEqual(TEXT_1, text)
1985
self.assertEqual({}, k._data._cache)
1988
class TestKnitIndex(KnitTests):
1990
def test_add_versions_dictionary_compresses(self):
1991
"""Adding versions to the index should update the lookup dict"""
1992
knit = self.make_test_knit()
1994
idx.add_version('a-1', ['fulltext'], (None, 0, 0), [])
1995
self.check_file_contents('test.kndx',
1996
'# bzr knit index 8\n'
1998
'a-1 fulltext 0 0 :'
2000
idx.add_versions([('a-2', ['fulltext'], (None, 0, 0), ['a-1']),
2001
('a-3', ['fulltext'], (None, 0, 0), ['a-2']),
2003
self.check_file_contents('test.kndx',
2004
'# bzr knit index 8\n'
2006
'a-1 fulltext 0 0 :\n'
2007
'a-2 fulltext 0 0 0 :\n'
2008
'a-3 fulltext 0 0 1 :'
2010
self.assertEqual(['a-1', 'a-2', 'a-3'], idx._history)
2011
self.assertEqual({'a-1':('a-1', ['fulltext'], 0, 0, [], 0),
2012
'a-2':('a-2', ['fulltext'], 0, 0, ['a-1'], 1),
2013
'a-3':('a-3', ['fulltext'], 0, 0, ['a-2'], 2),
2016
def test_add_versions_fails_clean(self):
2017
"""If add_versions fails in the middle, it restores a pristine state.
2019
Any modifications that are made to the index are reset if all versions
2022
# This cheats a little bit by passing in a generator which will
2023
# raise an exception before the processing finishes
2024
# Other possibilities would be to have an version with the wrong number
2025
# of entries, or to make the backing transport unable to write any
2028
knit = self.make_test_knit()
2030
idx.add_version('a-1', ['fulltext'], (None, 0, 0), [])
2032
class StopEarly(Exception):
2035
def generate_failure():
2036
"""Add some entries and then raise an exception"""
2037
yield ('a-2', ['fulltext'], (None, 0, 0), ['a-1'])
2038
yield ('a-3', ['fulltext'], (None, 0, 0), ['a-2'])
2041
# Assert the pre-condition
2042
self.assertEqual(['a-1'], idx._history)
2043
self.assertEqual({'a-1':('a-1', ['fulltext'], 0, 0, [], 0)}, idx._cache)
2045
self.assertRaises(StopEarly, idx.add_versions, generate_failure())
2047
# And it shouldn't be modified
2048
self.assertEqual(['a-1'], idx._history)
2049
self.assertEqual({'a-1':('a-1', ['fulltext'], 0, 0, [], 0)}, idx._cache)
2051
def test_knit_index_ignores_empty_files(self):
2052
# There was a race condition in older bzr, where a ^C at the right time
2053
# could leave an empty .kndx file, which bzr would later claim was a
2054
# corrupted file since the header was not present. In reality, the file
2055
# just wasn't created, so it should be ignored.
2056
t = get_transport('.')
2057
t.put_bytes('test.kndx', '')
2059
knit = self.make_test_knit()
2061
def test_knit_index_checks_header(self):
2062
t = get_transport('.')
2063
t.put_bytes('test.kndx', '# not really a knit header\n\n')
2065
self.assertRaises(KnitHeaderError, self.make_test_knit)
2068
class TestGraphIndexKnit(KnitTests):
2069
"""Tests for knits using a GraphIndex rather than a KnitIndex."""
2071
def make_g_index(self, name, ref_lists=0, nodes=[]):
2072
builder = GraphIndexBuilder(ref_lists)
2073
for node, references, value in nodes:
2074
builder.add_node(node, references, value)
2075
stream = builder.finish()
2076
trans = self.get_transport()
2077
trans.put_file(name, stream)
2078
return GraphIndex(trans, name)
2080
def two_graph_index(self, deltas=False, catch_adds=False):
2081
"""Build a two-graph index.
2083
:param deltas: If true, use underlying indices with two node-ref
2084
lists and 'parent' set to a delta-compressed against tail.
2086
# build a complex graph across several indices.
2088
# delta compression inn the index
2089
index1 = self.make_g_index('1', 2, [
2090
(('tip', ), 'N0 100', ([('parent', )], [], )),
2091
(('tail', ), '', ([], []))])
2092
index2 = self.make_g_index('2', 2, [
2093
(('parent', ), ' 100 78', ([('tail', ), ('ghost', )], [('tail', )])),
2094
(('separate', ), '', ([], []))])
2096
# just blob location and graph in the index.
2097
index1 = self.make_g_index('1', 1, [
2098
(('tip', ), 'N0 100', ([('parent', )], )),
2099
(('tail', ), '', ([], ))])
2100
index2 = self.make_g_index('2', 1, [
2101
(('parent', ), ' 100 78', ([('tail', ), ('ghost', )], )),
2102
(('separate', ), '', ([], ))])
2103
combined_index = CombinedGraphIndex([index1, index2])
2105
self.combined_index = combined_index
2106
self.caught_entries = []
2107
add_callback = self.catch_add
2110
return KnitGraphIndex(combined_index, deltas=deltas,
2111
add_callback=add_callback)
2113
def test_get_graph(self):
2114
index = self.two_graph_index()
2115
self.assertEqual(set([
2116
('tip', ('parent', )),
2118
('parent', ('tail', 'ghost')),
2120
]), set(index.get_graph()))
2122
def test_get_ancestry(self):
2123
# get_ancestry is defined as eliding ghosts, not erroring.
2124
index = self.two_graph_index()
2125
self.assertEqual([], index.get_ancestry([]))
2126
self.assertEqual(['separate'], index.get_ancestry(['separate']))
2127
self.assertEqual(['tail'], index.get_ancestry(['tail']))
2128
self.assertEqual(['tail', 'parent'], index.get_ancestry(['parent']))
2129
self.assertEqual(['tail', 'parent', 'tip'], index.get_ancestry(['tip']))
2130
self.assertTrue(index.get_ancestry(['tip', 'separate']) in
2131
(['tail', 'parent', 'tip', 'separate'],
2132
['separate', 'tail', 'parent', 'tip'],
2134
# and without topo_sort
2135
self.assertEqual(set(['separate']),
2136
set(index.get_ancestry(['separate'], topo_sorted=False)))
2137
self.assertEqual(set(['tail']),
2138
set(index.get_ancestry(['tail'], topo_sorted=False)))
2139
self.assertEqual(set(['tail', 'parent']),
2140
set(index.get_ancestry(['parent'], topo_sorted=False)))
2141
self.assertEqual(set(['tail', 'parent', 'tip']),
2142
set(index.get_ancestry(['tip'], topo_sorted=False)))
2143
self.assertEqual(set(['separate', 'tail', 'parent', 'tip']),
2144
set(index.get_ancestry(['tip', 'separate'])))
2145
# asking for a ghost makes it go boom.
2146
self.assertRaises(errors.RevisionNotPresent, index.get_ancestry, ['ghost'])
2148
def test_get_ancestry_with_ghosts(self):
2149
index = self.two_graph_index()
2150
self.assertEqual([], index.get_ancestry_with_ghosts([]))
2151
self.assertEqual(['separate'], index.get_ancestry_with_ghosts(['separate']))
2152
self.assertEqual(['tail'], index.get_ancestry_with_ghosts(['tail']))
2153
self.assertTrue(index.get_ancestry_with_ghosts(['parent']) in
2154
(['tail', 'ghost', 'parent'],
2155
['ghost', 'tail', 'parent'],
2157
self.assertTrue(index.get_ancestry_with_ghosts(['tip']) in
2158
(['tail', 'ghost', 'parent', 'tip'],
2159
['ghost', 'tail', 'parent', 'tip'],
2161
self.assertTrue(index.get_ancestry_with_ghosts(['tip', 'separate']) in
2162
(['tail', 'ghost', 'parent', 'tip', 'separate'],
2163
['ghost', 'tail', 'parent', 'tip', 'separate'],
2164
['separate', 'tail', 'ghost', 'parent', 'tip'],
2165
['separate', 'ghost', 'tail', 'parent', 'tip'],
2167
# asking for a ghost makes it go boom.
2168
self.assertRaises(errors.RevisionNotPresent, index.get_ancestry_with_ghosts, ['ghost'])
2170
def test_num_versions(self):
2171
index = self.two_graph_index()
2172
self.assertEqual(4, index.num_versions())
2174
def test_get_versions(self):
2175
index = self.two_graph_index()
2176
self.assertEqual(set(['tail', 'tip', 'parent', 'separate']),
2177
set(index.get_versions()))
2179
def test_has_version(self):
2180
index = self.two_graph_index()
2181
self.assertTrue(index.has_version('tail'))
2182
self.assertFalse(index.has_version('ghost'))
2184
def test_get_position(self):
2185
index = self.two_graph_index()
2186
self.assertEqual((index._graph_index._indices[0], 0, 100), index.get_position('tip'))
2187
self.assertEqual((index._graph_index._indices[1], 100, 78), index.get_position('parent'))
2189
def test_get_method_deltas(self):
2190
index = self.two_graph_index(deltas=True)
2191
self.assertEqual('fulltext', index.get_method('tip'))
2192
self.assertEqual('line-delta', index.get_method('parent'))
2194
def test_get_method_no_deltas(self):
2195
# check that the parent-history lookup is ignored with deltas=False.
2196
index = self.two_graph_index(deltas=False)
2197
self.assertEqual('fulltext', index.get_method('tip'))
2198
self.assertEqual('fulltext', index.get_method('parent'))
2200
def test_get_options_deltas(self):
2201
index = self.two_graph_index(deltas=True)
2202
self.assertEqual(['fulltext', 'no-eol'], index.get_options('tip'))
2203
self.assertEqual(['line-delta'], index.get_options('parent'))
2205
def test_get_options_no_deltas(self):
2206
# check that the parent-history lookup is ignored with deltas=False.
2207
index = self.two_graph_index(deltas=False)
2208
self.assertEqual(['fulltext', 'no-eol'], index.get_options('tip'))
2209
self.assertEqual(['fulltext'], index.get_options('parent'))
2211
def test_get_parents(self):
2212
# get_parents ignores ghosts
2213
index = self.two_graph_index()
2214
self.assertEqual(('tail', ), index.get_parents('parent'))
2215
# and errors on ghosts.
2216
self.assertRaises(errors.RevisionNotPresent,
2217
index.get_parents, 'ghost')
2219
def test_get_parents_with_ghosts(self):
2220
index = self.two_graph_index()
2221
self.assertEqual(('tail', 'ghost'), index.get_parents_with_ghosts('parent'))
2222
# and errors on ghosts.
2223
self.assertRaises(errors.RevisionNotPresent,
2224
index.get_parents_with_ghosts, 'ghost')
2226
def test_check_versions_present(self):
2227
# ghosts should not be considered present
2228
index = self.two_graph_index()
2229
self.assertRaises(RevisionNotPresent, index.check_versions_present,
2231
self.assertRaises(RevisionNotPresent, index.check_versions_present,
2233
index.check_versions_present(['tail', 'separate'])
2235
def catch_add(self, entries):
2236
self.caught_entries.append(entries)
2238
def test_add_no_callback_errors(self):
2239
index = self.two_graph_index()
2240
self.assertRaises(errors.ReadOnlyError, index.add_version,
2241
'new', 'fulltext,no-eol', (None, 50, 60), ['separate'])
2243
def test_add_version_smoke(self):
2244
index = self.two_graph_index(catch_adds=True)
2245
index.add_version('new', 'fulltext,no-eol', (None, 50, 60), ['separate'])
2246
self.assertEqual([[(('new', ), 'N50 60', ((('separate',),),))]],
2247
self.caught_entries)
2249
def test_add_version_delta_not_delta_index(self):
2250
index = self.two_graph_index(catch_adds=True)
2251
self.assertRaises(errors.KnitCorrupt, index.add_version,
2252
'new', 'no-eol,line-delta', (None, 0, 100), ['parent'])
2253
self.assertEqual([], self.caught_entries)
2255
def test_add_version_same_dup(self):
2256
index = self.two_graph_index(catch_adds=True)
2257
# options can be spelt two different ways
2258
index.add_version('tip', 'fulltext,no-eol', (None, 0, 100), ['parent'])
2259
index.add_version('tip', 'no-eol,fulltext', (None, 0, 100), ['parent'])
2260
# but neither should have added data.
2261
self.assertEqual([[], []], self.caught_entries)
2263
def test_add_version_different_dup(self):
2264
index = self.two_graph_index(deltas=True, catch_adds=True)
2266
self.assertRaises(errors.KnitCorrupt, index.add_version,
2267
'tip', 'no-eol,line-delta', (None, 0, 100), ['parent'])
2268
self.assertRaises(errors.KnitCorrupt, index.add_version,
2269
'tip', 'line-delta,no-eol', (None, 0, 100), ['parent'])
2270
self.assertRaises(errors.KnitCorrupt, index.add_version,
2271
'tip', 'fulltext', (None, 0, 100), ['parent'])
2273
self.assertRaises(errors.KnitCorrupt, index.add_version,
2274
'tip', 'fulltext,no-eol', (None, 50, 100), ['parent'])
2275
self.assertRaises(errors.KnitCorrupt, index.add_version,
2276
'tip', 'fulltext,no-eol', (None, 0, 1000), ['parent'])
2278
self.assertRaises(errors.KnitCorrupt, index.add_version,
2279
'tip', 'fulltext,no-eol', (None, 0, 100), [])
2280
self.assertEqual([], self.caught_entries)
2282
def test_add_versions_nodeltas(self):
2283
index = self.two_graph_index(catch_adds=True)
2284
index.add_versions([
2285
('new', 'fulltext,no-eol', (None, 50, 60), ['separate']),
2286
('new2', 'fulltext', (None, 0, 6), ['new']),
2288
self.assertEqual([(('new', ), 'N50 60', ((('separate',),),)),
2289
(('new2', ), ' 0 6', ((('new',),),))],
2290
sorted(self.caught_entries[0]))
2291
self.assertEqual(1, len(self.caught_entries))
2293
def test_add_versions_deltas(self):
2294
index = self.two_graph_index(deltas=True, catch_adds=True)
2295
index.add_versions([
2296
('new', 'fulltext,no-eol', (None, 50, 60), ['separate']),
2297
('new2', 'line-delta', (None, 0, 6), ['new']),
2299
self.assertEqual([(('new', ), 'N50 60', ((('separate',),), ())),
2300
(('new2', ), ' 0 6', ((('new',),), (('new',),), ))],
2301
sorted(self.caught_entries[0]))
2302
self.assertEqual(1, len(self.caught_entries))
2304
def test_add_versions_delta_not_delta_index(self):
2305
index = self.two_graph_index(catch_adds=True)
2306
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2307
[('new', 'no-eol,line-delta', (None, 0, 100), ['parent'])])
2308
self.assertEqual([], self.caught_entries)
2310
def test_add_versions_same_dup(self):
2311
index = self.two_graph_index(catch_adds=True)
2312
# options can be spelt two different ways
2313
index.add_versions([('tip', 'fulltext,no-eol', (None, 0, 100), ['parent'])])
2314
index.add_versions([('tip', 'no-eol,fulltext', (None, 0, 100), ['parent'])])
2315
# but neither should have added data.
2316
self.assertEqual([[], []], self.caught_entries)
2318
def test_add_versions_different_dup(self):
2319
index = self.two_graph_index(deltas=True, catch_adds=True)
2321
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2322
[('tip', 'no-eol,line-delta', (None, 0, 100), ['parent'])])
2323
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2324
[('tip', 'line-delta,no-eol', (None, 0, 100), ['parent'])])
2325
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2326
[('tip', 'fulltext', (None, 0, 100), ['parent'])])
2328
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2329
[('tip', 'fulltext,no-eol', (None, 50, 100), ['parent'])])
2330
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2331
[('tip', 'fulltext,no-eol', (None, 0, 1000), ['parent'])])
2333
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2334
[('tip', 'fulltext,no-eol', (None, 0, 100), [])])
2335
# change options in the second record
2336
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2337
[('tip', 'fulltext,no-eol', (None, 0, 100), ['parent']),
2338
('tip', 'no-eol,line-delta', (None, 0, 100), ['parent'])])
2339
self.assertEqual([], self.caught_entries)
2341
def test_iter_parents(self):
2342
index1 = self.make_g_index('1', 1, [
2344
(('r0', ), 'N0 100', ([], )),
2346
(('r1', ), '', ([('r0', )], ))])
2347
index2 = self.make_g_index('2', 1, [
2349
(('r2', ), 'N0 100', ([('r1', ), ('r0', )], )),
2351
combined_index = CombinedGraphIndex([index1, index2])
2352
index = KnitGraphIndex(combined_index)
2354
# cases: each sample data individually:
2355
self.assertEqual(set([('r0', ())]),
2356
set(index.iter_parents(['r0'])))
2357
self.assertEqual(set([('r1', ('r0', ))]),
2358
set(index.iter_parents(['r1'])))
2359
self.assertEqual(set([('r2', ('r1', 'r0'))]),
2360
set(index.iter_parents(['r2'])))
2361
# no nodes returned for a missing node
2362
self.assertEqual(set(),
2363
set(index.iter_parents(['missing'])))
2364
# 1 node returned with missing nodes skipped
2365
self.assertEqual(set([('r1', ('r0', ))]),
2366
set(index.iter_parents(['ghost1', 'r1', 'ghost'])))
2368
self.assertEqual(set([('r0', ()), ('r1', ('r0', ))]),
2369
set(index.iter_parents(['r0', 'r1'])))
2370
# 2 nodes returned, missing skipped
2371
self.assertEqual(set([('r0', ()), ('r1', ('r0', ))]),
2372
set(index.iter_parents(['a', 'r0', 'b', 'r1', 'c'])))
2375
class TestNoParentsGraphIndexKnit(KnitTests):
2376
"""Tests for knits using KnitGraphIndex with no parents."""
2378
def make_g_index(self, name, ref_lists=0, nodes=[]):
2379
builder = GraphIndexBuilder(ref_lists)
2380
for node, references in nodes:
2381
builder.add_node(node, references)
2382
stream = builder.finish()
2383
trans = self.get_transport()
2384
trans.put_file(name, stream)
2385
return GraphIndex(trans, name)
2387
def test_parents_deltas_incompatible(self):
2388
index = CombinedGraphIndex([])
2389
self.assertRaises(errors.KnitError, KnitGraphIndex, index,
2390
deltas=True, parents=False)
2392
def two_graph_index(self, catch_adds=False):
2393
"""Build a two-graph index.
2395
:param deltas: If true, use underlying indices with two node-ref
2396
lists and 'parent' set to a delta-compressed against tail.
2398
# put several versions in the index.
2399
index1 = self.make_g_index('1', 0, [
2400
(('tip', ), 'N0 100'),
2402
index2 = self.make_g_index('2', 0, [
2403
(('parent', ), ' 100 78'),
2404
(('separate', ), '')])
2405
combined_index = CombinedGraphIndex([index1, index2])
2407
self.combined_index = combined_index
2408
self.caught_entries = []
2409
add_callback = self.catch_add
2412
return KnitGraphIndex(combined_index, parents=False,
2413
add_callback=add_callback)
2415
def test_get_graph(self):
2416
index = self.two_graph_index()
2417
self.assertEqual(set([
2422
]), set(index.get_graph()))
2424
def test_get_ancestry(self):
2425
# with no parents, ancestry is always just the key.
2426
index = self.two_graph_index()
2427
self.assertEqual([], index.get_ancestry([]))
2428
self.assertEqual(['separate'], index.get_ancestry(['separate']))
2429
self.assertEqual(['tail'], index.get_ancestry(['tail']))
2430
self.assertEqual(['parent'], index.get_ancestry(['parent']))
2431
self.assertEqual(['tip'], index.get_ancestry(['tip']))
2432
self.assertTrue(index.get_ancestry(['tip', 'separate']) in
2433
(['tip', 'separate'],
2434
['separate', 'tip'],
2436
# asking for a ghost makes it go boom.
2437
self.assertRaises(errors.RevisionNotPresent, index.get_ancestry, ['ghost'])
2439
def test_get_ancestry_with_ghosts(self):
2440
index = self.two_graph_index()
2441
self.assertEqual([], index.get_ancestry_with_ghosts([]))
2442
self.assertEqual(['separate'], index.get_ancestry_with_ghosts(['separate']))
2443
self.assertEqual(['tail'], index.get_ancestry_with_ghosts(['tail']))
2444
self.assertEqual(['parent'], index.get_ancestry_with_ghosts(['parent']))
2445
self.assertEqual(['tip'], index.get_ancestry_with_ghosts(['tip']))
2446
self.assertTrue(index.get_ancestry_with_ghosts(['tip', 'separate']) in
2447
(['tip', 'separate'],
2448
['separate', 'tip'],
2450
# asking for a ghost makes it go boom.
2451
self.assertRaises(errors.RevisionNotPresent, index.get_ancestry_with_ghosts, ['ghost'])
2453
def test_num_versions(self):
2454
index = self.two_graph_index()
2455
self.assertEqual(4, index.num_versions())
2457
def test_get_versions(self):
2458
index = self.two_graph_index()
2459
self.assertEqual(set(['tail', 'tip', 'parent', 'separate']),
2460
set(index.get_versions()))
2462
def test_has_version(self):
2463
index = self.two_graph_index()
2464
self.assertTrue(index.has_version('tail'))
2465
self.assertFalse(index.has_version('ghost'))
2467
def test_get_position(self):
2468
index = self.two_graph_index()
2469
self.assertEqual((index._graph_index._indices[0], 0, 100), index.get_position('tip'))
2470
self.assertEqual((index._graph_index._indices[1], 100, 78), index.get_position('parent'))
2472
def test_get_method(self):
2473
index = self.two_graph_index()
2474
self.assertEqual('fulltext', index.get_method('tip'))
2475
self.assertEqual(['fulltext'], index.get_options('parent'))
2477
def test_get_options(self):
2478
index = self.two_graph_index()
2479
self.assertEqual(['fulltext', 'no-eol'], index.get_options('tip'))
2480
self.assertEqual(['fulltext'], index.get_options('parent'))
2482
def test_get_parents(self):
2483
index = self.two_graph_index()
2484
self.assertEqual((), index.get_parents('parent'))
2485
# and errors on ghosts.
2486
self.assertRaises(errors.RevisionNotPresent,
2487
index.get_parents, 'ghost')
2489
def test_get_parents_with_ghosts(self):
2490
index = self.two_graph_index()
2491
self.assertEqual((), index.get_parents_with_ghosts('parent'))
2492
# and errors on ghosts.
2493
self.assertRaises(errors.RevisionNotPresent,
2494
index.get_parents_with_ghosts, 'ghost')
2496
def test_check_versions_present(self):
2497
index = self.two_graph_index()
2498
self.assertRaises(RevisionNotPresent, index.check_versions_present,
2500
self.assertRaises(RevisionNotPresent, index.check_versions_present,
2501
['tail', 'missing'])
2502
index.check_versions_present(['tail', 'separate'])
2504
def catch_add(self, entries):
2505
self.caught_entries.append(entries)
2507
def test_add_no_callback_errors(self):
2508
index = self.two_graph_index()
2509
self.assertRaises(errors.ReadOnlyError, index.add_version,
2510
'new', 'fulltext,no-eol', (None, 50, 60), ['separate'])
2512
def test_add_version_smoke(self):
2513
index = self.two_graph_index(catch_adds=True)
2514
index.add_version('new', 'fulltext,no-eol', (None, 50, 60), [])
2515
self.assertEqual([[(('new', ), 'N50 60')]],
2516
self.caught_entries)
2518
def test_add_version_delta_not_delta_index(self):
2519
index = self.two_graph_index(catch_adds=True)
2520
self.assertRaises(errors.KnitCorrupt, index.add_version,
2521
'new', 'no-eol,line-delta', (None, 0, 100), [])
2522
self.assertEqual([], self.caught_entries)
2524
def test_add_version_same_dup(self):
2525
index = self.two_graph_index(catch_adds=True)
2526
# options can be spelt two different ways
2527
index.add_version('tip', 'fulltext,no-eol', (None, 0, 100), [])
2528
index.add_version('tip', 'no-eol,fulltext', (None, 0, 100), [])
2529
# but neither should have added data.
2530
self.assertEqual([[], []], self.caught_entries)
2532
def test_add_version_different_dup(self):
2533
index = self.two_graph_index(catch_adds=True)
2535
self.assertRaises(errors.KnitCorrupt, index.add_version,
2536
'tip', 'no-eol,line-delta', (None, 0, 100), [])
2537
self.assertRaises(errors.KnitCorrupt, index.add_version,
2538
'tip', 'line-delta,no-eol', (None, 0, 100), [])
2539
self.assertRaises(errors.KnitCorrupt, index.add_version,
2540
'tip', 'fulltext', (None, 0, 100), [])
2542
self.assertRaises(errors.KnitCorrupt, index.add_version,
2543
'tip', 'fulltext,no-eol', (None, 50, 100), [])
2544
self.assertRaises(errors.KnitCorrupt, index.add_version,
2545
'tip', 'fulltext,no-eol', (None, 0, 1000), [])
2547
self.assertRaises(errors.KnitCorrupt, index.add_version,
2548
'tip', 'fulltext,no-eol', (None, 0, 100), ['parent'])
2549
self.assertEqual([], self.caught_entries)
2551
def test_add_versions(self):
2552
index = self.two_graph_index(catch_adds=True)
2553
index.add_versions([
2554
('new', 'fulltext,no-eol', (None, 50, 60), []),
2555
('new2', 'fulltext', (None, 0, 6), []),
2557
self.assertEqual([(('new', ), 'N50 60'), (('new2', ), ' 0 6')],
2558
sorted(self.caught_entries[0]))
2559
self.assertEqual(1, len(self.caught_entries))
2561
def test_add_versions_delta_not_delta_index(self):
2562
index = self.two_graph_index(catch_adds=True)
2563
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2564
[('new', 'no-eol,line-delta', (None, 0, 100), ['parent'])])
2565
self.assertEqual([], self.caught_entries)
2567
def test_add_versions_parents_not_parents_index(self):
2568
index = self.two_graph_index(catch_adds=True)
2569
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2570
[('new', 'no-eol,fulltext', (None, 0, 100), ['parent'])])
2571
self.assertEqual([], self.caught_entries)
2573
def test_add_versions_same_dup(self):
2574
index = self.two_graph_index(catch_adds=True)
2575
# options can be spelt two different ways
2576
index.add_versions([('tip', 'fulltext,no-eol', (None, 0, 100), [])])
2577
index.add_versions([('tip', 'no-eol,fulltext', (None, 0, 100), [])])
2578
# but neither should have added data.
2579
self.assertEqual([[], []], self.caught_entries)
2581
def test_add_versions_different_dup(self):
2582
index = self.two_graph_index(catch_adds=True)
2584
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2585
[('tip', 'no-eol,line-delta', (None, 0, 100), [])])
2586
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2587
[('tip', 'line-delta,no-eol', (None, 0, 100), [])])
2588
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2589
[('tip', 'fulltext', (None, 0, 100), [])])
2591
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2592
[('tip', 'fulltext,no-eol', (None, 50, 100), [])])
2593
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2594
[('tip', 'fulltext,no-eol', (None, 0, 1000), [])])
2596
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2597
[('tip', 'fulltext,no-eol', (None, 0, 100), ['parent'])])
2598
# change options in the second record
2599
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2600
[('tip', 'fulltext,no-eol', (None, 0, 100), []),
2601
('tip', 'no-eol,line-delta', (None, 0, 100), [])])
2602
self.assertEqual([], self.caught_entries)
2604
def test_iter_parents(self):
2605
index = self.two_graph_index()
2606
self.assertEqual(set([
2607
('tip', ()), ('tail', ()), ('parent', ()), ('separate', ())
2609
set(index.iter_parents(['tip', 'tail', 'ghost', 'parent', 'separate'])))
2610
self.assertEqual(set([('tip', ())]),
2611
set(index.iter_parents(['tip'])))
2612
self.assertEqual(set(),
2613
set(index.iter_parents([])))
added
removed
bzrlib/tests/test_knit.py
