/brz/remove-bazaar

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

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_weave.py

Cleanup

Show diffs side-by-side

added added

removed removed

Lines of Context:
 
1
#! /usr/bin/python2.4
 
2
 
 
3
# Copyright (C) 2005 by Canonical Ltd
 
4
 
 
5
# This program is free software; you can redistribute it and/or modify
 
6
# it under the terms of the GNU General Public License as published by
 
7
# the Free Software Foundation; either version 2 of the License, or
 
8
# (at your option) any later version.
 
9
 
 
10
# This program is distributed in the hope that it will be useful,
 
11
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
12
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
13
# GNU General Public License for more details.
 
14
 
 
15
# You should have received a copy of the GNU General Public License
 
16
# along with this program; if not, write to the Free Software
 
17
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
18
 
 
19
 
 
20
# TODO: tests regarding version names
 
21
# TODO: rbc 20050108 test that join does not leave an inconsistent weave 
 
22
#       if it fails.
 
23
 
 
24
"""test suite for weave algorithm"""
 
25
 
 
26
from pprint import pformat
 
27
 
 
28
import bzrlib.errors as errors
 
29
from bzrlib.weave import Weave, WeaveFormatError, WeaveError, reweave
 
30
from bzrlib.weavefile import write_weave, read_weave
 
31
from bzrlib.tests import TestCase
 
32
from bzrlib.osutils import sha_string
 
33
 
 
34
 
 
35
# texts for use in testing
 
36
TEXT_0 = ["Hello world"]
 
37
TEXT_1 = ["Hello world",
 
38
          "A second line"]
 
39
 
 
40
 
 
41
class TestBase(TestCase):
 
42
    def check_read_write(self, k):
 
43
        """Check the weave k can be written & re-read."""
 
44
        from tempfile import TemporaryFile
 
45
        tf = TemporaryFile()
 
46
 
 
47
        write_weave(k, tf)
 
48
        tf.seek(0)
 
49
        k2 = read_weave(tf)
 
50
 
 
51
        if k != k2:
 
52
            tf.seek(0)
 
53
            self.log('serialized weave:')
 
54
            self.log(tf.read())
 
55
 
 
56
            self.log('')
 
57
            self.log('parents: %s' % (k._parents == k2._parents))
 
58
            self.log('         %r' % k._parents)
 
59
            self.log('         %r' % k2._parents)
 
60
            self.log('')
 
61
            self.fail('read/write check failed')
 
62
 
 
63
 
 
64
class WeaveContains(TestBase):
 
65
    """Weave __contains__ operator"""
 
66
    def runTest(self):
 
67
        k = Weave()
 
68
        self.assertFalse('foo' in k)
 
69
        k.add_lines('foo', [], TEXT_1)
 
70
        self.assertTrue('foo' in k)
 
71
 
 
72
 
 
73
class Easy(TestBase):
 
74
    def runTest(self):
 
75
        k = Weave()
 
76
 
 
77
 
 
78
class StoreText(TestBase):
 
79
    """Store and retrieve a simple text."""
 
80
 
 
81
    def test_storing_text(self):
 
82
        k = Weave()
 
83
        idx = k.add_lines('text0', [], TEXT_0)
 
84
        self.assertEqual(k.get_lines(idx), TEXT_0)
 
85
        self.assertEqual(idx, 0)
 
86
 
 
87
 
 
88
class AnnotateOne(TestBase):
 
89
    def runTest(self):
 
90
        k = Weave()
 
91
        k.add_lines('text0', [], TEXT_0)
 
92
        self.assertEqual(k.annotate('text0'),
 
93
                         [('text0', TEXT_0[0])])
 
94
 
 
95
 
 
96
class StoreTwo(TestBase):
 
97
    def runTest(self):
 
98
        k = Weave()
 
99
 
 
100
        idx = k.add_lines('text0', [], TEXT_0)
 
101
        self.assertEqual(idx, 0)
 
102
 
 
103
        idx = k.add_lines('text1', [], TEXT_1)
 
104
        self.assertEqual(idx, 1)
 
105
 
 
106
        self.assertEqual(k.get_lines(0), TEXT_0)
 
107
        self.assertEqual(k.get_lines(1), TEXT_1)
 
108
 
 
109
 
 
110
class GetSha1(TestBase):
 
111
    def test_get_sha1(self):
 
112
        k = Weave()
 
113
        k.add_lines('text0', [], 'text0')
 
114
        self.assertEqual('34dc0e430c642a26c3dd1c2beb7a8b4f4445eb79',
 
115
                         k.get_sha1('text0'))
 
116
        self.assertRaises(errors.RevisionNotPresent,
 
117
                          k.get_sha1, 0)
 
118
        self.assertRaises(errors.RevisionNotPresent,
 
119
                          k.get_sha1, 'text1')
 
120
                        
 
121
 
 
122
class InvalidAdd(TestBase):
 
123
    """Try to use invalid version number during add."""
 
124
    def runTest(self):
 
125
        k = Weave()
 
126
 
 
127
        self.assertRaises(errors.RevisionNotPresent,
 
128
                          k.add_lines,
 
129
                          'text0',
 
130
                          ['69'],
 
131
                          ['new text!'])
 
132
 
 
133
 
 
134
class RepeatedAdd(TestBase):
 
135
    """Add the same version twice; harmless."""
 
136
    def runTest(self):
 
137
        k = Weave()
 
138
        idx = k.add_lines('text0', [], TEXT_0)
 
139
        idx2 = k.add_lines('text0', [], TEXT_0)
 
140
        self.assertEqual(idx, idx2)
 
141
 
 
142
 
 
143
class InvalidRepeatedAdd(TestBase):
 
144
    def runTest(self):
 
