~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_knit.py

MergeĀ fromĀ mainline

Show diffs side-by-side

added added

removed removed

Lines of Context:
 
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
        """Store and reload a knit"""
 
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
        self.assertEquals(k.get_lines('text-1'), ['a\n',    'b'  ])
 
93
        self.assertEquals(k.get_lines('text-2'), ['a\rb\n', 'b\n'])
 
94
 
 
95
    def test_delta(self):
 
96
        """Expression of knit delta as lines"""
 
97
        k = self.make_test_knit()
 
98
        td = list(line_delta(TEXT_1.splitlines(True),
 
99
                             TEXT_1A.splitlines(True)))
 
100
        self.assertEqualDiff(''.join(td), delta_1_1a)
 
101
        out = apply_line_delta(TEXT_1.splitlines(True), td)
 
102
        self.assertEqualDiff(''.join(out), TEXT_1A)
 
103
 
 
104
    def test_add_with_parents(self):
 
105
        """Store in knit with parents"""
 
106
        k = self.make_test_knit()
 
107
        self.add_stock_one_and_one_a(k)
 
108
        self.assertEquals(k.get_parents('text-1'), [])
 
109
        self.assertEquals(k.get_parents('text-1a'), ['text-1'])
 
110
 
 
111
    def test_ancestry(self):
 
112
        """Store in knit with parents"""
 
113
        k = self.make_test_knit()
 
114
        self.add_stock_one_and_one_a(k)
 
115
        self.assertEquals(set(k.get_ancestry(['text-1a'])), set(['text-1a', 'text-1']))
 
116
 
 
117
    def test_add_delta(self):
 
118
        """Store in knit with parents"""
 
119
        k = KnitVersionedFile('test', LocalTransport('.'), factory=KnitPlainFactory(),
 
120
            delta=True, create=True)
 
121
        self.add_stock_one_and_one_a(k)
 
122
        self.assertEqualDiff(''.join(k.get_lines('text-1a')), TEXT_1A)
 
123
 
 
124
    def test_annotate(self):
 
125
        """Annotations"""
 
126
        k = KnitVersionedFile('knit', LocalTransport('.'), factory=KnitAnnotateFactory(),
 
127
            delta=True, create=True)
 
128
        self.insert_and_test_small_annotate(k)
 
129
 
 
130
    def insert_and_test_small_annotate(self, k):
 
131
        """test annotation with k works correctly."""
 
132
        k.add_lines('text-1', [], ['a\n', 'b\n'])
 
133
        k.add_lines('text-2', ['text-1'], ['a\n', 'c\n'])
 
134
 
 
135
        origins = k.annotate('text-2')
 
136
        self.assertEquals(origins[0], ('text-1', 'a\n'))
 
137
        self.assertEquals(origins[1], ('text-2', 'c\n'))
 
138
 
 
139
    def test_annotate_fulltext(self):
 
140
        """Annotations"""
 
141
        k = KnitVersionedFile('knit', LocalTransport('.'), factory=KnitAnnotateFactory(),
 
142
            delta=False, create=True)
 
143
        self.insert_and_test_small_annotate(k)
 
144
 
 
145
    def test_annotate_merge_1(self):
 
146
        k = self.make_test_knit(True)
 
147
        k.add_lines('text-a1', [], ['a\n', 'b\n'])
 
148
        k.add_lines('text-a2', [], ['d\n', 'c\n'])
 
149
        k.add_lines('text-am', ['text-a1', 'text-a2'], ['d\n', 'b\n'])
 
150
        origins = k.annotate('text-am')
 
151
        self.assertEquals(origins[0], ('text-a2', 'd\n'))
 
152
        self.assertEquals(origins[1], ('text-a1', 'b\n'))
 
153
 
 
154
    def test_annotate_merge_2(self):
 
155
        k = self.make_test_knit(True)
 
156
        k.add_lines('text-a1', [], ['a\n', 'b\n', 'c\n'])
 
157
        k.add_lines('text-a2', [], ['x\n', 'y\n', 'z\n'])
 
158
        k.add_lines('text-am', ['text-a1', 'text-a2'], ['a\n', 'y\n', 'c\n'])
 
