~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_knit.py

Exclude more files from dumb-rsync upload

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2005, 2006 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
 
 
24
 
from bzrlib import (
25
 
    errors,
26
 
    )
27
 
from bzrlib.errors import (
28
 
    RevisionAlreadyPresent,
29
 
    KnitHeaderError,
30
 
    RevisionNotPresent,
31
 
    NoSuchFile,
32
 
    )
33
 
from bzrlib.knit import (
34
 
    KnitContent,
35
 
    KnitVersionedFile,
36
 
    KnitPlainFactory,
37
 
    KnitAnnotateFactory,
38
 
    _KnitData,
39
 
    _KnitIndex,
40
 
    WeaveToKnit,
41
 
    )
42
 
from bzrlib.osutils import split_lines
43
 
from bzrlib.tests import TestCase, TestCaseWithTransport
44
 
from bzrlib.transport import TransportLogger, get_transport
45
 
from bzrlib.transport.memory import MemoryTransport
46
 
from bzrlib.weave import Weave
47
 
 
48
 
 
49
 
class KnitContentTests(TestCase):
50
 
 
51
 
    def test_constructor(self):
52
 
        content = KnitContent([])
53
 
 
54
 
    def test_text(self):
55
 
        content = KnitContent([])
56
 
        self.assertEqual(content.text(), [])
57
 
 
58
 
        content = KnitContent([("origin1", "text1"), ("origin2", "text2")])
59
 
        self.assertEqual(content.text(), ["text1", "text2"])
60
 
 
61
 
    def test_annotate(self):
62
 
        content = KnitContent([])
63
 
        self.assertEqual(content.annotate(), [])
64
 
 
65
 
        content = KnitContent([("origin1", "text1"), ("origin2", "text2")])
66
 
        self.assertEqual(content.annotate(),
67
 
            [("origin1", "text1"), ("origin2", "text2")])
68
 
 
69
 
    def test_annotate_iter(self):
70
 
        content = KnitContent([])
71
 
        it = content.annotate_iter()
72
 
        self.assertRaises(StopIteration, it.next)
73
 
 
74
 
        content = KnitContent([("origin1", "text1"), ("origin2", "text2")])
75
 
        it = content.annotate_iter()
76
 
        self.assertEqual(it.next(), ("origin1", "text1"))
77
 
        self.assertEqual(it.next(), ("origin2", "text2"))
78
 
        self.assertRaises(StopIteration, it.next)
79
 
 
80
 
    def test_copy(self):
81
 
        content = KnitContent([("origin1", "text1"), ("origin2", "text2")])
82
 
        copy = content.copy()
83
 
        self.assertIsInstance(copy, KnitContent)
84
 
        self.assertEqual(copy.annotate(),
85
 
            [("origin1", "text1"), ("origin2", "text2")])
86
 
 
87
 
    def test_line_delta(self):
88
 
        content1 = KnitContent([("", "a"), ("", "b")])
89
 
        content2 = KnitContent([("", "a"), ("", "a"), ("", "c")])
90
 
        self.assertEqual(content1.line_delta(content2),
91
 
            [(1, 2, 2, [("", "a"), ("", "c")])])
92
 
 
93
 
    def test_line_delta_iter(self):
94
 
        content1 = KnitContent([("", "a"), ("", "b")])
95
 
        content2 = KnitContent([("", "a"), ("", "a"), ("", "c")])
96
 
        it = content1.line_delta_iter(content2)
97
 
        self.assertEqual(it.next(), (1, 2, 2, [("", "a"), ("", "c")]))
98
 
        self.assertRaises(StopIteration, it.next)
99
 
 
100
 
 
101
 
class MockTransport(object):
102
 
 
103
 
    def __init__(self, file_lines=None):
104
 
        self.file_lines = file_lines
105
 
        self.calls = []
106
 
        # We have no base directory for the MockTransport
107
 
        self.base = ''
108
 
 
109
 
    def get(self, filename):
110
 
        if self.file_lines is None:
111
 
            raise NoSuchFile(filename)
112
 
        else:
113
 
            return StringIO("\n".join(self.file_lines))
114
 
 
115
 
    def readv(self, relpath, offsets):
116
 
        fp = self.get(relpath)
117
 
        for offset, size in offsets:
118
 
            fp.seek(offset)
119
 
            yield offset, fp.read(size)
120
 
 
121
 
    def __getattr__(self, name):
122
 
        def queue_call(*args, **kwargs):
123
 
            self.calls.append((name, args, kwargs))
124
 
        return queue_call
125
 
 
126
 
 
127
 
class LowLevelKnitDataTests(TestCase):
128
 
 
129
 
    def create_gz_content(self, text):
130
 
        sio = StringIO()
131
 
        gz_file = gzip.GzipFile(mode='wb', fileobj=sio)
132
 
        gz_file.write(text)
133
 
        gz_file.close()
134
 
        return sio.getvalue()
135
 
 
136
 
    def test_valid_knit_data(self):
137
 
        sha1sum = sha.new('foo\nbar\n').hexdigest()
138
 
        gz_txt = self.create_gz_content('version rev-id-1 2 %s\n'
139
 
                                        'foo\n'
140
 
                                        'bar\n'
141
 
                                        'end rev-id-1\n'
142
 
                                        % (sha1sum,))
143
 
        transport = MockTransport([gz_txt])
144
 
        data = _KnitData(transport, 'filename', mode='r')
145
 
        records = [('rev-id-1', 0, len(gz_txt))]
146
 
 
147
 
        contents = data.read_records(records)
148
 
        self.assertEqual({'rev-id-1':(['foo\n', 'bar\n'], sha1sum)}, contents)
149
 
 
150
 
        raw_contents = list(data.read_records_iter_raw(records))
151
 
        self.assertEqual([('rev-id-1', gz_txt)], raw_contents)
152
 
 
153
 
    def test_not_enough_lines(self):
154
 
        sha1sum = sha.new('foo\n').hexdigest()
155
 
        # record says 2 lines data says 1
156
 
        gz_txt = self.create_gz_content('version rev-id-1 2 %s\n'
157
 
                                        'foo\n'
158
 
                                        'end rev-id-1\n'
159
 
                                        % (sha1sum,))
160
 
        transport = MockTransport([gz_txt])
161
 
        data = _KnitData(transport, 'filename', mode='r')
162
 
        records = [('rev-id-1', 0, len(gz_txt))]
163
 
        self.assertRaises(errors.KnitCorrupt, data.read_records, records)
164
 
 
165
 
        # read_records_iter_raw won't detect that sort of mismatch/corruption
166
 
        raw_contents = list(data.read_records_iter_raw(records))
167
 
        self.assertEqual([('rev-id-1', gz_txt)], raw_contents)
168
 
 
169
 
    def test_too_many_lines(self):
170
 
        sha1sum = sha.new('foo\nbar\n').hexdigest()
171
 
        # record says 1 lines data says 2
172
 
        gz_txt = self.create_gz_content('version rev-id-1 1 %s\n'
173
 
                                        'foo\n'
174
 
                                        'bar\n'
175
 
                                        'end rev-id-1\n'
176
 
                                        % (sha1sum,))
177
 
        transport = MockTransport([gz_txt])
178
 
        data = _KnitData(transport, 'filename', mode='r')
179
 
        records = [('rev-id-1', 0, len(gz_txt))]
180
 
        self.assertRaises(errors.KnitCorrupt, data.read_records, records)
181
 
 
182
 
        # read_records_iter_raw won't detect that sort of mismatch/corruption
183
 
        raw_contents = list(data.read_records_iter_raw(records))
184
 
        self.assertEqual([('rev-id-1', gz_txt)], raw_contents)
185
 
 
186
 
    def test_mismatched_version_id(self):
187
 
        sha1sum = sha.new('foo\nbar\n').hexdigest()
188
 
        gz_txt = self.create_gz_content('version rev-id-1 2 %s\n'
189
 
                                        'foo\n'
190
 
                                        'bar\n'
191
 
                                        'end rev-id-1\n'
192
 
                                        % (sha1sum,))
193
 
        transport = MockTransport([gz_txt])
194
 
        data = _KnitData(transport, 'filename', mode='r')
195
 
        # We are asking for rev-id-2, but the data is rev-id-1
196
 
        records = [('rev-id-2', 0, len(gz_txt))]
197
 
        self.assertRaises(errors.KnitCorrupt, data.read_records, records)
