← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/selftest/teststore.py

  • Committer: Robert Collins
  • Date: 2005-10-08 00:02:33 UTC
  • mfrom: (1421)
  • mto: This revision was merged to the branch mainline in revision 1422.
  • Revision ID: robertc@robertcollins.net-20051008000233-c7c5d0d2b7000da0
merge from newformat stuff and upgrade

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/selftest/teststore.py
16
16
 
17
17
"""Test Store implementation
18
18
"""
19
 
 
20
 
from StringIO import StringIO
 
19
from cStringIO import StringIO
21
20
import os
22
21
 
23
 
from bzrlib.remotebranch import RemoteStore
24
 
from bzrlib.store import ImmutableStore, copy_all
25
 
from bzrlib.selftest import TestCaseInTempDir
 
22
from bzrlib.store import copy_all
 
23
from bzrlib.transport.local import LocalTransport
 
24
from bzrlib.transport import NoSuchFile
 
25
from bzrlib.store.compressed_text import CompressedTextStore
 
26
from bzrlib.store.text import TextStore
 
27
from bzrlib.selftest import TestCase, TestCaseInTempDir
26
28
from bzrlib.errors import BzrError, UnlistableStore
27
 
 
28
 
 
29
 
class TestStore(TestCaseInTempDir):
30
 
    def test_multiple_add(self):
31
 
        """Multiple add with same ID should raise a BzrError"""
32
 
        store = ImmutableStore('.')
33
 
        store.add(StringIO('goodbye'), '123123')
34
 
        self.assertRaises(BzrError, store.add, StringIO('goodbye'), '123123')
35
 
 
36
 
    def test_copy_all(self):
37
 
        """Test copying"""
38
 
        os.mkdir('a')
39
 
        store_a = ImmutableStore('a')
40
 
        store_a.add('foo', '1')
41
 
        os.mkdir('b')
42
 
        store_b = ImmutableStore('b')
43
 
        copy_all(store_a, store_b)
44
 
        self.assertEqual(store_a['1'].read(), 'foo')
45
 
        self.assertEqual(store_b['1'].read(), 'foo')
46
 
        store_c = RemoteStore('http://example.com/')
47
 
        self.assertRaises(UnlistableStore, copy_all, store_c, store_b)
 
29
import bzrlib.store
 
30
 
 
31
 
 
32
def fill_store(store):
 
33
    store.add(StringIO('hello'), 'a')
 
34
    store.add(StringIO('other'), 'b')
 
35
    store.add(StringIO('something'), 'c')
 
36
    store.add(StringIO('goodbye'), '123123')
 
37
 
 
38
def check_equals(tester, store, files, values, permit_failure=False):
 
39
    files = store.get(files, permit_failure=permit_failure)
 
40
    count = 0
 
41
    for f, v in zip(files, values):
 
42
        count += 1
 
43
        if v is None:
 
44
            tester.assert_(f is None)
 
45
        else:
 
46
            tester.assertEquals(f.read(), v)
 
47
    tester.assertEquals(count, len(values))
 
48
    # We need to check to make sure there are no more
 
49
    # files to be returned, I'm using a cheezy way
 
50
    # Convert to a list, and there shouldn't be any left
 
51
    tester.assertEquals(len(list(files)), 0)
 
52
 
 
53
def test_multiple_add(tester, store):
 
54
    fill_store(store)
 
55
    tester.assertRaises(BzrError, store.add, StringIO('goodbye'), '123123')
 
56
 
 
57
def test_get(tester, store):
 
58
    fill_store(store)
 
59
 
 
60
    check_equals(tester, store, ['a'], ['hello'])
 
61
    check_equals(tester, store, ['b', 'c'], ['other', 'something'])
 
62
 
 
63
    # Make sure that requesting a non-existing file fails
 
64
    tester.assertRaises(NoSuchFile, check_equals, tester, store,
 
65
            ['d'], [None])
 
66
    tester.assertRaises(NoSuchFile, check_equals, tester, store,
 
67
            ['a', 'd'], ['hello', None])
 
68
    tester.assertRaises(NoSuchFile, check_equals, tester, store,
 
69
            ['d', 'a'], [None, 'hello'])
 
70
    tester.assertRaises(NoSuchFile, check_equals, tester, store,
 
71
            ['d', 'd', 'd'], [None, None, None])
 
72
    tester.assertRaises(NoSuchFile, check_equals, tester, store,
 
73
            ['a', 'd', 'b'], ['hello', None, 'other'])
 
74
 
 
75
def test_ignore_get(tester, store):
 
76
    fill_store(store)
 
77
 
 
78
    files = store.get(['d'], permit_failure=True)
 
79
    files = list(files)
 
80
    tester.assertEquals(len(files), 1)
 
81
    tester.assert_(files[0] is None)
 
82
 
 
83
    check_equals(tester, store, ['a', 'd'], ['hello', None],
 
84
            permit_failure=True)
 
85
    check_equals(tester, store, ['d', 'a'], [None, 'hello'],
 
86
            permit_failure=True)
 
87
    check_equals(tester, store, ['d', 'd'], [None, None],
 
88
            permit_failure=True)
 
89
    check_equals(tester, store, ['a', 'd', 'b'], ['hello', None, 'other'],
 
90
            permit_failure=True)
 