159
        origins = k.annotate('text-am')
 
160
        self.assertEquals(origins[0], ('text-a1', 'a\n'))
 
161
        self.assertEquals(origins[1], ('text-a2', 'y\n'))
 
162
        self.assertEquals(origins[2], ('text-a1', 'c\n'))
 
163
 
 
164
    def test_annotate_merge_9(self):
 
165
        k = self.make_test_knit(True)
 
166
        k.add_lines('text-a1', [], ['a\n', 'b\n', 'c\n'])
 
167
        k.add_lines('text-a2', [], ['x\n', 'y\n', 'z\n'])
 
168
        k.add_lines('text-am', ['text-a1', 'text-a2'], ['k\n', 'y\n', 'c\n'])
 
169
        origins = k.annotate('text-am')
 
170
        self.assertEquals(origins[0], ('text-am', 'k\n'))
 
171
        self.assertEquals(origins[1], ('text-a2', 'y\n'))
 
172
        self.assertEquals(origins[2], ('text-a1', 'c\n'))
 
173
 
 
174
    def test_annotate_merge_3(self):
 
175
        k = self.make_test_knit(True)
 
176
        k.add_lines('text-a1', [], ['a\n', 'b\n', 'c\n'])
 
177
        k.add_lines('text-a2', [] ,['x\n', 'y\n', 'z\n'])
 
178
        k.add_lines('text-am', ['text-a1', 'text-a2'], ['k\n', 'y\n', 'z\n'])
 
179
        origins = k.annotate('text-am')
 
180
        self.assertEquals(origins[0], ('text-am', 'k\n'))
 
181
        self.assertEquals(origins[1], ('text-a2', 'y\n'))
 
182
        self.assertEquals(origins[2], ('text-a2', 'z\n'))
 
183
 
 
184
    def test_annotate_merge_4(self):
 
185
        k = self.make_test_knit(True)
 
186
        k.add_lines('text-a1', [], ['a\n', 'b\n', 'c\n'])
 
187
        k.add_lines('text-a2', [], ['x\n', 'y\n', 'z\n'])
 
188
        k.add_lines('text-a3', ['text-a1'], ['a\n', 'b\n', 'p\n'])
 
189
        k.add_lines('text-am', ['text-a2', 'text-a3'], ['a\n', 'b\n', 'z\n'])
 
190
        origins = k.annotate('text-am')
 
191
        self.assertEquals(origins[0], ('text-a1', 'a\n'))
 
192
        self.assertEquals(origins[1], ('text-a1', 'b\n'))
 
193
        self.assertEquals(origins[2], ('text-a2', 'z\n'))
 
194
 
 
195
    def test_annotate_merge_5(self):
 
196
        k = self.make_test_knit(True)
 
197
        k.add_lines('text-a1', [], ['a\n', 'b\n', 'c\n'])
 
198
        k.add_lines('text-a2', [], ['d\n', 'e\n', 'f\n'])
 
199
        k.add_lines('text-a3', [], ['x\n', 'y\n', 'z\n'])
 
200
        k.add_lines('text-am',
 
201
                    ['text-a1', 'text-a2', 'text-a3'],
 
202
                    ['a\n', 'e\n', 'z\n'])
 
203
        origins = k.annotate('text-am')
 
204
        self.assertEquals(origins[0], ('text-a1', 'a\n'))
 
205
        self.assertEquals(origins[1], ('text-a2', 'e\n'))
 
206
        self.assertEquals(origins[2], ('text-a3', 'z\n'))
 
207
 
 
208
    def test_annotate_file_cherry_pick(self):
 
209
        k = self.make_test_knit(True)
 
210
        k.add_lines('text-1', [], ['a\n', 'b\n', 'c\n'])
 
211
        k.add_lines('text-2', ['text-1'], ['d\n', 'e\n', 'f\n'])
 
212
        k.add_lines('text-3', ['text-2', 'text-1'], ['a\n', 'b\n', 'c\n'])
 
213
        origins = k.annotate('text-3')
 
214
        self.assertEquals(origins[0], ('text-1', 'a\n'))
 
215
        self.assertEquals(origins[1], ('text-1', 'b\n'))
 