198
 
 
199
 
        # read_records_iter_raw will notice if we request the wrong version.
200
 
        self.assertRaises(errors.KnitCorrupt, list,
201
 
                          data.read_records_iter_raw(records))
202
 
 
203
 
    def test_uncompressed_data(self):
204
 
        sha1sum = sha.new('foo\nbar\n').hexdigest()
205
 
        txt = ('version rev-id-1 2 %s\n'
206
 
               'foo\n'
207
 
               'bar\n'
208
 
               'end rev-id-1\n'
209
 
               % (sha1sum,))
210
 
        transport = MockTransport([txt])
211
 
        data = _KnitData(transport, 'filename', mode='r')
212
 
        records = [('rev-id-1', 0, len(txt))]
213
 
 
214
 
        # We don't have valid gzip data ==> corrupt
215
 
        self.assertRaises(errors.KnitCorrupt, data.read_records, records)
216
 
 
217
 
        # read_records_iter_raw will notice the bad data
218
 
        self.assertRaises(errors.KnitCorrupt, list,
219
 
                          data.read_records_iter_raw(records))
220
 
 
221
 
    def test_corrupted_data(self):
222
 
        sha1sum = sha.new('foo\nbar\n').hexdigest()
223
 
        gz_txt = self.create_gz_content('version rev-id-1 2 %s\n'
224
 
                                        'foo\n'
225
 
                                        'bar\n'
226
 
                                        'end rev-id-1\n'
227
 
                                        % (sha1sum,))
228
 
        # Change 2 bytes in the middle to \xff
229
 
        gz_txt = gz_txt[:10] + '\xff\xff' + gz_txt[12:]
230
 
        transport = MockTransport([gz_txt])
231
 
        data = _KnitData(transport, 'filename', mode='r')
232
 
        records = [('rev-id-1', 0, len(gz_txt))]
233
 
 
234
 
        self.assertRaises(errors.KnitCorrupt, data.read_records, records)
235
 
 
236
 
        # read_records_iter_raw will notice if we request the wrong version.
237
 
        self.assertRaises(errors.KnitCorrupt, list,
238
 
                          data.read_records_iter_raw(records))
239
 
 
240
 
 
241
 
class LowLevelKnitIndexTests(TestCase):
242
 
 
243
 
    def test_no_such_file(self):
244
 
        transport = MockTransport()
245
 
 
246
 
        self.assertRaises(NoSuchFile, _KnitIndex, transport, "filename", "r")
247
 
        self.assertRaises(NoSuchFile, _KnitIndex, transport,
248
 
            "filename", "w", create=False)
249
 
 
250
 
    def test_create_file(self):
251
 
        transport = MockTransport()
252
 
 
253
 
        index = _KnitIndex(transport, "filename", "w",
254
 
            file_mode="wb", create=True)
255
 
        self.assertEqual(
256
 
                ("put_bytes_non_atomic",
257
 
                    ("filename", index.HEADER), {"mode": "wb"}),
258
 
                transport.calls.pop(0))
259
 
 
260
 
    def test_delay_create_file(self):
261
 
        transport = MockTransport()
262
 
 
263
 
        index = _KnitIndex(transport, "filename", "w",
264
 
            create=True, file_mode="wb", create_parent_dir=True,
265
 
            delay_create=True, dir_mode=0777)
266
 
        self.assertEqual([], transport.calls)
267
 
 
268
 
        index.add_versions([])
269
 
        name, (filename, f), kwargs = transport.calls.pop(0)
270
 
        self.assertEqual("put_file_non_atomic", name)
271
 
        self.assertEqual(
272
 
            {"dir_mode": 0777, "create_parent_dir": True, "mode": "wb"},
273
 
            kwargs)
274
 
        self.assertEqual("filename", filename)
275
 
        self.assertEqual(index.HEADER, f.read())
276
 
 
277
 
        index.add_versions([])
278
 
        self.assertEqual(("append_bytes", ("filename", ""), {}),
279
 
            transport.calls.pop(0))
280
 
 
281
 
    def test_read_utf8_version_id(self):
282
 
        unicode_revision_id = u"version-\N{CYRILLIC CAPITAL LETTER A}"
283
 
        utf8_revision_id = unicode_revision_id.encode('utf-8')
284
 
        transport = MockTransport([
285
 
            _KnitIndex.HEADER,
286
 
            '%s option 0 1 :' % (utf8_revision_id,)
287
 
            ])
288
 
        index = _KnitIndex(transport, "filename", "r")
289
 
        # _KnitIndex is a private class, and deals in utf8 revision_ids, not
290
 
        # Unicode revision_ids.
291
 
        self.assertTrue(index.has_version(utf8_revision_id))
292
 
        self.assertFalse(index.has_version(unicode_revision_id))
293
 
 
294
 
    def test_read_utf8_parents(self):
295
 
        unicode_revision_id = u"version-\N{CYRILLIC CAPITAL LETTER A}"
296
 
        utf8_revision_id = unicode_revision_id.encode('utf-8')
297
 
        transport = MockTransport([
298
 
            _KnitIndex.HEADER,
299
 
            "version option 0 1 .%s :" % (utf8_revision_id,)
300
 
            ])
301
 
        index = _KnitIndex(transport, "filename", "r")
302
 
        self.assertEqual([utf8_revision_id],
303
 
            index.get_parents_with_ghosts("version"))
304
 
 
305
 
    def test_read_ignore_corrupted_lines(self):
306
 
        transport = MockTransport([
307
 
            _KnitIndex.HEADER,
308
 
            "corrupted",
309
 
            "corrupted options 0 1 .b .c ",
310
 
            "version options 0 1 :"
311
 
            ])
312
 
        index = _KnitIndex(transport, "filename", "r")
313
 
        self.assertEqual(1, index.num_versions())
314
 
        self.assertTrue(index.has_version("version"))
315
 
 
316
 
    def test_read_corrupted_header(self):
317
 
        transport = MockTransport(['not a bzr knit index header\n'])
318
 
        self.assertRaises(KnitHeaderError,
319
 
            _KnitIndex, transport, "filename", "r")
320
 
 
321
 
    def test_read_duplicate_entries(self):
322
 
        transport = MockTransport([
323
 
            _KnitIndex.HEADER,
324
 
            "parent options 0 1 :",
325
 
            "version options1 0 1 0 :",
326
 
            "version options2 1 2 .other :",
327
 
            "version options3 3 4 0 .other :"
328
 
            ])
329
 
        index = _KnitIndex(transport, "filename", "r")
330
 
        self.assertEqual(2, index.num_versions())
331
 
        self.assertEqual(1, index.lookup("version"))
332
 
        self.assertEqual((3, 4), index.get_position("version"))
333
 
        self.assertEqual(["options3"], index.get_options("version"))
334
 
        self.assertEqual(["parent", "other"],
335
 
            index.get_parents_with_ghosts("version"))
336
 
 
337
 
    def test_read_compressed_parents(self):
338
 
        transport = MockTransport([
339
 
            _KnitIndex.HEADER,
340
 
            "a option 0 1 :",
341
 
            "b option 0 1 0 :",
342
 
            "c option 0 1 1 0 :",
343
 
            ])
344
 
        index = _KnitIndex(transport, "filename", "r")
345
 
        self.assertEqual(["a"], index.get_parents("b"))
346
 
        self.assertEqual(["b", "a"], index.get_parents("c"))
347
 
 
348
 
    def test_write_utf8_version_id(self):
349
 
        unicode_revision_id = u"version-\N{CYRILLIC CAPITAL LETTER A}"
350
 
        utf8_revision_id = unicode_revision_id.encode('utf-8')
351
 
        transport = MockTransport([
352
 
            _KnitIndex.HEADER
353
 
            ])
354
 
        index = _KnitIndex(transport, "filename", "r")
355
 
        index.add_version(utf8_revision_id, ["option"], 0, 1, [])
356
 
        self.assertEqual(("append_bytes", ("filename",
357
 
            "\n%s option 0 1  :" % (utf8_revision_id,)),
358
 
            {}),
359
 
            transport.calls.pop(0))
360
 
 
361
 
    def test_write_utf8_parents(self):
362
 
        unicode_revision_id = u"version-\N{CYRILLIC CAPITAL LETTER A}"
363
 
        utf8_revision_id = unicode_revision_id.encode('utf-8')
364
 
        transport = MockTransport([
365
 
            _KnitIndex.HEADER
366
 
            ])