145
        k = Weave()
 
146
        k.add_lines('basis', [], TEXT_0)
 
147
        idx = k.add_lines('text0', [], TEXT_0)
 
148
        self.assertRaises(errors.RevisionAlreadyPresent,
 
149
                          k.add_lines,
 
150
                          'text0',
 
151
                          [],
 
152
                          ['not the same text'])
 
153
        self.assertRaises(errors.RevisionAlreadyPresent,
 
154
                          k.add_lines,
 
155
                          'text0',
 
156
                          ['basis'],         # not the right parents
 
157
                          TEXT_0)
 
158
        
 
159
 
 
160
class InsertLines(TestBase):
 
161
    """Store a revision that adds one line to the original.
 
162
 
 
163
    Look at the annotations to make sure that the first line is matched
 
164
    and not stored repeatedly."""
 
165
    def runTest(self):
 
166
        k = Weave()
 
167
 
 
168
        k.add_lines('text0', [], ['line 1'])
 
169
        k.add_lines('text1', ['text0'], ['line 1', 'line 2'])
 
170
 
 
171
        self.assertEqual(k.annotate('text0'),
 
172
                         [('text0', 'line 1')])
 
173
 
 
174
        self.assertEqual(k.get_lines(1),
 
175
                         ['line 1',
 
176
                          'line 2'])
 
177
 
 
178
        self.assertEqual(k.annotate('text1'),
 
179
                         [('text0', 'line 1'),
 
180
                          ('text1', 'line 2')])
 
181
 
 
182
        k.add_lines('text2', ['text0'], ['line 1', 'diverged line'])
 
183
 
 
184
        self.assertEqual(k.annotate('text2'),
 
185
                         [('text0', 'line 1'),
 
186
                          ('text2', 'diverged line')])
 
187
 
 
188
        text3 = ['line 1', 'middle line', 'line 2']
 
189
        k.add_lines('text3',
 
190
              ['text0', 'text1'],
 
191
              text3)
 
192
 
 
193
        # self.log("changes to text3: " + pformat(list(k._delta(set([0, 1]), text3))))
 
194
 
 
195
        self.log("k._weave=" + pformat(k._weave))
 
196
 
 
197
        self.assertEqual(k.annotate('text3'),
 
198
                         [('text0', 'line 1'),
 
199
                          ('text3', 'middle line'),
 
200
                          ('text1', 'line 2')])
 
201
 
 
202
        # now multiple insertions at different places
 
203
        k.add_lines('text4',
 
204
              ['text0', 'text1', 'text3'],
 
205
              ['line 1', 'aaa', 'middle line', 'bbb', 'line 2', 'ccc'])
 
206
 
 
207
        self.assertEqual(k.annotate('text4'), 
 
208
                         [('text0', 'line 1'),
 
209
                          ('text4', 'aaa'),
 
210
                          ('text3', 'middle line'),
 
211
                          ('text4', 'bbb'),
 
212
                          ('text1', 'line 2'),
 
213
                          ('text4', 'ccc')])
 
214
 
 
215
 
 
216
class DeleteLines(TestBase):
 
217
    """Deletion of lines from existing text.
 
218
 
 
219
    Try various texts all based on a common ancestor."""
 
220
    def runTest(self):
 
221
        k = Weave()
 
222
 
 
223
        base_text = ['one', 'two', 'three', 'four']
 
224
 
 
225
        k.add_lines('text0', [], base_text)
 
226
        
 
227
        texts = [['one', 'two', 'three'],
 
228
                 ['two', 'three', 'four'],
 
229
                 ['one', 'four'],
 
230
                 ['one', 'two', 'three', 'four'],
 
231
                 ]
 
232
 
 
233
        i = 1
 
234
        for t in texts:
 
235
            ver = k.add_lines('text%d' % i,
 
236
                        ['text0'], t)
 
237
            i += 1
 
238
 
 
239
        self.log('final weave:')
 
240
        self.log('k._weave=' + pformat(k._weave))
 
241
 
 
242
        for i in range(len(texts)):
 
243
            self.assertEqual(k.get_lines(i+1),
 
244
                             texts[i])
 
245
 
 
246
 
 
247
class SuicideDelete(TestBase):
 
248
    """Invalid weave which tries to add and delete simultaneously."""
 
249
    def runTest(self):
 
250
        k = Weave()
 
251
 
 
252
        k._parents = [(),
 
253
                ]
 
254
        k._weave = [('{', 0),
 
255
                'first line',
 
256
                ('[', 0),
 
257
                'deleted in 0',
 
258
                (']', 0),
 
259
                ('}', 0),
 
260
                ]
 
261
        ################################### SKIPPED
 
262
        # Weave.get doesn't trap this anymore
 
263
        return 
 
264
 
 
265
        self.assertRaises(WeaveFormatError,
 
266
                          k.get_lines,
 
267
                          0)        
 
268
 
 
269
 
 
270
class CannedDelete(TestBase):
 
271
    """Unpack canned weave with deleted lines."""
 
272
    def runTest(self):
 
273
        k = Weave()
 
274
 
 
275
        k._parents = [(),
 
276
                frozenset([0]),
 
277
                ]
 
278
        k._weave = [('{', 0),
 
279
                'first line',
 
280
                ('[', 1),
 
281
                'line to be deleted',
 
282
                (']', 1),
 
283
                'last line',
 
284
                ('}', 0),
 
285
                ]
 
286
        k._sha1s = [sha_string('first lineline to be deletedlast line')
 
287
                  , sha_string('first linelast line')]
 
288
 
 
289
        self.assertEqual(k.get_lines(0),
 
290
                         ['first line',
 
291
                          'line to be deleted',
 
292
                          'last line',
 
293
                          ])
 
