/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

Raise SettingFileIdUnsupported

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])))