367
 
        index = _KnitIndex(transport, "filename", "r")
368
 
        index.add_version("version", ["option"], 0, 1, [utf8_revision_id])
369
 
        self.assertEqual(("append_bytes", ("filename",
370
 
            "\nversion option 0 1 .%s :" % (utf8_revision_id,)),
371
 
            {}),
372
 
            transport.calls.pop(0))
373
 
 
374
 
    def test_get_graph(self):
375
 
        transport = MockTransport()
376
 
        index = _KnitIndex(transport, "filename", "w", create=True)
377
 
        self.assertEqual([], index.get_graph())
378
 
 
379
 
        index.add_version("a", ["option"], 0, 1, ["b"])
380
 
        self.assertEqual([("a", ["b"])], index.get_graph())
381
 
 
382
 
        index.add_version("c", ["option"], 0, 1, ["d"])
383
 
        self.assertEqual([("a", ["b"]), ("c", ["d"])],
384
 
            sorted(index.get_graph()))
385
 
 
386
 
    def test_get_ancestry(self):
387
 
        transport = MockTransport([
388
 
            _KnitIndex.HEADER,
389
 
            "a option 0 1 :",
390
 
            "b option 0 1 0 .e :",
391
 
            "c option 0 1 1 0 :",
392
 
            "d option 0 1 2 .f :"
393
 
            ])
394
 
        index = _KnitIndex(transport, "filename", "r")
395
 
 
396
 
        self.assertEqual([], index.get_ancestry([]))
397
 
        self.assertEqual(["a"], index.get_ancestry(["a"]))
398
 
        self.assertEqual(["a", "b"], index.get_ancestry(["b"]))
399
 
        self.assertEqual(["a", "b", "c"], index.get_ancestry(["c"]))
400
 
        self.assertEqual(["a", "b", "c", "d"], index.get_ancestry(["d"]))
401
 
        self.assertEqual(["a", "b"], index.get_ancestry(["a", "b"]))
402
 
        self.assertEqual(["a", "b", "c"], index.get_ancestry(["a", "c"]))
403
 
 
404
 
        self.assertRaises(RevisionNotPresent, index.get_ancestry, ["e"])
405
 
 
406
 
    def test_get_ancestry_with_ghosts(self):
407
 
        transport = MockTransport([
408
 
            _KnitIndex.HEADER,
409
 
            "a option 0 1 :",
410
 
            "b option 0 1 0 .e :",
411
 
            "c option 0 1 0 .f .g :",
412
 
            "d option 0 1 2 .h .j .k :"
413
 
            ])
414
 
        index = _KnitIndex(transport, "filename", "r")
415
 
 
416
 
        self.assertEqual([], index.get_ancestry_with_ghosts([]))
417
 
        self.assertEqual(["a"], index.get_ancestry_with_ghosts(["a"]))
418
 
        self.assertEqual(["a", "e", "b"],
419
 
            index.get_ancestry_with_ghosts(["b"]))
420
 
        self.assertEqual(["a", "g", "f", "c"],
421
 
            index.get_ancestry_with_ghosts(["c"]))
422
 
        self.assertEqual(["a", "g", "f", "c", "k", "j", "h", "d"],
423
 
            index.get_ancestry_with_ghosts(["d"]))
424
 
        self.assertEqual(["a", "e", "b"],
425
 
            index.get_ancestry_with_ghosts(["a", "b"]))
426
 
        self.assertEqual(["a", "g", "f", "c"],
427
 
            index.get_ancestry_with_ghosts(["a", "c"]))
428
 
        self.assertEqual(
429
 
            ["a", "g", "f", "c", "e", "b", "k", "j", "h", "d"],
430
 
            index.get_ancestry_with_ghosts(["b", "d"]))
431
 
 
432
 
        self.assertRaises(RevisionNotPresent,
433
 
            index.get_ancestry_with_ghosts, ["e"])
434
 
 
435
 
    def test_num_versions(self):
436
 
        transport = MockTransport([
437
 
            _KnitIndex.HEADER
438
 
            ])
439
 
        index = _KnitIndex(transport, "filename", "r")
440
 
 
441
 
        self.assertEqual(0, index.num_versions())
442
 
        self.assertEqual(0, len(index))
443
 
 
444
 
        index.add_version("a", ["option"], 0, 1, [])
445
 
        self.assertEqual(1, index.num_versions())
446
 
        self.assertEqual(1, len(index))
447
 
 
448
 
        index.add_version("a", ["option2"], 1, 2, [])
449
 
        self.assertEqual(1, index.num_versions())
450
 
        self.assertEqual(1, len(index))
451
 
 
452
 
        index.add_version("b", ["option"], 0, 1, [])
453
 
        self.assertEqual(2, index.num_versions())
454
 
        self.assertEqual(2, len(index))
455
 
 
456
 
    def test_get_versions(self):
457
 
        transport = MockTransport([
458
 
            _KnitIndex.HEADER
459
 
            ])
460
 
        index = _KnitIndex(transport, "filename", "r")
461
 
 
462
 
        self.assertEqual([], index.get_versions())
463
 
 
464
 
        index.add_version("a", ["option"], 0, 1, [])
465
 
        self.assertEqual(["a"], index.get_versions())
466
 
 
467
 
        index.add_version("a", ["option"], 0, 1, [])
468
 
        self.assertEqual(["a"], index.get_versions())
469
 
 
470
 
        index.add_version("b", ["option"], 0, 1, [])
471
 
        self.assertEqual(["a", "b"], index.get_versions())
472
 
 
473
 
    def test_idx_to_name(self):
474
 
        transport = MockTransport([
475
 
            _KnitIndex.HEADER,
476
 
            "a option 0 1 :",
477
 
            "b option 0 1 :"
478
 
            ])
479
 
        index = _KnitIndex(transport, "filename", "r")
480
 
 
481
 
        self.assertEqual("a", index.idx_to_name(0))
482
 
        self.assertEqual("b", index.idx_to_name(1))
483
 
        self.assertEqual("b", index.idx_to_name(-1))
484
 
        self.assertEqual("a", index.idx_to_name(-2))
485
 
 
486
 
    def test_lookup(self):
487
 
        transport = MockTransport([
488
 
            _KnitIndex.HEADER,
489
 
            "a option 0 1 :",
490
 
            "b option 0 1 :"
491
 
            ])
492
 
        index = _KnitIndex(transport, "filename", "r")
493
 
 
494
 
        self.assertEqual(0, index.lookup("a"))
495
 
        self.assertEqual(1, index.lookup("b"))
496
 
 
497
 
    def test_add_version(self):
498
 
        transport = MockTransport([
499
 
            _KnitIndex.HEADER
500
 
            ])
501
 
        index = _KnitIndex(transport, "filename", "r")
502
 
 
503
 
        index.add_version("a", ["option"], 0, 1, ["b"])
504
 
        self.assertEqual(("append_bytes",
505
 
            ("filename", "\na option 0 1 .b :"),
506
 
            {}), transport.calls.pop(0))
507
 
        self.assertTrue(index.has_version("a"))
508
 
        self.assertEqual(1, index.num_versions())
509
 
        self.assertEqual((0, 1), index.get_position("a"))
510
 
        self.assertEqual(["option"], index.get_options("a"))
511
 
        self.assertEqual(["b"], index.get_parents_with_ghosts("a"))
512
 
 
513
 
        index.add_version("a", ["opt"], 1, 2, ["c"])
514
 
        self.assertEqual(("append_bytes",
515
 
            ("filename", "\na opt 1 2 .c :"),
516
 
            {}), transport.calls.pop(0))
517
 
        self.assertTrue(index.has_version("a"))
518
 
        self.assertEqual(1, index.num_versions())
519
 
        self.assertEqual((1, 2), index.get_position("a"))
520
 
        self.assertEqual(["opt"], index.get_options("a"))
521
 
        self.assertEqual(["c"], index.get_parents_with_ghosts("a"))
522
 
 
523
 
        index.add_version("b", ["option"], 2, 3, ["a"])
524
 
        self.assertEqual(("append_bytes",
525
 
            ("filename", "\nb option 2 3 0 :"),
526
 
            {}), transport.calls.pop(0))
527
 
        self.assertTrue(index.has_version("b"))
528
 
        self.assertEqual(2, index.num_versions())
529
 
        self.assertEqual((2, 3), index.get_position("b"))