294
 
 
295
        self.assertEqual(k.get_lines(1),
 
296
                         ['first line',
 
297
                          'last line',
 
298
                          ])
 
299
 
 
300
 
 
301
class CannedReplacement(TestBase):
 
302
    """Unpack canned weave with deleted lines."""
 
303
    def runTest(self):
 
304
        k = Weave()
 
305
 
 
306
        k._parents = [frozenset(),
 
307
                frozenset([0]),
 
308
                ]
 
309
        k._weave = [('{', 0),
 
310
                'first line',
 
311
                ('[', 1),
 
312
                'line to be deleted',
 
313
                (']', 1),
 
314
                ('{', 1),
 
315
                'replacement line',                
 
316
                ('}', 1),
 
317
                'last line',
 
318
                ('}', 0),
 
319
                ]
 
320
        k._sha1s = [sha_string('first lineline to be deletedlast line')
 
321
                  , sha_string('first linereplacement linelast line')]
 
322
 
 
323
        self.assertEqual(k.get_lines(0),
 
324
                         ['first line',
 
325
                          'line to be deleted',
 
326
                          'last line',
 
327
                          ])
 
328
 
 
329
        self.assertEqual(k.get_lines(1),
 
330
                         ['first line',
 
331
                          'replacement line',
 
332
                          'last line',
 
333
                          ])
 
334
 
 
335
 
 
336
class BadWeave(TestBase):
 
337
    """Test that we trap an insert which should not occur."""
 
338
    def runTest(self):
 
339
        k = Weave()
 
340
 
 
341
        k._parents = [frozenset(),
 
342
                ]
 
343
        k._weave = ['bad line',
 
344
                ('{', 0),
 
345
                'foo {',
 
346
                ('{', 1),
 
347
                '  added in version 1',
 
348
                ('{', 2),
 
349
                '  added in v2',
 
350
                ('}', 2),
 
351
                '  also from v1',
 
352
                ('}', 1),
 
353
                '}',
 
354
                ('}', 0)]
 
355
 
 
356
        ################################### SKIPPED
 
357
        # Weave.get doesn't trap this anymore
 
358
        return 
 
359
 
 
360
 
 
361
        self.assertRaises(WeaveFormatError,
 
362
                          k.get,
 
363
                          0)
 
364
 
 
365
 
 
366
class BadInsert(TestBase):
 
367
    """Test that we trap an insert which should not occur."""
 
368
    def runTest(self):
 
369
        k = Weave()
 
370
 
 
371
        k._parents = [frozenset(),
 
372
                frozenset([0]),
 
373
                frozenset([0]),
 
374
                frozenset([0,1,2]),
 
375
                ]
 
376
        k._weave = [('{', 0),
 
377
                'foo {',
 
378
                ('{', 1),
 
379
                '  added in version 1',
 
380
                ('{', 1),
 
381
                '  more in 1',
 
382
                ('}', 1),
 
383
                ('}', 1),
 
384
                ('}', 0)]
 
385
 
 
386
 
 
387
        # this is not currently enforced by get
 
388
        return  ##########################################
 
389
 
 
390
        self.assertRaises(WeaveFormatError,
 
391
                          k.get,
 
392
                          0)
 
393
 
 
394
        self.assertRaises(WeaveFormatError,
 
395
                          k.get,
 
396
                          1)
 
397
 
 
398
 
 
399
class InsertNested(TestBase):
 
400
    """Insertion with nested instructions."""
 
401
    def runTest(self):
 
402
        k = Weave()
 
403
 
 
404
        k._parents = [frozenset(),
 
405
                frozenset([0]),
 
406
                frozenset([0]),
 
407
                frozenset([0,1,2]),
 
408
                ]
 
409
        k._weave = [('{', 0),
 
410
                'foo {',
 
411
                ('{', 1),
 
412
                '  added in version 1',
 
413
                ('{', 2),
 
414
                '  added in v2',
 
415
                ('}', 2),
 
416
                '  also from v1',
 
417
                ('}', 1),
 
418
                '}',
 
419
                ('}', 0)]
 
420
 
 
421
        k._sha1s = [sha_string('foo {}')
 
422
                  , sha_string('foo {  added in version 1  also from v1}')
 
423
                  , sha_string('foo {  added in v2}')
 
424
                  , sha_string('foo {  added in version 1  added in v2  also from v1}')
 
425
                  ]
 
426
 
 
427
        self.assertEqual(k.get_lines(0),
 
428
                         ['foo {',
 
429
                          '}'])
 
430
 
 
431
        self.assertEqual(k.get_lines(1),
 
432
                         ['foo {',
 
433
                          '  added in version 1',
 
434
                          '  also from v1',
 
435
                          '}'])
 
436
                       
 
437
        self.assertEqual(k.get_lines(2),
 
438
                         ['foo {',
 
439
                          '  added in v2',
 
440
                          '}'])
 
441
 
 
442
        self.assertEqual(k.get_lines(3),
 
443
                         ['foo {',
 
444
                          '  added in version 1',
 
445
                          '  added in v2',
 
446
                          '  also from v1',
 
447
                          '}'])
 
448
                         
 
449
 
 
450
class DeleteLines2(TestBase):
 
451
    """Test recording revisions that delete lines.
 
452
 
 
453
    This relies on the weave having a way to represent lines knocked
 
454
    out by a later revision."""
 
455
    def runTest(self):
 
456
        k = Weave()
 
457
 
 
458
        k.add_lines('text0', [], ["line the first",
 
459
                   "line 2",
 
460
                   "line 3",
 
461
                   "fine"])
 
462
 
 
463
        self.assertEqual(len(k.get_lines(0)), 4)
 
