/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_knit.py

  • Committer: Robert Collins
  • Date: 2007-07-15 07:31:37 UTC
  • mto: (2592.3.29 repository)
  • mto: This revision was merged to the branch mainline in revision 2624.
  • Revision ID: robertc@robertcollins.net-20070715073137-cd9kb764q4e40o0f
Tweak documentation as per Aaron's review.

Show diffs side-by-side

added added

removed removed

Lines of Context:
 
1
# Copyright (C) 2005, 2006, 2007 Canonical Ltd
 
2
#
 
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.
 
7
#
 
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.
 
12
#
 
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
 
16
 
 
17
"""Tests for Knit data structure"""
 
18
 
 
19
from cStringIO import StringIO
 
20
import difflib
 
21
import gzip
 
22
import sha
 
23
import sys
 
24
 
 
25
from bzrlib import (
 
26
    errors,
 
27
    generate_ids,
 
28
    knit,
 
29
    )
 
30
from bzrlib.errors import (
 
31
    RevisionAlreadyPresent,
 
32
    KnitHeaderError,
 
33
    RevisionNotPresent,
 
34
    NoSuchFile,
 
35
    )
 
36
from bzrlib.knit import (
 
37
    KnitContent,
 
38
    KnitVersionedFile,
 
39
    KnitPlainFactory,
 
40
    KnitAnnotateFactory,
 
41
    _KnitData,
 
42
    _KnitIndex,
 
43
    WeaveToKnit,
 
44
    )
 
45
from bzrlib.osutils import split_lines
 
46
from bzrlib.tests import TestCase, TestCaseWithTransport, Feature
 
47
from bzrlib.transport import TransportLogger, get_transport
 
48
from bzrlib.transport.memory import MemoryTransport
 
49
from bzrlib.weave import Weave
 
50
 
 
51
 
 
52
class _CompiledKnitFeature(Feature):
 
53
 
 
54
    def _probe(self):
 
55
        try:
 
56
            import bzrlib._knit_load_data_c
 
57
        except ImportError:
 
58
            return False
 
59
        return True
 
60
 
 
61
    def feature_name(self):
 
62
        return 'bzrlib._knit_load_data_c'
 
63
 
 
64
CompiledKnitFeature = _CompiledKnitFeature()
 
65
 
 
66
 
 
67
class KnitContentTests(TestCase):
 
68
 
 
69
    def test_constructor(self):
 
70
        content = KnitContent([])
 
71
 
 
72
    def test_text(self):
 
73
        content = KnitContent([])
 
74
        self.assertEqual(content.text(), [])
 
75
 
 
76
        content = KnitContent([("origin1", "text1"), ("origin2", "text2")])
 
77
        self.assertEqual(content.text(), ["text1", "text2"])
 
78
 
 
79
    def test_annotate(self):
 
80
        content = KnitContent([])
 
81
        self.assertEqual(content.annotate(), [])
 
82
 
 
83
        content = KnitContent([("origin1", "text1"), ("origin2", "text2")])
 
84
        self.assertEqual(content.annotate(),
 
85
            [("origin1", "text1"), ("origin2", "text2")])
 
86
 
 
87
    def test_annotate_iter(self):
 
88
        content = KnitContent([])
 
89
        it = content.annotate_iter()
 
90
        self.assertRaises(StopIteration, it.next)
 
91
 
 
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)
 
97
 
 
98
    def test_copy(self):
 
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")])
 
104
 
 
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")])])
 
110
 
 
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)
 
117
 
 
118
 
 
119
class MockTransport(object):
 
120
 
 
121
    def __init__(self, file_lines=None):
 
122
        self.file_lines = file_lines
 
123
        self.calls = []
 
124
        # We have no base directory for the MockTransport
 
125
        self.base = ''
 
126
 
 
127
    def get(self, filename):
 
128
        if self.file_lines is None:
 
129
            raise NoSuchFile(filename)
 
130
        else:
 
131
            return StringIO("\n".join(self.file_lines))
 
132
 
 
133
    def readv(self, relpath, offsets):
 
134
        fp = self.get(relpath)
 
135
        for offset, size in offsets:
 
136
            fp.seek(offset)
 
137
            yield offset, fp.read(size)
 
138
 
 
139
    def __getattr__(self, name):
 
140
        def queue_call(*args, **kwargs):
 
141
            self.calls.append((name, args, kwargs))
 
142
        return queue_call
 
143
 
 
144
 
 
145
class LowLevelKnitDataTests(TestCase):
 
146
 
 
147
    def create_gz_content(self, text):
 
148
        sio = StringIO()
 
149
        gz_file = gzip.GzipFile(mode='wb', fileobj=sio)
 
150
        gz_file.write(text)
 
151
        gz_file.close()
 
152
        return sio.getvalue()
 
153
 
 
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'
 
157
                                        'foo\n'
 
158
                                        'bar\n'
 
159
                                        'end rev-id-1\n'
 
160
                                        % (sha1sum,))
 
161
        transport = MockTransport([gz_txt])
 
162
        data = _KnitData(transport, 'filename', mode='r')
 
163
        records = [('rev-id-1', 0, len(gz_txt))]
 
164
 
 
165
        contents = data.read_records(records)
 
166
        self.assertEqual({'rev-id-1':(['foo\n', 'bar\n'], sha1sum)}, contents)
 
167
 
 
168
        raw_contents = list(data.read_records_iter_raw(records))
 
169
        self.assertEqual([('rev-id-1', gz_txt)], raw_contents)
 
170
 
 
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'
 
175
                                        'foo\n'
 
176
                                        'end rev-id-1\n'
 
177
                                        % (sha1sum,))
 
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)
 
182
 
 
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)
 
186
 
 
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'
 
191
                                        'foo\n'
 
192
                                        'bar\n'
 
193
                                        'end rev-id-1\n'
 
194
                                        % (sha1sum,))
 
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)
 
199
 
 
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)
 
203
 
 
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'
 
207
                                        'foo\n'
 
208
                                        'bar\n'
 
209
                                        'end rev-id-1\n'
 
210
                                        % (sha1sum,))
 
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)
 
216
 
 
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))
 
220
 
 
221
    def test_uncompressed_data(self):
 
222
        sha1sum = sha.new('foo\nbar\n').hexdigest()
 
223
        txt = ('version rev-id-1 2 %s\n'
 
224
               'foo\n'
 
225
               'bar\n'
 
226
               'end rev-id-1\n'
 
227
               % (sha1sum,))
 
228
        transport = MockTransport([txt])
 
229
        data = _KnitData(transport, 'filename', mode='r')
 
230
        records = [('rev-id-1', 0, len(txt))]
 
231
 
 
232
        # We don't have valid gzip data ==> corrupt
 
233
        self.assertRaises(errors.KnitCorrupt, data.read_records, records)
 
234
 
 
235
        # read_records_iter_raw will notice the bad data
 
236
        self.assertRaises(errors.KnitCorrupt, list,
 
237
                          data.read_records_iter_raw(records))
 
238
 
 
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'
 
242
                                        'foo\n'
 
243
                                        'bar\n'
 
244
                                        'end rev-id-1\n'
 
245
                                        % (sha1sum,))
 
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))]
 
251
 
 
252
        self.assertRaises(errors.KnitCorrupt, data.read_records, records)
 
253
 
 
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))
 
257
 
 
258
 
 
259
class LowLevelKnitIndexTests(TestCase):
 
260
 
 
261
    def get_knit_index(self, *args, **kwargs):
 
262
        orig = knit._load_data
 
263
        def reset():
 
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)
 
269
 
 
270
    def test_no_such_file(self):
 