91
    check_equals(tester, store, ['a', 'd', 'b'], ['hello', None, 'other'],
 
92
            permit_failure=True)
 
93
    check_equals(tester, store, ['b', 'd', 'c'], ['other', None, 'something'],
 
94
            permit_failure=True)
 
95
 
 
96
 
 
97
def get_compressed_store(path='.'):
 
98
    t = LocalTransport(path)
 
99
    return CompressedTextStore(t)
 
100
 
 
101
 
 
102
def get_text_store(path='.'):
 
103
    t = LocalTransport(path)
 
104
    return TextStore(t)
 
105
 
 
106
 
 
107
class TestCompressedTextStore(TestCaseInTempDir):
 
108
 
 
109
    def test_multiple_add(self):
 
110
        """Multiple add with same ID should raise a BzrError"""
 
111
        store = get_compressed_store()
 
112
        test_multiple_add(self, store)
 
113
 
 
114
    def test_get(self):
 
115
        store = get_compressed_store()
 
116
        test_get(self, store)
 
117
 
 
118
    def test_ignore_get(self):
 
119
        store = get_compressed_store()
 
120
        test_ignore_get(self, store)
 
121
 
 
122
    def test_total_size(self):
 
123
        store = get_compressed_store('.')
 
124
        store.add(StringIO('goodbye'), '123123')
 
125
        store.add(StringIO('goodbye2'), '123123.dsc')
 
126
        # these get gzipped - content should be stable
 
127
        self.assertEqual(store.total_size(), (2, 55))
 
128
        
 
129
    def test_copy_all(self):
 
130
        """Test copying"""
 
131
        os.mkdir('a')
 
132
        store_a = get_text_store('a')
 
133
        store_a.add('foo', '1')
 
134
        os.mkdir('b')
 
135
        store_b = get_text_store('b')
 
136
        copy_all(store_a, store_b)
 
137
        self.assertEqual(store_a['1'].read(), 'foo')
 
138
        self.assertEqual(store_b['1'].read(), 'foo')
 
139
 
 
140
 
 
141
class TestMemoryStore(TestCase):
 
142
    
 
143
    def get_store(self):
 
144
        return bzrlib.store.ImmutableMemoryStore()
 
145
    
 
146
    def test_imports(self):
 
147
        from bzrlib.store import ImmutableMemoryStore
 
148
 
 
149
    def test_add_and_retrieve(self):
 
150
        store = self.get_store()
 
151
        store.add(StringIO('hello'), 'aa')
 
152
        self.assertNotEqual(store['aa'], None)
 
153
        self.assertEqual(store['aa'].read(), 'hello')
 
154
        store.add(StringIO('hello world'), 'bb')
 
155
        self.assertNotEqual(store['bb'], None)
 
156
        self.assertEqual(store['bb'].read(), 'hello world')
 
157
 
 
158
    def test_missing_is_absent(self):
 
159
        store = self.get_store()
 
160
        self.failIf('aa' in store)
 
161
 
 
162
    def test_adding_fails_when_present(self):
 
163
        store = self.get_store()
 
164
        store.add(StringIO('hello'), 'aa')
 
165
        self.assertRaises(bzrlib.store.StoreError,
 
166
                          store.add, StringIO('hello'), 'aa')
 
167
 
 
168
    def test_total_size(self):
 
169
        store = self.get_store()
 
170
        store.add(StringIO('goodbye'), '123123')
 
171
        store.add(StringIO('goodbye2'), '123123.dsc')
 
172
        self.assertEqual(store.total_size(), (2, 15))
 
173
        # TODO: Switch the exception form UnlistableStore to
 
174
        #       or make Stores throw UnlistableStore if their
 
175
        #       Transport doesn't support listing
 
176
        # store_c = RemoteStore('http://example.com/')
 
177
        # self.assertRaises(UnlistableStore, copy_all, store_c, store_b)
 
178
 
 
179
 
 
180
class TestTextStore(TestCaseInTempDir):
 
181
    def test_multiple_add(self):
 
182
        """Multiple add with same ID should raise a BzrError"""
 
183
        store = get_text_store()
 
184
        test_multiple_add(self, store)
 
185
 
 
186
    def test_get(self):
 
187
        store = get_text_store()
 
188
        test_get(self, store)
 
189
 
 
190
    def test_ignore_get(self):
 
191
        store = get_text_store()
 
192
        test_ignore_get(self, store)
 
193
 
 
194
    def test_copy_all(self):
 
195
        """Test copying"""
 
196
        os.mkdir('a')
 
197
        store_a = get_text_store('a')
 
198
        store_a.add('foo', '1')
 
199
        os.mkdir('b')
 
200
        store_b = get_text_store('b')
 
201
        copy_all(store_a, store_b)
 
202
        self.assertEqual(store_a['1'].read(), 'foo')
 
203
        self.assertEqual(store_b['1'].read(), 'foo')
 
204
        # TODO: Switch the exception form UnlistableStore to
 
205
        #       or make Stores throw UnlistableStore if their
 
206
        #       Transport doesn't support listing
 
207
        # store_c = RemoteStore('http://example.com/')
 
208
        # self.assertRaises(UnlistableStore, copy_all, store_c, store_b)