← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_weave.py

  • Committer: John Arbash Meinel
  • Date: 2008-08-18 22:34:21 UTC
  • mto: (3606.5.6 1.6)
  • mto: This revision was merged to the branch mainline in revision 3641.
  • Revision ID: john@arbash-meinel.com-20080818223421-todjny24vj4faj4t
Add tests for the fetching behavior.

The proper parameter passed is 'unordered' add an assert for it, and
fix callers that were passing 'unsorted' instead.
Add tests that we make the right get_record_stream call based
on the value of _fetch_uses_deltas.
Fix the fetch request for signatures.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_weave.py
1
 
#! /usr/bin/python2.4
2
 
 
3
 
# Copyright (C) 2005 by Canonical Ltd
4
 
 
 
1
# Copyright (C) 2005 Canonical Ltd
 
2
#
5
3
# This program is free software; you can redistribute it and/or modify
6
4
# it under the terms of the GNU General Public License as published by
7
5
# the Free Software Foundation; either version 2 of the License, or
8
6
# (at your option) any later version.
9
 
 
 
7
#
10
8
# This program is distributed in the hope that it will be useful,
11
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
12
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
13
11
# GNU General Public License for more details.
14
 
 
 
12
#
15
13
# You should have received a copy of the GNU General Public License
16
14
# along with this program; if not, write to the Free Software
17
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
25
23
 
26
24
from pprint import pformat
27
25
 
28
 
import bzrlib.errors as errors
29
 
from bzrlib.weave import Weave, WeaveFormatError, WeaveError, reweave
 
26
from bzrlib import (
 
27
    errors,
 
28
    )
 
29
from bzrlib.osutils import sha_string
 
30
from bzrlib.tests import TestCase, TestCaseInTempDir
 
31
from bzrlib.weave import Weave, WeaveFormatError, WeaveError
30
32
from bzrlib.weavefile import write_weave, read_weave
31
 
from bzrlib.tests import TestCase
32
 
from bzrlib.osutils import sha_string
33
33
 
34
34
 
35
35
# texts for use in testing
39
39
 
40
40
 
41
41
class TestBase(TestCase):
 
42
 
42
43
    def check_read_write(self, k):
43
44
        """Check the weave k can be written & re-read."""
44
45
        from tempfile import TemporaryFile
64
65
class WeaveContains(TestBase):
65
66
    """Weave __contains__ operator"""
66
67
    def runTest(self):
67
 
        k = Weave()
 
68
        k = Weave(get_scope=lambda:None)
68
69
        self.assertFalse('foo' in k)
69
70
        k.add_lines('foo', [], TEXT_1)
70
71
        self.assertTrue('foo' in k)
75
76
        k = Weave()
76
77
 
77
78
 
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
79
class AnnotateOne(TestBase):
89
80
    def runTest(self):
90
81
        k = Weave()
93
84
                         [('text0', TEXT_0[0])])
94
85
 
95
86
 
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
87
class InvalidAdd(TestBase):
123
88
    """Try to use invalid version number during add."""
124
89
    def runTest(self):
133
98
 
134
99
class RepeatedAdd(TestBase):
135
100
    """Add the same version twice; harmless."""
136
 
    def runTest(self):
 
101
 
 
102
    def test_duplicate_add(self):
137
103
        k = Weave()
138
104
        idx = k.add_lines('text0', [], TEXT_0)
139
105
        idx2 = k.add_lines('text0', [], TEXT_0)
691
657
        self.weave1.add_lines('v1', [], self.lines1)
692
658
        self.weave1.add_lines('v2', ['v1'], ['hello\n', 'world\n'])
693
659
        self.weave1.add_lines('v3', ['v2'], self.lines3)
694
 
        
695
 
    def test_join_empty(self):
696
 
        """Join two empty weaves."""
697
 
        eq = self.assertEqual
698
 
        w1 = Weave()
