← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_knit.py

  • Committer: John Arbash Meinel
  • Date: 2006-04-25 15:05:42 UTC
  • mfrom: (1185.85.85 bzr-encoding)
  • mto: This revision was merged to the branch mainline in revision 1752.
  • Revision ID: john@arbash-meinel.com-20060425150542-c7b518dca9928691
[merge] the old bzr-encoding changes, reparenting them on bzr.dev

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_knit.py
 
1
# Copyright (C) 2005, 2006 by 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
 
 
20
import difflib
 
21
 
 
22
 
 
23
from bzrlib.errors import KnitError, RevisionAlreadyPresent
 
24
from bzrlib.knit import KnitVersionedFile, KnitPlainFactory, KnitAnnotateFactory
 
25
from bzrlib.osutils import split_lines
 
26
from bzrlib.tests import TestCaseInTempDir
 
27
from bzrlib.transport import TransportLogger
 
28
from bzrlib.transport.local import LocalTransport
 
29
from bzrlib.transport.memory import MemoryTransport
 
30
 
 
31
 
 
32
class KnitTests(TestCaseInTempDir):
 
33
 
 
34
    def add_stock_one_and_one_a(self, k):
 
35
        k.add_lines('text-1', [], split_lines(TEXT_1))
 
36
        k.add_lines('text-1a', ['text-1'], split_lines(TEXT_1A))
 
37
 
 
38
    def test_knit_constructor(self):
 
39
        """Construct empty k"""
 
40
        self.make_test_knit()
 
41
 
 
42
    def make_test_knit(self, annotate=False):
 
43
        if not annotate:
 
44
            factory = KnitPlainFactory()
 
45
        else:
 
46
            factory = None
 
47
        return KnitVersionedFile('test', LocalTransport('.'), access_mode='w', factory=factory, create=True)
 
48
 
 
49
    def test_knit_add(self):
 
50
        """Store one text in knit and retrieve"""
 
51
        k = self.make_test_knit()
 
52
        k.add_lines('text-1', [], split_lines(TEXT_1))
 
53
        self.assertTrue(k.has_version('text-1'))
 
54
        self.assertEqualDiff(''.join(k.get_lines('text-1')), TEXT_1)
 
55
 
 
56
    def test_knit_reload(self):
 
57
        # test that the content in a reloaded knit is correct
 
58
        k = self.make_test_knit()
 
59
        k.add_lines('text-1', [], split_lines(TEXT_1))
 
60
        del k
 
61
        k2 = KnitVersionedFile('test', LocalTransport('.'), access_mode='r', factory=KnitPlainFactory(), create=True)
 
62
        self.assertTrue(k2.has_version('text-1'))
 
63
        self.assertEqualDiff(''.join(k2.get_lines('text-1')), TEXT_1)
 
64
 
 
65
    def test_knit_several(self):
 
66
        """Store several texts in a knit"""
 
67
        k = self.make_test_knit()
 
68
        k.add_lines('text-1', [], split_lines(TEXT_1))
 
69
        k.add_lines('text-2', [], split_lines(TEXT_2))
 
70
        self.assertEqualDiff(''.join(k.get_lines('text-1')), TEXT_1)
 
71
        self.assertEqualDiff(''.join(k.get_lines('text-2')), TEXT_2)
 
72
        
 
73
    def test_repeated_add(self):
 
74
        """Knit traps attempt to replace existing version"""
 
75
        k = self.make_test_knit()
 
76
        k.add_lines('text-1', [], split_lines(TEXT_1))
 
77
        self.assertRaises(RevisionAlreadyPresent, 
 
78
                k.add_lines,
 
79
                'text-1', [], split_lines(TEXT_1))
 
80
 
 
81
    def test_empty(self):
 
82
        k = self.make_test_knit(True)
 
83
        k.add_lines('text-1', [], [])
 
84
        self.assertEquals(k.get_lines('text-1'), [])
 
85
 
 
86
    def test_incomplete(self):
 
87
        """Test if texts without a ending line-end can be inserted and
 
88
        extracted."""
 
89
        k = KnitVersionedFile('test', LocalTransport('.'), delta=False, create=True)
 