530
 
        self.assertEqual(["option"], index.get_options("b"))
531
 
        self.assertEqual(["a"], index.get_parents_with_ghosts("b"))
532
 
 
533
 
    def test_add_versions(self):
534
 
        transport = MockTransport([
535
 
            _KnitIndex.HEADER
536
 
            ])
537
 
        index = _KnitIndex(transport, "filename", "r")
538
 
 
539
 
        index.add_versions([
540
 
            ("a", ["option"], 0, 1, ["b"]),
541
 
            ("a", ["opt"], 1, 2, ["c"]),
542
 
            ("b", ["option"], 2, 3, ["a"])
543
 
            ])
544
 
        self.assertEqual(("append_bytes", ("filename",
545
 
            "\na option 0 1 .b :"
546
 
            "\na opt 1 2 .c :"
547
 
            "\nb option 2 3 0 :"
548
 
            ), {}), transport.calls.pop(0))
549
 
        self.assertTrue(index.has_version("a"))
550
 
        self.assertTrue(index.has_version("b"))
551
 
        self.assertEqual(2, index.num_versions())
552
 
        self.assertEqual((1, 2), index.get_position("a"))
553
 
        self.assertEqual((2, 3), index.get_position("b"))
554
 
        self.assertEqual(["opt"], index.get_options("a"))
555
 
        self.assertEqual(["option"], index.get_options("b"))
556
 
        self.assertEqual(["c"], index.get_parents_with_ghosts("a"))
557
 
        self.assertEqual(["a"], index.get_parents_with_ghosts("b"))
558
 
 
559
 
    def test_delay_create_and_add_versions(self):
560
 
        transport = MockTransport()
561
 
 
562
 
        index = _KnitIndex(transport, "filename", "w",
563
 
            create=True, file_mode="wb", create_parent_dir=True,
564
 
            delay_create=True, dir_mode=0777)
565
 
        self.assertEqual([], transport.calls)
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
 
        name, (filename, f), kwargs = transport.calls.pop(0)
573
 
        self.assertEqual("put_file_non_atomic", name)
574
 
        self.assertEqual(
575
 
            {"dir_mode": 0777, "create_parent_dir": True, "mode": "wb"},
576
 
            kwargs)
577
 
        self.assertEqual("filename", filename)
578
 
        self.assertEqual(
579
 
            index.HEADER +
580
 
            "\na option 0 1 .b :"
581
 
            "\na opt 1 2 .c :"
582
 
            "\nb option 2 3 0 :",
583
 
            f.read())
584
 
 
585
 
    def test_has_version(self):
586
 
        transport = MockTransport([
587
 
            _KnitIndex.HEADER,
588
 
            "a option 0 1 :"
589
 
            ])
590
 
        index = _KnitIndex(transport, "filename", "r")
591
 
 
592
 
        self.assertTrue(index.has_version("a"))
593
 
        self.assertFalse(index.has_version("b"))
594
 
 
595
 
    def test_get_position(self):
596
 
        transport = MockTransport([
597
 
            _KnitIndex.HEADER,
598
 
            "a option 0 1 :",
599
 
            "b option 1 2 :"
600
 
            ])
601
 
        index = _KnitIndex(transport, "filename", "r")
602
 
 
603
 
        self.assertEqual((0, 1), index.get_position("a"))
604
 
        self.assertEqual((1, 2), index.get_position("b"))
605
 
 
606
 
    def test_get_method(self):
607
 
        transport = MockTransport([
608
 
            _KnitIndex.HEADER,
609
 
            "a fulltext,unknown 0 1 :",
610
 
            "b unknown,line-delta 1 2 :",
611
 
            "c bad 3 4 :"
612
 
            ])
613
 
        index = _KnitIndex(transport, "filename", "r")
614
 
 
615
 
        self.assertEqual("fulltext", index.get_method("a"))
616
 
        self.assertEqual("line-delta", index.get_method("b"))
617
 
        self.assertRaises(errors.KnitIndexUnknownMethod, index.get_method, "c")
618
 
 
619
 
    def test_get_options(self):
620
 
        transport = MockTransport([
621
 
            _KnitIndex.HEADER,
622
 
            "a opt1 0 1 :",
623
 
            "b opt2,opt3 1 2 :"
624
 
            ])
625
 
        index = _KnitIndex(transport, "filename", "r")
626
 
 
627
 
        self.assertEqual(["opt1"], index.get_options("a"))
628
 
        self.assertEqual(["opt2", "opt3"], index.get_options("b"))
629
 
 
630
 
    def test_get_parents(self):
631
 
        transport = MockTransport([
632
 
            _KnitIndex.HEADER,
633
 
            "a option 0 1 :",
634
 
            "b option 1 2 0 .c :",
635
 
            "c option 1 2 1 0 .e :"
636
 
            ])
637
 
        index = _KnitIndex(transport, "filename", "r")
638
 
 
639
 
        self.assertEqual([], index.get_parents("a"))
640
 
        self.assertEqual(["a", "c"], index.get_parents("b"))
641
 
        self.assertEqual(["b", "a"], index.get_parents("c"))
642
 
 
643
 
    def test_get_parents_with_ghosts(self):
644
 
        transport = MockTransport([
645
 
            _KnitIndex.HEADER,
646
 
            "a option 0 1 :",
647
 
            "b option 1 2 0 .c :",
648
 
            "c option 1 2 1 0 .e :"
649
 
            ])
650
 
        index = _KnitIndex(transport, "filename", "r")
651
 
 
652
 
        self.assertEqual([], index.get_parents_with_ghosts("a"))
653
 
        self.assertEqual(["a", "c"], index.get_parents_with_ghosts("b"))
654
 
        self.assertEqual(["b", "a", "e"],
655
 
            index.get_parents_with_ghosts("c"))
656
 
 
657
 
    def test_check_versions_present(self):
658
 
        transport = MockTransport([
659
 
            _KnitIndex.HEADER,
660
 
            "a option 0 1 :",
661
 
            "b option 0 1 :"
662
 
            ])
663
 
        index = _KnitIndex(transport, "filename", "r")
664
 
 
665
 
        check = index.check_versions_present
666
 
 
667
 
        check([])
668
 
        check(["a"])
669
 
        check(["b"])
670
 
        check(["a", "b"])
671
 
        self.assertRaises(RevisionNotPresent, check, ["c"])
672
 
        self.assertRaises(RevisionNotPresent, check, ["a", "b", "c"])
673
 
 
674
 
 
675
 
class KnitTests(TestCaseWithTransport):
676
 
    """Class containing knit test helper routines."""
677
 
 
678
 
    def make_test_knit(self, annotate=False, delay_create=False):
679
 
        if not annotate:
680
 
            factory = KnitPlainFactory()
681
 
        else:
682
 
            factory = None
683
 
        return KnitVersionedFile('test', get_transport('.'), access_mode='w',
684
 
                                 factory=factory, create=True,
685
 
                                 delay_create=delay_create)
686
 
 
687
 
 
688
 
class BasicKnitTests(KnitTests):
689
 
 
690
 
    def add_stock_one_and_one_a(self, k):
691
 
        k.add_lines('text-1', [], split_lines(TEXT_1))
692
 
        k.add_lines('text-1a', ['text-1'], split_lines(TEXT_1A))
693
 
 
694
 
    def test_knit_constructor(self):
695
 
        """Construct empty k"""
696
 
        self.make_test_knit()
697
 
 
698
 
    def test_knit_add(self):
699
 
        """Store one text in knit and retrieve"""
700
 
        k = self.make_test_knit()
701
 
        k.add_lines('text-1', [], split_lines(TEXT_1))
702
 
        self.assertTrue(k.has_version('text-1'))
703
 
        self.assertEqualDiff(''.join(k.get_lines('text-1')), TEXT_1)
704
 
 
705
 
    def test_knit_reload(self):
706
 
        # test that the content in a reloaded knit is correct
707
 
        k = self.make_test_knit()
708
 
        k.add_lines('text-1', [], split_lines(TEXT_1))
709
 
        del k
710
 
        k2 = KnitVersionedFile('test', get_transport('.'), access_mode='r', factory=KnitPlainFactory(), create=True)
711
 
        self.assertTrue(k2.has_version('text-1'))
712
 
        self.assertEqualDiff(''.join(k2.get_lines('text-1')), TEXT_1)
713
 
 
714
 
    def test_knit_several(self):
