17
17
"""Tests for Knit data structure"""
19
from cStringIO import StringIO
30
from bzrlib.errors import (
31
RevisionAlreadyPresent,
23
from bzrlib.errors import KnitError, RevisionAlreadyPresent, NoSuchFile
36
24
from bzrlib.knit import (
40
27
KnitAnnotateFactory,
45
29
from bzrlib.osutils import split_lines
46
from bzrlib.tests import TestCase, TestCaseWithTransport, Feature
30
from bzrlib.tests import TestCaseWithTransport
47
31
from bzrlib.transport import TransportLogger, get_transport
48
32
from bzrlib.transport.memory import MemoryTransport
49
33
from bzrlib.weave import Weave
52
class _CompiledKnitFeature(Feature):
56
import bzrlib._knit_load_data_c
61
def feature_name(self):
62
return 'bzrlib._knit_load_data_c'
64
CompiledKnitFeature = _CompiledKnitFeature()
67
class KnitContentTests(TestCase):
69
def test_constructor(self):
70
content = KnitContent([])
73
content = KnitContent([])
74
self.assertEqual(content.text(), [])
76
content = KnitContent([("origin1", "text1"), ("origin2", "text2")])
77
self.assertEqual(content.text(), ["text1", "text2"])
79
def test_annotate(self):
80
content = KnitContent([])
81
self.assertEqual(content.annotate(), [])
83
content = KnitContent([("origin1", "text1"), ("origin2", "text2")])
84
self.assertEqual(content.annotate(),
85
[("origin1", "text1"), ("origin2", "text2")])
87
def test_annotate_iter(self):
88
content = KnitContent([])
89
it = content.annotate_iter()
90
self.assertRaises(StopIteration, it.next)
92
content = KnitContent([("origin1", "text1"), ("origin2", "text2")])
93
it = content.annotate_iter()
94
self.assertEqual(it.next(), ("origin1", "text1"))
95
self.assertEqual(it.next(), ("origin2", "text2"))
96
self.assertRaises(StopIteration, it.next)
99
content = KnitContent([("origin1", "text1"), ("origin2", "text2")])
100
copy = content.copy()
101
self.assertIsInstance(copy, KnitContent)
102
self.assertEqual(copy.annotate(),
103
[("origin1", "text1"), ("origin2", "text2")])
105
def test_line_delta(self):
106
content1 = KnitContent([("", "a"), ("", "b")])
107
content2 = KnitContent([("", "a"), ("", "a"), ("", "c")])
108
self.assertEqual(content1.line_delta(content2),
109
[(1, 2, 2, [("", "a"), ("", "c")])])
111
def test_line_delta_iter(self):
112
content1 = KnitContent([("", "a"), ("", "b")])
113
content2 = KnitContent([("", "a"), ("", "a"), ("", "c")])
114
it = content1.line_delta_iter(content2)
115
self.assertEqual(it.next(), (1, 2, 2, [("", "a"), ("", "c")]))
116
self.assertRaises(StopIteration, it.next)
119
class MockTransport(object):
121
def __init__(self, file_lines=None):
122
self.file_lines = file_lines
124
# We have no base directory for the MockTransport
127
def get(self, filename):
128
if self.file_lines is None:
129
raise NoSuchFile(filename)
131
return StringIO("\n".join(self.file_lines))
133
def readv(self, relpath, offsets):
134
fp = self.get(relpath)
135
for offset, size in offsets:
137
yield offset, fp.read(size)
139
def __getattr__(self, name):
140
def queue_call(*args, **kwargs):
141
self.calls.append((name, args, kwargs))
145
class LowLevelKnitDataTests(TestCase):
147
def create_gz_content(self, text):
149
gz_file = gzip.GzipFile(mode='wb', fileobj=sio)
152
return sio.getvalue()
154
def test_valid_knit_data(self):
155
sha1sum = sha.new('foo\nbar\n').hexdigest()
156
gz_txt = self.create_gz_content('version rev-id-1 2 %s\n'
161
transport = MockTransport([gz_txt])
162
data = _KnitData(transport, 'filename', mode='r')
163
records = [('rev-id-1', 0, len(gz_txt))]
165
contents = data.read_records(records)
166
self.assertEqual({'rev-id-1':(['foo\n', 'bar\n'], sha1sum)}, contents)
168
raw_contents = list(data.read_records_iter_raw(records))
169
self.assertEqual([('rev-id-1', gz_txt)], raw_contents)
171
def test_not_enough_lines(self):
172
sha1sum = sha.new('foo\n').hexdigest()
173
# record says 2 lines data says 1
174
gz_txt = self.create_gz_content('version rev-id-1 2 %s\n'
178
transport = MockTransport([gz_txt])
179
data = _KnitData(transport, 'filename', mode='r')
180
records = [('rev-id-1', 0, len(gz_txt))]
181
self.assertRaises(errors.KnitCorrupt, data.read_records, records)
183
# read_records_iter_raw won't detect that sort of mismatch/corruption
184
raw_contents = list(data.read_records_iter_raw(records))
185
self.assertEqual([('rev-id-1', gz_txt)], raw_contents)
187
def test_too_many_lines(self):
188
sha1sum = sha.new('foo\nbar\n').hexdigest()
189
# record says 1 lines data says 2
190
gz_txt = self.create_gz_content('version rev-id-1 1 %s\n'
195
transport = MockTransport([gz_txt])
196
data = _KnitData(transport, 'filename', mode='r')
197
records = [('rev-id-1', 0, len(gz_txt))]
198
self.assertRaises(errors.KnitCorrupt, data.read_records, records)
200
# read_records_iter_raw won't detect that sort of mismatch/corruption
201
raw_contents = list(data.read_records_iter_raw(records))
202
self.assertEqual([('rev-id-1', gz_txt)], raw_contents)
204
def test_mismatched_version_id(self):
205
sha1sum = sha.new('foo\nbar\n').hexdigest()
206
gz_txt = self.create_gz_content('version rev-id-1 2 %s\n'
211
transport = MockTransport([gz_txt])
212
data = _KnitData(transport, 'filename', mode='r')
213
# We are asking for rev-id-2, but the data is rev-id-1
214
records = [('rev-id-2', 0, len(gz_txt))]
215
self.assertRaises(errors.KnitCorrupt, data.read_records, records)
217
# read_records_iter_raw will notice if we request the wrong version.
218
self.assertRaises(errors.KnitCorrupt, list,
219
data.read_records_iter_raw(records))
221
def test_uncompressed_data(self):
222
sha1sum = sha.new('foo\nbar\n').hexdigest()
223
txt = ('version rev-id-1 2 %s\n'
228
transport = MockTransport([txt])
229
data = _KnitData(transport, 'filename', mode='r')
230
records = [('rev-id-1', 0, len(txt))]
232
# We don't have valid gzip data ==> corrupt
233
self.assertRaises(errors.KnitCorrupt, data.read_records, records)
235
# read_records_iter_raw will notice the bad data
236
self.assertRaises(errors.KnitCorrupt, list,
237
data.read_records_iter_raw(records))
239
def test_corrupted_data(self):
240
sha1sum = sha.new('foo\nbar\n').hexdigest()
241
gz_txt = self.create_gz_content('version rev-id-1 2 %s\n'
246
# Change 2 bytes in the middle to \xff
247
gz_txt = gz_txt[:10] + '\xff\xff' + gz_txt[12:]
248
transport = MockTransport([gz_txt])
249
data = _KnitData(transport, 'filename', mode='r')
250
records = [('rev-id-1', 0, len(gz_txt))]
252
self.assertRaises(errors.KnitCorrupt, data.read_records, records)
254
# read_records_iter_raw will notice if we request the wrong version.
255
self.assertRaises(errors.KnitCorrupt, list,
256
data.read_records_iter_raw(records))
259
class LowLevelKnitIndexTests(TestCase):
261
def get_knit_index(self, *args, **kwargs):
262
orig = knit._load_data
264
knit._load_data = orig
265
self.addCleanup(reset)
266
from bzrlib._knit_load_data_py import _load_data_py
267
knit._load_data = _load_data_py
268
return _KnitIndex(*args, **kwargs)
270
def test_no_such_file(self):
271
transport = MockTransport()
273
self.assertRaises(NoSuchFile, self.get_knit_index,
274
transport, "filename", "r")
275
self.assertRaises(NoSuchFile, self.get_knit_index,
276
transport, "filename", "w", create=False)
278
def test_create_file(self):
279
transport = MockTransport()
281
index = self.get_knit_index(transport, "filename", "w",
282
file_mode="wb", create=True)
284
("put_bytes_non_atomic",
285
("filename", index.HEADER), {"mode": "wb"}),
286
transport.calls.pop(0))
288
def test_delay_create_file(self):
289
transport = MockTransport()
291
index = self.get_knit_index(transport, "filename", "w",
292
create=True, file_mode="wb", create_parent_dir=True,
293
delay_create=True, dir_mode=0777)
294
self.assertEqual([], transport.calls)
296
index.add_versions([])
297
name, (filename, f), kwargs = transport.calls.pop(0)
298
self.assertEqual("put_file_non_atomic", name)
300
{"dir_mode": 0777, "create_parent_dir": True, "mode": "wb"},
302
self.assertEqual("filename", filename)
303
self.assertEqual(index.HEADER, f.read())
305
index.add_versions([])
306
self.assertEqual(("append_bytes", ("filename", ""), {}),
307
transport.calls.pop(0))
309
def test_read_utf8_version_id(self):
310
unicode_revision_id = u"version-\N{CYRILLIC CAPITAL LETTER A}"
311
utf8_revision_id = unicode_revision_id.encode('utf-8')
312
transport = MockTransport([
314
'%s option 0 1 :' % (utf8_revision_id,)
316
index = self.get_knit_index(transport, "filename", "r")
317
# _KnitIndex is a private class, and deals in utf8 revision_ids, not
318
# Unicode revision_ids.
319
self.assertTrue(index.has_version(utf8_revision_id))
320
self.assertFalse(index.has_version(unicode_revision_id))
322
def test_read_utf8_parents(self):
323
unicode_revision_id = u"version-\N{CYRILLIC CAPITAL LETTER A}"
324
utf8_revision_id = unicode_revision_id.encode('utf-8')
325
transport = MockTransport([
327
"version option 0 1 .%s :" % (utf8_revision_id,)
329
index = self.get_knit_index(transport, "filename", "r")
330
self.assertEqual([utf8_revision_id],
331
index.get_parents_with_ghosts("version"))
333
def test_read_ignore_corrupted_lines(self):
334
transport = MockTransport([
337
"corrupted options 0 1 .b .c ",
338
"version options 0 1 :"
340
index = self.get_knit_index(transport, "filename", "r")
341
self.assertEqual(1, index.num_versions())
342
self.assertTrue(index.has_version("version"))
344
def test_read_corrupted_header(self):
345
transport = MockTransport(['not a bzr knit index header\n'])
346
self.assertRaises(KnitHeaderError,
347
self.get_knit_index, transport, "filename", "r")
349
def test_read_duplicate_entries(self):
350
transport = MockTransport([
352
"parent options 0 1 :",
353
"version options1 0 1 0 :",
354
"version options2 1 2 .other :",
355
"version options3 3 4 0 .other :"
357
index = self.get_knit_index(transport, "filename", "r")
358
self.assertEqual(2, index.num_versions())
359
self.assertEqual(1, index.lookup("version"))
360
self.assertEqual((3, 4), index.get_position("version"))
361
self.assertEqual(["options3"], index.get_options("version"))
362
self.assertEqual(["parent", "other"],
363
index.get_parents_with_ghosts("version"))
365
def test_read_compressed_parents(self):
366
transport = MockTransport([
370
"c option 0 1 1 0 :",
372
index = self.get_knit_index(transport, "filename", "r")
373
self.assertEqual(["a"], index.get_parents("b"))
374
self.assertEqual(["b", "a"], index.get_parents("c"))
376
def test_write_utf8_version_id(self):
377
unicode_revision_id = u"version-\N{CYRILLIC CAPITAL LETTER A}"
378
utf8_revision_id = unicode_revision_id.encode('utf-8')
379
transport = MockTransport([
382
index = self.get_knit_index(transport, "filename", "r")
383
index.add_version(utf8_revision_id, ["option"], 0, 1, [])
384
self.assertEqual(("append_bytes", ("filename",
385
"\n%s option 0 1 :" % (utf8_revision_id,)),
387
transport.calls.pop(0))
389
def test_write_utf8_parents(self):
390
unicode_revision_id = u"version-\N{CYRILLIC CAPITAL LETTER A}"
391
utf8_revision_id = unicode_revision_id.encode('utf-8')
392
transport = MockTransport([
395
index = self.get_knit_index(transport, "filename", "r")
396
index.add_version("version", ["option"], 0, 1, [utf8_revision_id])
397
self.assertEqual(("append_bytes", ("filename",
398
"\nversion option 0 1 .%s :" % (utf8_revision_id,)),
400
transport.calls.pop(0))
402
def test_get_graph(self):
403
transport = MockTransport()
404
index = self.get_knit_index(transport, "filename", "w", create=True)
405
self.assertEqual([], index.get_graph())
407
index.add_version("a", ["option"], 0, 1, ["b"])
408
self.assertEqual([("a", ["b"])], index.get_graph())
410
index.add_version("c", ["option"], 0, 1, ["d"])
411
self.assertEqual([("a", ["b"]), ("c", ["d"])],
412
sorted(index.get_graph()))
414
def test_get_ancestry(self):
415
transport = MockTransport([
418
"b option 0 1 0 .e :",
419
"c option 0 1 1 0 :",
420
"d option 0 1 2 .f :"
422
index = self.get_knit_index(transport, "filename", "r")
424
self.assertEqual([], index.get_ancestry([]))
425
self.assertEqual(["a"], index.get_ancestry(["a"]))
426
self.assertEqual(["a", "b"], index.get_ancestry(["b"]))
427
self.assertEqual(["a", "b", "c"], index.get_ancestry(["c"]))
428
self.assertEqual(["a", "b", "c", "d"], index.get_ancestry(["d"]))
429
self.assertEqual(["a", "b"], index.get_ancestry(["a", "b"]))
430
self.assertEqual(["a", "b", "c"], index.get_ancestry(["a", "c"]))
432
self.assertRaises(RevisionNotPresent, index.get_ancestry, ["e"])
434
def test_get_ancestry_with_ghosts(self):
435
transport = MockTransport([
438
"b option 0 1 0 .e :",
439
"c option 0 1 0 .f .g :",
440
"d option 0 1 2 .h .j .k :"
442
index = self.get_knit_index(transport, "filename", "r")
444
self.assertEqual([], index.get_ancestry_with_ghosts([]))
445
self.assertEqual(["a"], index.get_ancestry_with_ghosts(["a"]))
446
self.assertEqual(["a", "e", "b"],
447
index.get_ancestry_with_ghosts(["b"]))
448
self.assertEqual(["a", "g", "f", "c"],
449
index.get_ancestry_with_ghosts(["c"]))
450
self.assertEqual(["a", "g", "f", "c", "k", "j", "h", "d"],
451
index.get_ancestry_with_ghosts(["d"]))
452
self.assertEqual(["a", "e", "b"],
453
index.get_ancestry_with_ghosts(["a", "b"]))
454
self.assertEqual(["a", "g", "f", "c"],
455
index.get_ancestry_with_ghosts(["a", "c"]))
457
["a", "g", "f", "c", "e", "b", "k", "j", "h", "d"],
458
index.get_ancestry_with_ghosts(["b", "d"]))
460
self.assertRaises(RevisionNotPresent,
461
index.get_ancestry_with_ghosts, ["e"])
463
def test_num_versions(self):
464
transport = MockTransport([
467
index = self.get_knit_index(transport, "filename", "r")
469
self.assertEqual(0, index.num_versions())
470
self.assertEqual(0, len(index))
472
index.add_version("a", ["option"], 0, 1, [])
473
self.assertEqual(1, index.num_versions())
474
self.assertEqual(1, len(index))
476
index.add_version("a", ["option2"], 1, 2, [])
477
self.assertEqual(1, index.num_versions())
478
self.assertEqual(1, len(index))
480
index.add_version("b", ["option"], 0, 1, [])
481
self.assertEqual(2, index.num_versions())
482
self.assertEqual(2, len(index))
484
def test_get_versions(self):
485
transport = MockTransport([
488
index = self.get_knit_index(transport, "filename", "r")
490
self.assertEqual([], index.get_versions())
492
index.add_version("a", ["option"], 0, 1, [])
493
self.assertEqual(["a"], index.get_versions())
495
index.add_version("a", ["option"], 0, 1, [])
496
self.assertEqual(["a"], index.get_versions())
498
index.add_version("b", ["option"], 0, 1, [])
499
self.assertEqual(["a", "b"], index.get_versions())
501
def test_idx_to_name(self):
502
transport = MockTransport([
507
index = self.get_knit_index(transport, "filename", "r")
509
self.assertEqual("a", index.idx_to_name(0))
510
self.assertEqual("b", index.idx_to_name(1))
511
self.assertEqual("b", index.idx_to_name(-1))
512
self.assertEqual("a", index.idx_to_name(-2))
514
def test_lookup(self):
515
transport = MockTransport([
520
index = self.get_knit_index(transport, "filename", "r")
522
self.assertEqual(0, index.lookup("a"))
523
self.assertEqual(1, index.lookup("b"))
525
def test_add_version(self):
526
transport = MockTransport([
529
index = self.get_knit_index(transport, "filename", "r")
531
index.add_version("a", ["option"], 0, 1, ["b"])
532
self.assertEqual(("append_bytes",
533
("filename", "\na option 0 1 .b :"),
534
{}), transport.calls.pop(0))
535
self.assertTrue(index.has_version("a"))
536
self.assertEqual(1, index.num_versions())
537
self.assertEqual((0, 1), index.get_position("a"))
538
self.assertEqual(["option"], index.get_options("a"))
539
self.assertEqual(["b"], index.get_parents_with_ghosts("a"))
541
index.add_version("a", ["opt"], 1, 2, ["c"])
542
self.assertEqual(("append_bytes",
543
("filename", "\na opt 1 2 .c :"),
544
{}), transport.calls.pop(0))
545
self.assertTrue(index.has_version("a"))
546
self.assertEqual(1, index.num_versions())
547
self.assertEqual((1, 2), index.get_position("a"))
548
self.assertEqual(["opt"], index.get_options("a"))
549
self.assertEqual(["c"], index.get_parents_with_ghosts("a"))
551
index.add_version("b", ["option"], 2, 3, ["a"])
552
self.assertEqual(("append_bytes",
553
("filename", "\nb option 2 3 0 :"),
554
{}), transport.calls.pop(0))
555
self.assertTrue(index.has_version("b"))
556
self.assertEqual(2, index.num_versions())
557
self.assertEqual((2, 3), index.get_position("b"))
558
self.assertEqual(["option"], index.get_options("b"))
559
self.assertEqual(["a"], index.get_parents_with_ghosts("b"))
561
def test_add_versions(self):
562
transport = MockTransport([
565
index = self.get_knit_index(transport, "filename", "r")
568
("a", ["option"], 0, 1, ["b"]),
569
("a", ["opt"], 1, 2, ["c"]),
570
("b", ["option"], 2, 3, ["a"])
572
self.assertEqual(("append_bytes", ("filename",
573
"\na option 0 1 .b :"
576
), {}), transport.calls.pop(0))
577
self.assertTrue(index.has_version("a"))
578
self.assertTrue(index.has_version("b"))
579
self.assertEqual(2, index.num_versions())
580
self.assertEqual((1, 2), index.get_position("a"))
581
self.assertEqual((2, 3), index.get_position("b"))
582
self.assertEqual(["opt"], index.get_options("a"))
583
self.assertEqual(["option"], index.get_options("b"))
584
self.assertEqual(["c"], index.get_parents_with_ghosts("a"))
585
self.assertEqual(["a"], index.get_parents_with_ghosts("b"))
587
def test_delay_create_and_add_versions(self):
588
transport = MockTransport()
590
index = self.get_knit_index(transport, "filename", "w",
591
create=True, file_mode="wb", create_parent_dir=True,
592
delay_create=True, dir_mode=0777)
593
self.assertEqual([], transport.calls)
596
("a", ["option"], 0, 1, ["b"]),
597
("a", ["opt"], 1, 2, ["c"]),
598
("b", ["option"], 2, 3, ["a"])
600
name, (filename, f), kwargs = transport.calls.pop(0)
601
self.assertEqual("put_file_non_atomic", name)
603
{"dir_mode": 0777, "create_parent_dir": True, "mode": "wb"},
605
self.assertEqual("filename", filename)
608
"\na option 0 1 .b :"
610
"\nb option 2 3 0 :",
613
def test_has_version(self):
614
transport = MockTransport([
618
index = self.get_knit_index(transport, "filename", "r")
620
self.assertTrue(index.has_version("a"))
621
self.assertFalse(index.has_version("b"))
623
def test_get_position(self):
624
transport = MockTransport([
629
index = self.get_knit_index(transport, "filename", "r")
631
self.assertEqual((0, 1), index.get_position("a"))
632
self.assertEqual((1, 2), index.get_position("b"))
634
def test_get_method(self):
635
transport = MockTransport([
637
"a fulltext,unknown 0 1 :",
638
"b unknown,line-delta 1 2 :",
641
index = self.get_knit_index(transport, "filename", "r")
643
self.assertEqual("fulltext", index.get_method("a"))
644
self.assertEqual("line-delta", index.get_method("b"))
645
self.assertRaises(errors.KnitIndexUnknownMethod, index.get_method, "c")
647
def test_get_options(self):
648
transport = MockTransport([
653
index = self.get_knit_index(transport, "filename", "r")
655
self.assertEqual(["opt1"], index.get_options("a"))
656
self.assertEqual(["opt2", "opt3"], index.get_options("b"))
658
def test_get_parents(self):
659
transport = MockTransport([
662
"b option 1 2 0 .c :",
663
"c option 1 2 1 0 .e :"
665
index = self.get_knit_index(transport, "filename", "r")
667
self.assertEqual([], index.get_parents("a"))
668
self.assertEqual(["a", "c"], index.get_parents("b"))
669
self.assertEqual(["b", "a"], index.get_parents("c"))
671
def test_get_parents_with_ghosts(self):
672
transport = MockTransport([
675
"b option 1 2 0 .c :",
676
"c option 1 2 1 0 .e :"
678
index = self.get_knit_index(transport, "filename", "r")
680
self.assertEqual([], index.get_parents_with_ghosts("a"))
681
self.assertEqual(["a", "c"], index.get_parents_with_ghosts("b"))
682
self.assertEqual(["b", "a", "e"],
683
index.get_parents_with_ghosts("c"))
685
def test_check_versions_present(self):
686
transport = MockTransport([
691
index = self.get_knit_index(transport, "filename", "r")
693
check = index.check_versions_present
699
self.assertRaises(RevisionNotPresent, check, ["c"])
700
self.assertRaises(RevisionNotPresent, check, ["a", "b", "c"])
702
def test_impossible_parent(self):
703
"""Test we get KnitCorrupt if the parent couldn't possibly exist."""
704
transport = MockTransport([
707
"b option 0 1 4 :" # We don't have a 4th record
710
self.assertRaises(errors.KnitCorrupt,
711
self.get_knit_index, transport, 'filename', 'r')
713
if (str(e) == ('exceptions must be strings, classes, or instances,'
714
' not exceptions.IndexError')
715
and sys.version_info[0:2] >= (2,5)):
716
self.knownFailure('Pyrex <0.9.5 fails with TypeError when'
717
' raising new style exceptions with python'
722
def test_corrupted_parent(self):
723
transport = MockTransport([
727
"c option 0 1 1v :", # Can't have a parent of '1v'
730
self.assertRaises(errors.KnitCorrupt,
731
self.get_knit_index, transport, 'filename', 'r')
733
if (str(e) == ('exceptions must be strings, classes, or instances,'
734
' not exceptions.ValueError')
735
and sys.version_info[0:2] >= (2,5)):
736
self.knownFailure('Pyrex <0.9.5 fails with TypeError when'
737
' raising new style exceptions with python'
742
def test_corrupted_parent_in_list(self):
743
transport = MockTransport([
747
"c option 0 1 1 v :", # Can't have a parent of 'v'
750
self.assertRaises(errors.KnitCorrupt,
751
self.get_knit_index, transport, 'filename', 'r')
753
if (str(e) == ('exceptions must be strings, classes, or instances,'
754
' not exceptions.ValueError')
755
and sys.version_info[0:2] >= (2,5)):
756
self.knownFailure('Pyrex <0.9.5 fails with TypeError when'
757
' raising new style exceptions with python'
762
def test_invalid_position(self):
763
transport = MockTransport([
768
self.assertRaises(errors.KnitCorrupt,
769
self.get_knit_index, transport, 'filename', 'r')
771
if (str(e) == ('exceptions must be strings, classes, or instances,'
772
' not exceptions.ValueError')
773
and sys.version_info[0:2] >= (2,5)):
774
self.knownFailure('Pyrex <0.9.5 fails with TypeError when'
775
' raising new style exceptions with python'
780
def test_invalid_size(self):
781
transport = MockTransport([
786
self.assertRaises(errors.KnitCorrupt,
787
self.get_knit_index, transport, 'filename', 'r')
789
if (str(e) == ('exceptions must be strings, classes, or instances,'
790
' not exceptions.ValueError')
791
and sys.version_info[0:2] >= (2,5)):
792
self.knownFailure('Pyrex <0.9.5 fails with TypeError when'
793
' raising new style exceptions with python'
798
def test_short_line(self):
799
transport = MockTransport([
802
"b option 10 10 0", # This line isn't terminated, ignored
804
index = self.get_knit_index(transport, "filename", "r")
805
self.assertEqual(['a'], index.get_versions())
807
def test_skip_incomplete_record(self):
808
# A line with bogus data should just be skipped
809
transport = MockTransport([
812
"b option 10 10 0", # This line isn't terminated, ignored
813
"c option 20 10 0 :", # Properly terminated, and starts with '\n'
815
index = self.get_knit_index(transport, "filename", "r")
816
self.assertEqual(['a', 'c'], index.get_versions())
818
def test_trailing_characters(self):
819
# A line with bogus data should just be skipped
820
transport = MockTransport([
823
"b option 10 10 0 :a", # This line has extra trailing characters
824
"c option 20 10 0 :", # Properly terminated, and starts with '\n'
826
index = self.get_knit_index(transport, "filename", "r")
827
self.assertEqual(['a', 'c'], index.get_versions())
830
class LowLevelKnitIndexTests_c(LowLevelKnitIndexTests):
832
_test_needs_features = [CompiledKnitFeature]
834
def get_knit_index(self, *args, **kwargs):
835
orig = knit._load_data
837
knit._load_data = orig
838
self.addCleanup(reset)
839
from bzrlib._knit_load_data_c import _load_data_c
840
knit._load_data = _load_data_c
841
return _KnitIndex(*args, **kwargs)
845
36
class KnitTests(TestCaseWithTransport):
846
37
"""Class containing knit test helper routines."""
added
removed
bzrlib/tests/test_knit.py
