17
17
"""Tests for Knit data structure"""
19
from cStringIO import StringIO
23
from bzrlib.errors import KnitError, RevisionAlreadyPresent
25
from bzrlib.errors import (
26
RevisionAlreadyPresent,
24
31
from bzrlib.knit import (
27
35
KnitAnnotateFactory,
29
39
from bzrlib.osutils import split_lines
30
from bzrlib.tests import TestCaseWithTransport
31
from bzrlib.transport import TransportLogger
32
from bzrlib.transport.local import LocalTransport
40
from bzrlib.tests import TestCase, TestCaseWithTransport
41
from bzrlib.transport import TransportLogger, get_transport
33
42
from bzrlib.transport.memory import MemoryTransport
34
43
from bzrlib.weave import Weave
46
class KnitContentTests(TestCase):
48
def test_constructor(self):
49
content = KnitContent([])
52
content = KnitContent([])
53
self.assertEqual(content.text(), [])
55
content = KnitContent([("origin1", "text1"), ("origin2", "text2")])
56
self.assertEqual(content.text(), ["text1", "text2"])
58
def test_annotate(self):
59
content = KnitContent([])
60
self.assertEqual(content.annotate(), [])
62
content = KnitContent([("origin1", "text1"), ("origin2", "text2")])
63
self.assertEqual(content.annotate(),
64
[("origin1", "text1"), ("origin2", "text2")])
66
def test_annotate_iter(self):
67
content = KnitContent([])
68
it = content.annotate_iter()
69
self.assertRaises(StopIteration, it.next)
71
content = KnitContent([("origin1", "text1"), ("origin2", "text2")])
72
it = content.annotate_iter()
73
self.assertEqual(it.next(), ("origin1", "text1"))
74
self.assertEqual(it.next(), ("origin2", "text2"))
75
self.assertRaises(StopIteration, it.next)
78
content = KnitContent([("origin1", "text1"), ("origin2", "text2")])
80
self.assertIsInstance(copy, KnitContent)
81
self.assertEqual(copy.annotate(),
82
[("origin1", "text1"), ("origin2", "text2")])
84
def test_line_delta(self):
85
content1 = KnitContent([("", "a"), ("", "b")])
86
content2 = KnitContent([("", "a"), ("", "a"), ("", "c")])
87
self.assertEqual(content1.line_delta(content2),
88
[(1, 2, 2, [("", "a"), ("", "c")])])
90
def test_line_delta_iter(self):
91
content1 = KnitContent([("", "a"), ("", "b")])
92
content2 = KnitContent([("", "a"), ("", "a"), ("", "c")])
93
it = content1.line_delta_iter(content2)
94
self.assertEqual(it.next(), (1, 2, 2, [("", "a"), ("", "c")]))
95
self.assertRaises(StopIteration, it.next)
98
class MockTransport(object):
100
def __init__(self, file_lines=None):
101
self.file_lines = file_lines
103
# We have no base directory for the MockTransport
106
def get(self, filename):
107
if self.file_lines is None:
108
raise NoSuchFile(filename)
110
return StringIO("\n".join(self.file_lines))
112
def __getattr__(self, name):
113
def queue_call(*args, **kwargs):
114
self.calls.append((name, args, kwargs))
118
class LowLevelKnitIndexTests(TestCase):
120
def test_no_such_file(self):
121
transport = MockTransport()
123
self.assertRaises(NoSuchFile, _KnitIndex, transport, "filename", "r")
124
self.assertRaises(NoSuchFile, _KnitIndex, transport,
125
"filename", "w", create=False)
127
def test_create_file(self):
128
transport = MockTransport()
130
index = _KnitIndex(transport, "filename", "w",
131
file_mode="wb", create=True)
133
("put_bytes_non_atomic",
134
("filename", index.HEADER), {"mode": "wb"}),
135
transport.calls.pop(0))
137
def test_delay_create_file(self):
138
transport = MockTransport()
140
index = _KnitIndex(transport, "filename", "w",
141
create=True, file_mode="wb", create_parent_dir=True,
142
delay_create=True, dir_mode=0777)
143
self.assertEqual([], transport.calls)
145
index.add_versions([])
146
name, (filename, f), kwargs = transport.calls.pop(0)
147
self.assertEqual("put_file_non_atomic", name)
149
{"dir_mode": 0777, "create_parent_dir": True, "mode": "wb"},
151
self.assertEqual("filename", filename)
152
self.assertEqual(index.HEADER, f.read())
154
index.add_versions([])
155
self.assertEqual(("append_bytes", ("filename", ""), {}),
156
transport.calls.pop(0))
158
def test_read_utf8_version_id(self):
159
unicode_revision_id = u"version-\N{CYRILLIC CAPITAL LETTER A}"
160
utf8_revision_id = unicode_revision_id.encode('utf-8')
161
transport = MockTransport([
163
'%s option 0 1 :' % (utf8_revision_id,)
165
index = _KnitIndex(transport, "filename", "r")
166
# _KnitIndex is a private class, and deals in utf8 revision_ids, not
167
# Unicode revision_ids.
168
self.assertTrue(index.has_version(utf8_revision_id))
169
self.assertFalse(index.has_version(unicode_revision_id))
171
def test_read_utf8_parents(self):
172
unicode_revision_id = u"version-\N{CYRILLIC CAPITAL LETTER A}"
173
utf8_revision_id = unicode_revision_id.encode('utf-8')
174
transport = MockTransport([
176
"version option 0 1 .%s :" % (utf8_revision_id,)
178
index = _KnitIndex(transport, "filename", "r")
179
self.assertEqual([utf8_revision_id],
180
index.get_parents_with_ghosts("version"))
182
def test_read_ignore_corrupted_lines(self):
183
transport = MockTransport([
186
"corrupted options 0 1 .b .c ",
187
"version options 0 1 :"
189
index = _KnitIndex(transport, "filename", "r")
190
self.assertEqual(1, index.num_versions())
191
self.assertTrue(index.has_version("version"))
193
def test_read_corrupted_header(self):
194
transport = MockTransport(['not a bzr knit index header\n'])
195
self.assertRaises(KnitHeaderError,
196
_KnitIndex, transport, "filename", "r")
198
def test_read_duplicate_entries(self):
199
transport = MockTransport([
201
"parent options 0 1 :",
202
"version options1 0 1 0 :",
203
"version options2 1 2 .other :",
204
"version options3 3 4 0 .other :"
206
index = _KnitIndex(transport, "filename", "r")
207
self.assertEqual(2, index.num_versions())
208
self.assertEqual(1, index.lookup("version"))
209
self.assertEqual((3, 4), index.get_position("version"))
210
self.assertEqual(["options3"], index.get_options("version"))
211
self.assertEqual(["parent", "other"],
212
index.get_parents_with_ghosts("version"))
214
def test_read_compressed_parents(self):
215
transport = MockTransport([
219
"c option 0 1 1 0 :",
221
index = _KnitIndex(transport, "filename", "r")
222
self.assertEqual(["a"], index.get_parents("b"))
223
self.assertEqual(["b", "a"], index.get_parents("c"))
225
def test_write_utf8_version_id(self):
226
unicode_revision_id = u"version-\N{CYRILLIC CAPITAL LETTER A}"
227
utf8_revision_id = unicode_revision_id.encode('utf-8')
228
transport = MockTransport([
231
index = _KnitIndex(transport, "filename", "r")
232
index.add_version(utf8_revision_id, ["option"], 0, 1, [])
233
self.assertEqual(("append_bytes", ("filename",
234
"\n%s option 0 1 :" % (utf8_revision_id,)),
236
transport.calls.pop(0))
238
def test_write_utf8_parents(self):
239
unicode_revision_id = u"version-\N{CYRILLIC CAPITAL LETTER A}"
240
utf8_revision_id = unicode_revision_id.encode('utf-8')
241
transport = MockTransport([
244
index = _KnitIndex(transport, "filename", "r")
245
index.add_version("version", ["option"], 0, 1, [utf8_revision_id])
246
self.assertEqual(("append_bytes", ("filename",
247
"\nversion option 0 1 .%s :" % (utf8_revision_id,)),
249
transport.calls.pop(0))
251
def test_get_graph(self):
252
transport = MockTransport()
253
index = _KnitIndex(transport, "filename", "w", create=True)
254
self.assertEqual([], index.get_graph())
256
index.add_version("a", ["option"], 0, 1, ["b"])
257
self.assertEqual([("a", ["b"])], index.get_graph())
259
index.add_version("c", ["option"], 0, 1, ["d"])
260
self.assertEqual([("a", ["b"]), ("c", ["d"])],
261
sorted(index.get_graph()))
263
def test_get_ancestry(self):
264
transport = MockTransport([
267
"b option 0 1 0 .e :",
268
"c option 0 1 1 0 :",
269
"d option 0 1 2 .f :"
271
index = _KnitIndex(transport, "filename", "r")
273
self.assertEqual([], index.get_ancestry([]))
274
self.assertEqual(["a"], index.get_ancestry(["a"]))
275
self.assertEqual(["a", "b"], index.get_ancestry(["b"]))
276
self.assertEqual(["a", "b", "c"], index.get_ancestry(["c"]))
277
self.assertEqual(["a", "b", "c", "d"], index.get_ancestry(["d"]))
278
self.assertEqual(["a", "b"], index.get_ancestry(["a", "b"]))
279
self.assertEqual(["a", "b", "c"], index.get_ancestry(["a", "c"]))
281
self.assertRaises(RevisionNotPresent, index.get_ancestry, ["e"])
283
def test_get_ancestry_with_ghosts(self):
284
transport = MockTransport([
287
"b option 0 1 0 .e :",
288
"c option 0 1 0 .f .g :",
289
"d option 0 1 2 .h .j .k :"
291
index = _KnitIndex(transport, "filename", "r")
293
self.assertEqual([], index.get_ancestry_with_ghosts([]))
294
self.assertEqual(["a"], index.get_ancestry_with_ghosts(["a"]))
295
self.assertEqual(["a", "e", "b"],
296
index.get_ancestry_with_ghosts(["b"]))
297
self.assertEqual(["a", "g", "f", "c"],
298
index.get_ancestry_with_ghosts(["c"]))
299
self.assertEqual(["a", "g", "f", "c", "k", "j", "h", "d"],
300
index.get_ancestry_with_ghosts(["d"]))
301
self.assertEqual(["a", "e", "b"],
302
index.get_ancestry_with_ghosts(["a", "b"]))
303
self.assertEqual(["a", "g", "f", "c"],
304
index.get_ancestry_with_ghosts(["a", "c"]))
306
["a", "g", "f", "c", "e", "b", "k", "j", "h", "d"],
307
index.get_ancestry_with_ghosts(["b", "d"]))
309
self.assertRaises(RevisionNotPresent,
310
index.get_ancestry_with_ghosts, ["e"])
312
def test_num_versions(self):
313
transport = MockTransport([
316
index = _KnitIndex(transport, "filename", "r")
318
self.assertEqual(0, index.num_versions())
319
self.assertEqual(0, len(index))
321
index.add_version("a", ["option"], 0, 1, [])
322
self.assertEqual(1, index.num_versions())
323
self.assertEqual(1, len(index))
325
index.add_version("a", ["option2"], 1, 2, [])
326
self.assertEqual(1, index.num_versions())
327
self.assertEqual(1, len(index))
329
index.add_version("b", ["option"], 0, 1, [])
330
self.assertEqual(2, index.num_versions())
331
self.assertEqual(2, len(index))
333
def test_get_versions(self):
334
transport = MockTransport([
337
index = _KnitIndex(transport, "filename", "r")
339
self.assertEqual([], index.get_versions())
341
index.add_version("a", ["option"], 0, 1, [])
342
self.assertEqual(["a"], index.get_versions())
344
index.add_version("a", ["option"], 0, 1, [])
345
self.assertEqual(["a"], index.get_versions())
347
index.add_version("b", ["option"], 0, 1, [])
348
self.assertEqual(["a", "b"], index.get_versions())
350
def test_idx_to_name(self):
351
transport = MockTransport([
356
index = _KnitIndex(transport, "filename", "r")
358
self.assertEqual("a", index.idx_to_name(0))
359
self.assertEqual("b", index.idx_to_name(1))
360
self.assertEqual("b", index.idx_to_name(-1))
361
self.assertEqual("a", index.idx_to_name(-2))
363
def test_lookup(self):
364
transport = MockTransport([
369
index = _KnitIndex(transport, "filename", "r")
371
self.assertEqual(0, index.lookup("a"))
372
self.assertEqual(1, index.lookup("b"))
374
def test_add_version(self):
375
transport = MockTransport([
378
index = _KnitIndex(transport, "filename", "r")
380
index.add_version("a", ["option"], 0, 1, ["b"])
381
self.assertEqual(("append_bytes",
382
("filename", "\na option 0 1 .b :"),
383
{}), transport.calls.pop(0))
384
self.assertTrue(index.has_version("a"))
385
self.assertEqual(1, index.num_versions())
386
self.assertEqual((0, 1), index.get_position("a"))
387
self.assertEqual(["option"], index.get_options("a"))
388
self.assertEqual(["b"], index.get_parents_with_ghosts("a"))
390
index.add_version("a", ["opt"], 1, 2, ["c"])
391
self.assertEqual(("append_bytes",
392
("filename", "\na opt 1 2 .c :"),
393
{}), transport.calls.pop(0))
394
self.assertTrue(index.has_version("a"))
395
self.assertEqual(1, index.num_versions())
396
self.assertEqual((1, 2), index.get_position("a"))
397
self.assertEqual(["opt"], index.get_options("a"))
398
self.assertEqual(["c"], index.get_parents_with_ghosts("a"))
400
index.add_version("b", ["option"], 2, 3, ["a"])
401
self.assertEqual(("append_bytes",
402
("filename", "\nb option 2 3 0 :"),
403
{}), transport.calls.pop(0))
404
self.assertTrue(index.has_version("b"))
405
self.assertEqual(2, index.num_versions())
406
self.assertEqual((2, 3), index.get_position("b"))
407
self.assertEqual(["option"], index.get_options("b"))
408
self.assertEqual(["a"], index.get_parents_with_ghosts("b"))
410
def test_add_versions(self):
411
transport = MockTransport([
414
index = _KnitIndex(transport, "filename", "r")
417
("a", ["option"], 0, 1, ["b"]),
418
("a", ["opt"], 1, 2, ["c"]),
419
("b", ["option"], 2, 3, ["a"])
421
self.assertEqual(("append_bytes", ("filename",
422
"\na option 0 1 .b :"
425
), {}), transport.calls.pop(0))
426
self.assertTrue(index.has_version("a"))
427
self.assertTrue(index.has_version("b"))
428
self.assertEqual(2, index.num_versions())
429
self.assertEqual((1, 2), index.get_position("a"))
430
self.assertEqual((2, 3), index.get_position("b"))
431
self.assertEqual(["opt"], index.get_options("a"))
432
self.assertEqual(["option"], index.get_options("b"))
433
self.assertEqual(["c"], index.get_parents_with_ghosts("a"))
434
self.assertEqual(["a"], index.get_parents_with_ghosts("b"))
436
def test_delay_create_and_add_versions(self):
437
transport = MockTransport()
439
index = _KnitIndex(transport, "filename", "w",
440
create=True, file_mode="wb", create_parent_dir=True,
441
delay_create=True, dir_mode=0777)
442
self.assertEqual([], transport.calls)
445
("a", ["option"], 0, 1, ["b"]),
446
("a", ["opt"], 1, 2, ["c"]),
447
("b", ["option"], 2, 3, ["a"])
449
name, (filename, f), kwargs = transport.calls.pop(0)
450
self.assertEqual("put_file_non_atomic", name)
452
{"dir_mode": 0777, "create_parent_dir": True, "mode": "wb"},
454
self.assertEqual("filename", filename)
457
"\na option 0 1 .b :"
459
"\nb option 2 3 0 :",
462
def test_has_version(self):
463
transport = MockTransport([
467
index = _KnitIndex(transport, "filename", "r")
469
self.assertTrue(index.has_version("a"))
470
self.assertFalse(index.has_version("b"))
472
def test_get_position(self):
473
transport = MockTransport([
478
index = _KnitIndex(transport, "filename", "r")
480
self.assertEqual((0, 1), index.get_position("a"))
481
self.assertEqual((1, 2), index.get_position("b"))
483
def test_get_method(self):
484
transport = MockTransport([
486
"a fulltext,unknown 0 1 :",
487
"b unknown,line-delta 1 2 :",
490
index = _KnitIndex(transport, "filename", "r")
492
self.assertEqual("fulltext", index.get_method("a"))
493
self.assertEqual("line-delta", index.get_method("b"))
494
self.assertRaises(errors.KnitIndexUnknownMethod, index.get_method, "c")
496
def test_get_options(self):
497
transport = MockTransport([
502
index = _KnitIndex(transport, "filename", "r")
504
self.assertEqual(["opt1"], index.get_options("a"))
505
self.assertEqual(["opt2", "opt3"], index.get_options("b"))
507
def test_get_parents(self):
508
transport = MockTransport([
511
"b option 1 2 0 .c :",
512
"c option 1 2 1 0 .e :"
514
index = _KnitIndex(transport, "filename", "r")
516
self.assertEqual([], index.get_parents("a"))
517
self.assertEqual(["a", "c"], index.get_parents("b"))
518
self.assertEqual(["b", "a"], index.get_parents("c"))
520
def test_get_parents_with_ghosts(self):
521
transport = MockTransport([
524
"b option 1 2 0 .c :",
525
"c option 1 2 1 0 .e :"
527
index = _KnitIndex(transport, "filename", "r")
529
self.assertEqual([], index.get_parents_with_ghosts("a"))
530
self.assertEqual(["a", "c"], index.get_parents_with_ghosts("b"))
531
self.assertEqual(["b", "a", "e"],
532
index.get_parents_with_ghosts("c"))
534
def test_check_versions_present(self):
535
transport = MockTransport([
540
index = _KnitIndex(transport, "filename", "r")
542
check = index.check_versions_present
548
self.assertRaises(RevisionNotPresent, check, ["c"])
549
self.assertRaises(RevisionNotPresent, check, ["a", "b", "c"])
37
552
class KnitTests(TestCaseWithTransport):
38
553
"""Class containing knit test helper routines."""
40
def make_test_knit(self, annotate=False):
555
def make_test_knit(self, annotate=False, delay_create=False):
42
557
factory = KnitPlainFactory()
45
return KnitVersionedFile('test', LocalTransport('.'), access_mode='w', factory=factory, create=True)
560
return KnitVersionedFile('test', get_transport('.'), access_mode='w',
561
factory=factory, create=True,
562
delay_create=delay_create)
48
565
class BasicKnitTests(KnitTests):
363
845
"\nrevid2 line-delta 84 82 0 :",
365
847
# we should be able to load this file again
366
knit = KnitVersionedFile('test', LocalTransport('.'), access_mode='r')
848
knit = KnitVersionedFile('test', get_transport('.'), access_mode='r')
367
849
self.assertEqual(['revid', 'revid2'], knit.versions())
368
850
# write a short write to the file and ensure that its ignored
369
851
indexfile = file('test.kndx', 'at')
370
852
indexfile.write('\nrevid3 line-delta 166 82 1 2 3 4 5 .phwoar:demo ')
371
853
indexfile.close()
372
854
# we should be able to load this file again
373
knit = KnitVersionedFile('test', LocalTransport('.'), access_mode='w')
855
knit = KnitVersionedFile('test', get_transport('.'), access_mode='w')
374
856
self.assertEqual(['revid', 'revid2'], knit.versions())
375
857
# and add a revision with the same id the failed write had
376
858
knit.add_lines('revid3', ['revid2'], ['a\n'])
377
859
# and when reading it revid3 should now appear.
378
knit = KnitVersionedFile('test', LocalTransport('.'), access_mode='r')
860
knit = KnitVersionedFile('test', get_transport('.'), access_mode='r')
379
861
self.assertEqual(['revid', 'revid2', 'revid3'], knit.versions())
380
862
self.assertEqual(['revid2'], knit.get_parents('revid3'))
864
def test_delay_create(self):
865
"""Test that passing delay_create=True creates files late"""
866
knit = self.make_test_knit(annotate=True, delay_create=True)
867
self.failIfExists('test.knit')
868
self.failIfExists('test.kndx')
869
knit.add_lines_with_ghosts('revid', ['a_ghost'], ['a\n'])
870
self.failUnlessExists('test.knit')
871
self.assertFileEqual(
872
"# bzr knit index 8\n"
874
"revid fulltext 0 84 .a_ghost :",
877
def test_create_parent_dir(self):
878
"""create_parent_dir can create knits in nonexistant dirs"""
879
# Has no effect if we don't set 'delay_create'
880
trans = get_transport('.')
881
self.assertRaises(NoSuchFile, KnitVersionedFile, 'dir/test',
882
trans, access_mode='w', factory=None,
883
create=True, create_parent_dir=True)
884
# Nothing should have changed yet
885
knit = KnitVersionedFile('dir/test', trans, access_mode='w',
886
factory=None, create=True,
887
create_parent_dir=True,
889
self.failIfExists('dir/test.knit')
890
self.failIfExists('dir/test.kndx')
891
self.failIfExists('dir')
892
knit.add_lines('revid', [], ['a\n'])
893
self.failUnlessExists('dir')
894
self.failUnlessExists('dir/test.knit')
895
self.assertFileEqual(
896
"# bzr knit index 8\n"
898
"revid fulltext 0 84 :",
901
def test_create_mode_700(self):
902
trans = get_transport('.')
903
if not trans._can_roundtrip_unix_modebits():
904
# Can't roundtrip, so no need to run this test
906
knit = KnitVersionedFile('dir/test', trans, access_mode='w',
907
factory=None, create=True,
908
create_parent_dir=True,
912
knit.add_lines('revid', [], ['a\n'])
913
self.assertTransportMode(trans, 'dir', 0700)
914
self.assertTransportMode(trans, 'dir/test.knit', 0600)
915
self.assertTransportMode(trans, 'dir/test.kndx', 0600)
917
def test_create_mode_770(self):
918
trans = get_transport('.')
919
if not trans._can_roundtrip_unix_modebits():
920
# Can't roundtrip, so no need to run this test
922
knit = KnitVersionedFile('dir/test', trans, access_mode='w',
923
factory=None, create=True,
924
create_parent_dir=True,
928
knit.add_lines('revid', [], ['a\n'])
929
self.assertTransportMode(trans, 'dir', 0770)
930
self.assertTransportMode(trans, 'dir/test.knit', 0660)
931
self.assertTransportMode(trans, 'dir/test.kndx', 0660)
933
def test_create_mode_777(self):
934
trans = get_transport('.')
935
if not trans._can_roundtrip_unix_modebits():
936
# Can't roundtrip, so no need to run this test
938
knit = KnitVersionedFile('dir/test', trans, access_mode='w',
939
factory=None, create=True,
940
create_parent_dir=True,
944
knit.add_lines('revid', [], ['a\n'])
945
self.assertTransportMode(trans, 'dir', 0777)
946
self.assertTransportMode(trans, 'dir/test.knit', 0666)
947
self.assertTransportMode(trans, 'dir/test.kndx', 0666)
382
949
def test_plan_merge(self):
383
950
my_knit = self.make_test_knit(annotate=True)
384
951
my_knit.add_lines('text1', [], split_lines(TEXT_1))
487
1054
self.failIf(WeaveToKnit.is_compatible(k, w))
488
1055
self.failIf(WeaveToKnit.is_compatible(w, w))
489
1056
self.failIf(WeaveToKnit.is_compatible(k, k))
1059
class TestKnitCaching(KnitTests):
1061
def create_knit(self, cache_add=False):
1062
k = self.make_test_knit(True)
1066
k.add_lines('text-1', [], split_lines(TEXT_1))
1067
k.add_lines('text-2', [], split_lines(TEXT_2))
1070
def test_no_caching(self):
1071
k = self.create_knit()
1072
# Nothing should be cached without setting 'enable_cache'
1073
self.assertEqual({}, k._data._cache)
1075
def test_cache_add_and_clear(self):
1076
k = self.create_knit(True)
1078
self.assertEqual(['text-1', 'text-2'], sorted(k._data._cache.keys()))
1081
self.assertEqual({}, k._data._cache)
1083
def test_cache_data_read_raw(self):
1084
k = self.create_knit()
1086
# Now cache and read
1089
def read_one_raw(version):
1090
pos_map = k._get_components_positions([version])
1091
method, pos, size, next = pos_map[version]
1092
lst = list(k._data.read_records_iter_raw([(version, pos, size)]))
1093
self.assertEqual(1, len(lst))
1096
val = read_one_raw('text-1')
1097
self.assertEqual({'text-1':val[1]}, k._data._cache)
1100
# After clear, new reads are not cached
1101
self.assertEqual({}, k._data._cache)
1103
val2 = read_one_raw('text-1')
1104
self.assertEqual(val, val2)
1105
self.assertEqual({}, k._data._cache)
1107
def test_cache_data_read(self):
1108
k = self.create_knit()
1110
def read_one(version):
1111
pos_map = k._get_components_positions([version])
1112
method, pos, size, next = pos_map[version]
1113
lst = list(k._data.read_records_iter([(version, pos, size)]))
1114
self.assertEqual(1, len(lst))
1117
# Now cache and read
1120
val = read_one('text-2')
1121
self.assertEqual(['text-2'], k._data._cache.keys())
1122
self.assertEqual('text-2', val[0])
1123
content, digest = k._data._parse_record('text-2',
1124
k._data._cache['text-2'])
1125
self.assertEqual(content, val[1])
1126
self.assertEqual(digest, val[2])
1129
self.assertEqual({}, k._data._cache)
1131
val2 = read_one('text-2')
1132
self.assertEqual(val, val2)
1133
self.assertEqual({}, k._data._cache)
1135
def test_cache_read(self):
1136
k = self.create_knit()
1139
text = k.get_text('text-1')
1140
self.assertEqual(TEXT_1, text)
1141
self.assertEqual(['text-1'], k._data._cache.keys())
1144
self.assertEqual({}, k._data._cache)
1146
text = k.get_text('text-1')
1147
self.assertEqual(TEXT_1, text)
1148
self.assertEqual({}, k._data._cache)
1151
class TestKnitIndex(KnitTests):
1153
def test_add_versions_dictionary_compresses(self):
1154
"""Adding versions to the index should update the lookup dict"""
1155
knit = self.make_test_knit()
1157
idx.add_version('a-1', ['fulltext'], 0, 0, [])
1158
self.check_file_contents('test.kndx',
1159
'# bzr knit index 8\n'
1161
'a-1 fulltext 0 0 :'
1163
idx.add_versions([('a-2', ['fulltext'], 0, 0, ['a-1']),
1164
('a-3', ['fulltext'], 0, 0, ['a-2']),
1166
self.check_file_contents('test.kndx',
1167
'# bzr knit index 8\n'
1169
'a-1 fulltext 0 0 :\n'
1170
'a-2 fulltext 0 0 0 :\n'
1171
'a-3 fulltext 0 0 1 :'
1173
self.assertEqual(['a-1', 'a-2', 'a-3'], idx._history)
1174
self.assertEqual({'a-1':('a-1', ['fulltext'], 0, 0, [], 0),
1175
'a-2':('a-2', ['fulltext'], 0, 0, ['a-1'], 1),
1176
'a-3':('a-3', ['fulltext'], 0, 0, ['a-2'], 2),
1179
def test_add_versions_fails_clean(self):
1180
"""If add_versions fails in the middle, it restores a pristine state.
1182
Any modifications that are made to the index are reset if all versions
1185
# This cheats a little bit by passing in a generator which will
1186
# raise an exception before the processing finishes
1187
# Other possibilities would be to have an version with the wrong number
1188
# of entries, or to make the backing transport unable to write any
1191
knit = self.make_test_knit()
1193
idx.add_version('a-1', ['fulltext'], 0, 0, [])
1195
class StopEarly(Exception):
1198
def generate_failure():
1199
"""Add some entries and then raise an exception"""
1200
yield ('a-2', ['fulltext'], 0, 0, ['a-1'])
1201
yield ('a-3', ['fulltext'], 0, 0, ['a-2'])
1204
# Assert the pre-condition
1205
self.assertEqual(['a-1'], idx._history)
1206
self.assertEqual({'a-1':('a-1', ['fulltext'], 0, 0, [], 0)}, idx._cache)
1208
self.assertRaises(StopEarly, idx.add_versions, generate_failure())
1210
# And it shouldn't be modified
1211
self.assertEqual(['a-1'], idx._history)
1212
self.assertEqual({'a-1':('a-1', ['fulltext'], 0, 0, [], 0)}, idx._cache)
1214
def test_knit_index_ignores_empty_files(self):
1215
# There was a race condition in older bzr, where a ^C at the right time
1216
# could leave an empty .kndx file, which bzr would later claim was a
1217
# corrupted file since the header was not present. In reality, the file
1218
# just wasn't created, so it should be ignored.
1219
t = get_transport('.')
1220
t.put_bytes('test.kndx', '')
1222
knit = self.make_test_knit()
1224
def test_knit_index_checks_header(self):
1225
t = get_transport('.')
1226
t.put_bytes('test.kndx', '# not really a knit header\n\n')
1228
self.assertRaises(KnitHeaderError, self.make_test_knit)
added
removed
bzrlib/tests/test_knit.py
