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 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_add_versions_random_id_is_accepted(self):
820
transport = MockTransport([
823
index = self.get_knit_index(transport, "filename", "r")
826
("a", ["option"], (None, 0, 1), ["b"]),
827
("a", ["opt"], (None, 1, 2), ["c"]),
828
("b", ["option"], (None, 2, 3), ["a"])
831
def test_delay_create_and_add_versions(self):
832
transport = MockTransport()
834
index = self.get_knit_index(transport, "filename", "w",
835
create=True, file_mode="wb", create_parent_dir=True,
836
delay_create=True, dir_mode=0777)
837
self.assertEqual([], transport.calls)
840
("a", ["option"], (None, 0, 1), ["b"]),
841
("a", ["opt"], (None, 1, 2), ["c"]),
842
("b", ["option"], (None, 2, 3), ["a"])
844
name, (filename, f), kwargs = transport.calls.pop(0)
845
self.assertEqual("put_file_non_atomic", name)
847
{"dir_mode": 0777, "create_parent_dir": True, "mode": "wb"},
849
self.assertEqual("filename", filename)
852
"\na option 0 1 .b :"
854
"\nb option 2 3 0 :",
857
def test_has_version(self):
858
transport = MockTransport([
862
index = self.get_knit_index(transport, "filename", "r")
864
self.assertTrue(index.has_version("a"))
865
self.assertFalse(index.has_version("b"))
867
def test_get_position(self):
868
transport = MockTransport([
873
index = self.get_knit_index(transport, "filename", "r")
875
self.assertEqual((None, 0, 1), index.get_position("a"))
876
self.assertEqual((None, 1, 2), index.get_position("b"))
878
def test_get_method(self):
879
transport = MockTransport([
881
"a fulltext,unknown 0 1 :",
882
"b unknown,line-delta 1 2 :",
885
index = self.get_knit_index(transport, "filename", "r")
887
self.assertEqual("fulltext", index.get_method("a"))
888
self.assertEqual("line-delta", index.get_method("b"))
889
self.assertRaises(errors.KnitIndexUnknownMethod, index.get_method, "c")
891
def test_get_options(self):
892
transport = MockTransport([
897
index = self.get_knit_index(transport, "filename", "r")
899
self.assertEqual(["opt1"], index.get_options("a"))
900
self.assertEqual(["opt2", "opt3"], index.get_options("b"))
902
def test_get_parents(self):
903
transport = MockTransport([
906
"b option 1 2 0 .c :",
907
"c option 1 2 1 0 .e :"
909
index = self.get_knit_index(transport, "filename", "r")
911
self.assertEqual([], index.get_parents("a"))
912
self.assertEqual(["a", "c"], index.get_parents("b"))
913
self.assertEqual(["b", "a"], index.get_parents("c"))
915
def test_get_parents_with_ghosts(self):
916
transport = MockTransport([
919
"b option 1 2 0 .c :",
920
"c option 1 2 1 0 .e :"
922
index = self.get_knit_index(transport, "filename", "r")
924
self.assertEqual([], index.get_parents_with_ghosts("a"))
925
self.assertEqual(["a", "c"], index.get_parents_with_ghosts("b"))
926
self.assertEqual(["b", "a", "e"],
927
index.get_parents_with_ghosts("c"))
929
def test_check_versions_present(self):
930
transport = MockTransport([
935
index = self.get_knit_index(transport, "filename", "r")
937
check = index.check_versions_present
943
self.assertRaises(RevisionNotPresent, check, ["c"])
944
self.assertRaises(RevisionNotPresent, check, ["a", "b", "c"])
946
def test_impossible_parent(self):
947
"""Test we get KnitCorrupt if the parent couldn't possibly exist."""
948
transport = MockTransport([
951
"b option 0 1 4 :" # We don't have a 4th record
954
self.assertRaises(errors.KnitCorrupt,
955
self.get_knit_index, transport, 'filename', 'r')
957
if (str(e) == ('exceptions must be strings, classes, or instances,'
958
' not exceptions.IndexError')
959
and sys.version_info[0:2] >= (2,5)):
960
self.knownFailure('Pyrex <0.9.5 fails with TypeError when'
961
' raising new style exceptions with python'
966
def test_corrupted_parent(self):
967
transport = MockTransport([
971
"c option 0 1 1v :", # Can't have a parent of '1v'
974
self.assertRaises(errors.KnitCorrupt,
975
self.get_knit_index, transport, 'filename', 'r')
977
if (str(e) == ('exceptions must be strings, classes, or instances,'
978
' not exceptions.ValueError')
979
and sys.version_info[0:2] >= (2,5)):
980
self.knownFailure('Pyrex <0.9.5 fails with TypeError when'
981
' raising new style exceptions with python'
986
def test_corrupted_parent_in_list(self):
987
transport = MockTransport([
991
"c option 0 1 1 v :", # Can't have a parent of 'v'
994
self.assertRaises(errors.KnitCorrupt,
995
self.get_knit_index, transport, 'filename', 'r')
997
if (str(e) == ('exceptions must be strings, classes, or instances,'
998
' not exceptions.ValueError')
999
and sys.version_info[0:2] >= (2,5)):
1000
self.knownFailure('Pyrex <0.9.5 fails with TypeError when'
1001
' raising new style exceptions with python'
1006
def test_invalid_position(self):
1007
transport = MockTransport([
1012
self.assertRaises(errors.KnitCorrupt,
1013
self.get_knit_index, transport, 'filename', 'r')
1014
except TypeError, e:
1015
if (str(e) == ('exceptions must be strings, classes, or instances,'
1016
' not exceptions.ValueError')
1017
and sys.version_info[0:2] >= (2,5)):
1018
self.knownFailure('Pyrex <0.9.5 fails with TypeError when'
1019
' raising new style exceptions with python'
1024
def test_invalid_size(self):
1025
transport = MockTransport([
1030
self.assertRaises(errors.KnitCorrupt,
1031
self.get_knit_index, transport, 'filename', 'r')
1032
except TypeError, e:
1033
if (str(e) == ('exceptions must be strings, classes, or instances,'
1034
' not exceptions.ValueError')
1035
and sys.version_info[0:2] >= (2,5)):
1036
self.knownFailure('Pyrex <0.9.5 fails with TypeError when'
1037
' raising new style exceptions with python'
1042
def test_short_line(self):
1043
transport = MockTransport([
1046
"b option 10 10 0", # This line isn't terminated, ignored
1048
index = self.get_knit_index(transport, "filename", "r")
1049
self.assertEqual(['a'], index.get_versions())
1051
def test_skip_incomplete_record(self):
1052
# A line with bogus data should just be skipped
1053
transport = MockTransport([
1056
"b option 10 10 0", # This line isn't terminated, ignored
1057
"c option 20 10 0 :", # Properly terminated, and starts with '\n'
1059
index = self.get_knit_index(transport, "filename", "r")
1060
self.assertEqual(['a', 'c'], index.get_versions())
1062
def test_trailing_characters(self):
1063
# A line with bogus data should just be skipped
1064
transport = MockTransport([
1067
"b option 10 10 0 :a", # This line has extra trailing characters
1068
"c option 20 10 0 :", # Properly terminated, and starts with '\n'
1070
index = self.get_knit_index(transport, "filename", "r")
1071
self.assertEqual(['a', 'c'], index.get_versions())
1074
class LowLevelKnitIndexTests_c(LowLevelKnitIndexTests):
1076
_test_needs_features = [CompiledKnitFeature]
1078
def get_knit_index(self, *args, **kwargs):
1079
orig = knit._load_data
1081
knit._load_data = orig
1082
self.addCleanup(reset)
1083
from bzrlib._knit_load_data_c import _load_data_c
1084
knit._load_data = _load_data_c
1085
return _KnitIndex(*args, **kwargs)
1089
class KnitTests(TestCaseWithTransport):
1090
"""Class containing knit test helper routines."""
1092
def make_test_knit(self, annotate=False, delay_create=False, index=None,
1095
factory = KnitPlainFactory()
1098
return KnitVersionedFile(name, get_transport('.'), access_mode='w',
1099
factory=factory, create=True,
1100
delay_create=delay_create, index=index)
1102
def assertRecordContentEqual(self, knit, version_id, candidate_content):
1103
"""Assert that some raw record content matches the raw record content
1104
for a particular version_id in the given knit.
1106
index_memo = knit._index.get_position(version_id)
1107
record = (version_id, index_memo)
1108
[(_, expected_content)] = list(knit._data.read_records_iter_raw([record]))
1109
self.assertEqual(expected_content, candidate_content)
1112
class BasicKnitTests(KnitTests):
1114
def add_stock_one_and_one_a(self, k):
1115
k.add_lines('text-1', [], split_lines(TEXT_1))
1116
k.add_lines('text-1a', ['text-1'], split_lines(TEXT_1A))
1118
def test_knit_constructor(self):
1119
"""Construct empty k"""
1120
self.make_test_knit()
1122
def test_make_explicit_index(self):
1123
"""We can supply an index to use."""
1124
knit = KnitVersionedFile('test', get_transport('.'),
1125
index='strangelove')
1126
self.assertEqual(knit._index, 'strangelove')
1128
def test_knit_add(self):
1129
"""Store one text in knit and retrieve"""
1130
k = self.make_test_knit()
1131
k.add_lines('text-1', [], split_lines(TEXT_1))
1132
self.assertTrue(k.has_version('text-1'))
1133
self.assertEqualDiff(''.join(k.get_lines('text-1')), TEXT_1)
1135
def test_knit_reload(self):
1136
# test that the content in a reloaded knit is correct
1137
k = self.make_test_knit()
1138
k.add_lines('text-1', [], split_lines(TEXT_1))
1140
k2 = KnitVersionedFile('test', get_transport('.'), access_mode='r', factory=KnitPlainFactory(), create=True)
1141
self.assertTrue(k2.has_version('text-1'))
1142
self.assertEqualDiff(''.join(k2.get_lines('text-1')), TEXT_1)
1144
def test_knit_several(self):
1145
"""Store several texts in a knit"""
1146
k = self.make_test_knit()
1147
k.add_lines('text-1', [], split_lines(TEXT_1))
1148
k.add_lines('text-2', [], split_lines(TEXT_2))
1149
self.assertEqualDiff(''.join(k.get_lines('text-1')), TEXT_1)
1150
self.assertEqualDiff(''.join(k.get_lines('text-2')), TEXT_2)
1152
def test_repeated_add(self):
1153
"""Knit traps attempt to replace existing version"""
1154
k = self.make_test_knit()
1155
k.add_lines('text-1', [], split_lines(TEXT_1))
1156
self.assertRaises(RevisionAlreadyPresent,
1158
'text-1', [], split_lines(TEXT_1))
1160
def test_empty(self):
1161
k = self.make_test_knit(True)
1162
k.add_lines('text-1', [], [])
1163
self.assertEquals(k.get_lines('text-1'), [])
1165
def test_incomplete(self):
1166
"""Test if texts without a ending line-end can be inserted and
1168
k = KnitVersionedFile('test', get_transport('.'), delta=False, create=True)
1169
k.add_lines('text-1', [], ['a\n', 'b' ])
1170
k.add_lines('text-2', ['text-1'], ['a\rb\n', 'b\n'])
1171
# reopening ensures maximum room for confusion
1172
k = KnitVersionedFile('test', get_transport('.'), delta=False, create=True)
1173
self.assertEquals(k.get_lines('text-1'), ['a\n', 'b' ])
1174
self.assertEquals(k.get_lines('text-2'), ['a\rb\n', 'b\n'])
1176
def test_delta(self):
1177
"""Expression of knit delta as lines"""
1178
k = self.make_test_knit()
1179
td = list(line_delta(TEXT_1.splitlines(True),
1180
TEXT_1A.splitlines(True)))
1181
self.assertEqualDiff(''.join(td), delta_1_1a)
1182
out = apply_line_delta(TEXT_1.splitlines(True), td)
1183
self.assertEqualDiff(''.join(out), TEXT_1A)
1185
def test_add_with_parents(self):
1186
"""Store in knit with parents"""
1187
k = self.make_test_knit()
1188
self.add_stock_one_and_one_a(k)
1189
self.assertEquals(k.get_parents('text-1'), [])
1190
self.assertEquals(k.get_parents('text-1a'), ['text-1'])
1192
def test_ancestry(self):
1193
"""Store in knit with parents"""
1194
k = self.make_test_knit()
1195
self.add_stock_one_and_one_a(k)
1196
self.assertEquals(set(k.get_ancestry(['text-1a'])), set(['text-1a', 'text-1']))
1198
def test_add_delta(self):
1199
"""Store in knit with parents"""
1200
k = KnitVersionedFile('test', get_transport('.'), factory=KnitPlainFactory(),
1201
delta=True, create=True)
1202
self.add_stock_one_and_one_a(k)
1204
self.assertEqualDiff(''.join(k.get_lines('text-1a')), TEXT_1A)
1206
def test_add_delta_knit_graph_index(self):
1207
"""Does adding work with a KnitGraphIndex."""
1208
index = InMemoryGraphIndex(2)
1209
knit_index = KnitGraphIndex(index, add_callback=index.add_nodes,
1211
k = KnitVersionedFile('test', get_transport('.'),
1212
delta=True, create=True, index=knit_index)
1213
self.add_stock_one_and_one_a(k)
1215
self.assertEqualDiff(''.join(k.get_lines('text-1a')), TEXT_1A)
1216
# check the index had the right data added.
1217
self.assertEqual(set([
1218
(index, ('text-1', ), ' 0 127', ((), ())),
1219
(index, ('text-1a', ), ' 127 140', ((('text-1', ),), (('text-1', ),))),
1220
]), set(index.iter_all_entries()))
1221
# we should not have a .kndx file
1222
self.assertFalse(get_transport('.').has('test.kndx'))
1224
def test_annotate(self):
1226
k = KnitVersionedFile('knit', get_transport('.'), factory=KnitAnnotateFactory(),
1227
delta=True, create=True)
1228
self.insert_and_test_small_annotate(k)
1230
def insert_and_test_small_annotate(self, k):
1231
"""test annotation with k works correctly."""
1232
k.add_lines('text-1', [], ['a\n', 'b\n'])
1233
k.add_lines('text-2', ['text-1'], ['a\n', 'c\n'])
1235
origins = k.annotate('text-2')
1236
self.assertEquals(origins[0], ('text-1', 'a\n'))
1237
self.assertEquals(origins[1], ('text-2', 'c\n'))
1239
def test_annotate_fulltext(self):
1241
k = KnitVersionedFile('knit', get_transport('.'), factory=KnitAnnotateFactory(),
1242
delta=False, create=True)
1243
self.insert_and_test_small_annotate(k)
1245
def test_annotate_merge_1(self):
1246
k = self.make_test_knit(True)
1247
k.add_lines('text-a1', [], ['a\n', 'b\n'])
1248
k.add_lines('text-a2', [], ['d\n', 'c\n'])
1249
k.add_lines('text-am', ['text-a1', 'text-a2'], ['d\n', 'b\n'])
1250
origins = k.annotate('text-am')
1251
self.assertEquals(origins[0], ('text-a2', 'd\n'))
1252
self.assertEquals(origins[1], ('text-a1', 'b\n'))
1254
def test_annotate_merge_2(self):
1255
k = self.make_test_knit(True)
1256
k.add_lines('text-a1', [], ['a\n', 'b\n', 'c\n'])
1257
k.add_lines('text-a2', [], ['x\n', 'y\n', 'z\n'])
1258
k.add_lines('text-am', ['text-a1', 'text-a2'], ['a\n', 'y\n', 'c\n'])
1259
origins = k.annotate('text-am')
1260
self.assertEquals(origins[0], ('text-a1', 'a\n'))
1261
self.assertEquals(origins[1], ('text-a2', 'y\n'))
1262
self.assertEquals(origins[2], ('text-a1', 'c\n'))
1264
def test_annotate_merge_9(self):
1265
k = self.make_test_knit(True)
1266
k.add_lines('text-a1', [], ['a\n', 'b\n', 'c\n'])
1267
k.add_lines('text-a2', [], ['x\n', 'y\n', 'z\n'])
1268
k.add_lines('text-am', ['text-a1', 'text-a2'], ['k\n', 'y\n', 'c\n'])
1269
origins = k.annotate('text-am')
1270
self.assertEquals(origins[0], ('text-am', 'k\n'))
1271
self.assertEquals(origins[1], ('text-a2', 'y\n'))
1272
self.assertEquals(origins[2], ('text-a1', 'c\n'))
1274
def test_annotate_merge_3(self):
1275
k = self.make_test_knit(True)
1276
k.add_lines('text-a1', [], ['a\n', 'b\n', 'c\n'])
1277
k.add_lines('text-a2', [] ,['x\n', 'y\n', 'z\n'])
1278
k.add_lines('text-am', ['text-a1', 'text-a2'], ['k\n', 'y\n', 'z\n'])
1279
origins = k.annotate('text-am')
1280
self.assertEquals(origins[0], ('text-am', 'k\n'))
1281
self.assertEquals(origins[1], ('text-a2', 'y\n'))
1282
self.assertEquals(origins[2], ('text-a2', 'z\n'))
1284
def test_annotate_merge_4(self):
1285
k = self.make_test_knit(True)
1286
k.add_lines('text-a1', [], ['a\n', 'b\n', 'c\n'])
1287
k.add_lines('text-a2', [], ['x\n', 'y\n', 'z\n'])
1288
k.add_lines('text-a3', ['text-a1'], ['a\n', 'b\n', 'p\n'])
1289
k.add_lines('text-am', ['text-a2', 'text-a3'], ['a\n', 'b\n', 'z\n'])
1290
origins = k.annotate('text-am')
1291
self.assertEquals(origins[0], ('text-a1', 'a\n'))
1292
self.assertEquals(origins[1], ('text-a1', 'b\n'))
1293
self.assertEquals(origins[2], ('text-a2', 'z\n'))
1295
def test_annotate_merge_5(self):
1296
k = self.make_test_knit(True)
1297
k.add_lines('text-a1', [], ['a\n', 'b\n', 'c\n'])
1298
k.add_lines('text-a2', [], ['d\n', 'e\n', 'f\n'])
1299
k.add_lines('text-a3', [], ['x\n', 'y\n', 'z\n'])
1300
k.add_lines('text-am',
1301
['text-a1', 'text-a2', 'text-a3'],
1302
['a\n', 'e\n', 'z\n'])
1303
origins = k.annotate('text-am')
1304
self.assertEquals(origins[0], ('text-a1', 'a\n'))
1305
self.assertEquals(origins[1], ('text-a2', 'e\n'))
1306
self.assertEquals(origins[2], ('text-a3', 'z\n'))
1308
def test_annotate_file_cherry_pick(self):
1309
k = self.make_test_knit(True)
1310
k.add_lines('text-1', [], ['a\n', 'b\n', 'c\n'])
1311
k.add_lines('text-2', ['text-1'], ['d\n', 'e\n', 'f\n'])
1312
k.add_lines('text-3', ['text-2', 'text-1'], ['a\n', 'b\n', 'c\n'])
1313
origins = k.annotate('text-3')
1314
self.assertEquals(origins[0], ('text-1', 'a\n'))
1315
self.assertEquals(origins[1], ('text-1', 'b\n'))
1316
self.assertEquals(origins[2], ('text-1', 'c\n'))
1318
def _test_join_with_factories(self, k1_factory, k2_factory):
1319
k1 = KnitVersionedFile('test1', get_transport('.'), factory=k1_factory, create=True)
1320
k1.add_lines('text-a', [], ['a1\n', 'a2\n', 'a3\n'])
1321
k1.add_lines('text-b', ['text-a'], ['a1\n', 'b2\n', 'a3\n'])
1322
k1.add_lines('text-c', [], ['c1\n', 'c2\n', 'c3\n'])
1323
k1.add_lines('text-d', ['text-c'], ['c1\n', 'd2\n', 'd3\n'])
1324
k1.add_lines('text-m', ['text-b', 'text-d'], ['a1\n', 'b2\n', 'd3\n'])
1325
k2 = KnitVersionedFile('test2', get_transport('.'), factory=k2_factory, create=True)
1326
count = k2.join(k1, version_ids=['text-m'])
1327
self.assertEquals(count, 5)
1328
self.assertTrue(k2.has_version('text-a'))
1329
self.assertTrue(k2.has_version('text-c'))
1330
origins = k2.annotate('text-m')
1331
self.assertEquals(origins[0], ('text-a', 'a1\n'))
1332
self.assertEquals(origins[1], ('text-b', 'b2\n'))
1333
self.assertEquals(origins[2], ('text-d', 'd3\n'))
1335
def test_knit_join_plain_to_plain(self):
1336
"""Test joining a plain knit with a plain knit."""
1337
self._test_join_with_factories(KnitPlainFactory(), KnitPlainFactory())
1339
def test_knit_join_anno_to_anno(self):
1340
"""Test joining an annotated knit with an annotated knit."""
1341
self._test_join_with_factories(None, None)
1343
def test_knit_join_anno_to_plain(self):
1344
"""Test joining an annotated knit with a plain knit."""
1345
self._test_join_with_factories(None, KnitPlainFactory())
1347
def test_knit_join_plain_to_anno(self):
1348
"""Test joining a plain knit with an annotated knit."""
1349
self._test_join_with_factories(KnitPlainFactory(), None)
1351
def test_reannotate(self):
1352
k1 = KnitVersionedFile('knit1', get_transport('.'),
1353
factory=KnitAnnotateFactory(), create=True)
1355
k1.add_lines('text-a', [], ['a\n', 'b\n'])
1357
k1.add_lines('text-b', ['text-a'], ['a\n', 'c\n'])
1359
k2 = KnitVersionedFile('test2', get_transport('.'),
1360
factory=KnitAnnotateFactory(), create=True)
1361
k2.join(k1, version_ids=['text-b'])
1364
k1.add_lines('text-X', ['text-b'], ['a\n', 'b\n'])
1366
k2.add_lines('text-c', ['text-b'], ['z\n', 'c\n'])
1368
k2.add_lines('text-Y', ['text-b'], ['b\n', 'c\n'])
1370
# test-c will have index 3
1371
k1.join(k2, version_ids=['text-c'])
1373
lines = k1.get_lines('text-c')
1374
self.assertEquals(lines, ['z\n', 'c\n'])
1376
origins = k1.annotate('text-c')
1377
self.assertEquals(origins[0], ('text-c', 'z\n'))
1378
self.assertEquals(origins[1], ('text-b', 'c\n'))
1380
def test_get_line_delta_texts(self):
1381
"""Make sure we can call get_texts on text with reused line deltas"""
1382
k1 = KnitVersionedFile('test1', get_transport('.'),
1383
factory=KnitPlainFactory(), create=True)
1388
parents = ['%d' % (t-1)]
1389
k1.add_lines('%d' % t, parents, ['hello\n'] * t)
1390
k1.get_texts(('%d' % t) for t in range(3))
1392
def test_iter_lines_reads_in_order(self):
1393
instrumented_t = get_transport('trace+memory:///')
1394
k1 = KnitVersionedFile('id', instrumented_t, create=True, delta=True)
1395
self.assertEqual([('get', 'id.kndx',)], instrumented_t._activity)
1396
# add texts with no required ordering
1397
k1.add_lines('base', [], ['text\n'])
1398
k1.add_lines('base2', [], ['text2\n'])
1400
# clear the logged activity, but preserve the list instance in case of
1401
# clones pointing at it.
1402
del instrumented_t._activity[:]
1403
# request a last-first iteration
1404
results = list(k1.iter_lines_added_or_present_in_versions(
1407
[('readv', 'id.knit', [(0, 87), (87, 89)], False, None)],
1408
instrumented_t._activity)
1409
self.assertEqual([('text\n', 'base'), ('text2\n', 'base2')], results)
1411
def test_create_empty_annotated(self):
1412
k1 = self.make_test_knit(True)
1414
k1.add_lines('text-a', [], ['a\n', 'b\n'])
1415
k2 = k1.create_empty('t', MemoryTransport())
1416
self.assertTrue(isinstance(k2.factory, KnitAnnotateFactory))
1417
self.assertEqual(k1.delta, k2.delta)
1418
# the generic test checks for empty content and file class
1420
def test_knit_format(self):
1421
# this tests that a new knit index file has the expected content
1422
# and that is writes the data we expect as records are added.
1423
knit = self.make_test_knit(True)
1424
# Now knit files are not created until we first add data to them
1425
self.assertFileEqual("# bzr knit index 8\n", 'test.kndx')
1426
knit.add_lines_with_ghosts('revid', ['a_ghost'], ['a\n'])
1427
self.assertFileEqual(
1428
"# bzr knit index 8\n"
1430
"revid fulltext 0 84 .a_ghost :",
1432
knit.add_lines_with_ghosts('revid2', ['revid'], ['a\n'])
1433
self.assertFileEqual(
1434
"# bzr knit index 8\n"
1435
"\nrevid fulltext 0 84 .a_ghost :"
1436
"\nrevid2 line-delta 84 82 0 :",
1438
# we should be able to load this file again
1439
knit = KnitVersionedFile('test', get_transport('.'), access_mode='r')
1440
self.assertEqual(['revid', 'revid2'], knit.versions())
1441
# write a short write to the file and ensure that its ignored
1442
indexfile = file('test.kndx', 'ab')
1443
indexfile.write('\nrevid3 line-delta 166 82 1 2 3 4 5 .phwoar:demo ')
1445
# we should be able to load this file again
1446
knit = KnitVersionedFile('test', get_transport('.'), access_mode='w')
1447
self.assertEqual(['revid', 'revid2'], knit.versions())
1448
# and add a revision with the same id the failed write had
1449
knit.add_lines('revid3', ['revid2'], ['a\n'])
1450
# and when reading it revid3 should now appear.
1451
knit = KnitVersionedFile('test', get_transport('.'), access_mode='r')
1452
self.assertEqual(['revid', 'revid2', 'revid3'], knit.versions())
1453
self.assertEqual(['revid2'], knit.get_parents('revid3'))
1455
def test_delay_create(self):
1456
"""Test that passing delay_create=True creates files late"""
1457
knit = self.make_test_knit(annotate=True, delay_create=True)
1458
self.failIfExists('test.knit')
1459
self.failIfExists('test.kndx')
1460
knit.add_lines_with_ghosts('revid', ['a_ghost'], ['a\n'])
1461
self.failUnlessExists('test.knit')
1462
self.assertFileEqual(
1463
"# bzr knit index 8\n"
1465
"revid fulltext 0 84 .a_ghost :",
1468
def test_create_parent_dir(self):
1469
"""create_parent_dir can create knits in nonexistant dirs"""
1470
# Has no effect if we don't set 'delay_create'
1471
trans = get_transport('.')
1472
self.assertRaises(NoSuchFile, KnitVersionedFile, 'dir/test',
1473
trans, access_mode='w', factory=None,
1474
create=True, create_parent_dir=True)
1475
# Nothing should have changed yet
1476
knit = KnitVersionedFile('dir/test', trans, access_mode='w',
1477
factory=None, create=True,
1478
create_parent_dir=True,
1480
self.failIfExists('dir/test.knit')
1481
self.failIfExists('dir/test.kndx')
1482
self.failIfExists('dir')
1483
knit.add_lines('revid', [], ['a\n'])
1484
self.failUnlessExists('dir')
1485
self.failUnlessExists('dir/test.knit')
1486
self.assertFileEqual(
1487
"# bzr knit index 8\n"
1489
"revid fulltext 0 84 :",
1492
def test_create_mode_700(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', 0700)
1505
self.assertTransportMode(trans, 'dir/test.knit', 0600)
1506
self.assertTransportMode(trans, 'dir/test.kndx', 0600)
1508
def test_create_mode_770(self):
1509
trans = get_transport('.')
1510
if not trans._can_roundtrip_unix_modebits():
1511
# Can't roundtrip, so no need to run this test
1513
knit = KnitVersionedFile('dir/test', trans, access_mode='w',
1514
factory=None, create=True,
1515
create_parent_dir=True,
1519
knit.add_lines('revid', [], ['a\n'])
1520
self.assertTransportMode(trans, 'dir', 0770)
1521
self.assertTransportMode(trans, 'dir/test.knit', 0660)
1522
self.assertTransportMode(trans, 'dir/test.kndx', 0660)
1524
def test_create_mode_777(self):
1525
trans = get_transport('.')
1526
if not trans._can_roundtrip_unix_modebits():
1527
# Can't roundtrip, so no need to run this test
1529
knit = KnitVersionedFile('dir/test', trans, access_mode='w',
1530
factory=None, create=True,
1531
create_parent_dir=True,
1535
knit.add_lines('revid', [], ['a\n'])
1536
self.assertTransportMode(trans, 'dir', 0777)
1537
self.assertTransportMode(trans, 'dir/test.knit', 0666)
1538
self.assertTransportMode(trans, 'dir/test.kndx', 0666)
1540
def test_plan_merge(self):
1541
my_knit = self.make_test_knit(annotate=True)
1542
my_knit.add_lines('text1', [], split_lines(TEXT_1))
1543
my_knit.add_lines('text1a', ['text1'], split_lines(TEXT_1A))
1544
my_knit.add_lines('text1b', ['text1'], split_lines(TEXT_1B))
1545
plan = list(my_knit.plan_merge('text1a', 'text1b'))
1546
for plan_line, expected_line in zip(plan, AB_MERGE):
1547
self.assertEqual(plan_line, expected_line)
1549
def test_get_stream_empty(self):
1550
"""Get a data stream for an empty knit file."""
1551
k1 = self.make_test_knit()
1552
format, data_list, reader_callable = k1.get_data_stream([])
1553
self.assertEqual('knit-plain', format)
1554
self.assertEqual([], data_list)
1555
content = reader_callable(None)
1556
self.assertEqual('', content)
1557
self.assertIsInstance(content, str)
1559
def test_get_stream_one_version(self):
1560
"""Get a data stream for a single record out of a knit containing just
1563
k1 = self.make_test_knit()
1565
('text-a', [], TEXT_1),
1567
expected_data_list = [
1568
# version, options, length, parents
1569
('text-a', ['fulltext'], 122, []),
1571
for version_id, parents, lines in test_data:
1572
k1.add_lines(version_id, parents, split_lines(lines))
1574
format, data_list, reader_callable = k1.get_data_stream(['text-a'])
1575
self.assertEqual('knit-plain', format)
1576
self.assertEqual(expected_data_list, data_list)
1577
# There's only one record in the knit, so the content should be the
1578
# entire knit data file's contents.
1579
self.assertEqual(k1.transport.get_bytes(k1._data._access._filename),
1580
reader_callable(None))
1582
def test_get_stream_get_one_version_of_many(self):
1583
"""Get a data stream for just one version out of a knit containing many
1586
k1 = self.make_test_knit()
1587
# Insert the same data as test_knit_join, as they seem to cover a range
1588
# of cases (no parents, one parent, multiple parents).
1590
('text-a', [], TEXT_1),
1591
('text-b', ['text-a'], TEXT_1),
1592
('text-c', [], TEXT_1),
1593
('text-d', ['text-c'], TEXT_1),
1594
('text-m', ['text-b', 'text-d'], TEXT_1),
1596
expected_data_list = [
1597
# version, options, length, parents
1598
('text-m', ['line-delta'], 84, ['text-b', 'text-d']),
1600
for version_id, parents, lines in test_data:
1601
k1.add_lines(version_id, parents, split_lines(lines))
1603
format, data_list, reader_callable = k1.get_data_stream(['text-m'])
1604
self.assertEqual('knit-plain', format)
1605
self.assertEqual(expected_data_list, data_list)
1606
self.assertRecordContentEqual(k1, 'text-m', reader_callable(None))
1608
def test_get_data_stream_unordered_index(self):
1609
"""Get a data stream when the knit index reports versions out of order.
1611
https://bugs.launchpad.net/bzr/+bug/164637
1613
k1 = self.make_test_knit()
1615
('text-a', [], TEXT_1),
1616
('text-b', ['text-a'], TEXT_1),
1617
('text-c', [], TEXT_1),
1618
('text-d', ['text-c'], TEXT_1),
1619
('text-m', ['text-b', 'text-d'], TEXT_1),
1621
for version_id, parents, lines in test_data:
1622
k1.add_lines(version_id, parents, split_lines(lines))
1623
# monkey-patch versions method to return out of order, as if coming
1624
# from multiple independently indexed packs
1625
original_versions = k1.versions
1626
k1.versions = lambda: reversed(original_versions())
1627
expected_data_list = [
1628
('text-a', ['fulltext'], 122, []),
1629
('text-b', ['line-delta'], 84, ['text-a'])]
1630
# now check the fulltext is first and the delta second
1631
format, data_list, _ = k1.get_data_stream(['text-a', 'text-b'])
1632
self.assertEqual('knit-plain', format)
1633
self.assertEqual(expected_data_list, data_list)
1634
# and that's true if we ask for them in the opposite order too
1635
format, data_list, _ = k1.get_data_stream(['text-b', 'text-a'])
1636
self.assertEqual(expected_data_list, data_list)
1637
# also try requesting more versions
1638
format, data_list, _ = k1.get_data_stream([
1639
'text-m', 'text-b', 'text-a'])
1641
('text-a', ['fulltext'], 122, []),
1642
('text-b', ['line-delta'], 84, ['text-a']),
1643
('text-m', ['line-delta'], 84, ['text-b', 'text-d']),
1646
def test_get_stream_ghost_parent(self):
1647
"""Get a data stream for a version with a ghost parent."""
1648
k1 = self.make_test_knit()
1650
k1.add_lines('text-a', [], split_lines(TEXT_1))
1651
k1.add_lines_with_ghosts('text-b', ['text-a', 'text-ghost'],
1652
split_lines(TEXT_1))
1654
expected_data_list = [
1655
# version, options, length, parents
1656
('text-b', ['line-delta'], 84, ['text-a', 'text-ghost']),
1659
format, data_list, reader_callable = k1.get_data_stream(['text-b'])
1660
self.assertEqual('knit-plain', format)
1661
self.assertEqual(expected_data_list, data_list)
1662
self.assertRecordContentEqual(k1, 'text-b', reader_callable(None))
1664
def test_get_stream_get_multiple_records(self):
1665
"""Get a stream for multiple records of a knit."""
1666
k1 = self.make_test_knit()
1667
# Insert the same data as test_knit_join, as they seem to cover a range
1668
# of cases (no parents, one parent, multiple parents).
1670
('text-a', [], TEXT_1),
1671
('text-b', ['text-a'], TEXT_1),
1672
('text-c', [], TEXT_1),
1673
('text-d', ['text-c'], TEXT_1),
1674
('text-m', ['text-b', 'text-d'], TEXT_1),
1676
for version_id, parents, lines in test_data:
1677
k1.add_lines(version_id, parents, split_lines(lines))
1679
# This test is actually a bit strict as the order in which they're
1680
# returned is not defined. This matches the current (deterministic)
1682
expected_data_list = [
1683
# version, options, length, parents
1684
('text-d', ['line-delta'], 84, ['text-c']),
1685
('text-b', ['line-delta'], 84, ['text-a']),
1687
# Note that even though we request the revision IDs in a particular
1688
# order, the data stream may return them in any order it likes. In this
1689
# case, they'll be in the order they were inserted into the knit.
1690
format, data_list, reader_callable = k1.get_data_stream(
1691
['text-d', 'text-b'])
1692
self.assertEqual('knit-plain', format)
1693
self.assertEqual(expected_data_list, data_list)
1694
# must match order they're returned
1695
self.assertRecordContentEqual(k1, 'text-d', reader_callable(84))
1696
self.assertRecordContentEqual(k1, 'text-b', reader_callable(84))
1697
self.assertEqual('', reader_callable(None),
1698
"There should be no more bytes left to read.")
1700
def test_get_stream_all(self):
1701
"""Get a data stream for all the records in a knit.
1703
This exercises fulltext records, line-delta records, records with
1704
various numbers of parents, and reading multiple records out of the
1705
callable. These cases ought to all be exercised individually by the
1706
other test_get_stream_* tests; this test is basically just paranoia.
1708
k1 = self.make_test_knit()
1709
# Insert the same data as test_knit_join, as they seem to cover a range
1710
# of cases (no parents, one parent, multiple parents).
1712
('text-a', [], TEXT_1),
1713
('text-b', ['text-a'], TEXT_1),
1714
('text-c', [], TEXT_1),
1715
('text-d', ['text-c'], TEXT_1),
1716
('text-m', ['text-b', 'text-d'], TEXT_1),
1718
for version_id, parents, lines in test_data:
1719
k1.add_lines(version_id, parents, split_lines(lines))
1721
# This test is actually a bit strict as the order in which they're
1722
# returned is not defined. This matches the current (deterministic)
1724
expected_data_list = [
1725
# version, options, length, parents
1726
('text-a', ['fulltext'], 122, []),
1727
('text-b', ['line-delta'], 84, ['text-a']),
1728
('text-m', ['line-delta'], 84, ['text-b', 'text-d']),
1729
('text-c', ['fulltext'], 121, []),
1730
('text-d', ['line-delta'], 84, ['text-c']),
1732
format, data_list, reader_callable = k1.get_data_stream(
1733
['text-a', 'text-b', 'text-c', 'text-d', 'text-m'])
1734
self.assertEqual('knit-plain', format)
1735
self.assertEqual(expected_data_list, data_list)
1736
for version_id, options, length, parents in expected_data_list:
1737
bytes = reader_callable(length)
1738
self.assertRecordContentEqual(k1, version_id, bytes)
1740
def assertKnitFilesEqual(self, knit1, knit2):
1741
"""Assert that the contents of the index and data files of two knits are
1745
knit1.transport.get_bytes(knit1._data._access._filename),
1746
knit2.transport.get_bytes(knit2._data._access._filename))
1748
knit1.transport.get_bytes(knit1._index._filename),
1749
knit2.transport.get_bytes(knit2._index._filename))
1751
def test_insert_data_stream_empty(self):
1752
"""Inserting a data stream with no records should not put any data into
1755
k1 = self.make_test_knit()
1756
k1.insert_data_stream(
1757
(k1.get_format_signature(), [], lambda ignored: ''))
1758
self.assertEqual('', k1.transport.get_bytes(k1._data._access._filename),
1759
"The .knit should be completely empty.")
1760
self.assertEqual(k1._index.HEADER,
1761
k1.transport.get_bytes(k1._index._filename),
1762
"The .kndx should have nothing apart from the header.")
1764
def test_insert_data_stream_one_record(self):
1765
"""Inserting a data stream with one record from a knit with one record
1766
results in byte-identical files.
1768
source = self.make_test_knit(name='source')
1769
source.add_lines('text-a', [], split_lines(TEXT_1))
1770
data_stream = source.get_data_stream(['text-a'])
1772
target = self.make_test_knit(name='target')
1773
target.insert_data_stream(data_stream)
1775
self.assertKnitFilesEqual(source, target)
1777
def test_insert_data_stream_records_already_present(self):
1778
"""Insert a data stream where some records are alreday present in the
1779
target, and some not. Only the new records are inserted.
1781
source = self.make_test_knit(name='source')
1782
target = self.make_test_knit(name='target')
1783
# Insert 'text-a' into both source and target
1784
source.add_lines('text-a', [], split_lines(TEXT_1))
1785
target.insert_data_stream(source.get_data_stream(['text-a']))
1786
# Insert 'text-b' into just the source.
1787
source.add_lines('text-b', ['text-a'], split_lines(TEXT_1))
1788
# Get a data stream of both text-a and text-b, and insert it.
1789
data_stream = source.get_data_stream(['text-a', 'text-b'])
1790
target.insert_data_stream(data_stream)
1791
# The source and target will now be identical. This means the text-a
1792
# record was not added a second time.
1793
self.assertKnitFilesEqual(source, target)
1795
def test_insert_data_stream_multiple_records(self):
1796
"""Inserting a data stream of all records from a knit with multiple
1797
records results in byte-identical files.
1799
source = self.make_test_knit(name='source')
1800
source.add_lines('text-a', [], split_lines(TEXT_1))
1801
source.add_lines('text-b', ['text-a'], split_lines(TEXT_1))
1802
source.add_lines('text-c', [], split_lines(TEXT_1))
1803
data_stream = source.get_data_stream(['text-a', 'text-b', 'text-c'])
1805
target = self.make_test_knit(name='target')
1806
target.insert_data_stream(data_stream)
1808
self.assertKnitFilesEqual(source, target)
1810
def test_insert_data_stream_ghost_parent(self):
1811
"""Insert a data stream with a record that has a ghost parent."""
1812
# Make a knit with a record, text-a, that has a ghost parent.
1813
source = self.make_test_knit(name='source')
1814
source.add_lines_with_ghosts('text-a', ['text-ghost'],
1815
split_lines(TEXT_1))
1816
data_stream = source.get_data_stream(['text-a'])
1818
target = self.make_test_knit(name='target')
1819
target.insert_data_stream(data_stream)
1821
self.assertKnitFilesEqual(source, target)
1823
# The target knit object is in a consistent state, i.e. the record we
1824
# just added is immediately visible.
1825
self.assertTrue(target.has_version('text-a'))
1826
self.assertTrue(target.has_ghost('text-ghost'))
1827
self.assertEqual(split_lines(TEXT_1), target.get_lines('text-a'))
1829
def test_insert_data_stream_inconsistent_version_lines(self):
1830
"""Inserting a data stream which has different content for a version_id
1831
than already exists in the knit will raise KnitCorrupt.
1833
source = self.make_test_knit(name='source')
1834
target = self.make_test_knit(name='target')
1835
# Insert a different 'text-a' into both source and target
1836
source.add_lines('text-a', [], split_lines(TEXT_1))
1837
target.add_lines('text-a', [], split_lines(TEXT_2))
1838
# Insert a data stream with conflicting content into the target
1839
data_stream = source.get_data_stream(['text-a'])
1841
errors.KnitCorrupt, target.insert_data_stream, data_stream)
1843
def test_insert_data_stream_inconsistent_version_parents(self):
1844
"""Inserting a data stream which has different parents for a version_id
1845
than already exists in the knit will raise KnitCorrupt.
1847
source = self.make_test_knit(name='source')
1848
target = self.make_test_knit(name='target')
1849
# Insert a different 'text-a' into both source and target. They differ
1850
# only by the parents list, the content is the same.
1851
source.add_lines_with_ghosts('text-a', [], split_lines(TEXT_1))
1852
target.add_lines_with_ghosts('text-a', ['a-ghost'], split_lines(TEXT_1))
1853
# Insert a data stream with conflicting content into the target
1854
data_stream = source.get_data_stream(['text-a'])
1856
errors.KnitCorrupt, target.insert_data_stream, data_stream)
1858
def test_insert_data_stream_incompatible_format(self):
1859
"""A data stream in a different format to the target knit cannot be
1862
It will raise KnitDataStreamIncompatible.
1864
data_stream = ('fake-format-signature', [], lambda _: '')
1865
target = self.make_test_knit(name='target')
1867
errors.KnitDataStreamIncompatible,
1868
target.insert_data_stream, data_stream)
1870
# * test that a stream of "already present version, then new version"
1871
# inserts correctly.
1882
Banana cup cake recipe
1888
- self-raising flour
1892
Banana cup cake recipe
1894
- bananas (do not use plantains!!!)
1901
Banana cup cake recipe
1904
- self-raising flour
1917
AB_MERGE_TEXT="""unchanged|Banana cup cake recipe
1922
new-b|- bananas (do not use plantains!!!)
1923
unchanged|- broken tea cups
1924
new-a|- self-raising flour
1927
AB_MERGE=[tuple(l.split('|')) for l in AB_MERGE_TEXT.splitlines(True)]
1930
def line_delta(from_lines, to_lines):
1931
"""Generate line-based delta from one text to another"""
1932
s = difflib.SequenceMatcher(None, from_lines, to_lines)
1933
for op in s.get_opcodes():
1934
if op[0] == 'equal':
1936
yield '%d,%d,%d\n' % (op[1], op[2], op[4]-op[3])
1937
for i in range(op[3], op[4]):
1941
def apply_line_delta(basis_lines, delta_lines):
1942
"""Apply a line-based perfect diff
1944
basis_lines -- text to apply the patch to
1945
delta_lines -- diff instructions and content
1947
out = basis_lines[:]
1950
while i < len(delta_lines):
1952
a, b, c = map(long, l.split(','))
1954
out[offset+a:offset+b] = delta_lines[i:i+c]
1956
offset = offset + (b - a) + c
1960
class TestWeaveToKnit(KnitTests):
1962
def test_weave_to_knit_matches(self):
1963
# check that the WeaveToKnit is_compatible function
1964
# registers True for a Weave to a Knit.
1966
k = self.make_test_knit()
1967
self.failUnless(WeaveToKnit.is_compatible(w, k))
1968
self.failIf(WeaveToKnit.is_compatible(k, w))
1969
self.failIf(WeaveToKnit.is_compatible(w, w))
1970
self.failIf(WeaveToKnit.is_compatible(k, k))
1973
class TestKnitCaching(KnitTests):
1975
def create_knit(self):
1976
k = self.make_test_knit(True)
1977
k.add_lines('text-1', [], split_lines(TEXT_1))
1978
k.add_lines('text-2', [], split_lines(TEXT_2))
1981
def test_no_caching(self):
1982
k = self.create_knit()
1983
# Nothing should be cached without setting 'enable_cache'
1984
self.assertEqual({}, k._data._cache)
1986
def test_cache_data_read_raw(self):
1987
k = self.create_knit()
1989
# Now cache and read
1992
def read_one_raw(version):
1993
pos_map = k._get_components_positions([version])
1994
method, index_memo, next = pos_map[version]
1995
lst = list(k._data.read_records_iter_raw([(version, index_memo)]))
1996
self.assertEqual(1, len(lst))
1999
val = read_one_raw('text-1')
2000
self.assertEqual({'text-1':val[1]}, k._data._cache)
2003
# After clear, new reads are not cached
2004
self.assertEqual({}, k._data._cache)
2006
val2 = read_one_raw('text-1')
2007
self.assertEqual(val, val2)
2008
self.assertEqual({}, k._data._cache)
2010
def test_cache_data_read(self):
2011
k = self.create_knit()
2013
def read_one(version):
2014
pos_map = k._get_components_positions([version])
2015
method, index_memo, next = pos_map[version]
2016
lst = list(k._data.read_records_iter([(version, index_memo)]))
2017
self.assertEqual(1, len(lst))
2020
# Now cache and read
2023
val = read_one('text-2')
2024
self.assertEqual(['text-2'], k._data._cache.keys())
2025
self.assertEqual('text-2', val[0])
2026
content, digest = k._data._parse_record('text-2',
2027
k._data._cache['text-2'])
2028
self.assertEqual(content, val[1])
2029
self.assertEqual(digest, val[2])
2032
self.assertEqual({}, k._data._cache)
2034
val2 = read_one('text-2')
2035
self.assertEqual(val, val2)
2036
self.assertEqual({}, k._data._cache)
2038
def test_cache_read(self):
2039
k = self.create_knit()
2042
text = k.get_text('text-1')
2043
self.assertEqual(TEXT_1, text)
2044
self.assertEqual(['text-1'], k._data._cache.keys())
2047
self.assertEqual({}, k._data._cache)
2049
text = k.get_text('text-1')
2050
self.assertEqual(TEXT_1, text)
2051
self.assertEqual({}, k._data._cache)
2054
class TestKnitIndex(KnitTests):
2056
def test_add_versions_dictionary_compresses(self):
2057
"""Adding versions to the index should update the lookup dict"""
2058
knit = self.make_test_knit()
2060
idx.add_version('a-1', ['fulltext'], (None, 0, 0), [])
2061
self.check_file_contents('test.kndx',
2062
'# bzr knit index 8\n'
2064
'a-1 fulltext 0 0 :'
2066
idx.add_versions([('a-2', ['fulltext'], (None, 0, 0), ['a-1']),
2067
('a-3', ['fulltext'], (None, 0, 0), ['a-2']),
2069
self.check_file_contents('test.kndx',
2070
'# bzr knit index 8\n'
2072
'a-1 fulltext 0 0 :\n'
2073
'a-2 fulltext 0 0 0 :\n'
2074
'a-3 fulltext 0 0 1 :'
2076
self.assertEqual(['a-1', 'a-2', 'a-3'], idx._history)
2077
self.assertEqual({'a-1':('a-1', ['fulltext'], 0, 0, [], 0),
2078
'a-2':('a-2', ['fulltext'], 0, 0, ['a-1'], 1),
2079
'a-3':('a-3', ['fulltext'], 0, 0, ['a-2'], 2),
2082
def test_add_versions_fails_clean(self):
2083
"""If add_versions fails in the middle, it restores a pristine state.
2085
Any modifications that are made to the index are reset if all versions
2088
# This cheats a little bit by passing in a generator which will
2089
# raise an exception before the processing finishes
2090
# Other possibilities would be to have an version with the wrong number
2091
# of entries, or to make the backing transport unable to write any
2094
knit = self.make_test_knit()
2096
idx.add_version('a-1', ['fulltext'], (None, 0, 0), [])
2098
class StopEarly(Exception):
2101
def generate_failure():
2102
"""Add some entries and then raise an exception"""
2103
yield ('a-2', ['fulltext'], (None, 0, 0), ['a-1'])
2104
yield ('a-3', ['fulltext'], (None, 0, 0), ['a-2'])
2107
# Assert the pre-condition
2108
self.assertEqual(['a-1'], idx._history)
2109
self.assertEqual({'a-1':('a-1', ['fulltext'], 0, 0, [], 0)}, idx._cache)
2111
self.assertRaises(StopEarly, idx.add_versions, generate_failure())
2113
# And it shouldn't be modified
2114
self.assertEqual(['a-1'], idx._history)
2115
self.assertEqual({'a-1':('a-1', ['fulltext'], 0, 0, [], 0)}, idx._cache)
2117
def test_knit_index_ignores_empty_files(self):
2118
# There was a race condition in older bzr, where a ^C at the right time
2119
# could leave an empty .kndx file, which bzr would later claim was a
2120
# corrupted file since the header was not present. In reality, the file
2121
# just wasn't created, so it should be ignored.
2122
t = get_transport('.')
2123
t.put_bytes('test.kndx', '')
2125
knit = self.make_test_knit()
2127
def test_knit_index_checks_header(self):
2128
t = get_transport('.')
2129
t.put_bytes('test.kndx', '# not really a knit header\n\n')
2131
self.assertRaises(KnitHeaderError, self.make_test_knit)
2134
class TestGraphIndexKnit(KnitTests):
2135
"""Tests for knits using a GraphIndex rather than a KnitIndex."""
2137
def make_g_index(self, name, ref_lists=0, nodes=[]):
2138
builder = GraphIndexBuilder(ref_lists)
2139
for node, references, value in nodes:
2140
builder.add_node(node, references, value)
2141
stream = builder.finish()
2142
trans = self.get_transport()
2143
size = trans.put_file(name, stream)
2144
return GraphIndex(trans, name, size)
2146
def two_graph_index(self, deltas=False, catch_adds=False):
2147
"""Build a two-graph index.
2149
:param deltas: If true, use underlying indices with two node-ref
2150
lists and 'parent' set to a delta-compressed against tail.
2152
# build a complex graph across several indices.
2154
# delta compression inn the index
2155
index1 = self.make_g_index('1', 2, [
2156
(('tip', ), 'N0 100', ([('parent', )], [], )),
2157
(('tail', ), '', ([], []))])
2158
index2 = self.make_g_index('2', 2, [
2159
(('parent', ), ' 100 78', ([('tail', ), ('ghost', )], [('tail', )])),
2160
(('separate', ), '', ([], []))])
2162
# just blob location and graph in the index.
2163
index1 = self.make_g_index('1', 1, [
2164
(('tip', ), 'N0 100', ([('parent', )], )),
2165
(('tail', ), '', ([], ))])
2166
index2 = self.make_g_index('2', 1, [
2167
(('parent', ), ' 100 78', ([('tail', ), ('ghost', )], )),
2168
(('separate', ), '', ([], ))])
2169
combined_index = CombinedGraphIndex([index1, index2])
2171
self.combined_index = combined_index
2172
self.caught_entries = []
2173
add_callback = self.catch_add
2176
return KnitGraphIndex(combined_index, deltas=deltas,
2177
add_callback=add_callback)
2179
def test_get_graph(self):
2180
index = self.two_graph_index()
2181
self.assertEqual(set([
2182
('tip', ('parent', )),
2184
('parent', ('tail', 'ghost')),
2186
]), set(index.get_graph()))
2188
def test_get_ancestry(self):
2189
# get_ancestry is defined as eliding ghosts, not erroring.
2190
index = self.two_graph_index()
2191
self.assertEqual([], index.get_ancestry([]))
2192
self.assertEqual(['separate'], index.get_ancestry(['separate']))
2193
self.assertEqual(['tail'], index.get_ancestry(['tail']))
2194
self.assertEqual(['tail', 'parent'], index.get_ancestry(['parent']))
2195
self.assertEqual(['tail', 'parent', 'tip'], index.get_ancestry(['tip']))
2196
self.assertTrue(index.get_ancestry(['tip', 'separate']) in
2197
(['tail', 'parent', 'tip', 'separate'],
2198
['separate', 'tail', 'parent', 'tip'],
2200
# and without topo_sort
2201
self.assertEqual(set(['separate']),
2202
set(index.get_ancestry(['separate'], topo_sorted=False)))
2203
self.assertEqual(set(['tail']),
2204
set(index.get_ancestry(['tail'], topo_sorted=False)))
2205
self.assertEqual(set(['tail', 'parent']),
2206
set(index.get_ancestry(['parent'], topo_sorted=False)))
2207
self.assertEqual(set(['tail', 'parent', 'tip']),
2208
set(index.get_ancestry(['tip'], topo_sorted=False)))
2209
self.assertEqual(set(['separate', 'tail', 'parent', 'tip']),
2210
set(index.get_ancestry(['tip', 'separate'])))
2211
# asking for a ghost makes it go boom.
2212
self.assertRaises(errors.RevisionNotPresent, index.get_ancestry, ['ghost'])
2214
def test_get_ancestry_with_ghosts(self):
2215
index = self.two_graph_index()
2216
self.assertEqual([], index.get_ancestry_with_ghosts([]))
2217
self.assertEqual(['separate'], index.get_ancestry_with_ghosts(['separate']))
2218
self.assertEqual(['tail'], index.get_ancestry_with_ghosts(['tail']))
2219
self.assertTrue(index.get_ancestry_with_ghosts(['parent']) in
2220
(['tail', 'ghost', 'parent'],
2221
['ghost', 'tail', 'parent'],
2223
self.assertTrue(index.get_ancestry_with_ghosts(['tip']) in
2224
(['tail', 'ghost', 'parent', 'tip'],
2225
['ghost', 'tail', 'parent', 'tip'],
2227
self.assertTrue(index.get_ancestry_with_ghosts(['tip', 'separate']) in
2228
(['tail', 'ghost', 'parent', 'tip', 'separate'],
2229
['ghost', 'tail', 'parent', 'tip', 'separate'],
2230
['separate', 'tail', 'ghost', 'parent', 'tip'],
2231
['separate', 'ghost', 'tail', 'parent', 'tip'],
2233
# asking for a ghost makes it go boom.
2234
self.assertRaises(errors.RevisionNotPresent, index.get_ancestry_with_ghosts, ['ghost'])
2236
def test_num_versions(self):
2237
index = self.two_graph_index()
2238
self.assertEqual(4, index.num_versions())
2240
def test_get_versions(self):
2241
index = self.two_graph_index()
2242
self.assertEqual(set(['tail', 'tip', 'parent', 'separate']),
2243
set(index.get_versions()))
2245
def test_has_version(self):
2246
index = self.two_graph_index()
2247
self.assertTrue(index.has_version('tail'))
2248
self.assertFalse(index.has_version('ghost'))
2250
def test_get_position(self):
2251
index = self.two_graph_index()
2252
self.assertEqual((index._graph_index._indices[0], 0, 100), index.get_position('tip'))
2253
self.assertEqual((index._graph_index._indices[1], 100, 78), index.get_position('parent'))
2255
def test_get_method_deltas(self):
2256
index = self.two_graph_index(deltas=True)
2257
self.assertEqual('fulltext', index.get_method('tip'))
2258
self.assertEqual('line-delta', index.get_method('parent'))
2260
def test_get_method_no_deltas(self):
2261
# check that the parent-history lookup is ignored with deltas=False.
2262
index = self.two_graph_index(deltas=False)
2263
self.assertEqual('fulltext', index.get_method('tip'))
2264
self.assertEqual('fulltext', index.get_method('parent'))
2266
def test_get_options_deltas(self):
2267
index = self.two_graph_index(deltas=True)
2268
self.assertEqual(['fulltext', 'no-eol'], index.get_options('tip'))
2269
self.assertEqual(['line-delta'], index.get_options('parent'))
2271
def test_get_options_no_deltas(self):
2272
# check that the parent-history lookup is ignored with deltas=False.
2273
index = self.two_graph_index(deltas=False)
2274
self.assertEqual(['fulltext', 'no-eol'], index.get_options('tip'))
2275
self.assertEqual(['fulltext'], index.get_options('parent'))
2277
def test_get_parents(self):
2278
# get_parents ignores ghosts
2279
index = self.two_graph_index()
2280
self.assertEqual(('tail', ), index.get_parents('parent'))
2281
# and errors on ghosts.
2282
self.assertRaises(errors.RevisionNotPresent,
2283
index.get_parents, 'ghost')
2285
def test_get_parents_with_ghosts(self):
2286
index = self.two_graph_index()
2287
self.assertEqual(('tail', 'ghost'), index.get_parents_with_ghosts('parent'))
2288
# and errors on ghosts.
2289
self.assertRaises(errors.RevisionNotPresent,
2290
index.get_parents_with_ghosts, 'ghost')
2292
def test_check_versions_present(self):
2293
# ghosts should not be considered present
2294
index = self.two_graph_index()
2295
self.assertRaises(RevisionNotPresent, index.check_versions_present,
2297
self.assertRaises(RevisionNotPresent, index.check_versions_present,
2299
index.check_versions_present(['tail', 'separate'])
2301
def catch_add(self, entries):
2302
self.caught_entries.append(entries)
2304
def test_add_no_callback_errors(self):
2305
index = self.two_graph_index()
2306
self.assertRaises(errors.ReadOnlyError, index.add_version,
2307
'new', 'fulltext,no-eol', (None, 50, 60), ['separate'])
2309
def test_add_version_smoke(self):
2310
index = self.two_graph_index(catch_adds=True)
2311
index.add_version('new', 'fulltext,no-eol', (None, 50, 60), ['separate'])
2312
self.assertEqual([[(('new', ), 'N50 60', ((('separate',),),))]],
2313
self.caught_entries)
2315
def test_add_version_delta_not_delta_index(self):
2316
index = self.two_graph_index(catch_adds=True)
2317
self.assertRaises(errors.KnitCorrupt, index.add_version,
2318
'new', 'no-eol,line-delta', (None, 0, 100), ['parent'])
2319
self.assertEqual([], self.caught_entries)
2321
def test_add_version_same_dup(self):
2322
index = self.two_graph_index(catch_adds=True)
2323
# options can be spelt two different ways
2324
index.add_version('tip', 'fulltext,no-eol', (None, 0, 100), ['parent'])
2325
index.add_version('tip', 'no-eol,fulltext', (None, 0, 100), ['parent'])
2326
# but neither should have added data.
2327
self.assertEqual([[], []], self.caught_entries)
2329
def test_add_version_different_dup(self):
2330
index = self.two_graph_index(deltas=True, catch_adds=True)
2332
self.assertRaises(errors.KnitCorrupt, index.add_version,
2333
'tip', 'no-eol,line-delta', (None, 0, 100), ['parent'])
2334
self.assertRaises(errors.KnitCorrupt, index.add_version,
2335
'tip', 'line-delta,no-eol', (None, 0, 100), ['parent'])
2336
self.assertRaises(errors.KnitCorrupt, index.add_version,
2337
'tip', 'fulltext', (None, 0, 100), ['parent'])
2339
self.assertRaises(errors.KnitCorrupt, index.add_version,
2340
'tip', 'fulltext,no-eol', (None, 50, 100), ['parent'])
2341
self.assertRaises(errors.KnitCorrupt, index.add_version,
2342
'tip', 'fulltext,no-eol', (None, 0, 1000), ['parent'])
2344
self.assertRaises(errors.KnitCorrupt, index.add_version,
2345
'tip', 'fulltext,no-eol', (None, 0, 100), [])
2346
self.assertEqual([], self.caught_entries)
2348
def test_add_versions_nodeltas(self):
2349
index = self.two_graph_index(catch_adds=True)
2350
index.add_versions([
2351
('new', 'fulltext,no-eol', (None, 50, 60), ['separate']),
2352
('new2', 'fulltext', (None, 0, 6), ['new']),
2354
self.assertEqual([(('new', ), 'N50 60', ((('separate',),),)),
2355
(('new2', ), ' 0 6', ((('new',),),))],
2356
sorted(self.caught_entries[0]))
2357
self.assertEqual(1, len(self.caught_entries))
2359
def test_add_versions_deltas(self):
2360
index = self.two_graph_index(deltas=True, catch_adds=True)
2361
index.add_versions([
2362
('new', 'fulltext,no-eol', (None, 50, 60), ['separate']),
2363
('new2', 'line-delta', (None, 0, 6), ['new']),
2365
self.assertEqual([(('new', ), 'N50 60', ((('separate',),), ())),
2366
(('new2', ), ' 0 6', ((('new',),), (('new',),), ))],
2367
sorted(self.caught_entries[0]))
2368
self.assertEqual(1, len(self.caught_entries))
2370
def test_add_versions_delta_not_delta_index(self):
2371
index = self.two_graph_index(catch_adds=True)
2372
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2373
[('new', 'no-eol,line-delta', (None, 0, 100), ['parent'])])
2374
self.assertEqual([], self.caught_entries)
2376
def test_add_versions_random_id_accepted(self):
2377
index = self.two_graph_index(catch_adds=True)
2378
index.add_versions([], random_id=True)
2380
def test_add_versions_same_dup(self):
2381
index = self.two_graph_index(catch_adds=True)
2382
# options can be spelt two different ways
2383
index.add_versions([('tip', 'fulltext,no-eol', (None, 0, 100), ['parent'])])
2384
index.add_versions([('tip', 'no-eol,fulltext', (None, 0, 100), ['parent'])])
2385
# but neither should have added data.
2386
self.assertEqual([[], []], self.caught_entries)
2388
def test_add_versions_different_dup(self):
2389
index = self.two_graph_index(deltas=True, catch_adds=True)
2391
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2392
[('tip', 'no-eol,line-delta', (None, 0, 100), ['parent'])])
2393
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2394
[('tip', 'line-delta,no-eol', (None, 0, 100), ['parent'])])
2395
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2396
[('tip', 'fulltext', (None, 0, 100), ['parent'])])
2398
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2399
[('tip', 'fulltext,no-eol', (None, 50, 100), ['parent'])])
2400
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2401
[('tip', 'fulltext,no-eol', (None, 0, 1000), ['parent'])])
2403
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2404
[('tip', 'fulltext,no-eol', (None, 0, 100), [])])
2405
# change options in the second record
2406
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2407
[('tip', 'fulltext,no-eol', (None, 0, 100), ['parent']),
2408
('tip', 'no-eol,line-delta', (None, 0, 100), ['parent'])])
2409
self.assertEqual([], self.caught_entries)
2411
def test_iter_parents(self):
2412
index1 = self.make_g_index('1', 1, [
2414
(('r0', ), 'N0 100', ([], )),
2416
(('r1', ), '', ([('r0', )], ))])
2417
index2 = self.make_g_index('2', 1, [
2419
(('r2', ), 'N0 100', ([('r1', ), ('r0', )], )),
2421
combined_index = CombinedGraphIndex([index1, index2])
2422
index = KnitGraphIndex(combined_index)
2424
# cases: each sample data individually:
2425
self.assertEqual(set([('r0', ())]),
2426
set(index.iter_parents(['r0'])))
2427
self.assertEqual(set([('r1', ('r0', ))]),
2428
set(index.iter_parents(['r1'])))
2429
self.assertEqual(set([('r2', ('r1', 'r0'))]),
2430
set(index.iter_parents(['r2'])))
2431
# no nodes returned for a missing node
2432
self.assertEqual(set(),
2433
set(index.iter_parents(['missing'])))
2434
# 1 node returned with missing nodes skipped
2435
self.assertEqual(set([('r1', ('r0', ))]),
2436
set(index.iter_parents(['ghost1', 'r1', 'ghost'])))
2438
self.assertEqual(set([('r0', ()), ('r1', ('r0', ))]),
2439
set(index.iter_parents(['r0', 'r1'])))
2440
# 2 nodes returned, missing skipped
2441
self.assertEqual(set([('r0', ()), ('r1', ('r0', ))]),
2442
set(index.iter_parents(['a', 'r0', 'b', 'r1', 'c'])))
2445
class TestNoParentsGraphIndexKnit(KnitTests):
2446
"""Tests for knits using KnitGraphIndex with no parents."""
2448
def make_g_index(self, name, ref_lists=0, nodes=[]):
2449
builder = GraphIndexBuilder(ref_lists)
2450
for node, references in nodes:
2451
builder.add_node(node, references)
2452
stream = builder.finish()
2453
trans = self.get_transport()
2454
size = trans.put_file(name, stream)
2455
return GraphIndex(trans, name, size)
2457
def test_parents_deltas_incompatible(self):
2458
index = CombinedGraphIndex([])
2459
self.assertRaises(errors.KnitError, KnitGraphIndex, index,
2460
deltas=True, parents=False)
2462
def two_graph_index(self, catch_adds=False):
2463
"""Build a two-graph index.
2465
:param deltas: If true, use underlying indices with two node-ref
2466
lists and 'parent' set to a delta-compressed against tail.
2468
# put several versions in the index.
2469
index1 = self.make_g_index('1', 0, [
2470
(('tip', ), 'N0 100'),
2472
index2 = self.make_g_index('2', 0, [
2473
(('parent', ), ' 100 78'),
2474
(('separate', ), '')])
2475
combined_index = CombinedGraphIndex([index1, index2])
2477
self.combined_index = combined_index
2478
self.caught_entries = []
2479
add_callback = self.catch_add
2482
return KnitGraphIndex(combined_index, parents=False,
2483
add_callback=add_callback)
2485
def test_get_graph(self):
2486
index = self.two_graph_index()
2487
self.assertEqual(set([
2492
]), set(index.get_graph()))
2494
def test_get_ancestry(self):
2495
# with no parents, ancestry is always just the key.
2496
index = self.two_graph_index()
2497
self.assertEqual([], index.get_ancestry([]))
2498
self.assertEqual(['separate'], index.get_ancestry(['separate']))
2499
self.assertEqual(['tail'], index.get_ancestry(['tail']))
2500
self.assertEqual(['parent'], index.get_ancestry(['parent']))
2501
self.assertEqual(['tip'], index.get_ancestry(['tip']))
2502
self.assertTrue(index.get_ancestry(['tip', 'separate']) in
2503
(['tip', 'separate'],
2504
['separate', 'tip'],
2506
# asking for a ghost makes it go boom.
2507
self.assertRaises(errors.RevisionNotPresent, index.get_ancestry, ['ghost'])
2509
def test_get_ancestry_with_ghosts(self):
2510
index = self.two_graph_index()
2511
self.assertEqual([], index.get_ancestry_with_ghosts([]))
2512
self.assertEqual(['separate'], index.get_ancestry_with_ghosts(['separate']))
2513
self.assertEqual(['tail'], index.get_ancestry_with_ghosts(['tail']))
2514
self.assertEqual(['parent'], index.get_ancestry_with_ghosts(['parent']))
2515
self.assertEqual(['tip'], index.get_ancestry_with_ghosts(['tip']))
2516
self.assertTrue(index.get_ancestry_with_ghosts(['tip', 'separate']) in
2517
(['tip', 'separate'],
2518
['separate', 'tip'],
2520
# asking for a ghost makes it go boom.
2521
self.assertRaises(errors.RevisionNotPresent, index.get_ancestry_with_ghosts, ['ghost'])
2523
def test_num_versions(self):
2524
index = self.two_graph_index()
2525
self.assertEqual(4, index.num_versions())
2527
def test_get_versions(self):
2528
index = self.two_graph_index()
2529
self.assertEqual(set(['tail', 'tip', 'parent', 'separate']),
2530
set(index.get_versions()))
2532
def test_has_version(self):
2533
index = self.two_graph_index()
2534
self.assertTrue(index.has_version('tail'))
2535
self.assertFalse(index.has_version('ghost'))
2537
def test_get_position(self):
2538
index = self.two_graph_index()
2539
self.assertEqual((index._graph_index._indices[0], 0, 100), index.get_position('tip'))
2540
self.assertEqual((index._graph_index._indices[1], 100, 78), index.get_position('parent'))
2542
def test_get_method(self):
2543
index = self.two_graph_index()
2544
self.assertEqual('fulltext', index.get_method('tip'))
2545
self.assertEqual(['fulltext'], index.get_options('parent'))
2547
def test_get_options(self):
2548
index = self.two_graph_index()
2549
self.assertEqual(['fulltext', 'no-eol'], index.get_options('tip'))
2550
self.assertEqual(['fulltext'], index.get_options('parent'))
2552
def test_get_parents(self):
2553
index = self.two_graph_index()
2554
self.assertEqual((), index.get_parents('parent'))
2555
# and errors on ghosts.
2556
self.assertRaises(errors.RevisionNotPresent,
2557
index.get_parents, 'ghost')
2559
def test_get_parents_with_ghosts(self):
2560
index = self.two_graph_index()
2561
self.assertEqual((), index.get_parents_with_ghosts('parent'))
2562
# and errors on ghosts.
2563
self.assertRaises(errors.RevisionNotPresent,
2564
index.get_parents_with_ghosts, 'ghost')
2566
def test_check_versions_present(self):
2567
index = self.two_graph_index()
2568
self.assertRaises(RevisionNotPresent, index.check_versions_present,
2570
self.assertRaises(RevisionNotPresent, index.check_versions_present,
2571
['tail', 'missing'])
2572
index.check_versions_present(['tail', 'separate'])
2574
def catch_add(self, entries):
2575
self.caught_entries.append(entries)
2577
def test_add_no_callback_errors(self):
2578
index = self.two_graph_index()
2579
self.assertRaises(errors.ReadOnlyError, index.add_version,
2580
'new', 'fulltext,no-eol', (None, 50, 60), ['separate'])
2582
def test_add_version_smoke(self):
2583
index = self.two_graph_index(catch_adds=True)
2584
index.add_version('new', 'fulltext,no-eol', (None, 50, 60), [])
2585
self.assertEqual([[(('new', ), 'N50 60')]],
2586
self.caught_entries)
2588
def test_add_version_delta_not_delta_index(self):
2589
index = self.two_graph_index(catch_adds=True)
2590
self.assertRaises(errors.KnitCorrupt, index.add_version,
2591
'new', 'no-eol,line-delta', (None, 0, 100), [])
2592
self.assertEqual([], self.caught_entries)
2594
def test_add_version_same_dup(self):
2595
index = self.two_graph_index(catch_adds=True)
2596
# options can be spelt two different ways
2597
index.add_version('tip', 'fulltext,no-eol', (None, 0, 100), [])
2598
index.add_version('tip', 'no-eol,fulltext', (None, 0, 100), [])
2599
# but neither should have added data.
2600
self.assertEqual([[], []], self.caught_entries)
2602
def test_add_version_different_dup(self):
2603
index = self.two_graph_index(catch_adds=True)
2605
self.assertRaises(errors.KnitCorrupt, index.add_version,
2606
'tip', 'no-eol,line-delta', (None, 0, 100), [])
2607
self.assertRaises(errors.KnitCorrupt, index.add_version,
2608
'tip', 'line-delta,no-eol', (None, 0, 100), [])
2609
self.assertRaises(errors.KnitCorrupt, index.add_version,
2610
'tip', 'fulltext', (None, 0, 100), [])
2612
self.assertRaises(errors.KnitCorrupt, index.add_version,
2613
'tip', 'fulltext,no-eol', (None, 50, 100), [])
2614
self.assertRaises(errors.KnitCorrupt, index.add_version,
2615
'tip', 'fulltext,no-eol', (None, 0, 1000), [])
2617
self.assertRaises(errors.KnitCorrupt, index.add_version,
2618
'tip', 'fulltext,no-eol', (None, 0, 100), ['parent'])
2619
self.assertEqual([], self.caught_entries)
2621
def test_add_versions(self):
2622
index = self.two_graph_index(catch_adds=True)
2623
index.add_versions([
2624
('new', 'fulltext,no-eol', (None, 50, 60), []),
2625
('new2', 'fulltext', (None, 0, 6), []),
2627
self.assertEqual([(('new', ), 'N50 60'), (('new2', ), ' 0 6')],
2628
sorted(self.caught_entries[0]))
2629
self.assertEqual(1, len(self.caught_entries))
2631
def test_add_versions_delta_not_delta_index(self):
2632
index = self.two_graph_index(catch_adds=True)
2633
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2634
[('new', 'no-eol,line-delta', (None, 0, 100), ['parent'])])
2635
self.assertEqual([], self.caught_entries)
2637
def test_add_versions_parents_not_parents_index(self):
2638
index = self.two_graph_index(catch_adds=True)
2639
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2640
[('new', 'no-eol,fulltext', (None, 0, 100), ['parent'])])
2641
self.assertEqual([], self.caught_entries)
2643
def test_add_versions_random_id_accepted(self):
2644
index = self.two_graph_index(catch_adds=True)
2645
index.add_versions([], random_id=True)
2647
def test_add_versions_same_dup(self):
2648
index = self.two_graph_index(catch_adds=True)
2649
# options can be spelt two different ways
2650
index.add_versions([('tip', 'fulltext,no-eol', (None, 0, 100), [])])
2651
index.add_versions([('tip', 'no-eol,fulltext', (None, 0, 100), [])])
2652
# but neither should have added data.
2653
self.assertEqual([[], []], self.caught_entries)
2655
def test_add_versions_different_dup(self):
2656
index = self.two_graph_index(catch_adds=True)
2658
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2659
[('tip', 'no-eol,line-delta', (None, 0, 100), [])])
2660
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2661
[('tip', 'line-delta,no-eol', (None, 0, 100), [])])
2662
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2663
[('tip', 'fulltext', (None, 0, 100), [])])
2665
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2666
[('tip', 'fulltext,no-eol', (None, 50, 100), [])])
2667
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2668
[('tip', 'fulltext,no-eol', (None, 0, 1000), [])])
2670
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2671
[('tip', 'fulltext,no-eol', (None, 0, 100), ['parent'])])
2672
# change options in the second record
2673
self.assertRaises(errors.KnitCorrupt, index.add_versions,
2674
[('tip', 'fulltext,no-eol', (None, 0, 100), []),
2675
('tip', 'no-eol,line-delta', (None, 0, 100), [])])
2676
self.assertEqual([], self.caught_entries)
2678
def test_iter_parents(self):
2679
index = self.two_graph_index()
2680
self.assertEqual(set([
2681
('tip', ()), ('tail', ()), ('parent', ()), ('separate', ())
2683
set(index.iter_parents(['tip', 'tail', 'ghost', 'parent', 'separate'])))
2684
self.assertEqual(set([('tip', ())]),
2685
set(index.iter_parents(['tip'])))
2686
self.assertEqual(set(),
2687
set(index.iter_parents([])))
2690
class TestPackKnits(KnitTests):
2691
"""Tests that use a _PackAccess and KnitGraphIndex."""
2693
def test_get_data_stream_packs_ignores_pack_overhead(self):
2694
# Packs have an encoding overhead that should not be included in the
2695
# 'size' field of a data stream, because it is not returned by the
2696
# raw_reading functions - it is why index_memo's are opaque, and
2697
# get_data_stream was abusing this.
2698
packname = 'test.pack'
2699
transport = self.get_transport()
2700
def write_data(bytes):
2701
transport.append_bytes(packname, bytes)
2702
writer = pack.ContainerWriter(write_data)
2704
index = InMemoryGraphIndex(2)
2705
knit_index = KnitGraphIndex(index, add_callback=index.add_nodes,
2707
indices = {index:(transport, packname)}
2708
access = _PackAccess(indices, writer=(writer, index))
2709
k = KnitVersionedFile('test', get_transport('.'),
2710
delta=True, create=True, index=knit_index, access_method=access)
2711
# insert something into the knit
2712
k.add_lines('text-1', [], ["foo\n"])
2713
# get a data stream for it
2714
stream = k.get_data_stream(['text-1'])
2715
# if the stream has been incorrectly assembled, we will get a short read
2716
# reading from the stream (as streams have no trailer)
2717
expected_length = stream[1][0][2]
2718
# we use -1 to do the read, so that if a trailer is added this test
2719
# will fail and we'll adjust it to handle that case correctly, rather
2720
# than allowing an over-read that is bogus.
2721
self.assertEqual(expected_length, len(stream[2](-1)))
added
removed
bzrlib/tests/test_knit.py