271
        transport = MockTransport()
 
272
 
 
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)
 
277
 
 
278
    def test_create_file(self):
 
279
        transport = MockTransport()
 
280
 
 
281
        index = self.get_knit_index(transport, "filename", "w",
 
282
            file_mode="wb", create=True)
 
283
        self.assertEqual(
 
284
                ("put_bytes_non_atomic",
 
285
                    ("filename", index.HEADER), {"mode": "wb"}),
 
286
                transport.calls.pop(0))
 
287
 
 
288
    def test_delay_create_file(self):
 
289
        transport = MockTransport()
 
290
 
 
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)
 
295
 
 
296
        index.add_versions([])
 
297
        name, (filename, f), kwargs = transport.calls.pop(0)
 
298
        self.assertEqual("put_file_non_atomic", name)
 
299
        self.assertEqual(
 
300
            {"dir_mode": 0777, "create_parent_dir": True, "mode": "wb"},
 
301
            kwargs)
 
302
        self.assertEqual("filename", filename)
 
303
        self.assertEqual(index.HEADER, f.read())
 
304
 
 
305
        index.add_versions([])
 
306
        self.assertEqual(("append_bytes", ("filename", ""), {}),
 
307
            transport.calls.pop(0))
 
308
 
 
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([
 
313
            _KnitIndex.HEADER,
 
314
            '%s option 0 1 :' % (utf8_revision_id,)
 
315
            ])
 
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))
 
321
 
 
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([
 
326
            _KnitIndex.HEADER,
 
327
            "version option 0 1 .%s :" % (utf8_revision_id,)
 
328
            ])
 
329
        index = self.get_knit_index(transport, "filename", "r")
 
330
        self.assertEqual([utf8_revision_id],
 
331
            index.get_parents_with_ghosts("version"))
 
332
 
 
333
    def test_read_ignore_corrupted_lines(self):
 
334
        transport = MockTransport([
 
335
            _KnitIndex.HEADER,
 
336
            "corrupted",
 
337
            "corrupted options 0 1 .b .c ",
 
338
            "version options 0 1 :"
 
339
            ])
 
340
        index = self.get_knit_index(transport, "filename", "r")
 
341
        self.assertEqual(1, index.num_versions())
 
342
        self.assertTrue(index.has_version("version"))
 
343
 
 
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")
 
348
 
 
349
    def test_read_duplicate_entries(self):
 
350
        transport = MockTransport([
 
351
            _KnitIndex.HEADER,
 
352
            "parent options 0 1 :",
 
353
            "version options1 0 1 0 :",
 
354
            "version options2 1 2 .other :",
 
355
            "version options3 3 4 0 .other :"
 
356
            ])
 
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"))
 
364
 
 
365
    def test_read_compressed_parents(self):
 
366
        transport = MockTransport([
 
367
            _KnitIndex.HEADER,
 
368
            "a option 0 1 :",
 
369
            "b option 0 1 0 :",
 
370
            "c option 0 1 1 0 :",
 
371
            ])
 
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"))
 
375
 
 
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([
 
380
            _KnitIndex.HEADER
 
381
            ])
 
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,)),
 
386
            {}),
 
387
            transport.calls.pop(0))
 
388
 
 
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([
 
393
            _KnitIndex.HEADER
 
394
            ])
 
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,)),
 
399
            {}),
 
400
            transport.calls.pop(0))
 
401
 
 
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())
 
406
 
 
407
        index.add_version("a", ["option"], 0, 1, ["b"])
 
408
        self.assertEqual([("a", ["b"])], index.get_graph())
 
409
 
 
410
        index.add_version("c", ["option"], 0, 1, ["d"])
 
411
        self.assertEqual([("a", ["b"]), ("c", ["d"])],
 
412
            sorted(index.get_graph()))
 
413
 
 
414
    def test_get_ancestry(self):
 
415
        transport = MockTransport([
 
416
            _KnitIndex.HEADER,
 
417
            "a option 0 1 :",
 
418
            "b option 0 1 0 .e :",
 
419
            "c option 0 1 1 0 :",
 
420
            "d option 0 1 2 .f :"
 
421
            ])
 
422
        index = self.get_knit_index(transport, "filename", "r")
 
423
 
 
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"]))
 
431
 
 
432
        self.assertRaises(RevisionNotPresent, index.get_ancestry, ["e"])
 
433
 
 
434
    def test_get_ancestry_with_ghosts(self):
 
435
        transport = MockTransport([
 
436
            _KnitIndex.HEADER,
 
437
            "a option 0 1 :",
 
438
            "b option 0 1 0 .e :",
 
439
            "c option 0 1 0 .f .g :",
 
440
            "d option 0 1 2 .h .j .k :"
 
441
            ])
 
442
        index = self.get_knit_index(transport, "filename", "r")
 
443
 
 
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"]))
 
456
        self.assertEqual(
 
457
            ["a", "g", "f", "c", "e", "b", "k", "j", "h", "d"],
 
458
            index.get_ancestry_with_ghosts(["b", "d"]))
 
459
 
 
460
        self.assertRaises(RevisionNotPresent,
 
461
            index.get_ancestry_with_ghosts, ["e"])
 
462
 
 
463
    def test_num_versions(self):
 
464
        transport = MockTransport([
 
465
            _KnitIndex.HEADER
 
466
            ])
 
467
        index = self.get_knit_index(transport, "filename", "r")
 
468
 
 
469
        self.assertEqual(0, index.num_versions())
 
470
        self.assertEqual(0, len(index))
 
471
 
 
472
        index.add_version("a", ["option"], 0, 1, [])
 
473
        self.assertEqual(1, index.num_versions())
 
474
        self.assertEqual(1, len(index))
 
475
 
 
476
        index.add_version("a", ["option2"], 1, 2, [])
 
477
        self.assertEqual(1, index.num_versions())
 
478
        self.assertEqual(1, len(index))
 
479
 
 
480
        index.add_version("b", ["option"], 0, 1, [])
 
481
        self.assertEqual(2, index.num_versions())
 
482
        self.assertEqual(2, len(index))
 
483
 
 
484
    def test_get_versions(self):
 
485
        transport = MockTransport([
 
486
            _KnitIndex.HEADER
 
487
            ])
 
488
        index = self.get_knit_index(transport, "filename", "r")
 
489
 
 
490
        self.assertEqual([], index.get_versions())
 
491
 
 
492
        index.add_version("a", ["option"], 0, 1, [])
 
493
        self.assertEqual(["a"], index.get_versions())
 
494
 
 
495
        index.add_version("a", ["option"], 0, 1, [])
 
496
        self.assertEqual(["a"], index.get_versions())
 
497
 
 
498
        index.add_version("b", ["option"], 0, 1, [])
 
499
        self.assertEqual(["a", "b"], index.get_versions())
 
500
 
 
501
    def test_idx_to_name(self):
 
502
        transport = MockTransport([
 
503
            _KnitIndex.HEADER,
 
504
            "a option 0 1 :",
 
505
            "b option 0 1 :"
 
506
            ])
 
507
        index = self.get_knit_index(transport, "filename", "r")
 
508
 
 
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))
 
513
 
 
514
    def test_lookup(self):
 
515
        transport = MockTransport([
 
516
            _KnitIndex.HEADER,
 
517
            "a option 0 1 :",
 
518
            "b option 0 1 :"
 
519
            ])
 
520
        index = self.get_knit_index(transport, "filename", "r")
 
521
 
 
522
        self.assertEqual(0, index.lookup("a"))
 
523
        self.assertEqual(1, index.lookup("b"))
 
524
 
 
525
    def test_add_version(self):
 