216
        self.assertEquals(origins[2], ('text-1', 'c\n'))
 
217
 
 
218
    def test_knit_join(self):
 
219
        """Store in knit with parents"""
 
220
        k1 = KnitVersionedFile('test1', LocalTransport('.'), factory=KnitPlainFactory(), create=True)
 
221
        k1.add_lines('text-a', [], split_lines(TEXT_1))
 
222
        k1.add_lines('text-b', ['text-a'], split_lines(TEXT_1))
 
223
 
 
224
        k1.add_lines('text-c', [], split_lines(TEXT_1))
 
225
        k1.add_lines('text-d', ['text-c'], split_lines(TEXT_1))
 
226
 
 
227
        k1.add_lines('text-m', ['text-b', 'text-d'], split_lines(TEXT_1))
 
228
 
 
229
        k2 = KnitVersionedFile('test2', LocalTransport('.'), factory=KnitPlainFactory(), create=True)
 
230
        count = k2.join(k1, version_ids=['text-m'])
 
231
        self.assertEquals(count, 5)
 
232
        self.assertTrue(k2.has_version('text-a'))
 
233
        self.assertTrue(k2.has_version('text-c'))
 
234
 
 
235
    def test_reannotate(self):
 
236
        k1 = KnitVersionedFile('knit1', LocalTransport('.'),
 
237
                               factory=KnitAnnotateFactory(), create=True)
 
238
        # 0
 
239
        k1.add_lines('text-a', [], ['a\n', 'b\n'])
 
240
        # 1
 
241
        k1.add_lines('text-b', ['text-a'], ['a\n', 'c\n'])
 
242
 
 
243
        k2 = KnitVersionedFile('test2', LocalTransport('.'),
 
244
                               factory=KnitAnnotateFactory(), create=True)
 
245
        k2.join(k1, version_ids=['text-b'])
 
246
 
 
247
        # 2
 
248
        k1.add_lines('text-X', ['text-b'], ['a\n', 'b\n'])
 
249
        # 2
 
250
        k2.add_lines('text-c', ['text-b'], ['z\n', 'c\n'])
 
251
        # 3
 
252
        k2.add_lines('text-Y', ['text-b'], ['b\n', 'c\n'])
 
253
 
 
254
        # test-c will have index 3
 
255
        k1.join(k2, version_ids=['text-c'])
 
256
 
 
257
        lines = k1.get_lines('text-c')
 
258
        self.assertEquals(lines, ['z\n', 'c\n'])
 
259
 
 
260
        origins = k1.annotate('text-c')
 
261
        self.assertEquals(origins[0], ('text-c', 'z\n'))
 
262
        self.assertEquals(origins[1], ('text-b', 'c\n'))
 
263
 
 
264
    def test_extraction_reads_components_once(self):
 
265
        t = MemoryTransport()
 
266
        instrumented_t = TransportLogger(t)
 
267
        k1 = KnitVersionedFile('id', instrumented_t, create=True, delta=True)
 
268
        # should read the index
 
269
        self.assertEqual([('id.kndx',)], instrumented_t._calls)
 
270
        instrumented_t._calls = []
 
271
        # add a text       
 
272
        k1.add_lines('base', [], ['text\n'])
 
273
        # should not have read at all
 
274
        self.assertEqual([], instrumented_t._calls)
 
275
 
 
276
        # add a text
 
277
        k1.add_lines('sub', ['base'], ['text\n', 'text2\n'])
 
278
        # should not have read at all
 
279
        self.assertEqual([], instrumented_t._calls)
 
280
        
 
281
        # read a text
 
282
        k1.get_lines('sub')
 
283
        # should not have read at all
 
284
        self.assertEqual([], instrumented_t._calls)
 
285
 
 
286
        # clear the cache
 
287
        k1.clear_cache()
 
288
 
 
289
        # read a text
 
290
        k1.get_lines('base')
 
291
        # should have read a component
 
292
        # should not have read the first component only
 
293
        self.assertEqual([('id.knit', [(0, 87)])], instrumented_t._calls)
 
294
        instrumented_t._calls = []
 
295
        # read again
 
296
        k1.get_lines('base')
 
297
        # should not have read at all
 
298
        self.assertEqual([], instrumented_t._calls)
 