90
        k.add_lines('text-1', [], ['a\n',    'b'  ])
 
91
        k.add_lines('text-2', ['text-1'], ['a\rb\n', 'b\n'])
 
92
        # reopening ensures maximum room for confusion
 
93
        k = KnitVersionedFile('test', LocalTransport('.'), delta=False, create=True)
 
94
        self.assertEquals(k.get_lines('text-1'), ['a\n',    'b'  ])
 
95
        self.assertEquals(k.get_lines('text-2'), ['a\rb\n', 'b\n'])
 
96
 
 
97
    def test_delta(self):
 
98
        """Expression of knit delta as lines"""
 
99
        k = self.make_test_knit()
 
100
        td = list(line_delta(TEXT_1.splitlines(True),
 
101
                             TEXT_1A.splitlines(True)))
 
102
        self.assertEqualDiff(''.join(td), delta_1_1a)
 
103
        out = apply_line_delta(TEXT_1.splitlines(True), td)
 
104
        self.assertEqualDiff(''.join(out), TEXT_1A)
 
105
 
 
106
    def test_add_with_parents(self):
 
107
        """Store in knit with parents"""
 
108
        k = self.make_test_knit()
 
109
        self.add_stock_one_and_one_a(k)
 
110
        self.assertEquals(k.get_parents('text-1'), [])
 
111
        self.assertEquals(k.get_parents('text-1a'), ['text-1'])
 
112
 
 
113
    def test_ancestry(self):
 
114
        """Store in knit with parents"""
 
115
        k = self.make_test_knit()
 
116
        self.add_stock_one_and_one_a(k)
 
117
        self.assertEquals(set(k.get_ancestry(['text-1a'])), set(['text-1a', 'text-1']))
 
118
 
 
119
    def test_add_delta(self):
 
120
        """Store in knit with parents"""
 
121
        k = KnitVersionedFile('test', LocalTransport('.'), factory=KnitPlainFactory(),
 
122
            delta=True, create=True)
 
123
        self.add_stock_one_and_one_a(k)
 
124
        k.clear_cache()
 
125
        self.assertEqualDiff(''.join(k.get_lines('text-1a')), TEXT_1A)
 
126
 
 
127
    def test_annotate(self):
 
128
        """Annotations"""
 
129
        k = KnitVersionedFile('knit', LocalTransport('.'), factory=KnitAnnotateFactory(),
 
130
            delta=True, create=True)
 
131
        self.insert_and_test_small_annotate(k)
 
132
 
 
133
    def insert_and_test_small_annotate(self, k):
 
134
        """test annotation with k works correctly."""
 
135
        k.add_lines('text-1', [], ['a\n', 'b\n'])
 
136
        k.add_lines('text-2', ['text-1'], ['a\n', 'c\n'])
 
137
 
 
138
        origins = k.annotate('text-2')
 
139
        self.assertEquals(origins[0], ('text-1', 'a\n'))
 
140
        self.assertEquals(origins[1], ('text-2', 'c\n'))
 
141
 
 
142
    def test_annotate_fulltext(self):
 
143
        """Annotations"""
 
144
        k = KnitVersionedFile('knit', LocalTransport('.'), factory=KnitAnnotateFactory(),
 
145
            delta=False, create=True)
 
146
        self.insert_and_test_small_annotate(k)
 
147
 
 
148
    def test_annotate_merge_1(self):
 
149
        k = self.make_test_knit(True)
 
150
        k.add_lines('text-a1', [], ['a\n', 'b\n'])
 
151
        k.add_lines('text-a2', [], ['d\n', 'c\n'])
 
152
        k.add_lines('text-am', ['text-a1', 'text-a2'], ['d\n', 'b\n'])
 
153
        origins = k.annotate('text-am')
 
154
        self.assertEquals(origins[0], ('text-a2', 'd\n'))
 
155
        self.assertEquals(origins[1], ('text-a1', 'b\n'))
 
156
 
 
157
    def test_annotate_merge_2(self):
 
158
        k = self.make_test_knit(True)
 