464
 
 
465
        k.add_lines('text1', ['text0'], ["line the first",
 
466
                   "fine"])
 
467
 
 
468
        self.assertEqual(k.get_lines(1),
 
469
                         ["line the first",
 
470
                          "fine"])
 
471
 
 
472
        self.assertEqual(k.annotate('text1'),
 
473
                         [('text0', "line the first"),
 
474
                          ('text0', "fine")])
 
475
 
 
476
 
 
477
class IncludeVersions(TestBase):
 
478
    """Check texts that are stored across multiple revisions.
 
479
 
 
480
    Here we manually create a weave with particular encoding and make
 
481
    sure it unpacks properly.
 
482
 
 
483
    Text 0 includes nothing; text 1 includes text 0 and adds some
 
484
    lines.
 
485
    """
 
486
 
 
487
    def runTest(self):
 
488
        k = Weave()
 
489
 
 
490
        k._parents = [frozenset(), frozenset([0])]
 
491
        k._weave = [('{', 0),
 
492
                "first line",
 
493
                ('}', 0),
 
494
                ('{', 1),
 
495
                "second line",
 
496
                ('}', 1)]
 
497
 
 
498
        k._sha1s = [sha_string('first line')
 
499
                  , sha_string('first linesecond line')]
 
500
 
 
501
        self.assertEqual(k.get_lines(1),
 
502
                         ["first line",
 
503
                          "second line"])
 
504
 
 
505
        self.assertEqual(k.get_lines(0),
 
506
                         ["first line"])
 
507
 
 
508
 
 
509
class DivergedIncludes(TestBase):
 
510
    """Weave with two diverged texts based on version 0.
 
511
    """
 
512
    def runTest(self):
 
513
        # FIXME make the weave, dont poke at it.
 
514
        k = Weave()
 
515
 
 
516
        k._names = ['0', '1', '2']
 
517
        k._name_map = {'0':0, '1':1, '2':2}
 
518
        k._parents = [frozenset(),
 
519
                frozenset([0]),
 
520
                frozenset([0]),
 
521
                ]
 
522
        k._weave = [('{', 0),
 
523
                "first line",
 
524
                ('}', 0),
 
525
                ('{', 1),
 
526
                "second line",
 
527
                ('}', 1),
 
528
                ('{', 2),
 
529
                "alternative second line",
 
530
                ('}', 2),                
 
531
                ]
 
532
 
 
533
        k._sha1s = [sha_string('first line')
 
534
                  , sha_string('first linesecond line')
 
535
                  , sha_string('first linealternative second line')]
 
536
 
 
537
        self.assertEqual(k.get_lines(0),
 
538
                         ["first line"])
 
539
 
 
540
        self.assertEqual(k.get_lines(1),
 
541
                         ["first line",
 
542
                          "second line"])
 
543
 
 
544
        self.assertEqual(k.get_lines('2'),
 
545
                         ["first line",
 
546
                          "alternative second line"])
 
547
 
 
548
        self.assertEqual(list(k.get_ancestry(['2'])),
 
549
                         ['0', '2'])
 
550
 
 
551
 
 
552
class ReplaceLine(TestBase):
 
553
    def runTest(self):
 
554
        k = Weave()
 
555
 
 
556
        text0 = ['cheddar', 'stilton', 'gruyere']
 
557
        text1 = ['cheddar', 'blue vein', 'neufchatel', 'chevre']
 
558
        
 
559
        k.add_lines('text0', [], text0)
 
560
        k.add_lines('text1', ['text0'], text1)
 
561
 
 
562
        self.log('k._weave=' + pformat(k._weave))
 
563
 
 
564
        self.assertEqual(k.get_lines(0), text0)
 
565
        self.assertEqual(k.get_lines(1), text1)
 
566
 
 
567
 
 
568
class Merge(TestBase):
 
569
    """Storage of versions that merge diverged parents"""
 
570
    def runTest(self):
 
571
        k = Weave()
 
572
 
 
573
        texts = [['header'],
 
574
                 ['header', '', 'line from 1'],
 
575
                 ['header', '', 'line from 2', 'more from 2'],
 
576
                 ['header', '', 'line from 1', 'fixup line', 'line from 2'],
 
577
                 ]
 
578
 
 
579
        k.add_lines('text0', [], texts[0])
 
580
        k.add_lines('text1', ['text0'], texts[1])
 
581
        k.add_lines('text2', ['text0'], texts[2])
 
582
        k.add_lines('merge', ['text0', 'text1', 'text2'], texts[3])
 
583
 
 
584
        for i, t in enumerate(texts):
 
585
            self.assertEqual(k.get_lines(i), t)
 
586
 
 
587
        self.assertEqual(k.annotate('merge'),
 
588
                         [('text0', 'header'),
 
589
                          ('text1', ''),
 
590
                          ('text1', 'line from 1'),
 
591
                          ('merge', 'fixup line'),
 
592
                          ('text2', 'line from 2'),
 
593
                          ])
 
594
 
 
595
        self.assertEqual(list(k.get_ancestry(['merge'])),
 
596
                         ['text0', 'text1', 'text2', 'merge'])
 
597
 
 
598
        self.log('k._weave=' + pformat(k._weave))
 
599
 
 
600
        self.check_read_write(k)
 
601
 
 
602
 
 
603
class Conflicts(TestBase):
 
604
    """Test detection of conflicting regions during a merge.
 
605
 
 
606
    A base version is inserted, then two descendents try to
 
607
    insert different lines in the same place.  These should be
 
608
    reported as a possible conflict and forwarded to the user."""
 
609
    def runTest(self):
 
610
        return  # NOT RUN
 
611
        k = Weave()
 
612
 
 
613
        k.add_lines([], ['aaa', 'bbb'])
 