715
 
        """Store several texts in a knit"""
716
 
        k = self.make_test_knit()
717
 
        k.add_lines('text-1', [], split_lines(TEXT_1))
718
 
        k.add_lines('text-2', [], split_lines(TEXT_2))
719
 
        self.assertEqualDiff(''.join(k.get_lines('text-1')), TEXT_1)
720
 
        self.assertEqualDiff(''.join(k.get_lines('text-2')), TEXT_2)
721
 
        
722
 
    def test_repeated_add(self):
723
 
        """Knit traps attempt to replace existing version"""
724
 
        k = self.make_test_knit()
725
 
        k.add_lines('text-1', [], split_lines(TEXT_1))
726
 
        self.assertRaises(RevisionAlreadyPresent, 
727
 
                k.add_lines,
728
 
                'text-1', [], split_lines(TEXT_1))
729
 
 
730
 
    def test_empty(self):
731
 
        k = self.make_test_knit(True)
732
 
        k.add_lines('text-1', [], [])
733
 
        self.assertEquals(k.get_lines('text-1'), [])
734
 
 
735
 
    def test_incomplete(self):
736
 
        """Test if texts without a ending line-end can be inserted and
737
 
        extracted."""
738
 
        k = KnitVersionedFile('test', get_transport('.'), delta=False, create=True)
739
 
        k.add_lines('text-1', [], ['a\n',    'b'  ])
740
 
        k.add_lines('text-2', ['text-1'], ['a\rb\n', 'b\n'])
741
 
        # reopening ensures maximum room for confusion
742
 
        k = KnitVersionedFile('test', get_transport('.'), delta=False, create=True)
743
 
        self.assertEquals(k.get_lines('text-1'), ['a\n',    'b'  ])
744
 
        self.assertEquals(k.get_lines('text-2'), ['a\rb\n', 'b\n'])
745
 
 
746
 
    def test_delta(self):
747
 
        """Expression of knit delta as lines"""
748
 
        k = self.make_test_knit()
749
 
        td = list(line_delta(TEXT_1.splitlines(True),
750
 
                             TEXT_1A.splitlines(True)))
751
 
        self.assertEqualDiff(''.join(td), delta_1_1a)
752
 
        out = apply_line_delta(TEXT_1.splitlines(True), td)
753
 
        self.assertEqualDiff(''.join(out), TEXT_1A)
754
 
 
755
 
    def test_add_with_parents(self):
756
 
        """Store in knit with parents"""
757
 
        k = self.make_test_knit()
758
 
        self.add_stock_one_and_one_a(k)
759
 
        self.assertEquals(k.get_parents('text-1'), [])
760
 
        self.assertEquals(k.get_parents('text-1a'), ['text-1'])
761
 
 
762
 
    def test_ancestry(self):
763
 
        """Store in knit with parents"""
764
 
        k = self.make_test_knit()
765
 
        self.add_stock_one_and_one_a(k)
766
 
        self.assertEquals(set(k.get_ancestry(['text-1a'])), set(['text-1a', 'text-1']))
767
 
 
768
 
    def test_add_delta(self):
769
 
        """Store in knit with parents"""
770
 
        k = KnitVersionedFile('test', get_transport('.'), factory=KnitPlainFactory(),
771
 
            delta=True, create=True)
772
 
        self.add_stock_one_and_one_a(k)
773
 
        k.clear_cache()
774
 
        self.assertEqualDiff(''.join(k.get_lines('text-1a')), TEXT_1A)
775
 
 
776
 
    def test_annotate(self):
777
 
        """Annotations"""
778
 
        k = KnitVersionedFile('knit', get_transport('.'), factory=KnitAnnotateFactory(),
779
 
            delta=True, create=True)
780
 
        self.insert_and_test_small_annotate(k)
781
 
 
782
 
    def insert_and_test_small_annotate(self, k):
783
 
        """test annotation with k works correctly."""
784
 
        k.add_lines('text-1', [], ['a\n', 'b\n'])
785
 
        k.add_lines('text-2', ['text-1'], ['a\n', 'c\n'])
786
 
 
787
 
        origins = k.annotate('text-2')
788
 
        self.assertEquals(origins[0], ('text-1', 'a\n'))
789
 
        self.assertEquals(origins[1], ('text-2', 'c\n'))
790
 
 
791
 
    def test_annotate_fulltext(self):
792
 
        """Annotations"""
793
 
        k = KnitVersionedFile('knit', get_transport('.'), factory=KnitAnnotateFactory(),
794
 
            delta=False, create=True)
795
 
        self.insert_and_test_small_annotate(k)
796
 
 
797
 
    def test_annotate_merge_1(self):
798
 
        k = self.make_test_knit(True)
799
 
        k.add_lines('text-a1', [], ['a\n', 'b\n'])
800
 
        k.add_lines('text-a2', [], ['d\n', 'c\n'])
801
 
        k.add_lines('text-am', ['text-a1', 'text-a2'], ['d\n', 'b\n'])
802
 
        origins = k.annotate('text-am')
803
 
        self.assertEquals(origins[0], ('text-a2', 'd\n'))
804
 
        self.assertEquals(origins[1], ('text-a1', 'b\n'))
805
 
 
806
 
    def test_annotate_merge_2(self):
807
 
        k = self.make_test_knit(True)
808
 
        k.add_lines('text-a1', [], ['a\n', 'b\n', 'c\n'])
809
 
        k.add_lines('text-a2', [], ['x\n', 'y\n', 'z\n'])
810
 
        k.add_lines('text-am', ['text-a1', 'text-a2'], ['a\n', 'y\n', 'c\n'])
811
 
        origins = k.annotate('text-am')
812
 
        self.assertEquals(origins[0], ('text-a1', 'a\n'))
813
 
        self.assertEquals(origins[1], ('text-a2', 'y\n'))
814
 
        self.assertEquals(origins[2], ('text-a1', 'c\n'))
815
 
 
816
 
    def test_annotate_merge_9(self):
817
 
        k = self.make_test_knit(True)
818
 
        k.add_lines('text-a1', [], ['a\n', 'b\n', 'c\n'])
819
 
        k.add_lines('text-a2', [], ['x\n', 'y\n', 'z\n'])
820
 
        k.add_lines('text-am', ['text-a1', 'text-a2'], ['k\n', 'y\n', 'c\n'])
821
 
        origins = k.annotate('text-am')
822
 
        self.assertEquals(origins[0], ('text-am', 'k\n'))
823
 
        self.assertEquals(origins[1], ('text-a2', 'y\n'))
824
 
        self.assertEquals(origins[2], ('text-a1', 'c\n'))
825
 
 
826
 
    def test_annotate_merge_3(self):
827
 
        k = self.make_test_knit(True)
828
 
        k.add_lines('text-a1', [], ['a\n', 'b\n', 'c\n'])
829
 
        k.add_lines('text-a2', [] ,['x\n', 'y\n', 'z\n'])
830
 
        k.add_lines('text-am', ['text-a1', 'text-a2'], ['k\n', 'y\n', 'z\n'])
831
 
        origins = k.annotate('text-am')
832
 
        self.assertEquals(origins[0], ('text-am', 'k\n'))
833
 
        self.assertEquals(origins[1], ('text-a2', 'y\n'))
834
 
        self.assertEquals(origins[2], ('text-a2', 'z\n'))
835
 
 
836
 
    def test_annotate_merge_4(self):
837
 
        k = self.make_test_knit(True)
838
 
        k.add_lines('text-a1', [], ['a\n', 'b\n', 'c\n'])
839
 
        k.add_lines('text-a2', [], ['x\n', 'y\n', 'z\n'])
840
 
        k.add_lines('text-a3', ['text-a1'], ['a\n', 'b\n', 'p\n'])
841
 
        k.add_lines('text-am', ['text-a2', 'text-a3'], ['a\n', 'b\n', 'z\n'])
842
 
        origins = k.annotate('text-am')
843
 
        self.assertEquals(origins[0], ('text-a1', 'a\n'))
844
 
        self.assertEquals(origins[1], ('text-a1', 'b\n'))
845
 
        self.assertEquals(origins[2], ('text-a2', 'z\n'))
846
 
 
847
 
    def test_annotate_merge_5(self):
848
 
        k = self.make_test_knit(True)
849
 
        k.add_lines('text-a1', [], ['a\n', 'b\n', 'c\n'])