159
        k.add_lines('text-a1', [], ['a\n', 'b\n', 'c\n'])
 
160
        k.add_lines('text-a2', [], ['x\n', 'y\n', 'z\n'])
 
161
        k.add_lines('text-am', ['text-a1', 'text-a2'], ['a\n', 'y\n', 'c\n'])
 
162
        origins = k.annotate('text-am')
 
163
        self.assertEquals(origins[0], ('text-a1', 'a\n'))
 
164
        self.assertEquals(origins[1], ('text-a2', 'y\n'))
 
165
        self.assertEquals(origins[2], ('text-a1', 'c\n'))
 
166
 
 
167
    def test_annotate_merge_9(self):
 
168
        k = self.make_test_knit(True)
 
169
        k.add_lines('text-a1', [], ['a\n', 'b\n', 'c\n'])
 
170
        k.add_lines('text-a2', [], ['x\n', 'y\n', 'z\n'])
 
171
        k.add_lines('text-am', ['text-a1', 'text-a2'], ['k\n', 'y\n', 'c\n'])
 
172
        origins = k.annotate('text-am')
 
173
        self.assertEquals(origins[0], ('text-am', 'k\n'))
 
174
        self.assertEquals(origins[1], ('text-a2', 'y\n'))
 
175
        self.assertEquals(origins[2], ('text-a1', 'c\n'))
 
176
 
 
177
    def test_annotate_merge_3(self):
 
178
        k = self.make_test_knit(True)
 
179
        k.add_lines('text-a1', [], ['a\n', 'b\n', 'c\n'])
 
180
        k.add_lines('text-a2', [] ,['x\n', 'y\n', 'z\n'])
 
181
        k.add_lines('text-am', ['text-a1', 'text-a2'], ['k\n', 'y\n', 'z\n'])
 
182
        origins = k.annotate('text-am')
 
183
        self.assertEquals(origins[0], ('text-am', 'k\n'))
 
184
        self.assertEquals(origins[1], ('text-a2', 'y\n'))
 
185
        self.assertEquals(origins[2], ('text-a2', 'z\n'))
 
186
 
 
187
    def test_annotate_merge_4(self):
 
188
        k = self.make_test_knit(True)
 
189
        k.add_lines('text-a1', [], ['a\n', 'b\n', 'c\n'])
 
190
        k.add_lines('text-a2', [], ['x\n', 'y\n', 'z\n'])
 
191
        k.add_lines('text-a3', ['text-a1'], ['a\n', 'b\n', 'p\n'])
 
192
        k.add_lines('text-am', ['text-a2', 'text-a3'], ['a\n', 'b\n', 'z\n'])
 
193
        origins = k.annotate('text-am')
 
194
        self.assertEquals(origins[0], ('text-a1', 'a\n'))
 
195
        self.assertEquals(origins[1], ('text-a1', 'b\n'))
 
196
        self.assertEquals(origins[2], ('text-a2', 'z\n'))
 
197
 
 
198
    def test_annotate_merge_5(self):
 
199
        k = self.make_test_knit(True)
 
200
        k.add_lines('text-a1', [], ['a\n', 'b\n', 'c\n'])
 
201
        k.add_lines('text-a2', [], ['d\n', 'e\n', 'f\n'])
 
202
        k.add_lines('text-a3', [], ['x\n', 'y\n', 'z\n'])
 
203
        k.add_lines('text-am',
 
204
                    ['text-a1', 'text-a2', 'text-a3'],
 
205
                    ['a\n', 'e\n', 'z\n'])
 
206
        origins = k.annotate('text-am')
 
207
        self.assertEquals(origins[0], ('text-a1', 'a\n'))
 
208
        self.assertEquals(origins[1], ('text-a2', 'e\n'))
 
209
        self.assertEquals(origins[2], ('text-a3', 'z\n'))
 
210
 
 
211
    def test_annotate_file_cherry_pick(self):
 
212
        k = self.make_test_knit(True)
 
213
        k.add_lines('text-1', [], ['a\n', 'b\n', 'c\n'])
 
214
        k.add_lines('text-2', ['text-1'], ['d\n', 'e\n', 'f\n'])
 