614
        k.add_lines([0], ['aaa', '111', 'bbb'])
 
615
        k.add_lines([1], ['aaa', '222', 'bbb'])
 
616
 
 
617
        merged = k.merge([1, 2])
 
618
 
 
619
        self.assertEquals([[['aaa']],
 
620
                           [['111'], ['222']],
 
621
                           [['bbb']]])
 
622
 
 
623
 
 
624
class NonConflict(TestBase):
 
625
    """Two descendants insert compatible changes.
 
626
 
 
627
    No conflict should be reported."""
 
628
    def runTest(self):
 
629
        return  # NOT RUN
 
630
        k = Weave()
 
631
 
 
632
        k.add_lines([], ['aaa', 'bbb'])
 
633
        k.add_lines([0], ['111', 'aaa', 'ccc', 'bbb'])
 
634
        k.add_lines([1], ['aaa', 'ccc', 'bbb', '222'])
 
635
 
 
636
 
 
637
class Khayyam(TestBase):
 
638
    """Test changes to multi-line texts, and read/write"""
 
639
 
 
640
    def test_multi_line_merge(self):
 
641
        rawtexts = [
 
642
            """A Book of Verses underneath the Bough,
 
643
            A Jug of Wine, a Loaf of Bread, -- and Thou
 
644
            Beside me singing in the Wilderness --
 
645
            Oh, Wilderness were Paradise enow!""",
 
646
            
 
647
            """A Book of Verses underneath the Bough,
 
648
            A Jug of Wine, a Loaf of Bread, -- and Thou
 
649
            Beside me singing in the Wilderness --
 
650
            Oh, Wilderness were Paradise now!""",
 
651
 
 
652
            """A Book of poems underneath the tree,
 
653
            A Jug of Wine, a Loaf of Bread,
 
654
            and Thou
 
655
            Beside me singing in the Wilderness --
 
656
            Oh, Wilderness were Paradise now!
 
657
 
 
658
            -- O. Khayyam""",
 
659
 
 
660
            """A Book of Verses underneath the Bough,
 
661
            A Jug of Wine, a Loaf of Bread,
 
662
            and Thou
 
663
            Beside me singing in the Wilderness --
 
664
            Oh, Wilderness were Paradise now!""",
 
665
            ]
 
666
        texts = [[l.strip() for l in t.split('\n')] for t in rawtexts]
 
667
 
 
668
        k = Weave()
 
669
        parents = set()
 
670
        i = 0
 
671
        for t in texts:
 
672
            ver = k.add_lines('text%d' % i,
 
673
                        list(parents), t)
 
674
            parents.add('text%d' % i)
 
675
            i += 1
 
676
 
 
677
        self.log("k._weave=" + pformat(k._weave))
 
678
 
 
679
        for i, t in enumerate(texts):
 
680
            self.assertEqual(k.get_lines(i), t)
 
681
 
 
682
        self.check_read_write(k)
 
683
 
 
684
 
 
685
class MergeCases(TestBase):
 
686
    def doMerge(self, base, a, b, mp):
 
687
        from cStringIO import StringIO
 
688
        from textwrap import dedent
 
689
 
 
690
        def addcrlf(x):
 
691
            return x + '\n'
 
692
        
 
693
        w = Weave()
 
694
        w.add_lines('text0', [], map(addcrlf, base))
 
695
        w.add_lines('text1', ['text0'], map(addcrlf, a))
 
696
        w.add_lines('text2', ['text0'], map(addcrlf, b))
 
697
 
 
698
        self.log('weave is:')
 
699
        tmpf = StringIO()
 
700
        write_weave(w, tmpf)
 
701
        self.log(tmpf.getvalue())
 
702
 
 
703
        self.log('merge plan:')
 
704
        p = list(w.plan_merge('text1', 'text2'))
 
705
        for state, line in p:
 
706
            if line:
 
707
                self.log('%12s | %s' % (state, line[:-1]))
 
708
 
 
709
        self.log('merge:')
 
710
        mt = StringIO()
 
711
        mt.writelines(w.weave_merge(p))
 
712
        mt.seek(0)
 
713
        self.log(mt.getvalue())
 
714
 
 
715
        mp = map(addcrlf, mp)
 
716
        self.assertEqual(mt.readlines(), mp)
 
717
        
 
718
        
 
719
    def testOneInsert(self):
 
720
        self.doMerge([],
 
721
                     ['aa'],
 
722
                     [],
 
723
                     ['aa'])
 
724
 
 
725
    def testSeparateInserts(self):
 
726
        self.doMerge(['aaa', 'bbb', 'ccc'],
 
727
                     ['aaa', 'xxx', 'bbb', 'ccc'],
 
728
                     ['aaa', 'bbb', 'yyy', 'ccc'],
 
729
                     ['aaa', 'xxx', 'bbb', 'yyy', 'ccc'])
 
730
 
 
731
    def testSameInsert(self):
 
732
        self.doMerge(['aaa', 'bbb', 'ccc'],
 
733
                     ['aaa', 'xxx', 'bbb', 'ccc'],
 
734
                     ['aaa', 'xxx', 'bbb', 'yyy', 'ccc'],
 
735
                     ['aaa', 'xxx', 'bbb', 'yyy', 'ccc'])
 
736
 
 
737
    def testOverlappedInsert(self):
 
738
        self.doMerge(['aaa', 'bbb'],
 
739
                     ['aaa', 'xxx', 'yyy', 'bbb'],
 
740
                     ['aaa', 'xxx', 'bbb'],
 
741
                     ['aaa', '<<<<<<< ', 'xxx', 'yyy', '=======', 'xxx', 
 
742
                      '>>>>>>> ', 'bbb'])
 