850
 
        k.add_lines('text-a2', [], ['d\n', 'e\n', 'f\n'])
851
 
        k.add_lines('text-a3', [], ['x\n', 'y\n', 'z\n'])
852
 
        k.add_lines('text-am',
853
 
                    ['text-a1', 'text-a2', 'text-a3'],
854
 
                    ['a\n', 'e\n', 'z\n'])
855
 
        origins = k.annotate('text-am')
856
 
        self.assertEquals(origins[0], ('text-a1', 'a\n'))
857
 
        self.assertEquals(origins[1], ('text-a2', 'e\n'))
858
 
        self.assertEquals(origins[2], ('text-a3', 'z\n'))
859
 
 
860
 
    def test_annotate_file_cherry_pick(self):
861
 
        k = self.make_test_knit(True)
862
 
        k.add_lines('text-1', [], ['a\n', 'b\n', 'c\n'])
863
 
        k.add_lines('text-2', ['text-1'], ['d\n', 'e\n', 'f\n'])
864
 
        k.add_lines('text-3', ['text-2', 'text-1'], ['a\n', 'b\n', 'c\n'])
865
 
        origins = k.annotate('text-3')
866
 
        self.assertEquals(origins[0], ('text-1', 'a\n'))
867
 
        self.assertEquals(origins[1], ('text-1', 'b\n'))
868
 
        self.assertEquals(origins[2], ('text-1', 'c\n'))
869
 
 
870
 
    def test_knit_join(self):
871
 
        """Store in knit with parents"""
872
 
        k1 = KnitVersionedFile('test1', get_transport('.'), factory=KnitPlainFactory(), create=True)
873
 
        k1.add_lines('text-a', [], split_lines(TEXT_1))
874
 
        k1.add_lines('text-b', ['text-a'], split_lines(TEXT_1))
875
 
 
876
 
        k1.add_lines('text-c', [], split_lines(TEXT_1))
877
 
        k1.add_lines('text-d', ['text-c'], split_lines(TEXT_1))
878
 
 
879
 
        k1.add_lines('text-m', ['text-b', 'text-d'], split_lines(TEXT_1))
880
 
 
881
 
        k2 = KnitVersionedFile('test2', get_transport('.'), factory=KnitPlainFactory(), create=True)
882
 
        count = k2.join(k1, version_ids=['text-m'])
883
 
        self.assertEquals(count, 5)
884
 
        self.assertTrue(k2.has_version('text-a'))
885
 
        self.assertTrue(k2.has_version('text-c'))
886
 
 
887
 
    def test_reannotate(self):
888
 
        k1 = KnitVersionedFile('knit1', get_transport('.'),
889
 
                               factory=KnitAnnotateFactory(), create=True)
890
 
        # 0
891
 
        k1.add_lines('text-a', [], ['a\n', 'b\n'])
892
 
        # 1
893
 
        k1.add_lines('text-b', ['text-a'], ['a\n', 'c\n'])
894
 
 
895
 
        k2 = KnitVersionedFile('test2', get_transport('.'),
896
 
                               factory=KnitAnnotateFactory(), create=True)
897
 
        k2.join(k1, version_ids=['text-b'])
898
 
 
899
 
        # 2
900
 
        k1.add_lines('text-X', ['text-b'], ['a\n', 'b\n'])
901
 
        # 2
902
 
        k2.add_lines('text-c', ['text-b'], ['z\n', 'c\n'])
903
 
        # 3
904
 
        k2.add_lines('text-Y', ['text-b'], ['b\n', 'c\n'])
905
 
 
906
 
        # test-c will have index 3
907
 
        k1.join(k2, version_ids=['text-c'])
908
 
 
909
 
        lines = k1.get_lines('text-c')
910
 
        self.assertEquals(lines, ['z\n', 'c\n'])
911
 
 
912
 
        origins = k1.annotate('text-c')
913
 
        self.assertEquals(origins[0], ('text-c', 'z\n'))
914
 
        self.assertEquals(origins[1], ('text-b', 'c\n'))
915
 
 
916
 
    def test_get_line_delta_texts(self):
917
 
        """Make sure we can call get_texts on text with reused line deltas"""
918
 
        k1 = KnitVersionedFile('test1', get_transport('.'), 
919
 
                               factory=KnitPlainFactory(), create=True)
920
 
        for t in range(3):
921
 
            if t == 0:
922
 
                parents = []
923
 
            else:
924
 
                parents = ['%d' % (t-1)]
925
 
            k1.add_lines('%d' % t, parents, ['hello\n'] * t)
926
 
        k1.get_texts(('%d' % t) for t in range(3))
927
 
        
928
 
    def test_iter_lines_reads_in_order(self):
929
 
        t = MemoryTransport()
930
 
        instrumented_t = TransportLogger(t)
931
 
        k1 = KnitVersionedFile('id', instrumented_t, create=True, delta=True)
932
 
        self.assertEqual([('id.kndx',)], instrumented_t._calls)
933
 
        # add texts with no required ordering
934
 
        k1.add_lines('base', [], ['text\n'])
935
 
        k1.add_lines('base2', [], ['text2\n'])
936
 
        k1.clear_cache()
937
 
        instrumented_t._calls = []
938
 
        # request a last-first iteration
939
 
        results = list(k1.iter_lines_added_or_present_in_versions(['base2', 'base']))
940
 
        self.assertEqual([('id.knit', [(0, 87), (87, 89)])], instrumented_t._calls)
941
 
        self.assertEqual(['text\n', 'text2\n'], results)
942
 
 
943
 
    def test_create_empty_annotated(self):
944
 
        k1 = self.make_test_knit(True)
945
 
        # 0
946
 
        k1.add_lines('text-a', [], ['a\n', 'b\n'])
947
 
        k2 = k1.create_empty('t', MemoryTransport())
948
 
        self.assertTrue(isinstance(k2.factory, KnitAnnotateFactory))
949
 
        self.assertEqual(k1.delta, k2.delta)
950
 
        # the generic test checks for empty content and file class
951
 
 
952
 
    def test_knit_format(self):
953
 
        # this tests that a new knit index file has the expected content
954
 
        # and that is writes the data we expect as records are added.
955
 
        knit = self.make_test_knit(True)
956
 
        # Now knit files are not created until we first add data to them
957
 
        self.assertFileEqual("# bzr knit index 8\n", 'test.kndx')
958
 
        knit.add_lines_with_ghosts('revid', ['a_ghost'], ['a\n'])
959
 
        self.assertFileEqual(
960
 
            "# bzr knit index 8\n"
961
 
            "\n"
962
 
            "revid fulltext 0 84 .a_ghost :",
963
 
            'test.kndx')
964
 
        knit.add_lines_with_ghosts('revid2', ['revid'], ['a\n'])
965
 
        self.assertFileEqual(
966
 
            "# bzr knit index 8\n"
967
 
            "\nrevid fulltext 0 84 .a_ghost :"
968
 
            "\nrevid2 line-delta 84 82 0 :",
969
 
            'test.kndx')
970
 
        # we should be able to load this file again
971
 
        knit = KnitVersionedFile('test', get_transport('.'), access_mode='r')
972
 
        self.assertEqual(['revid', 'revid2'], knit.versions())
973
 
        # write a short write to the file and ensure that its ignored
974
 
        indexfile = file('test.kndx', 'at')
975
 
        indexfile.write('\nrevid3 line-delta 166 82 1 2 3 4 5 .phwoar:demo ')
976
 
        indexfile.close()
977
 
        # we should be able to load this file again
978
 
        knit = KnitVersionedFile('test', get_transport('.'), access_mode='w')
979
 
        self.assertEqual(['revid', 'revid2'], knit.versions())
980
 
        # and add a revision with the same id the failed write had
981
 
        knit.add_lines('revid3', ['revid2'], ['a\n'])
982
 
        # and when reading it revid3 should now appear.
983
 
        knit = KnitVersionedFile('test', get_transport('.'), access_mode='r')
984
 
        self.assertEqual(['revid', 'revid2', 'revid3'], knit.versions())
985
 
        self.assertEqual(['revid2'], knit.get_parents('revid3'))
986
 
 
987
 
    def test_delay_create(self):
988
 
        """Test that passing delay_create=True creates files late"""
989
 
        knit = self.make_test_knit(annotate=True, delay_create=True)
990
 
        self.failIfExists('test.knit')
991
 
        self.failIfExists('test.kndx')