215
        k.add_lines('text-3', ['text-2', 'text-1'], ['a\n', 'b\n', 'c\n'])
 
216
        origins = k.annotate('text-3')
 
217
        self.assertEquals(origins[0], ('text-1', 'a\n'))
 
218
        self.assertEquals(origins[1], ('text-1', 'b\n'))
 
219
        self.assertEquals(origins[2], ('text-1', 'c\n'))
 
220
 
 
221
    def test_knit_join(self):
 
222
        """Store in knit with parents"""
 
223
        k1 = KnitVersionedFile('test1', LocalTransport('.'), factory=KnitPlainFactory(), create=True)
 
224
        k1.add_lines('text-a', [], split_lines(TEXT_1))
 
225
        k1.add_lines('text-b', ['text-a'], split_lines(TEXT_1))
 
226
 
 
227
        k1.add_lines('text-c', [], split_lines(TEXT_1))
 
228
        k1.add_lines('text-d', ['text-c'], split_lines(TEXT_1))
 
229
 
 
230
        k1.add_lines('text-m', ['text-b', 'text-d'], split_lines(TEXT_1))
 
231
 
 
232
        k2 = KnitVersionedFile('test2', LocalTransport('.'), factory=KnitPlainFactory(), create=True)
 
233
        count = k2.join(k1, version_ids=['text-m'])
 
234
        self.assertEquals(count, 5)
 
235
        self.assertTrue(k2.has_version('text-a'))
 
236
        self.assertTrue(k2.has_version('text-c'))
 
237
 
 
238
    def test_reannotate(self):
 
239
        k1 = KnitVersionedFile('knit1', LocalTransport('.'),
 
240
                               factory=KnitAnnotateFactory(), create=True)
 
241
        # 0
 
242
        k1.add_lines('text-a', [], ['a\n', 'b\n'])
 
243
        # 1
 
244
        k1.add_lines('text-b', ['text-a'], ['a\n', 'c\n'])
 
245
 
 
246
        k2 = KnitVersionedFile('test2', LocalTransport('.'),
 
247
                               factory=KnitAnnotateFactory(), create=True)
 
248
        k2.join(k1, version_ids=['text-b'])
 
249
 
 
250
        # 2
 
251
        k1.add_lines('text-X', ['text-b'], ['a\n', 'b\n'])
 
252
        # 2
 
253
        k2.add_lines('text-c', ['text-b'], ['z\n', 'c\n'])
 
254
        # 3
 
255
        k2.add_lines('text-Y', ['text-b'], ['b\n', 'c\n'])
 
256
 
 
257
        # test-c will have index 3
 
258
        k1.join(k2, version_ids=['text-c'])
 
259
 
 
260
        lines = k1.get_lines('text-c')
 
261
        self.assertEquals(lines, ['z\n', 'c\n'])
 
262
 
 
263
        origins = k1.annotate('text-c')
 
264
        self.assertEquals(origins[0], ('text-c', 'z\n'))
 
265
        self.assertEquals(origins[1], ('text-b', 'c\n'))
 
266
 
 
267
    def test_extraction_reads_components_once(self):
 
268
        t = MemoryTransport()
 
269
        instrumented_t = TransportLogger(t)
 
270
        k1 = KnitVersionedFile('id', instrumented_t, create=True, delta=True)
 
271
        # should read the index
 
272
        self.assertEqual([('id.kndx',)], instrumented_t._calls)
 
273
        instrumented_t._calls = []
 
274
        # add a text       
 
275
        k1.add_lines('base', [], ['text\n'])
 
276
        # should not have read at all
 
277
        self.assertEqual([], instrumented_t._calls)
 
278
 
 
279
        # add a text
 
280
        k1.add_lines('sub', ['base'], ['text\n', 'text2\n'])
 
281
        # should not have read at all
 
282
        self.assertEqual([], instrumented_t._calls)
 
283
        
 
284
        # read a text
 
285
        k1.get_lines('sub')
 
286
        # should not have read at all
 
287
        self.assertEqual([], instrumented_t._calls)
 
288
 
 
289
        # clear the cache
 