743
 
 
744
        # really it ought to reduce this to 
 
745
        # ['aaa', 'xxx', 'yyy', 'bbb']
 
746
 
 
747
 
 
748
    def testClashReplace(self):
 
749
        self.doMerge(['aaa'],
 
750
                     ['xxx'],
 
751
                     ['yyy', 'zzz'],
 
752
                     ['<<<<<<< ', 'xxx', '=======', 'yyy', 'zzz', 
 
753
                      '>>>>>>> '])
 
754
 
 
755
    def testNonClashInsert(self):
 
756
        self.doMerge(['aaa'],
 
757
                     ['xxx', 'aaa'],
 
758
                     ['yyy', 'zzz'],
 
759
                     ['<<<<<<< ', 'xxx', 'aaa', '=======', 'yyy', 'zzz', 
 
760
                      '>>>>>>> '])
 
761
 
 
762
        self.doMerge(['aaa'],
 
763
                     ['aaa'],
 
764
                     ['yyy', 'zzz'],
 
765
                     ['yyy', 'zzz'])
 
766
 
 
767
 
 
768
    def testDeleteAndModify(self):
 
769
        """Clashing delete and modification.
 
770
 
 
771
        If one side modifies a region and the other deletes it then
 
772
        there should be a conflict with one side blank.
 
773
        """
 
774
 
 
775
        #######################################
 
776
        # skippd, not working yet
 
777
        return
 
778
        
 
779
        self.doMerge(['aaa', 'bbb', 'ccc'],
 
780
                     ['aaa', 'ddd', 'ccc'],
 
781
                     ['aaa', 'ccc'],
 
782
                     ['<<<<<<<< ', 'aaa', '=======', '>>>>>>> ', 'ccc'])
 
783
 
 
784
    def _test_merge_from_strings(self, base, a, b, expected):
 
785
        w = Weave()
 
786
        w.add_lines('text0', [], base.splitlines(True))
 
787
        w.add_lines('text1', ['text0'], a.splitlines(True))
 
788
        w.add_lines('text2', ['text0'], b.splitlines(True))
 
789
        self.log('merge plan:')
 
790
        p = list(w.plan_merge('text1', 'text2'))
 
791
        for state, line in p:
 
792
            if line:
 
793
                self.log('%12s | %s' % (state, line[:-1]))
 
794
        self.log('merge result:')
 
795
        result_text = ''.join(w.weave_merge(p))
 
796
        self.log(result_text)
 
797
        self.assertEqualDiff(result_text, expected)
 
798
 
 
799
    def test_deletion_extended(self):
 
800
        """One side deletes, the other deletes more.
 
801
        """
 
802
        base = """\
 
803
            line 1
 
804
            line 2
 
805
            line 3
 
806
            """
 
807
        a = """\
 
808
            line 1
 
809
            line 2
 
810
            """
 
811
        b = """\
 
812
            line 1
 
813
            """
 
814
        result = """\
 
815
            line 1
 
816
            """
 
817
        self._test_merge_from_strings(base, a, b, result)
 
818
 
 
819
    def test_deletion_overlap(self):
 
820
        """Delete overlapping regions with no other conflict.
 
821
 
 
822
        Arguably it'd be better to treat these as agreement, rather than 
 
823
        conflict, but for now conflict is safer.
 
824
        """
 
825
        base = """\
 
826
            start context
 
827
            int a() {}
 
828
            int b() {}
 
829
            int c() {}
 
830
            end context
 
831
            """
 
832
        a = """\
 
833
            start context
 
834
            int a() {}
 
835
            end context
 
836
            """
 
837
        b = """\
 
838
            start context
 
839
            int c() {}
 
840
            end context
 
841
            """
 
842
        result = """\
 
843
            start context
 
844
<<<<<<< 
 
845
            int a() {}
 
846
=======
 
847
            int c() {}
 
848
>>>>>>> 
 
849
            end context
 
850
            """
 
851
        self._test_merge_from_strings(base, a, b, result)
 
852
 
 
853
    def test_agreement_deletion(self):
 
854
        """Agree to delete some lines, without conflicts."""
 
855
        base = """\
 
856
            start context
 
857
            base line 1
 
858
            base line 2
 
859
            end context
 
860
            """
 
861
        a = """\
 
862
            start context
 
863
            base line 1
 
864
            end context
 
865
            """
 
866
        b = """\
 
867
            start context
 
868
            base line 1
 
869
            end context
 
870
            """
 
871
        result = """\
 
872
            start context
 
873
            base line 1
 
874
            end context
 
875
            """
 
876
        self._test_merge_from_strings(base, a, b, result)
 
877
 
 
878
    def test_sync_on_deletion(self):
 
879
        """Specific case of merge where we can synchronize incorrectly.
 
880
        
 
881
        A previous version of the weave merge concluded that the two versions
 
882
        agreed on deleting line 2, and this could be a synchronization point.
 
883
        Line 1 was then considered in isolation, and thought to be deleted on 
 
884
        both sides.
 
885
 
 
886
        It's better to consider the whole thing as a disagreement region.
 
887
        """
 
888
        base = """\
 
889
            start context
 
890
            base line 1
 
891
            base line 2
 
892
            end context
 
893
            """
 
894
        a = """\
 
895
            start context
 
896
            base line 1
 
897
            a's replacement line 2
 
898
            end context
 
899
            """
 
900
        b = """\
 
901
            start context
 
902
            b replaces
 
903
            both lines
 
904
            end context
 
905
            """
 
906
        result = """\
 
907
            start context
 
908
<<<<<<< 
 
909
            base line 1
 
910
            a's replacement line 2
 
911
=======
 
912
            b replaces
 
913
            both lines
 
914
>>>>>>> 
 
915
            end context
 
916
            """
 