992
 
        knit.add_lines_with_ghosts('revid', ['a_ghost'], ['a\n'])
993
 
        self.failUnlessExists('test.knit')
994
 
        self.assertFileEqual(
995
 
            "# bzr knit index 8\n"
996
 
            "\n"
997
 
            "revid fulltext 0 84 .a_ghost :",
998
 
            'test.kndx')
999
 
 
1000
 
    def test_create_parent_dir(self):
1001
 
        """create_parent_dir can create knits in nonexistant dirs"""
1002
 
        # Has no effect if we don't set 'delay_create'
1003
 
        trans = get_transport('.')
1004
 
        self.assertRaises(NoSuchFile, KnitVersionedFile, 'dir/test',
1005
 
                          trans, access_mode='w', factory=None,
1006
 
                          create=True, create_parent_dir=True)
1007
 
        # Nothing should have changed yet
1008
 
        knit = KnitVersionedFile('dir/test', trans, access_mode='w',
1009
 
                                 factory=None, create=True,
1010
 
                                 create_parent_dir=True,
1011
 
                                 delay_create=True)
1012
 
        self.failIfExists('dir/test.knit')
1013
 
        self.failIfExists('dir/test.kndx')
1014
 
        self.failIfExists('dir')
1015
 
        knit.add_lines('revid', [], ['a\n'])
1016
 
        self.failUnlessExists('dir')
1017
 
        self.failUnlessExists('dir/test.knit')
1018
 
        self.assertFileEqual(
1019
 
            "# bzr knit index 8\n"
1020
 
            "\n"
1021
 
            "revid fulltext 0 84  :",
1022
 
            'dir/test.kndx')
1023
 
 
1024
 
    def test_create_mode_700(self):
1025
 
        trans = get_transport('.')
1026
 
        if not trans._can_roundtrip_unix_modebits():
1027
 
            # Can't roundtrip, so no need to run this test
1028
 
            return
1029
 
        knit = KnitVersionedFile('dir/test', trans, access_mode='w',
1030
 
                                 factory=None, create=True,
1031
 
                                 create_parent_dir=True,
1032
 
                                 delay_create=True,
1033
 
                                 file_mode=0600,
1034
 
                                 dir_mode=0700)
1035
 
        knit.add_lines('revid', [], ['a\n'])
1036
 
        self.assertTransportMode(trans, 'dir', 0700)
1037
 
        self.assertTransportMode(trans, 'dir/test.knit', 0600)
1038
 
        self.assertTransportMode(trans, 'dir/test.kndx', 0600)
1039
 
 
1040
 
    def test_create_mode_770(self):
1041
 
        trans = get_transport('.')
1042
 
        if not trans._can_roundtrip_unix_modebits():
1043
 
            # Can't roundtrip, so no need to run this test
1044
 
            return
1045
 
        knit = KnitVersionedFile('dir/test', trans, access_mode='w',
1046
 
                                 factory=None, create=True,
1047
 
                                 create_parent_dir=True,
1048
 
                                 delay_create=True,
1049
 
                                 file_mode=0660,
1050
 
                                 dir_mode=0770)
1051
 
        knit.add_lines('revid', [], ['a\n'])
1052
 
        self.assertTransportMode(trans, 'dir', 0770)
1053
 
        self.assertTransportMode(trans, 'dir/test.knit', 0660)
1054
 
        self.assertTransportMode(trans, 'dir/test.kndx', 0660)
1055
 
 
1056
 
    def test_create_mode_777(self):
1057
 
        trans = get_transport('.')
1058
 
        if not trans._can_roundtrip_unix_modebits():
1059
 
            # Can't roundtrip, so no need to run this test
1060
 
            return
1061
 
        knit = KnitVersionedFile('dir/test', trans, access_mode='w',
1062
 
                                 factory=None, create=True,
1063
 
                                 create_parent_dir=True,
1064
 
                                 delay_create=True,
1065
 
                                 file_mode=0666,
1066
 
                                 dir_mode=0777)
1067
 
        knit.add_lines('revid', [], ['a\n'])
1068
 
        self.assertTransportMode(trans, 'dir', 0777)
1069
 
        self.assertTransportMode(trans, 'dir/test.knit', 0666)
1070
 
        self.assertTransportMode(trans, 'dir/test.kndx', 0666)
1071
 
 
1072
 
    def test_plan_merge(self):
1073
 
        my_knit = self.make_test_knit(annotate=True)
1074
 
        my_knit.add_lines('text1', [], split_lines(TEXT_1))
1075
 
        my_knit.add_lines('text1a', ['text1'], split_lines(TEXT_1A))
1076
 
        my_knit.add_lines('text1b', ['text1'], split_lines(TEXT_1B))
1077
 
        plan = list(my_knit.plan_merge('text1a', 'text1b'))
1078
 
        for plan_line, expected_line in zip(plan, AB_MERGE):
1079
 
            self.assertEqual(plan_line, expected_line)
1080
 
 
1081
 
 
1082
 
TEXT_1 = """\
1083
 
Banana cup cakes:
1084
 
 
1085
 
- bananas
1086
 
- eggs
1087
 
- broken tea cups
1088
 
"""
1089
 
 
1090
 
TEXT_1A = """\
1091
 
Banana cup cake recipe
1092
 
(serves 6)
1093
 
 
1094
 
- bananas
1095
 
- eggs
1096
 
- broken tea cups
1097
 
- self-raising flour
1098
 
"""
1099
 
 
1100
 
TEXT_1B = """\
1101
 
Banana cup cake recipe
1102
 
 
1103
 
- bananas (do not use plantains!!!)
1104
 
- broken tea cups
1105
 
- flour
1106
 
"""
1107
 
 
1108
 
delta_1_1a = """\
1109
 
0,1,2
1110
 
Banana cup cake recipe
1111
 
(serves 6)
1112
 
5,5,1
1113
 
- self-raising flour
1114
 
"""
1115
 
 
1116
 
TEXT_2 = """\
1117
 
Boeuf bourguignon
1118
 
 
1119
 
- beef
1120
 
- red wine
1121
 
- small onions
1122
 
- carrot
1123
 
- mushrooms
1124
 
"""
1125
 
 
1126
 
AB_MERGE_TEXT="""unchanged|Banana cup cake recipe
1127
 
new-a|(serves 6)
1128
 
unchanged|
1129
 
killed-b|- bananas
1130
 
killed-b|- eggs
1131
 
new-b|- bananas (do not use plantains!!!)
1132
 
unchanged|- broken tea cups
1133
 
new-a|- self-raising flour
1134
 
new-b|- flour
1135
 
"""
1136
 
AB_MERGE=[tuple(l.split('|')) for l in AB_MERGE_TEXT.splitlines(True)]
1137
 
 
1138
 
 
1139
 
def line_delta(from_lines, to_lines):
1140
 
    """Generate line-based delta from one text to another"""
1141
 
    s = difflib.SequenceMatcher(None, from_lines, to_lines)
1142
 
    for op in s.get_opcodes():
1143
 
        if op[0] == 'equal':
1144
 
            continue
1145
 
        yield '%d,%d,%d\n' % (op[1], op[2], op[4]-op[3])
1146
 
        for i in range(op[3], op[4]):
1147
 
            yield to_lines[i]
1148
 
 
1149
 
 
1150
 
def apply_line_delta(basis_lines, delta_lines):
1151
 
    """Apply a line-based perfect diff
1152
 
    
1153
 
    basis_lines -- text to apply the patch to
1154
 
    delta_lines -- diff instructions and content
1155
 
    """
1156
 
    out = basis_lines[:]
1157
 
    i = 0
1158
 
    offset = 0
1159
 
    while i < len(delta_lines):
1160
 
        l = delta_lines[i]
1161
 
        a, b, c = map(long, l.split(','))
1162
 
        i = i + 1
1163
 
        out[offset+a:offset+b] = delta_lines[i:i+c]
1164
 
        i = i + c
1165
 
        offset = offset + (b - a) + c
1166
 
    return out
1167
 
 
1168
 
 
1169
 
class TestWeaveToKnit(KnitTests):
1170
 
 
1171
 
    def test_weave_to_knit_matches(self):
1172
 
        # check that the WeaveToKnit is_compatible function
1173
 
        # registers True for a Weave to a Knit.
1174
 
        w = Weave()
1175
 
        k = self.make_test_knit()
1176
 
        self.failUnless(WeaveToKnit.is_compatible(w, k))