526
        transport = MockTransport([
 
527
            _KnitIndex.HEADER
 
528
            ])
 
529
        index = self.get_knit_index(transport, "filename", "r")
 
530
 
 
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"))
 
540
 
 
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"))
 
550
 
 
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"))
 
560
 
 
561
    def test_add_versions(self):
 
562
        transport = MockTransport([
 
563
            _KnitIndex.HEADER
 
564
            ])
 
565
        index = self.get_knit_index(transport, "filename", "r")
 
566
 
 
567
        index.add_versions([
 
568
            ("a", ["option"], 0, 1, ["b"]),
 
569
            ("a", ["opt"], 1, 2, ["c"]),
 
570
            ("b", ["option"], 2, 3, ["a"])
 
571
            ])
 
572
        self.assertEqual(("append_bytes", ("filename",
 
573
            "\na option 0 1 .b :"
 
574
            "\na opt 1 2 .c :"
 
575
            "\nb option 2 3 0 :"
 
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"))
 
586
 
 
587
    def test_delay_create_and_add_versions(self):
 
588
        transport = MockTransport()
 
589
 
 
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)
 
594
 
 
595
        index.add_versions([
 
596
            ("a", ["option"], 0, 1, ["b"]),
 
597
            ("a", ["opt"], 1, 2, ["c"]),
 
598
            ("b", ["option"], 2, 3, ["a"])
 
599
            ])
 
600
        name, (filename, f), kwargs = transport.calls.pop(0)
 
601
        self.assertEqual("put_file_non_atomic", name)
 
602
        self.assertEqual(
 
603
            {"dir_mode": 0777, "create_parent_dir": True, "mode": "wb"},
 
604
            kwargs)
 
605
        self.assertEqual("filename", filename)
 
606
        self.assertEqual(
 
607
            index.HEADER +
 
608
            "\na option 0 1 .b :"
 
609
            "\na opt 1 2 .c :"
 
610
            "\nb option 2 3 0 :",
 
611
            f.read())
 
612
 
 
613
    def test_has_version(self):
 
614
        transport = MockTransport([
 
615
            _KnitIndex.HEADER,
 
616
            "a option 0 1 :"
 
617
            ])
 
618
        index = self.get_knit_index(transport, "filename", "r")
 
619
 
 
620
        self.assertTrue(index.has_version("a"))
 
621
        self.assertFalse(index.has_version("b"))
 
622
 
 
623
    def test_get_position(self):
 
624
        transport = MockTransport([
 
625
            _KnitIndex.HEADER,
 
626
            "a option 0 1 :",
 
627
            "b option 1 2 :"
 
628
            ])
 
629
        index = self.get_knit_index(transport, "filename", "r")
 
630
 
 
631
        self.assertEqual((0, 1), index.get_position("a"))
 
632
        self.assertEqual((1, 2), index.get_position("b"))
 
633
 
 
634
    def test_get_method(self):
 
635
        transport = MockTransport([
 
636
            _KnitIndex.HEADER,
 
637
            "a fulltext,unknown 0 1 :",
 
638
            "b unknown,line-delta 1 2 :",
 
639
            "c bad 3 4 :"
 
640
            ])
 
641
        index = self.get_knit_index(transport, "filename", "r")
 
642
 
 
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")
 
646
 
 
647
    def test_get_options(self):
 
648
        transport = MockTransport([
 
649
            _KnitIndex.HEADER,
 
650
            "a opt1 0 1 :",
 
651
            "b opt2,opt3 1 2 :"
 
652
            ])
 
653
        index = self.get_knit_index(transport, "filename", "r")
 
654
 
 
655
        self.assertEqual(["opt1"], index.get_options("a"))
 
656
        self.assertEqual(["opt2", "opt3"], index.get_options("b"))
 
657
 
 
658
    def test_get_parents(self):
 
659
        transport = MockTransport([
 
660
            _KnitIndex.HEADER,
 
661
            "a option 0 1 :",
 
662
            "b option 1 2 0 .c :",
 
663
            "c option 1 2 1 0 .e :"
 
664
            ])
 
665
        index = self.get_knit_index(transport, "filename", "r")
 
666
 
 
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"))
 
670
 
 
671
    def test_get_parents_with_ghosts(self):
 
672
        transport = MockTransport([
 
673
            _KnitIndex.HEADER,
 
674
            "a option 0 1 :",
 
675
            "b option 1 2 0 .c :",
 
676
            "c option 1 2 1 0 .e :"
 
677
            ])
 
678
        index = self.get_knit_index(transport, "filename", "r")
 
679
 
 
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"))
 
684
 
 
685
    def test_check_versions_present(self):
 
686
        transport = MockTransport([
 
687
            _KnitIndex.HEADER,
 
688
            "a option 0 1 :",
 
689
            "b option 0 1 :"
 
690
            ])
 
691
        index = self.get_knit_index(transport, "filename", "r")
 
692
 
 
693
        check = index.check_versions_present
 
694
 
 
695
        check([])
 
696
        check(["a"])
 
697
        check(["b"])
 
698
        check(["a", "b"])
 
699
        self.assertRaises(RevisionNotPresent, check, ["c"])
 
700
        self.assertRaises(RevisionNotPresent, check, ["a", "b", "c"])
 
701
 
 
702
    def test_impossible_parent(self):
 
703
        """Test we get KnitCorrupt if the parent couldn't possibly exist."""
 
704
        transport = MockTransport([
 
705
            _KnitIndex.HEADER,
 
706
            "a option 0 1 :",
 
707
            "b option 0 1 4 :"  # We don't have a 4th record
 
708
            ])
 
709
        try:
 
710
            self.assertRaises(errors.KnitCorrupt,
 
711
                              self.get_knit_index, transport, 'filename', 'r')
 
712
        except TypeError, e:
 
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'
 
718
                                  ' >=2.5')
 
719
            else:
 
720
                raise
 
721
 
 
722
    def test_corrupted_parent(self):
 
723
        transport = MockTransport([
 
724
            _KnitIndex.HEADER,
 
725
            "a option 0 1 :",
 
726
            "b option 0 1 :",
 
727
            "c option 0 1 1v :", # Can't have a parent of '1v'
 
728
            ])
 
729
        try:
 
730
            self.assertRaises(errors.KnitCorrupt,
 
731
                              self.get_knit_index, transport, 'filename', 'r')
 
732
        except TypeError, e:
 
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'
 
738
                                  ' >=2.5')
 
739
            else:
 
740
                raise
 
741
 
 
742
    def test_corrupted_parent_in_list(self):
 
743
        transport = MockTransport([
 
744
            _KnitIndex.HEADER,
 
745
            "a option 0 1 :",
 
746
            "b option 0 1 :",
 
747
            "c option 0 1 1 v :", # Can't have a parent of 'v'
 
748
            ])
 
749
        try:
 
750
            self.assertRaises(errors.KnitCorrupt,
 
751
                              self.get_knit_index, transport, 'filename', 'r')
 
752
        except TypeError, e:
 
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'
 
758
                                  ' >=2.5')
 
759
            else:
 
760
                raise
 
761
 
 
762
    def test_invalid_position(self):
 
763
        transport = MockTransport([
 
764
            _KnitIndex.HEADER,
 
765
            "a option 1v 1 :",
 
766
            ])
 
767
        try:
 
768
            self.assertRaises(errors.KnitCorrupt,
 
769
                              self.get_knit_index, transport, 'filename', 'r')
 
770
        except TypeError, e:
 
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'
 
776
                                  ' >=2.5')
 
777
            else:
 
778
                raise
 
779
 
 
780
    def test_invalid_size(self):
 