699
 
        w2 = Weave()
700
 
        w1.join(w2)
701
 
        eq(len(w1), 0)
702
 
        
703
 
    def test_join_empty_to_nonempty(self):
704
 
        """Join empty weave onto nonempty."""
705
 
        self.weave1.join(Weave())
706
 
        self.assertEqual(len(self.weave1), 3)
707
 
 
708
 
    def test_join_unrelated(self):
709
 
        """Join two weaves with no history in common."""
710
 
        wb = Weave()
711
 
        wb.add_lines('b1', [], ['line from b\n'])
712
 
        w1 = self.weave1
713
 
        w1.join(wb)
714
 
        eq = self.assertEqual
715
 
        eq(len(w1), 4)
716
 
        eq(sorted(w1.versions()),
717
 
           ['b1', 'v1', 'v2', 'v3'])
718
 
 
719
 
    def test_join_related(self):
720
 
        wa = self.weave1.copy()
721
 
        wb = self.weave1.copy()
722
 
        wa.add_lines('a1', ['v3'], ['hello\n', 'sweet\n', 'world\n'])
723
 
        wb.add_lines('b1', ['v3'], ['hello\n', 'pale blue\n', 'world\n'])
724
 
        eq = self.assertEquals
725
 
        eq(len(wa), 4)
726
 
        eq(len(wb), 4)
727
 
        wa.join(wb)
728
 
        eq(len(wa), 5)
729
 
        eq(wa.get_lines('b1'),
730
 
           ['hello\n', 'pale blue\n', 'world\n'])
731
 
 
732
 
    def test_join_parent_disagreement(self):
733
 
        #join reconciles differening parents into a union.
734
 
        wa = Weave()
735
 
        wb = Weave()
736
 
        wa.add_lines('v1', [], ['hello\n'])
737
 
        wb.add_lines('v0', [], [])
738
 
        wb.add_lines('v1', ['v0'], ['hello\n'])
739
 
        wa.join(wb)
740
 
        self.assertEqual(['v0'], wa.get_parents('v1'))
741
 
 
742
 
    def test_join_text_disagreement(self):
743
 
        """Cannot join weaves with different texts for a version."""
744
 
        wa = Weave()
745
 
        wb = Weave()
746
 
        wa.add_lines('v1', [], ['hello\n'])
747
 
        wb.add_lines('v1', [], ['not\n', 'hello\n'])
748
 
        self.assertRaises(WeaveError,
749
 
                          wa.join, wb)
750
 
 
751
 
    def test_join_unordered(self):
752
 
        """Join weaves where indexes differ.
753
 
        
754
 
        The source weave contains a different version at index 0."""
755
 
        wa = self.weave1.copy()
756
 
        wb = Weave()
757
 
        wb.add_lines('x1', [], ['line from x1\n'])
758
 
        wb.add_lines('v1', [], ['hello\n'])
759
 
        wb.add_lines('v2', ['v1'], ['hello\n', 'world\n'])
760
 
        wa.join(wb)
761
 
        eq = self.assertEquals
762
 
        eq(sorted(wa.versions()), ['v1', 'v2', 'v3', 'x1',])
763
 
        eq(wa.get_text('x1'), 'line from x1\n')
764
660
 
765
661
    def test_written_detection(self):
766
662
        # Test detection of weave file corruption.
811
707
        self.assertRaises(errors.WeaveInvalidChecksum, w.check)
812
708
 
813
709
 
 
710
class TestWeave(TestCase):
 
711
 
 
712
    def test_allow_reserved_false(self):
 
713
        w = Weave('name', allow_reserved=False)
 
714
        # Add lines is checked at the WeaveFile level, not at the Weave level
 
715
        w.add_lines('name:', [], TEXT_1)
 
716
        # But get_lines is checked at this level
 
717
        self.assertRaises(errors.ReservedId, w.get_lines, 'name:')
 