290
        k1.clear_cache()
 
291
 
 
292
        # read a text
 
293
        k1.get_lines('base')
 
294
        # should have read a component
 
295
        # should not have read the first component only
 
296
        self.assertEqual([('id.knit', [(0, 87)])], instrumented_t._calls)
 
297
        instrumented_t._calls = []
 
298
        # read again
 
299
        k1.get_lines('base')
 
300
        # should not have read at all
 
301
        self.assertEqual([], instrumented_t._calls)
 
302
        # and now read the other component
 
303
        k1.get_lines('sub')
 
304
        # should have read the second component
 
305
        self.assertEqual([('id.knit', [(87, 93)])], instrumented_t._calls)
 
306
        instrumented_t._calls = []
 
307
 
 
308
        # clear the cache
 
309
        k1.clear_cache()
 
310
        # add a text cold 
 
311
        k1.add_lines('sub2', ['base'], ['text\n', 'text3\n'])
 
312
        # should read the first component only
 
313
        self.assertEqual([('id.knit', [(0, 87)])], instrumented_t._calls)
 
314
        
 
315
    def test_iter_lines_reads_in_order(self):
 
316
        t = MemoryTransport()
 
317
        instrumented_t = TransportLogger(t)
 
318
        k1 = KnitVersionedFile('id', instrumented_t, create=True, delta=True)
 
319
        self.assertEqual([('id.kndx',)], instrumented_t._calls)
 
320
        # add texts with no required ordering
 
321
        k1.add_lines('base', [], ['text\n'])
 
322
        k1.add_lines('base2', [], ['text2\n'])
 
323
        k1.clear_cache()
 
324
        instrumented_t._calls = []
 
325
        # request a last-first iteration
 
326
        results = list(k1.iter_lines_added_or_present_in_versions(['base2', 'base']))
 
327
        self.assertEqual([('id.knit', [(0, 87), (87, 89)])], instrumented_t._calls)
 
328
        self.assertEqual(['text\n', 'text2\n'], results)
 
329
 
 
330
    def test_create_empty_annotated(self):
 
331
        k1 = self.make_test_knit(True)
 
332
        # 0
 
333
        k1.add_lines('text-a', [], ['a\n', 'b\n'])
 
334
        k2 = k1.create_empty('t', MemoryTransport())
 
335
        self.assertTrue(isinstance(k2.factory, KnitAnnotateFactory))
 
336
        self.assertEqual(k1.delta, k2.delta)
 
337
        # the generic test checks for empty content and file class
 
338
 
 
339
    def test_knit_format(self):
 
340
        # this tests that a new knit index file has the expected content
 
341
        # and that is writes the data we expect as records are added.
 
342
        knit = self.make_test_knit(True)
 
343
        self.assertFileEqual("# bzr knit index 8\n", 'test.kndx')
 
344
        knit.add_lines_with_ghosts('revid', ['a_ghost'], ['a\n'])
 
345
        self.assertFileEqual(
 
346
            "# bzr knit index 8\n"
 
347
            "\n"
 
348
            "revid fulltext 0 84 .a_ghost :",
 
349
            'test.kndx')
 
350
        knit.add_lines_with_ghosts('revid2', ['revid'], ['a\n'])
 
351
        self.assertFileEqual(
 
352
            "# bzr knit index 8\n"
 
353
            "\nrevid fulltext 0 84 .a_ghost :"
 
354
            "\nrevid2 line-delta 84 82 0 :",
 
355
            'test.kndx')
 
356
        # we should be able to load this file again
 
357
        knit = KnitVersionedFile('test', LocalTransport('.'), access_mode='r')
 
358
        self.assertEqual(['revid', 'revid2'], knit.versions())
 
359
        # write a short write to the file and ensure that its ignored
 
360
        indexfile = file('test.kndx', 'at')
 
361
        indexfile.write('\nrevid3 line-delta 166 82 1 2 3 4 5 .phwoar:demo ')
 
362
        indexfile.close()
 
363
        # we should be able to load this file again
 
364
        knit = KnitVersionedFile('test', LocalTransport('.'), access_mode='w')
 