781
        transport = MockTransport([
 
782
            _KnitIndex.HEADER,
 
783
            "a option 1 1v :",
 
784
            ])
 
785
        try:
 
786
            self.assertRaises(errors.KnitCorrupt,
 
787
                              self.get_knit_index, transport, 'filename', 'r')
 
788
        except TypeError, e:
 
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'
 
794
                                  ' >=2.5')
 
795
            else:
 
796
                raise
 
797
 
 
798
    def test_short_line(self):
 
799
        transport = MockTransport([
 
800
            _KnitIndex.HEADER,
 
801
            "a option 0 10  :",
 
802
            "b option 10 10 0", # This line isn't terminated, ignored
 
803
            ])
 
804
        index = self.get_knit_index(transport, "filename", "r")
 
805
        self.assertEqual(['a'], index.get_versions())
 
806
 
 
807
    def test_skip_incomplete_record(self):
 
808
        # A line with bogus data should just be skipped
 
809
        transport = MockTransport([
 
810
            _KnitIndex.HEADER,
 
811
            "a option 0 10  :",
 
812
            "b option 10 10 0", # This line isn't terminated, ignored
 
813
            "c option 20 10 0 :", # Properly terminated, and starts with '\n'
 
814
            ])
 
815
        index = self.get_knit_index(transport, "filename", "r")
 
816
        self.assertEqual(['a', 'c'], index.get_versions())
 
817
 
 
818
    def test_trailing_characters(self):
 
819
        # A line with bogus data should just be skipped
 
820
        transport = MockTransport([
 
821
            _KnitIndex.HEADER,
 
822
            "a option 0 10  :",
 
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'
 
825
            ])
 
826
        index = self.get_knit_index(transport, "filename", "r")
 
827
        self.assertEqual(['a', 'c'], index.get_versions())
 
828
 
 
829
 
 
830
class LowLevelKnitIndexTests_c(LowLevelKnitIndexTests):
 
831
 
 
832
    _test_needs_features = [CompiledKnitFeature]
 
833
 
 
834
    def get_knit_index(self, *args, **kwargs):
 
835
        orig = knit._load_data
 
836
        def reset():
 
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)
 
842
 
 
843
 
 
844
 
 
845
class KnitTests(TestCaseWithTransport):
 
846
    """Class containing knit test helper routines."""
 
847
 
 
848
    def make_test_knit(self, annotate=False, delay_create=False):
 
849
        if not annotate:
 
850
            factory = KnitPlainFactory()
 
851
        else:
 
852
            factory = None
 
853
        return KnitVersionedFile('test', get_transport('.'), access_mode='w',
 
854
                                 factory=factory, create=True,
 
855
                                 delay_create=delay_create)
 
856
 
 
857
 
 
858
class BasicKnitTests(KnitTests):
 
859
 
 
860
    def add_stock_one_and_one_a(self, k):
 
861
        k.add_lines('text-1', [], split_lines(TEXT_1))
 
862
        k.add_lines('text-1a', ['text-1'], split_lines(TEXT_1A))
 
863
 
 
864
    def test_knit_constructor(self):
 
865
        """Construct empty k"""
 
866
        self.make_test_knit()
 
867
 
 
868
    def test_knit_add(self):
 
869
        """Store one text in knit and retrieve"""
 
870
        k = self.make_test_knit()
 
871
        k.add_lines('text-1', [], split_lines(TEXT_1))
 
872
        self.assertTrue(k.has_version('text-1'))
 
873
        self.assertEqualDiff(''.join(k.get_lines('text-1')), TEXT_1)
 
874
 
 
875
    def test_knit_reload(self):
 
876
        # test that the content in a reloaded knit is correct
 
877
        k = self.make_test_knit()
 
878
        k.add_lines('text-1', [], split_lines(TEXT_1))
 
879
        del k
 
880
        k2 = KnitVersionedFile('test', get_transport('.'), access_mode='r', factory=KnitPlainFactory(), create=True)
 
881
        self.assertTrue(k2.has_version('text-1'))
 
882
        self.assertEqualDiff(''.join(k2.get_lines('text-1')), TEXT_1)
 
883
 
 
884
    def test_knit_several(self):
 
885
        """Store several texts in a knit"""
 
886
        k = self.make_test_knit()
 
887
        k.add_lines('text-1', [], split_lines(TEXT_1))
 
888
        k.add_lines('text-2', [], split_lines(TEXT_2))
 
889
        self.assertEqualDiff(''.join(k.get_lines('text-1')), TEXT_1)
 
890
        self.assertEqualDiff(''.join(k.get_lines('text-2')), TEXT_2)
 
891
        
 
892
    def test_repeated_add(self):
 
893
        """Knit traps attempt to replace existing version"""
 
894
        k = self.make_test_knit()
 
895
        k.add_lines('text-1', [], split_lines(TEXT_1))
 
896
        self.assertRaises(RevisionAlreadyPresent, 
 
897
                k.add_lines,
 
898
                'text-1', [], split_lines(TEXT_1))
 
899
 
 
900
    def test_empty(self):
 
901
        k = self.make_test_knit(True)
 
902
        k.add_lines('text-1', [], [])
 
903
        self.assertEquals(k.get_lines('text-1'), [])
 
904
 
 
905
    def test_incomplete(self):
 
906
        """Test if texts without a ending line-end can be inserted and
 
907
        extracted."""
 
908
        k = KnitVersionedFile('test', get_transport('.'), delta=False, create=True)
 
909
        k.add_lines('text-1', [], ['a\n',    'b'  ])
 
910
        k.add_lines('text-2', ['text-1'], ['a\rb\n', 'b\n'])
 
911
        # reopening ensures maximum room for confusion
 
912
        k = KnitVersionedFile('test', get_transport('.'), delta=False, create=True)
 
913
        self.assertEquals(k.get_lines('text-1'), ['a\n',    'b'  ])
 
914
        self.assertEquals(k.get_lines('text-2'), ['a\rb\n', 'b\n'])
 
915
 
 
916
    def test_delta(self):
 
917
        """Expression of knit delta as lines"""
 
918
        k = self.make_test_knit()
 
919
        td = list(line_delta(TEXT_1.splitlines(True),
 
920
                             TEXT_1A.splitlines(True)))
 
921
        self.assertEqualDiff(''.join(td), delta_1_1a)
 
922
        out = apply_line_delta(TEXT_1.splitlines(True), td)
 
923
        self.assertEqualDiff(''.join(out), TEXT_1A)
 
924
 
 
925
    def test_add_with_parents(self):
 
926
        """Store in knit with parents"""
 
927
        k = self.make_test_knit()
 
928
        self.add_stock_one_and_one_a(k)
 
929
        self.assertEquals(k.get_parents('text-1'), [])
 
930
        self.assertEquals(k.get_parents('text-1a'), ['text-1'])
 
931
 
 
932
    def test_ancestry(self):
 
933
        """Store in knit with parents"""
 
934
        k = self.make_test_knit()
 
935
        self.add_stock_one_and_one_a(k)
 
936
        self.assertEquals(set(k.get_ancestry(['text-1a'])), set(['text-1a', 'text-1']))
 
937
 
 
938
    def test_add_delta(self):
 
939
        """Store in knit with parents"""
 
940
        k = KnitVersionedFile('test', get_transport('.'), factory=KnitPlainFactory(),
 
941
            delta=True, create=True)
 
942
        self.add_stock_one_and_one_a(k)
 
943
        k.clear_cache()
 
944
        self.assertEqualDiff(''.join(k.get_lines('text-1a')), TEXT_1A)
 
945
 
 
946
    def test_annotate(self):
 
947
        """Annotations"""
 