718
 
 
719
    def test_allow_reserved_true(self):
 
720
        w = Weave('name', allow_reserved=True)
 
721
        w.add_lines('name:', [], TEXT_1)
 
722
        self.assertEqual(TEXT_1, w.get_lines('name:'))
 
723
 
 
724
 
814
725
class InstrumentedWeave(Weave):
815
726
    """Keep track of how many times functions are called."""
816
727
    
823
734
        return Weave._extract(self, versions)
824
735
 
825
736
 
826
 
class JoinOptimization(TestCase):
827
 
    """Test that Weave.join() doesn't extract all texts, only what must be done."""
828
 
 
829
 
    def test_join(self):
830
 
        w1 = InstrumentedWeave()
831
 
        w2 = InstrumentedWeave()
832
 
 
833
 
        txt0 = ['a\n']
834
 
        txt1 = ['a\n', 'b\n']
835
 
        txt2 = ['a\n', 'c\n']
836
 
        txt3 = ['a\n', 'b\n', 'c\n']
837
 
 
838
 
        w1.add_lines('txt0', [], txt0) # extract 1a
839
 
        w2.add_lines('txt0', [], txt0) # extract 1b
840
 
        w1.add_lines('txt1', ['txt0'], txt1)# extract 2a
841
 
        w2.add_lines('txt2', ['txt0'], txt2)# extract 2b
842
 
        w1.join(w2) # extract 3a to add txt2 
843
 
        w2.join(w1) # extract 3b to add txt1 
844
 
 
845
 
        w1.add_lines('txt3', ['txt1', 'txt2'], txt3) # extract 4a 
846
 
        w2.add_lines('txt3', ['txt2', 'txt1'], txt3) # extract 4b
847
 
        # These secretly have inverted parents
848
 
 
849
 
        # This should not have to do any extractions
850
 
        w1.join(w2) # NO extract, texts already present with same parents
851
 
        w2.join(w1) # NO extract, texts already present with same parents
852
 
 
853
 
        self.assertEqual(4, w1._extract_count)
854
 
        self.assertEqual(4, w2._extract_count)
855
 
 
856
 
    def test_double_parent(self):
857
 
        # It should not be considered illegal to add
858
 
        # a revision with the same parent twice
859
 
        w1 = InstrumentedWeave()
860
 
        w2 = InstrumentedWeave()
861
 
 
862
 
        txt0 = ['a\n']
863
 
        txt1 = ['a\n', 'b\n']
864
 
        txt2 = ['a\n', 'c\n']
865
 
        txt3 = ['a\n', 'b\n', 'c\n']
866
 
 
867
 
        w1.add_lines('txt0', [], txt0)
868
 
        w2.add_lines('txt0', [], txt0)
869
 
        w1.add_lines('txt1', ['txt0'], txt1)
870
 
        w2.add_lines('txt1', ['txt0', 'txt0'], txt1)
871
 
        # Same text, effectively the same, because the
872
 
        # parent is only repeated
873
 
        w1.join(w2) # extract 3a to add txt2 
874
 
        w2.join(w1) # extract 3b to add txt1 
875
 
 
876
 
 
877
737
class TestNeedsReweave(TestCase):
878
738
    """Internal corner cases for when reweave is needed."""
879
739
 
890
750
        self.assertFalse(w1._compatible_parents(set(), set([1])))
891
751
        self.assertFalse(w1._compatible_parents(my_parents, set([1, 2, 3, 4])))
892
752
        self.assertFalse(w1._compatible_parents(my_parents, set([4])))
 
753
 
 
754
 
 
755
class TestWeaveFile(TestCaseInTempDir):
 
756
    
 
757
    def test_empty_file(self):
 
758
        f = open('empty.weave', 'wb+')
 
759
        try:
 
760
            self.assertRaises(errors.WeaveFormatError,
 
761
                              read_weave, f)
 
762
        finally:
 
763
            f.close()