365
        self.assertEqual(['revid', 'revid2'], knit.versions())
 
366
        # and add a revision with the same id the failed write had
 
367
        knit.add_lines('revid3', ['revid2'], ['a\n'])
 
368
        # and when reading it revid3 should now appear.
 
369
        knit = KnitVersionedFile('test', LocalTransport('.'), access_mode='r')
 
370
        self.assertEqual(['revid', 'revid2', 'revid3'], knit.versions())
 
371
        self.assertEqual(['revid2'], knit.get_parents('revid3'))
 
372
 
 
373
    def test_plan_merge(self):
 
374
        my_knit = self.make_test_knit(annotate=True)
 
375
        my_knit.add_lines('text1', [], split_lines(TEXT_1))
 
376
        my_knit.add_lines('text1a', ['text1'], split_lines(TEXT_1A))
 
377
        my_knit.add_lines('text1b', ['text1'], split_lines(TEXT_1B))
 
378
        plan = list(my_knit.plan_merge('text1a', 'text1b'))
 
379
        for plan_line, expected_line in zip(plan, AB_MERGE):
 
380
            self.assertEqual(plan_line, expected_line)
 
381
 
 
382
 
 
383
TEXT_1 = """\
 
384
Banana cup cakes:
 
385
 
 
386
- bananas
 
387
- eggs
 
388
- broken tea cups
 
389
"""
 
390
 
 
391
TEXT_1A = """\
 
392
Banana cup cake recipe
 
393
(serves 6)
 
394
 
 
395
- bananas
 
396
- eggs
 
397
- broken tea cups
 
398
- self-raising flour
 
399
"""
 
400
 
 
401
TEXT_1B = """\
 
402
Banana cup cake recipe
 
403
 
 
404
- bananas (do not use plantains!!!)
 
405
- broken tea cups
 
406
- flour
 
407
"""
 
408
 
 
409
delta_1_1a = """\
 
410
0,1,2
 
411
Banana cup cake recipe
 
412
(serves 6)
 
413
5,5,1
 
414
- self-raising flour
 
415
"""
 
416
 
 
417
TEXT_2 = """\
 
418
Boeuf bourguignon
 
419
 
 
420
- beef
 
421
- red wine
 
422
- small onions
 
423
- carrot
 
424
- mushrooms
 
425
"""
 
426
 
 
427
AB_MERGE_TEXT="""unchanged|Banana cup cake recipe
 
428
new-a|(serves 6)
 
429
unchanged|
 
430
killed-b|- bananas
 
431
killed-b|- eggs
 
432
new-b|- bananas (do not use plantains!!!)
 
433
unchanged|- broken tea cups
 
434
new-a|- self-raising flour
 
435
new-b|- flour
 
436
"""
 
437
AB_MERGE=[tuple(l.split('|')) for l in AB_MERGE_TEXT.splitlines(True)]
 
438
 
 
439
 
 
440
def line_delta(from_lines, to_lines):
 
441
    """Generate line-based delta from one text to another"""
 
442
    s = difflib.SequenceMatcher(None, from_lines, to_lines)
 
443
    for op in s.get_opcodes():
 
444
        if op[0] == 'equal':
 
445
            continue
 
446
        yield '%d,%d,%d\n' % (op[1], op[2], op[4]-op[3])
 
447
        for i in range(op[3], op[4]):
 
448
            yield to_lines[i]
 
449
 
 
450
 
 
451
def apply_line_delta(basis_lines, delta_lines):
 
452
    """Apply a line-based perfect diff
 
453
    
 
454
    basis_lines -- text to apply the patch to
 
455
    delta_lines -- diff instructions and content
 
456
    """
 
457
    out = basis_lines[:]
 
458
    i = 0
 
459
    offset = 0
 
460
    while i < len(delta_lines):
 
461
        l = delta_lines[i]
 
462
        a, b, c = map(long, l.split(','))
 
463
        i = i + 1
 
464
        out[offset+a:offset+b] = delta_lines[i:i+c]
 
465
        i = i + c
 
466
        offset = offset + (b - a) + c
 
467
    return out