948
        k = KnitVersionedFile('knit', get_transport('.'), factory=KnitAnnotateFactory(),
 
949
            delta=True, create=True)
 
950
        self.insert_and_test_small_annotate(k)
 
951
 
 
952
    def insert_and_test_small_annotate(self, k):
 
953
        """test annotation with k works correctly."""
 
954
        k.add_lines('text-1', [], ['a\n', 'b\n'])
 
955
        k.add_lines('text-2', ['text-1'], ['a\n', 'c\n'])
 
956
 
 
957
        origins = k.annotate('text-2')
 
958
        self.assertEquals(origins[0], ('text-1', 'a\n'))
 
959
        self.assertEquals(origins[1], ('text-2', 'c\n'))
 
960
 
 
961
    def test_annotate_fulltext(self):
 
962
        """Annotations"""
 
963
        k = KnitVersionedFile('knit', get_transport('.'), factory=KnitAnnotateFactory(),
 
964
            delta=False, create=True)
 
965
        self.insert_and_test_small_annotate(k)
 
966
 
 
967
    def test_annotate_merge_1(self):
 
968
        k = self.make_test_knit(True)
 
969
        k.add_lines('text-a1', [], ['a\n', 'b\n'])
 
970
        k.add_lines('text-a2', [], ['d\n', 'c\n'])
 
971
        k.add_lines('text-am', ['text-a1', 'text-a2'], ['d\n', 'b\n'])
 
972
        origins = k.annotate('text-am')
 
973
        self.assertEquals(origins[0], ('text-a2', 'd\n'))
 
974
        self.assertEquals(origins[1], ('text-a1', 'b\n'))
 
975
 
 
976
    def test_annotate_merge_2(self):
 
977
        k = self.make_test_knit(True)
 
978
        k.add_lines('text-a1', [], ['a\n', 'b\n', 'c\n'])
 
979
        k.add_lines('text-a2', [], ['x\n', 'y\n', 'z\n'])
 
980
        k.add_lines('text-am', ['text-a1', 'text-a2'], ['a\n', 'y\n', 'c\n'])
 
981
        origins = k.annotate('text-am')
 
982
        self.assertEquals(origins[0], ('text-a1', 'a\n'))
 
983
        self.assertEquals(origins[1], ('text-a2', 'y\n'))
 
984
        self.assertEquals(origins[2], ('text-a1', 'c\n'))
 
985
 
 
986
    def test_annotate_merge_9(self):
 
987
        k = self.make_test_knit(True)
 
988
        k.add_lines('text-a1', [], ['a\n', 'b\n', 'c\n'])
 
989
        k.add_lines('text-a2', [], ['x\n', 'y\n', 'z\n'])
 
990
        k.add_lines('text-am', ['text-a1', 'text-a2'], ['k\n', 'y\n', 'c\n'])
 
991
        origins = k.annotate('text-am')
 
992
        self.assertEquals(origins[0], ('text-am', 'k\n'))
 
993
        self.assertEquals(origins[1], ('text-a2', 'y\n'))
 
994
        self.assertEquals(origins[2], ('text-a1', 'c\n'))
 
995
 
 
996
    def test_annotate_merge_3(self):
 
997
        k = self.make_test_knit(True)
 
998
        k.add_lines('text-a1', [], ['a\n', 'b\n', 'c\n'])
 
999
        k.add_lines('text-a2', [] ,['x\n', 'y\n', 'z\n'])
 
1000
        k.add_lines('text-am', ['text-a1', 'text-a2'], ['k\n', 'y\n', 'z\n'])
 
1001
        origins = k.annotate('text-am')
 
1002
        self.assertEquals(origins[0], ('text-am', 'k\n'))
 
1003
        self.assertEquals(origins[1], ('text-a2', 'y\n'))
 
1004
        self.assertEquals(origins[2], ('text-a2', 'z\n'))
 
1005
 
 
1006
    def test_annotate_merge_4(self):
 
1007
        k = self.make_test_knit(True)
 
1008
        k.add_lines('text-a1', [], ['a\n', 'b\n', 'c\n'])
 
1009
        k.add_lines('text-a2', [], ['x\n', 'y\n', 'z\n'])
 
1010
        k.add_lines('text-a3', ['text-a1'], ['a\n', 'b\n', 'p\n'])
 
1011
        k.add_lines('text-am', ['text-a2', 'text-a3'], ['a\n', 'b\n', 'z\n'])
 
1012
        origins = k.annotate('text-am')
 
1013
        self.assertEquals(origins[0], ('text-a1', 'a\n'))
 
1014
        self.assertEquals(origins[1], ('text-a1', 'b\n'))
 
1015
        self.assertEquals(origins[2], ('text-a2', 'z\n'))
 
1016
 
 
1017
    def test_annotate_merge_5(self):
 
1018
        k = self.make_test_knit(True)
 
1019
        k.add_lines('text-a1', [], ['a\n', 'b\n', 'c\n'])
 
1020
        k.add_lines('text-a2', [], ['d\n', 'e\n', 'f\n'])
 
1021
        k.add_lines('text-a3', [], ['x\n', 'y\n', 'z\n'])
 
1022
        k.add_lines('text-am',
 
1023
                    ['text-a1', 'text-a2', 'text-a3'],
 
1024
                    ['a\n', 'e\n', 'z\n'])
 
1025
        origins = k.annotate('text-am')
 
1026
        self.assertEquals(origins[0], ('text-a1', 'a\n'))
 
1027
        self.assertEquals(origins[1], ('text-a2', 'e\n'))
 
1028
        self.assertEquals(origins[2], ('text-a3', 'z\n'))
 
1029
 
 
1030
    def test_annotate_file_cherry_pick(self):
 
1031
        k = self.make_test_knit(True)
 
1032
        k.add_lines('text-1', [], ['a\n', 'b\n', 'c\n'])
 
1033
        k.add_lines('text-2', ['text-1'], ['d\n', 'e\n', 'f\n'])
 
1034
        k.add_lines('text-3', ['text-2', 'text-1'], ['a\n', 'b\n', 'c\n'])
 
1035
        origins = k.annotate('text-3')
 
1036
        self.assertEquals(origins[0], ('text-1', 'a\n'))
 
1037
        self.assertEquals(origins[1], ('text-1', 'b\n'))
 
1038
        self.assertEquals(origins[2], ('text-1', 'c\n'))
 
1039
 
 
1040
    def test_knit_join(self):
 
1041
        """Store in knit with parents"""
 
1042
        k1 = KnitVersionedFile('test1', get_transport('.'), factory=KnitPlainFactory(), create=True)
 
1043
        k1.add_lines('text-a', [], split_lines(TEXT_1))
 
1044
        k1.add_lines('text-b', ['text-a'], split_lines(TEXT_1))
 
1045
 
 
1046
        k1.add_lines('text-c', [], split_lines(TEXT_1))
 
1047
        k1.add_lines('text-d', ['text-c'], split_lines(TEXT_1))
 
1048
 
 
1049
        k1.add_lines('text-m', ['text-b', 'text-d'], split_lines(TEXT_1))
 
1050
 
 
1051
        k2 = KnitVersionedFile('test2', get_transport('.'), factory=KnitPlainFactory(), create=True)
 
1052
        count = k2.join(k1, version_ids=['text-m'])
 
1053
        self.assertEquals(count, 5)
 
1054
        self.assertTrue(k2.has_version('text-a'))
 
1055
        self.assertTrue(k2.has_version('text-c'))
 
1056
 
 
1057
    def test_reannotate(self):
 
1058
        k1 = KnitVersionedFile('knit1', get_transport('.'),
 
1059
                               factory=KnitAnnotateFactory(), create=True)
 