1177
 
        self.failIf(WeaveToKnit.is_compatible(k, w))
1178
 
        self.failIf(WeaveToKnit.is_compatible(w, w))
1179
 
        self.failIf(WeaveToKnit.is_compatible(k, k))
1180
 
 
1181
 
 
1182
 
class TestKnitCaching(KnitTests):
1183
 
    
1184
 
    def create_knit(self, cache_add=False):
1185
 
        k = self.make_test_knit(True)
1186
 
        if cache_add:
1187
 
            k.enable_cache()
1188
 
 
1189
 
        k.add_lines('text-1', [], split_lines(TEXT_1))
1190
 
        k.add_lines('text-2', [], split_lines(TEXT_2))
1191
 
        return k
1192
 
 
1193
 
    def test_no_caching(self):
1194
 
        k = self.create_knit()
1195
 
        # Nothing should be cached without setting 'enable_cache'
1196
 
        self.assertEqual({}, k._data._cache)
1197
 
 
1198
 
    def test_cache_add_and_clear(self):
1199
 
        k = self.create_knit(True)
1200
 
 
1201
 
        self.assertEqual(['text-1', 'text-2'], sorted(k._data._cache.keys()))
1202
 
 
1203
 
        k.clear_cache()
1204
 
        self.assertEqual({}, k._data._cache)
1205
 
 
1206
 
    def test_cache_data_read_raw(self):
1207
 
        k = self.create_knit()
1208
 
 
1209
 
        # Now cache and read
1210
 
        k.enable_cache()
1211
 
 
1212
 
        def read_one_raw(version):
1213
 
            pos_map = k._get_components_positions([version])
1214
 
            method, pos, size, next = pos_map[version]
1215
 
            lst = list(k._data.read_records_iter_raw([(version, pos, size)]))
1216
 
            self.assertEqual(1, len(lst))
1217
 
            return lst[0]
1218
 
 
1219
 
        val = read_one_raw('text-1')
1220
 
        self.assertEqual({'text-1':val[1]}, k._data._cache)
1221
 
 
1222
 
        k.clear_cache()
1223
 
        # After clear, new reads are not cached
1224
 
        self.assertEqual({}, k._data._cache)
1225
 
 
1226
 
        val2 = read_one_raw('text-1')
1227
 
        self.assertEqual(val, val2)
1228
 
        self.assertEqual({}, k._data._cache)
1229
 
 
1230
 
    def test_cache_data_read(self):
1231
 
        k = self.create_knit()
1232
 
 
1233
 
        def read_one(version):
1234
 
            pos_map = k._get_components_positions([version])
1235
 
            method, pos, size, next = pos_map[version]
1236
 
            lst = list(k._data.read_records_iter([(version, pos, size)]))
1237
 
            self.assertEqual(1, len(lst))
1238
 
            return lst[0]
1239
 
 
1240
 
        # Now cache and read
1241
 
        k.enable_cache()
1242
 
 
1243
 
        val = read_one('text-2')
1244
 
        self.assertEqual(['text-2'], k._data._cache.keys())
1245
 
        self.assertEqual('text-2', val[0])
1246
 
        content, digest = k._data._parse_record('text-2',
1247
 
                                                k._data._cache['text-2'])
1248
 
        self.assertEqual(content, val[1])
1249
 
        self.assertEqual(digest, val[2])
1250
 
 
1251
 
        k.clear_cache()
1252
 
        self.assertEqual({}, k._data._cache)
1253
 
 
1254
 
        val2 = read_one('text-2')
1255
 
        self.assertEqual(val, val2)
1256
 
        self.assertEqual({}, k._data._cache)
1257
 
 
1258
 
    def test_cache_read(self):
1259
 
        k = self.create_knit()
1260
 
        k.enable_cache()
1261
 
 
1262
 
        text = k.get_text('text-1')
1263
 
        self.assertEqual(TEXT_1, text)
1264
 
        self.assertEqual(['text-1'], k._data._cache.keys())
1265
 
 
1266
 
        k.clear_cache()
1267
 
        self.assertEqual({}, k._data._cache)
1268
 
 
1269
 
        text = k.get_text('text-1')
1270
 
        self.assertEqual(TEXT_1, text)
1271
 
        self.assertEqual({}, k._data._cache)
1272
 
 
1273
 
 
1274
 
class TestKnitIndex(KnitTests):
1275
 
 
1276
 
    def test_add_versions_dictionary_compresses(self):
1277
 
        """Adding versions to the index should update the lookup dict"""
1278
 
        knit = self.make_test_knit()
1279
 
        idx = knit._index
1280
 
        idx.add_version('a-1', ['fulltext'], 0, 0, [])
1281
 
        self.check_file_contents('test.kndx',
1282
 
            '# bzr knit index 8\n'
1283
 
            '\n'
1284
 
            'a-1 fulltext 0 0  :'
1285
 
            )
1286
 
        idx.add_versions([('a-2', ['fulltext'], 0, 0, ['a-1']),
1287
 
                          ('a-3', ['fulltext'], 0, 0, ['a-2']),
1288
 
                         ])
1289
 
        self.check_file_contents('test.kndx',
1290
 
            '# bzr knit index 8\n'
1291
 
            '\n'
1292
 
            'a-1 fulltext 0 0  :\n'
1293
 
            'a-2 fulltext 0 0 0 :\n'
1294
 
            'a-3 fulltext 0 0 1 :'
1295
 
            )
1296
 
        self.assertEqual(['a-1', 'a-2', 'a-3'], idx._history)
1297
 
        self.assertEqual({'a-1':('a-1', ['fulltext'], 0, 0, [], 0),
1298
 
                          'a-2':('a-2', ['fulltext'], 0, 0, ['a-1'], 1),
1299
 
                          'a-3':('a-3', ['fulltext'], 0, 0, ['a-2'], 2),
1300
 
                         }, idx._cache)
1301
 
 
1302
 
    def test_add_versions_fails_clean(self):
1303
 
        """If add_versions fails in the middle, it restores a pristine state.
1304
 
 
1305
 
        Any modifications that are made to the index are reset if all versions
1306
 
        cannot be added.
1307
 
        """
1308
 
        # This cheats a little bit by passing in a generator which will
1309
 
        # raise an exception before the processing finishes
1310
 
        # Other possibilities would be to have an version with the wrong number
1311
 
        # of entries, or to make the backing transport unable to write any
1312
 
        # files.
1313
 
 
1314
 
        knit = self.make_test_knit()
1315
 
        idx = knit._index
1316
 
        idx.add_version('a-1', ['fulltext'], 0, 0, [])
1317
 
 
1318
 
        class StopEarly(Exception):
1319
 
            pass
1320
 
 
1321
 
        def generate_failure():
1322
 
            """Add some entries and then raise an exception"""
1323
 
            yield ('a-2', ['fulltext'], 0, 0, ['a-1'])
1324
 
            yield ('a-3', ['fulltext'], 0, 0, ['a-2'])
1325
 
            raise StopEarly()
1326
 
 
1327
 
        # Assert the pre-condition
1328
 
        self.assertEqual(['a-1'], idx._history)
1329
 
        self.assertEqual({'a-1':('a-1', ['fulltext'], 0, 0, [], 0)}, idx._cache)
1330
 
 
1331
 
        self.assertRaises(StopEarly, idx.add_versions, generate_failure())
1332
 
 
1333
 
        # And it shouldn't be modified
1334
 
        self.assertEqual(['a-1'], idx._history)
1335
 
        self.assertEqual({'a-1':('a-1', ['fulltext'], 0, 0, [], 0)}, idx._cache)
1336
 
 
1337
 
    def test_knit_index_ignores_empty_files(self):
1338
 
        # There was a race condition in older bzr, where a ^C at the right time
1339
 
        # could leave an empty .kndx file, which bzr would later claim was a
1340
 
        # corrupted file since the header was not present. In reality, the file
1341
 
        # just wasn't created, so it should be ignored.
1342
 
        t = get_transport('.')
1343
 
        t.put_bytes('test.kndx', '')
1344
 
 
1345
 
        knit = self.make_test_knit()
1346
 
 
1347
 
    def test_knit_index_checks_header(self):
1348
 
        t = get_transport('.')
1349
 
        t.put_bytes('test.kndx', '# not really a knit header\n\n')
1350
 
 
1351
 
        self.assertRaises(KnitHeaderError, self.make_test_knit)