917
        self._test_merge_from_strings(base, a, b, result)
 
918
 
 
919
 
 
920
class JoinWeavesTests(TestBase):
 
921
    def setUp(self):
 
922
        super(JoinWeavesTests, self).setUp()
 
923
        self.weave1 = Weave()
 
924
        self.lines1 = ['hello\n']
 
925
        self.lines3 = ['hello\n', 'cruel\n', 'world\n']
 
926
        self.weave1.add_lines('v1', [], self.lines1)
 
927
        self.weave1.add_lines('v2', ['v1'], ['hello\n', 'world\n'])
 
928
        self.weave1.add_lines('v3', ['v2'], self.lines3)
 
929
        
 
930
    def test_join_empty(self):
 
931
        """Join two empty weaves."""
 
932
        eq = self.assertEqual
 
933
        w1 = Weave()
 
934
        w2 = Weave()
 
935
        w1.join(w2)
 
936
        eq(len(w1), 0)
 
937
        
 
938
    def test_join_empty_to_nonempty(self):
 
939
        """Join empty weave onto nonempty."""
 
940
        self.weave1.join(Weave())
 
941
        self.assertEqual(len(self.weave1), 3)
 
942
 
 
943
    def test_join_unrelated(self):
 
944
        """Join two weaves with no history in common."""
 
945
        wb = Weave()
 
946
        wb.add_lines('b1', [], ['line from b\n'])
 
947
        w1 = self.weave1
 
948
        w1.join(wb)
 
949
        eq = self.assertEqual
 
950
        eq(len(w1), 4)
 
951
        eq(sorted(w1.versions()),
 
952
           ['b1', 'v1', 'v2', 'v3'])
 
953
 
 
954
    def test_join_related(self):
 
955
        wa = self.weave1.copy()
 
956
        wb = self.weave1.copy()
 
957
        wa.add_lines('a1', ['v3'], ['hello\n', 'sweet\n', 'world\n'])
 
958
        wb.add_lines('b1', ['v3'], ['hello\n', 'pale blue\n', 'world\n'])
 
959
        eq = self.assertEquals
 
960
        eq(len(wa), 4)
 
961
        eq(len(wb), 4)
 
962
        wa.join(wb)
 
963
        eq(len(wa), 5)
 
964
        eq(wa.get_lines('b1'),
 
965
           ['hello\n', 'pale blue\n', 'world\n'])
 
966
 
 
967
    def test_join_parent_disagreement(self):
 
968
        #join reconciles differening parents into a union.
 
969
        wa = Weave()
 
970
        wb = Weave()
 
971
        wa.add_lines('v1', [], ['hello\n'])
 
972
        wb.add_lines('v0', [], [])
 
973
        wb.add_lines('v1', ['v0'], ['hello\n'])
 
974
        wa.join(wb)
 
975
        self.assertEqual(['v0'], wa.get_parents('v1'))
 
976
 
 
977
    def test_join_text_disagreement(self):
 
978
        """Cannot join weaves with different texts for a version."""
 
979
        wa = Weave()
 
980
        wb = Weave()
 
981
        wa.add_lines('v1', [], ['hello\n'])
 
982
        wb.add_lines('v1', [], ['not\n', 'hello\n'])
 
983
        self.assertRaises(WeaveError,
 
984
                          wa.join, wb)
 
985
 
 
986
    def test_join_unordered(self):
 
987
        """Join weaves where indexes differ.
 
988
        
 
989
        The source weave contains a different version at index 0."""
 
990
        wa = self.weave1.copy()
 
991
        wb = Weave()
 
992
        wb.add_lines('x1', [], ['line from x1\n'])
 
993
        wb.add_lines('v1', [], ['hello\n'])
 
994
        wb.add_lines('v2', ['v1'], ['hello\n', 'world\n'])
 
995
        wa.join(wb)
 
996
        eq = self.assertEquals
 
997
        eq(sorted(wa.versions()), ['v1', 'v2', 'v3', 'x1',])
 
998
        eq(wa.get_text('x1'), 'line from x1\n')
 
999
 
 
1000
    def test_written_detection(self):
 
1001
        # Test detection of weave file corruption.
 
1002
        #
 
1003
        # Make sure that we can detect if a weave file has
 
1004
        # been corrupted. This doesn't test all forms of corruption,
 
1005
        # but it at least helps verify the data you get, is what you want.
 
1006
        from cStringIO import StringIO
 
1007
 
 
1008
        w = Weave()
 
1009
        w.add_lines('v1', [], ['hello\n'])
 
1010
        w.add_lines('v2', ['v1'], ['hello\n', 'there\n'])
 
1011
 
 
1012
        tmpf = StringIO()
 
1013
        write_weave(w, tmpf)
 
1014
 
 
1015
        # Because we are corrupting, we need to make sure we have the exact text
 
1016
        self.assertEquals('# bzr weave file v5\n'
 
1017
                          'i\n1 f572d396fae9206628714fb2ce00f72e94f2258f\nn v1\n\n'
 
1018
                          'i 0\n1 90f265c6e75f1c8f9ab76dcf85528352c5f215ef\nn v2\n\n'
 
1019
                          'w\n{ 0\n. hello\n}\n{ 1\n. there\n}\nW\n',
 
1020
                          tmpf.getvalue())
 
1021
 
 
1022
        # Change a single letter
 
1023
        tmpf = StringIO('# bzr weave file v5\n'
 
1024
                        'i\n1 f572d396fae9206628714fb2ce00f72e94f2258f\nn v1\n\n'
 
1025
                        'i 0\n1 90f265c6e75f1c8f9ab76dcf85528352c5f215ef\nn v2\n\n'
 
1026
                        'w\n{ 0\n. hello\n}\n{ 1\n. There\n}\nW\n')
 