299
        # and now read the other component
 
300
        k1.get_lines('sub')
 
301
        # should have read the second component
 
302
        self.assertEqual([('id.knit', [(87, 92)])], instrumented_t._calls)
 
303
        instrumented_t._calls = []
 
304
 
 
305
        # clear the cache
 
306
        k1.clear_cache()
 
307
        # add a text cold 
 
308
        k1.add_lines('sub2', ['base'], ['text\n', 'text3\n'])
 
309
        # should read the first component only
 
310
        self.assertEqual([('id.knit', [(0, 87)])], instrumented_t._calls)
 
311
        
 
312
    def test_iter_lines_reads_in_order(self):
 
313
        t = MemoryTransport()
 
314
        instrumented_t = TransportLogger(t)
 
315
        k1 = KnitVersionedFile('id', instrumented_t, create=True, delta=True)
 
316
        self.assertEqual([('id.kndx',)], instrumented_t._calls)
 
317
        # add texts with no required ordering
 
318
        k1.add_lines('base', [], ['text\n'])
 
319
        k1.add_lines('base2', [], ['text2\n'])
 
320
        k1.clear_cache()
 
321
        instrumented_t._calls = []
 
322
        # request a last-first iteration
 
323
        results = list(k1.iter_lines_added_or_present_in_versions(['base2', 'base']))
 
324
        self.assertEqual([('id.knit', [(0, 87), (87, 89)])], instrumented_t._calls)
 
325
        self.assertEqual(['text\n', 'text2\n'], results)
 
326
 
 
327
    def test_create_empty_annotated(self):
 
328
        k1 = self.make_test_knit(True)
 
329
        # 0
 
330
        k1.add_lines('text-a', [], ['a\n', 'b\n'])
 
331
        k2 = k1.create_empty('t', MemoryTransport())
 
332
        self.assertTrue(isinstance(k2.factory, KnitAnnotateFactory))
 
333
        self.assertEqual(k1.delta, k2.delta)
 
334
        # the generic test checks for empty content and file class
 
335
 
 
336
 
 
337
TEXT_1 = """\
 
338
Banana cup cakes:
 
339
 
 
340
- bananas
 
341
- eggs
 
342
- broken tea cups
 
343
"""
 
344
 
 
345
TEXT_1A = """\
 
346
Banana cup cake recipe
 
347
(serves 6)
 
348
 
 
349
- bananas
 
350
- eggs
 
351
- broken tea cups
 
352
- self-raising flour
 
353
"""
 
354
 
 
355
delta_1_1a = """\
 
356
0,1,2
 
357
Banana cup cake recipe
 
358
(serves 6)
 
359
5,5,1
 
360
- self-raising flour
 
361
"""
 
362
 
 
363
TEXT_2 = """\
 
364
Boeuf bourguignon
 
365
 
 
366
- beef
 
367
- red wine
 
368
- small onions
 
369
- carrot
 
370
- mushrooms
 
371
"""
 
372
 
 
373
def line_delta(from_lines, to_lines):
 
374
    """Generate line-based delta from one text to another"""
 
375
    s = difflib.SequenceMatcher(None, from_lines, to_lines)
 
376
    for op in s.get_opcodes():
 
377
        if op[0] == 'equal':
 
378
            continue
 
379
        yield '%d,%d,%d\n' % (op[1], op[2], op[4]-op[3])
 
380
        for i in range(op[3], op[4]):
 
381
            yield to_lines[i]
 
382
 
 
383
 
 
384
def apply_line_delta(basis_lines, delta_lines):
 
385
    """Apply a line-based perfect diff
 
386
    
 
387
    basis_lines -- text to apply the patch to
 
388
    delta_lines -- diff instructions and content
 
389
    """
 
390
    out = basis_lines[:]
 
391
    i = 0
 
392
    offset = 0
 
393
    while i < len(delta_lines):
 
394
        l = delta_lines[i]
 
395
        a, b, c = map(long, l.split(','))
 
396
        i = i + 1
 
397
        out[offset+a:offset+b] = delta_lines[i:i+c]
 
398
        i = i + c
 
399
        offset = offset + (b - a) + c
 
400
    return out