1060
        # 0
 
1061
        k1.add_lines('text-a', [], ['a\n', 'b\n'])
 
1062
        # 1
 
1063
        k1.add_lines('text-b', ['text-a'], ['a\n', 'c\n'])
 
1064
 
 
1065
        k2 = KnitVersionedFile('test2', get_transport('.'),
 
1066
                               factory=KnitAnnotateFactory(), create=True)
 
1067
        k2.join(k1, version_ids=['text-b'])
 
1068
 
 
1069
        # 2
 
1070
        k1.add_lines('text-X', ['text-b'], ['a\n', 'b\n'])
 
1071
        # 2
 
1072
        k2.add_lines('text-c', ['text-b'], ['z\n', 'c\n'])
 
1073
        # 3
 
1074
        k2.add_lines('text-Y', ['text-b'], ['b\n', 'c\n'])
 
1075
 
 
1076
        # test-c will have index 3
 
1077
        k1.join(k2, version_ids=['text-c'])
 
1078
 
 
1079
        lines = k1.get_lines('text-c')
 
1080
        self.assertEquals(lines, ['z\n', 'c\n'])
 
1081
 
 
1082
        origins = k1.annotate('text-c')
 
1083
        self.assertEquals(origins[0], ('text-c', 'z\n'))
 
1084
        self.assertEquals(origins[1], ('text-b', 'c\n'))
 
1085
 
 
1086
    def test_get_line_delta_texts(self):
 
1087
        """Make sure we can call get_texts on text with reused line deltas"""
 
1088
        k1 = KnitVersionedFile('test1', get_transport('.'), 
 
1089
                               factory=KnitPlainFactory(), create=True)
 
1090
        for t in range(3):
 
1091
            if t == 0:
 
1092
                parents = []
 
1093
            else:
 
1094
                parents = ['%d' % (t-1)]
 
1095
            k1.add_lines('%d' % t, parents, ['hello\n'] * t)
 
1096
        k1.get_texts(('%d' % t) for t in range(3))
 
1097
        
 
1098
    def test_iter_lines_reads_in_order(self):
 
1099
        t = MemoryTransport()
 
1100
        instrumented_t = TransportLogger(t)
 
1101
        k1 = KnitVersionedFile('id', instrumented_t, create=True, delta=True)
 
1102
        self.assertEqual([('id.kndx',)], instrumented_t._calls)
 
1103
        # add texts with no required ordering
 
1104
        k1.add_lines('base', [], ['text\n'])
 
1105
        k1.add_lines('base2', [], ['text2\n'])
 
1106
        k1.clear_cache()
 
1107
        instrumented_t._calls = []
 
1108
        # request a last-first iteration
 
1109
        results = list(k1.iter_lines_added_or_present_in_versions(['base2', 'base']))
 
1110
        self.assertEqual([('id.knit', [(0, 87), (87, 89)])], instrumented_t._calls)
 
1111
        self.assertEqual(['text\n', 'text2\n'], results)
 
1112
 
 
1113
    def test_create_empty_annotated(self):
 
1114
        k1 = self.make_test_knit(True)
 
1115
        # 0
 
1116
        k1.add_lines('text-a', [], ['a\n', 'b\n'])
 
1117
        k2 = k1.create_empty('t', MemoryTransport())
 
1118
        self.assertTrue(isinstance(k2.factory, KnitAnnotateFactory))
 
1119
        self.assertEqual(k1.delta, k2.delta)
 
1120
        # the generic test checks for empty content and file class
 
1121
 
 
1122
    def test_knit_format(self):
 
1123
        # this tests that a new knit index file has the expected content
 
1124
        # and that is writes the data we expect as records are added.
 
1125
        knit = self.make_test_knit(True)
 
1126
        # Now knit files are not created until we first add data to them
 
1127
        self.assertFileEqual("# bzr knit index 8\n", 'test.kndx')
 
1128
        knit.add_lines_with_ghosts('revid', ['a_ghost'], ['a\n'])
 
1129
        self.assertFileEqual(
 
1130
            "# bzr knit index 8\n"
 
1131
            "\n"
 
1132
            "revid fulltext 0 84 .a_ghost :",
 
1133
            'test.kndx')
 
1134
        knit.add_lines_with_ghosts('revid2', ['revid'], ['a\n'])
 
1135
        self.assertFileEqual(
 
1136
            "# bzr knit index 8\n"
 
1137
            "\nrevid fulltext 0 84 .a_ghost :"
 
1138
            "\nrevid2 line-delta 84 82 0 :",
 
1139
            'test.kndx')
 
1140
        # we should be able to load this file again
 
1141
        knit = KnitVersionedFile('test', get_transport('.'), access_mode='r')
 
1142
        self.assertEqual(['revid', 'revid2'], knit.versions())
 
1143
        # write a short write to the file and ensure that its ignored
 
1144
        indexfile = file('test.kndx', 'ab')
 
1145
        indexfile.write('\nrevid3 line-delta 166 82 1 2 3 4 5 .phwoar:demo ')
 
1146
        indexfile.close()
 
1147
        # we should be able to load this file again
 
1148
        knit = KnitVersionedFile('test', get_transport('.'), access_mode='w')
 
1149
        self.assertEqual(['revid', 'revid2'], knit.versions())
 
1150
        # and add a revision with the same id the failed write had
 
1151
        knit.add_lines('revid3', ['revid2'], ['a\n'])
 
1152
        # and when reading it revid3 should now appear.
 
1153
        knit = KnitVersionedFile('test', get_transport('.'), access_mode='r')
 
1154
        self.assertEqual(['revid', 'revid2', 'revid3'], knit.versions())
 
1155
        self.assertEqual(['revid2'], knit.get_parents('revid3'))
 
1156
 
 
1157
    def test_delay_create(self):
 
1158
        """Test that passing delay_create=True creates files late"""
 
1159
        knit = self.make_test_knit(annotate=True, delay_create=True)
 
1160
        self.failIfExists('test.knit')
 
1161
        self.failIfExists('test.kndx')
 
1162
        knit.add_lines_with_ghosts('revid', ['a_ghost'], ['a\n'])
 
1163
        self.failUnlessExists('test.knit')
 
1164
        self.assertFileEqual(
 
1165
            "# bzr knit index 8\n"
 
1166
            "\n"
 
1167
            "revid fulltext 0 84 .a_ghost :",
 
1168
            'test.kndx')
 
1169
 
 
1170
    def test_create_parent_dir(self):
 
1171
        """create_parent_dir can create knits in nonexistant dirs"""
 
1172
        # Has no effect if we don't set 'delay_create'
 
1173
        trans = get_transport('.')
 
1174
        self.assertRaises(NoSuchFile, KnitVersionedFile, 'dir/test',
 
1175
                          trans, access_mode='w', factory=None,
 
1176
                          create=True, create_parent_dir=True)
 
1177
        # Nothing should have changed yet
 
1178
        knit = KnitVersionedFile('dir/test', trans, access_mode='w',
 
1179
                                 factory=None, create=True,
 
1180
                                 create_parent_dir=True,
 
1181
                                 delay_create=True)
 
1182
        self.failIfExists('dir/test.knit')
 
1183
        self.failIfExists('dir/test.kndx')
 
1184
        self.failIfExists('dir')
 
1185
        knit.add_lines('revid', [], ['a\n'])
 
1186
        self.failUnlessExists('dir')
 
1187
        self.failUnlessExists('dir/test.knit')
 
1188
        self.assertFileEqual(
 
1189
            "# bzr knit index 8\n"
 
1190
            "\n"
 
1191
            "revid fulltext 0 84  :",
 
1192
            'dir/test.kndx')
 