1027
 
 
1028
        w = read_weave(tmpf)
 
1029
 
 
1030
        self.assertEqual('hello\n', w.get_text('v1'))
 
1031
        self.assertRaises(errors.WeaveInvalidChecksum, w.get_text, 'v2')
 
1032
        self.assertRaises(errors.WeaveInvalidChecksum, w.get_lines, 'v2')
 
1033
        self.assertRaises(errors.WeaveInvalidChecksum, w.check)
 
1034
 
 
1035
        # Change the sha checksum
 
1036
        tmpf = StringIO('# bzr weave file v5\n'
 
1037
                        'i\n1 f572d396fae9206628714fb2ce00f72e94f2258f\nn v1\n\n'
 
1038
                        'i 0\n1 f0f265c6e75f1c8f9ab76dcf85528352c5f215ef\nn v2\n\n'
 
1039
                        'w\n{ 0\n. hello\n}\n{ 1\n. there\n}\nW\n')
 
1040
 
 
1041
        w = read_weave(tmpf)
 
1042
 
 
1043
        self.assertEqual('hello\n', w.get_text('v1'))
 
1044
        self.assertRaises(errors.WeaveInvalidChecksum, w.get_text, 'v2')
 
1045
        self.assertRaises(errors.WeaveInvalidChecksum, w.get_lines, 'v2')
 
1046
        self.assertRaises(errors.WeaveInvalidChecksum, w.check)
 
1047
 
 
1048
 
 
1049
class InstrumentedWeave(Weave):
 
1050
    """Keep track of how many times functions are called."""
 
1051
    
 
1052
    def __init__(self, weave_name=None):
 
1053
        self._extract_count = 0
 
1054
        Weave.__init__(self, weave_name=weave_name)
 
1055
 
 
1056
    def _extract(self, versions):
 
1057
        self._extract_count += 1
 
1058
        return Weave._extract(self, versions)
 
1059
 
 
1060
 
 
1061
class JoinOptimization(TestCase):
 
1062
    """Test that Weave.join() doesn't extract all texts, only what must be done."""
 
1063
 
 
1064
    def test_join(self):
 
1065
        w1 = InstrumentedWeave()
 
1066
        w2 = InstrumentedWeave()
 
1067
 
 
1068
        txt0 = ['a\n']
 
1069
        txt1 = ['a\n', 'b\n']
 
1070
        txt2 = ['a\n', 'c\n']
 
1071
        txt3 = ['a\n', 'b\n', 'c\n']
 
1072
 
 
1073
        w1.add_lines('txt0', [], txt0) # extract 1a
 
1074
        w2.add_lines('txt0', [], txt0) # extract 1b
 
1075
        w1.add_lines('txt1', ['txt0'], txt1)# extract 2a
 
1076
        w2.add_lines('txt2', ['txt0'], txt2)# extract 2b
 
1077
        w1.join(w2) # extract 3a to add txt2 
 
1078
        w2.join(w1) # extract 3b to add txt1 
 
1079
 
 
1080
        w1.add_lines('txt3', ['txt1', 'txt2'], txt3) # extract 4a 
 
1081
        w2.add_lines('txt3', ['txt2', 'txt1'], txt3) # extract 4b
 
1082
        # These secretly have inverted parents
 
1083
 
 
1084
        # This should not have to do any extractions
 
1085
        w1.join(w2) # NO extract, texts already present with same parents
 
1086
        w2.join(w1) # NO extract, texts already present with same parents
 
1087
 
 
1088
        self.assertEqual(4, w1._extract_count)
 
1089
        self.assertEqual(4, w2._extract_count)
 
1090
 
 
1091
    def test_double_parent(self):
 
1092
        # It should not be considered illegal to add
 
1093
        # a revision with the same parent twice
 
1094
        w1 = InstrumentedWeave()
 
1095
        w2 = InstrumentedWeave()
 
1096
 
 
1097
        txt0 = ['a\n']
 
1098
        txt1 = ['a\n', 'b\n']
 
1099
        txt2 = ['a\n', 'c\n']
 
1100
        txt3 = ['a\n', 'b\n', 'c\n']
 
1101
 
 
1102
        w1.add_lines('txt0', [], txt0)
 
1103
        w2.add_lines('txt0', [], txt0)
 
1104
        w1.add_lines('txt1', ['txt0'], txt1)
 
1105
        w2.add_lines('txt1', ['txt0', 'txt0'], txt1)
 
1106
        # Same text, effectively the same, because the
 
1107
        # parent is only repeated
 
1108
        w1.join(w2) # extract 3a to add txt2 
 
1109
        w2.join(w1) # extract 3b to add txt1 
 
1110
 
 
1111
 
 
1112
class TestNeedsReweave(TestCase):
 
1113
    """Internal corner cases for when reweave is needed."""
 
1114
 
 
1115
    def test_compatible_parents(self):
 
1116
        w1 = Weave('a')
 
1117
        my_parents = set([1, 2, 3])
 
1118
        # subsets are ok
 
1119
        self.assertTrue(w1._compatible_parents(my_parents, set([3])))
 
1120
        # same sets
 
1121
        self.assertTrue(w1._compatible_parents(my_parents, set(my_parents)))
 
1122
        # same empty corner case
 
1123
        self.assertTrue(w1._compatible_parents(set(), set()))
 
1124
        # other cannot contain stuff my_parents does not
 
1125
        self.assertFalse(w1._compatible_parents(set(), set([1])))
 
1126
        self.assertFalse(w1._compatible_parents(my_parents, set([1, 2, 3, 4])))
 
1127
        self.assertFalse(w1._compatible_parents(my_parents, set([4])))