1193
 
 
1194
    def test_create_mode_700(self):
 
1195
        trans = get_transport('.')
 
1196
        if not trans._can_roundtrip_unix_modebits():
 
1197
            # Can't roundtrip, so no need to run this test
 
1198
            return
 
1199
        knit = KnitVersionedFile('dir/test', trans, access_mode='w',
 
1200
                                 factory=None, create=True,
 
1201
                                 create_parent_dir=True,
 
1202
                                 delay_create=True,
 
1203
                                 file_mode=0600,
 
1204
                                 dir_mode=0700)
 
1205
        knit.add_lines('revid', [], ['a\n'])
 
1206
        self.assertTransportMode(trans, 'dir', 0700)
 
1207
        self.assertTransportMode(trans, 'dir/test.knit', 0600)
 
1208
        self.assertTransportMode(trans, 'dir/test.kndx', 0600)
 
1209
 
 
1210
    def test_create_mode_770(self):
 
1211
        trans = get_transport('.')
 
1212
        if not trans._can_roundtrip_unix_modebits():
 
1213
            # Can't roundtrip, so no need to run this test
 
1214
            return
 
1215
        knit = KnitVersionedFile('dir/test', trans, access_mode='w',
 
1216
                                 factory=None, create=True,
 
1217
                                 create_parent_dir=True,
 
1218
                                 delay_create=True,
 
1219
                                 file_mode=0660,
 
1220
                                 dir_mode=0770)
 
1221
        knit.add_lines('revid', [], ['a\n'])
 
1222
        self.assertTransportMode(trans, 'dir', 0770)
 
1223
        self.assertTransportMode(trans, 'dir/test.knit', 0660)
 
1224
        self.assertTransportMode(trans, 'dir/test.kndx', 0660)
 
1225
 
 
1226
    def test_create_mode_777(self):
 
1227
        trans = get_transport('.')
 
1228
        if not trans._can_roundtrip_unix_modebits():
 
1229
            # Can't roundtrip, so no need to run this test
 
1230
            return
 
1231
        knit = KnitVersionedFile('dir/test', trans, access_mode='w',
 
1232
                                 factory=None, create=True,
 
1233
                                 create_parent_dir=True,
 
1234
                                 delay_create=True,
 
1235
                                 file_mode=0666,
 
1236
                                 dir_mode=0777)
 
1237
        knit.add_lines('revid', [], ['a\n'])
 
1238
        self.assertTransportMode(trans, 'dir', 0777)
 
1239
        self.assertTransportMode(trans, 'dir/test.knit', 0666)
 
1240
        self.assertTransportMode(trans, 'dir/test.kndx', 0666)
 
1241
 
 
1242
    def test_plan_merge(self):
 
1243
        my_knit = self.make_test_knit(annotate=True)
 
1244
        my_knit.add_lines('text1', [], split_lines(TEXT_1))
 
1245
        my_knit.add_lines('text1a', ['text1'], split_lines(TEXT_1A))
 
1246
        my_knit.add_lines('text1b', ['text1'], split_lines(TEXT_1B))
 
1247
        plan = list(my_knit.plan_merge('text1a', 'text1b'))
 
1248
        for plan_line, expected_line in zip(plan, AB_MERGE):
 
1249
            self.assertEqual(plan_line, expected_line)
 
1250
 
 
1251
 
 
1252
TEXT_1 = """\
 
1253
Banana cup cakes:
 
1254
 
 
1255
- bananas
 
1256
- eggs
 
1257
- broken tea cups
 
1258
"""
 
1259
 
 
1260
TEXT_1A = """\
 
1261
Banana cup cake recipe
 
1262
(serves 6)
 
1263
 
 
1264
- bananas
 
1265
- eggs
 
1266
- broken tea cups
 
1267
- self-raising flour
 
1268
"""
 
1269
 
 
1270
TEXT_1B = """\
 
1271
Banana cup cake recipe
 
1272
 
 
1273
- bananas (do not use plantains!!!)
 
1274
- broken tea cups
 
1275
- flour
 
1276
"""
 
1277
 
 
1278
delta_1_1a = """\
 
1279
0,1,2
 
1280
Banana cup cake recipe
 
1281
(serves 6)
 
1282
5,5,1
 
1283
- self-raising flour
 
1284
"""
 
1285
 
 
1286
TEXT_2 = """\
 
1287
Boeuf bourguignon
 
1288
 
 
1289
- beef
 
1290
- red wine
 
1291
- small onions
 
1292
- carrot
 
1293
- mushrooms
 
1294
"""
 
1295
 
 
1296
AB_MERGE_TEXT="""unchanged|Banana cup cake recipe
 
1297
new-a|(serves 6)
 
1298
unchanged|
 
1299
killed-b|- bananas
 
1300
killed-b|- eggs
 
1301
new-b|- bananas (do not use plantains!!!)
 
1302
unchanged|- broken tea cups
 
1303
new-a|- self-raising flour
 
1304
new-b|- flour
 
1305
"""
 
1306
AB_MERGE=[tuple(l.split('|')) for l in AB_MERGE_TEXT.splitlines(True)]
 
1307
 
 
1308
 
 
1309
def line_delta(from_lines, to_lines):
 
1310
    """Generate line-based delta from one text to another"""
 
1311
    s = difflib.SequenceMatcher(None, from_lines, to_lines)
 
1312
    for op in s.get_opcodes():
 
1313
        if op[0] == 'equal':
 
1314
            continue
 
1315
        yield '%d,%d,%d\n' % (op[1], op[2], op[4]-op[3])
 
1316
        for i in range(op[3], op[4]):
 
1317
            yield to_lines[i]
 
1318
 
 
1319
 
 
1320
def apply_line_delta(basis_lines, delta_lines):
 
1321
    """Apply a line-based perfect diff
 
1322
    
 
1323
    basis_lines -- text to apply the patch to
 
1324
    delta_lines -- diff instructions and content
 
1325
    """
 
1326
    out = basis_lines[:]
 
1327
    i = 0
 
1328
    offset = 0
 
1329
    while i < len(delta_lines):
 
1330
        l = delta_lines[i]
 
1331
        a, b, c = map(long, l.split(','))
 
1332
        i = i + 1
 
1333
        out[offset+a:offset+b] = delta_lines[i:i+c]
 
1334
        i = i + c
 
1335
        offset = offset + (b - a) + c
 
1336
    return out
 
1337
 
 
1338
 
 
1339
class TestWeaveToKnit(KnitTests):
 
1340
 
 
1341
    def test_weave_to_knit_matches(self):
 
1342
        # check that the WeaveToKnit is_compatible function
 
1343
        # registers True for a Weave to a Knit.
 
1344
        w = Weave()
 
1345
        k = self.make_test_knit()
 
1346
        self.failUnless(WeaveToKnit.is_compatible(w, k))
 
1347
        self.failIf(WeaveToKnit.is_compatible(k, w))
 
1348
        self.failIf(WeaveToKnit.is_compatible(w, w))
 
1349
        self.failIf(WeaveToKnit.is_compatible(k, k))
 
1350
 
 
1351
 
 
1352
class TestKnitCaching(KnitTests):
 
1353
    
 
1354
    def create_knit(self, cache_add=False):
 
1355
        k = self.make_test_knit(True)
 
1356
        if cache_add:
 
1357
            k.enable_cache()
 
1358
 
 
1359
        k.add_lines('text-1', [], split_lines(TEXT_1))
 
1360
        k.add_lines('text-2', [], split_lines(TEXT_2))
 
1361
        return k
 
1362
 
 
1363
    def test_no_caching(self):
 
1364
        k = self.create_knit()
 
1365
        # Nothing should be cached without setting 'enable_cache'
 
1366
        self.assertEqual({}, k._data._cache)
 
1367
 
 
1368
    def test_cache_add_and_clear(self):
 
1369
        k = self.create_knit(True)
 
1370
 
 
1371
        self.assertEqual(['text-1', 'text-2'], sorted(k._data._cache.keys()))
 
1372
 
 
1373
        k.clear_cache()
 
1374
        self.assertEqual({}, k._data._cache)
 
1375
 
 
1376
    def test_cache_data_read_raw(self):
 
1377
        k = self.create_knit()
 
1378
 
 
1379
        # Now cache and read
 
1380
        k.enable_cache()
 
1381
 
 
1382
        def read_one_raw(version):
 
1383
            pos_map = k._get_components_positions([version])
 
1384
            method, pos, size, next = pos_map[version]
 
1385
            lst = list(k._data.read_records_iter_raw([(version, pos, size)]))
 
1386
            self.assertEqual(1, len(lst))
 
1387
            return lst[0]
 
1388
 
 
1389
        val = read_one_raw('text-1')
 
1390
        self.assertEqual({'text-1':val[1]}, k._data._cache)
 
1391
 
 
1392
        k.clear_cache()
 
1393
        # After clear, new reads are not cached
 
1394
        self.assertEqual({}, k._data._cache)
 
1395
 
 
1396
        val2 = read_one_raw('text-1')
 
1397
        self.assertEqual(val, val2)
 
1398
        self.assertEqual({}, k._data._cache)
 
1399
 
 
1400
    def test_cache_data_read(self):
 
1401
        k = self.create_knit()
 
1402
 
 
1403
        def read_one(version):
 
1404
            pos_map = k._get_components_positions([version])
 
1405
            method, pos, size, next = pos_map[version]
 
1406
            lst = list(k._data.read_records_iter([(version, pos, size)]))
 
1407
            self.assertEqual(1, len(lst))
 
1408
            return lst[0]
 
1409
 
 
1410
        # Now cache and read
 
1411
        k.enable_cache()
 
1412
 
 
1413
        val = read_one('text-2')
 
1414
        self.assertEqual(['text-2'], k._data._cache.keys())
 
1415
        self.assertEqual('text-2', val[0])
 
1416
        content, digest = k._data._parse_record('text-2',
 
1417
                                                k._data._cache['text-2'])
 
1418
        self.assertEqual(content, val[1])
 
1419
        self.assertEqual(digest, val[2])
 
1420
 
 
1421
        k.clear_cache()
 
1422
        self.assertEqual({}, k._data._cache)
 
1423
 
 
1424
        val2 = read_one('text-2')
 
1425
        self.assertEqual(val, val2)
 
1426
        self.assertEqual({}, k._data._cache)
 
1427
 
 
1428
    def test_cache_read(self):
 
1429
        k = self.create_knit()
 
1430
        k.enable_cache()
 
1431
 
 
1432
        text = k.get_text('text-1')
 
1433
        self.assertEqual(TEXT_1, text)
 
1434
        self.assertEqual(['text-1'], k._data._cache.keys())
 
1435
 
 
1436
        k.clear_cache()
 
1437
        self.assertEqual({}, k._data._cache)
 
1438
 
 
1439
        text = k.get_text('text-1')
 
1440
        self.assertEqual(TEXT_1, text)
 
1441
        self.assertEqual({}, k._data._cache)
 
1442
 
 
1443
 
 
1444
class TestKnitIndex(KnitTests):
 
1445
 
 
1446
    def test_add_versions_dictionary_compresses(self):
 
1447
        """Adding versions to the index should update the lookup dict"""
 
1448
        knit = self.make_test_knit()
 
1449
        idx = knit._index
 
1450
        idx.add_version('a-1', ['fulltext'], 0, 0, [])
 
1451
        self.check_file_contents('test.kndx',
 
1452
            '# bzr knit index 8\n'
 
1453
            '\n'
 
1454
            'a-1 fulltext 0 0  :'
 
1455
            )
 
1456
        idx.add_versions([('a-2', ['fulltext'], 0, 0, ['a-1']),
 
1457
                          ('a-3', ['fulltext'], 0, 0, ['a-2']),
 
1458
                         ])
 
1459
        self.check_file_contents('test.kndx',
 
1460
            '# bzr knit index 8\n'
 
1461
            '\n'
 
1462
            'a-1 fulltext 0 0  :\n'
 
1463
            'a-2 fulltext 0 0 0 :\n'
 
1464
            'a-3 fulltext 0 0 1 :'
 
1465
            )
 
1466
        self.assertEqual(['a-1', 'a-2', 'a-3'], idx._history)
 
1467
        self.assertEqual({'a-1':('a-1', ['fulltext'], 0, 0, [], 0),
 
1468
                          'a-2':('a-2', ['fulltext'], 0, 0, ['a-1'], 1),
 
1469
                          'a-3':('a-3', ['fulltext'], 0, 0, ['a-2'], 2),
 
1470
                         }, idx._cache)
 
1471
 
 
1472
    def test_add_versions_fails_clean(self):
 
1473
        """If add_versions fails in the middle, it restores a pristine state.
 
1474
 
 
1475
        Any modifications that are made to the index are reset if all versions
 
1476
        cannot be added.
 
1477
        """
 
1478
        # This cheats a little bit by passing in a generator which will
 
1479
        # raise an exception before the processing finishes
 
1480
        # Other possibilities would be to have an version with the wrong number
 
1481
        # of entries, or to make the backing transport unable to write any
 
1482
        # files.
 
1483
 
 
1484
        knit = self.make_test_knit()
 
1485
        idx = knit._index
 
1486
        idx.add_version('a-1', ['fulltext'], 0, 0, [])
 
1487
 
 
1488
        class StopEarly(Exception):
 
1489
            pass
 
1490
 
 
1491
        def generate_failure():
 
1492
            """Add some entries and then raise an exception"""
 
1493
            yield ('a-2', ['fulltext'], 0, 0, ['a-1'])
 
1494
            yield ('a-3', ['fulltext'], 0, 0, ['a-2'])
 
1495
            raise StopEarly()
 
1496
 
 
1497
        # Assert the pre-condition
 
1498
        self.assertEqual(['a-1'], idx._history)
 
1499
        self.assertEqual({'a-1':('a-1', ['fulltext'], 0, 0, [], 0)}, idx._cache)
 
1500
 
 
1501
        self.assertRaises(StopEarly, idx.add_versions, generate_failure())
 
1502
 
 
1503
        # And it shouldn't be modified
 
1504
        self.assertEqual(['a-1'], idx._history)
 
1505
        self.assertEqual({'a-1':('a-1', ['fulltext'], 0, 0, [], 0)}, idx._cache)
 
1506
 
 
1507
    def test_knit_index_ignores_empty_files(self):
 
1508
        # There was a race condition in older bzr, where a ^C at the right time
 
1509
        # could leave an empty .kndx file, which bzr would later claim was a
 
1510
        # corrupted file since the header was not present. In reality, the file
 
1511
        # just wasn't created, so it should be ignored.
 
1512
        t = get_transport('.')
 
1513
        t.put_bytes('test.kndx', '')
 
1514
 
 
1515
        knit = self.make_test_knit()
 
1516
 
 
1517
    def test_knit_index_checks_header(self):
 
1518
        t = get_transport('.')
 
1519
        t.put_bytes('test.kndx', '# not really a knit header\n\n')
 
1520
 
 
1521
        self.assertRaises(KnitHeaderError, self.make_test_knit)